addressed comments

core/src/main/java/com/amazon/opendistroforelasticsearch/sql/expression/aggregation/Aggregator.java

@@ -60,13 +60,29 @@ public abstract class Aggregator<S extends AggregationState>
   public abstract S create();
 
   /**
-   * Iterate on the {@link BindingTuple}.
+   * Iterate on {@link ExprValue}.
+   * @param value {@link ExprValue}
+   * @param state {@link AggregationState}
+   * @return {@link AggregationState}
+   */
+  protected abstract S iterate(ExprValue value, S state);
+
+  /**
+   * Let the aggregator iterate on the {@link BindingTuple}
+   * To filter out ExprValues that are missing, null or cannot satisfy {@link #condition}
+   * Before the specific aggregator iterating ExprValue in the tuple.
    *
    * @param tuple {@link BindingTuple}
    * @param state {@link AggregationState}
    * @return {@link AggregationState}
    */
-  public abstract S iterate(BindingTuple tuple, S state);
+  public S iterate(BindingTuple tuple, S state) {
+    ExprValue value = getArguments().get(0).valueOf(tuple);
+    if (value.isNull() || value.isMissing() || !conditionValue(tuple)) {
+      return state;
+    }
+    return iterate(value, state);
+  }
 
   @Override
   public ExprValue valueOf(Environment<Expression, ExprValue> valueEnv) {

core/src/main/java/com/amazon/opendistroforelasticsearch/sql/expression/aggregation/AvgAggregator.java

@@ -43,13 +43,9 @@ public AvgState create() {
   }
 
   @Override
-  public AvgState iterate(BindingTuple tuple, AvgState state) {
-    Expression expression = getArguments().get(0);
-    ExprValue value = expression.valueOf(tuple);
-    if (!(value.isNull() || value.isMissing()) && conditionValue(tuple)) {
-      state.count++;
-      state.total += ExprValueUtils.getDoubleValue(value);
-    }
+  protected AvgState iterate(ExprValue value, AvgState state) {
+    state.count++;
+    state.total += ExprValueUtils.getDoubleValue(value);
     return state;
   }
 

core/src/main/java/com/amazon/opendistroforelasticsearch/sql/expression/aggregation/CountAggregator.java

@@ -39,12 +39,8 @@ public CountAggregator.CountState create() {
   }
 
   @Override
-  public CountState iterate(BindingTuple tuple, CountState state) {
-    Expression expression = getArguments().get(0);
-    ExprValue value = expression.valueOf(tuple);
-    if (!(value.isNull() || value.isMissing()) && conditionValue(tuple)) {
-      state.count++;
-    }
+  protected CountState iterate(ExprValue value, CountState state) {
+    state.count++;
     return state;
   }
 

core/src/main/java/com/amazon/opendistroforelasticsearch/sql/expression/aggregation/MaxAggregator.java

@@ -37,12 +37,8 @@ public MaxState create() {
   }
 
   @Override
-  public MaxState iterate(BindingTuple tuple, MaxState state) {
-    Expression expression = getArguments().get(0);
-    ExprValue value = expression.valueOf(tuple);
-    if (!(value.isNull() || value.isMissing()) && conditionValue(tuple)) {
-      state.max(value);
-    }
+  protected MaxState iterate(ExprValue value, MaxState state) {
+    state.max(value);
     return state;
   }
 

core/src/main/java/com/amazon/opendistroforelasticsearch/sql/expression/aggregation/MinAggregator.java

@@ -42,12 +42,8 @@ public MinState create() {
   }
 
   @Override
-  public MinState iterate(BindingTuple tuple, MinState state) {
-    Expression expression = getArguments().get(0);
-    ExprValue value = expression.valueOf(tuple);
-    if (!(value.isNull() || value.isMissing()) && conditionValue(tuple)) {
-      state.min(value);
-    }
+  protected MinState iterate(ExprValue value, MinState state) {
+    state.min(value);
     return state;
   }
 

core/src/main/java/com/amazon/opendistroforelasticsearch/sql/expression/aggregation/NamedAggregator.java

@@ -17,6 +17,7 @@
 
 package com.amazon.opendistroforelasticsearch.sql.expression.aggregation;
 
+import com.amazon.opendistroforelasticsearch.sql.data.model.ExprValue;
 import com.amazon.opendistroforelasticsearch.sql.expression.ExpressionNodeVisitor;
 import com.amazon.opendistroforelasticsearch.sql.storage.bindingtuple.BindingTuple;
 import com.google.common.base.Strings;
@@ -63,8 +64,8 @@ public AggregationState create() {
   }
 
   @Override
-  public AggregationState iterate(BindingTuple tuple, AggregationState state) {
-    return delegated.iterate(tuple, state);
+  protected AggregationState iterate(ExprValue value, AggregationState state) {
+    return delegated.iterate(value, state);
   }
 
   /**

core/src/main/java/com/amazon/opendistroforelasticsearch/sql/expression/aggregation/SumAggregator.java

@@ -53,13 +53,9 @@ public SumState create() {
   }
 
   @Override
-  public SumState iterate(BindingTuple tuple, SumState state) {
-    Expression expression = getArguments().get(0);
-    ExprValue value = expression.valueOf(tuple);
-    if (!(value.isNull() || value.isMissing()) && conditionValue(tuple)) {
-      state.isEmptyCollection = false;
-      state.add(value);
-    }
+  protected SumState iterate(ExprValue value, SumState state) {
+    state.isEmptyCollection = false;
+    state.add(value);
     return state;
   }
 

integ-test/src/test/resources/correctness/queries/filter.txt

@@ -2,4 +2,5 @@ SELECT AVG(AvgTicketPrice) FILTER(WHERE Carrier = 'Kibana Airlines') AS filtered
 SELECT AVG(AvgTicketPrice) FILTER(WHERE Carrier = 'Kibana Airlines') AS filtered FROM kibana_sample_data_flights GROUP BY Origin ORDER BY Origin
 SELECT AVG(AvgTicketPrice + 1) FILTER(WHERE Carrier = 'Kibana Airlines') AS filtered FROM kibana_sample_data_flights
 SELECT AVG(AvgTicketPrice) FILTER(WHERE Carrier = 'Kibana Airlines') / 2 AS filtered FROM kibana_sample_data_flights
+SELECT AVG(AvgTicketPrice) FILTER(WHERE ABS(AvgTicketPrice) < 10000) AS filtered FROM kibana_sample_data_flights
 SELECT AVG(AvgTicketPrice) AS unfiltered, AVG(AvgTicketPrice) FILTER(WHERE Carrier = 'Kibana Airlines') AS filtered1, AVG(AvgTicketPrice) FILTER(WHERE Carrier = 'ES-Air') AS filtered2 FROM kibana_sample_data_flights WHERE DestWeather = 'Sunny'