feat: support SQL aggregate FILTER (WHERE ...) clause in native execution

Previously, Comet fell back to Spark for any aggregation containing
a FILTER (WHERE ...) clause (e.g. SUM(x) FILTER (WHERE y > 0)).
The native SumInt/SumDecimal accumulators already received opt_filter
support in the previous commit. This commit wires the full pipeline:

Proto (expr.proto):
- Add optional Expr filter = 89 to AggExpr message

Scala serialization (QueryPlanSerde.scala):
- In aggExprToProto, serialize aggExpr.filter into the proto when
  aggExpr.mode == Partial (filters are only meaningful in partial mode)
- If the filter expression cannot be serialized, fall back gracefully

Native planner (planner.rs):
- Build per-aggregate filter PhysicalExpr from agg_expr.filter
- Pass to AggregateExec::try_new instead of vec![None; num_agg]

Comet planner (operators.scala):
- Remove the blanket fallback guard for aggregate expressions with filter

Tests (aggregate_filter.sql):
- Update queries from expect_fallback to plain query mode now that
  native execution is supported; tests verify results match Spark

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: viirya

native/core/src/execution/planner.rs

@@ -975,12 +975,27 @@ impl PhysicalPlanner {
 
                 let num_agg = agg.agg_exprs.len();
                 let aggr_expr = agg_exprs?.into_iter().map(Arc::new).collect();
+
+                // Build per-aggregate filter expressions from the FILTER (WHERE ...) clause.
+                // Filters are only present in Partial mode; Final/PartialMerge always get None.
+                let filter_exprs: Result<Vec<Option<Arc<dyn PhysicalExpr>>>, ExecutionError> = agg
+                    .agg_exprs
+                    .iter()
+                    .map(|expr| {
+                        if let Some(f) = expr.filter.as_deref() {
+                            self.create_expr(f, Arc::clone(&schema)).map(Some)
+                        } else {
+                            Ok(None)
+                        }
+                    })
+                    .collect();
+
                 let aggregate: Arc<dyn ExecutionPlan> = Arc::new(
                     datafusion::physical_plan::aggregates::AggregateExec::try_new(
                         mode,
                         group_by,
                         aggr_expr,
-                        vec![None; num_agg], // no filter expressions
+                        filter_exprs?,
                         Arc::clone(&child.native_plan),
                         Arc::clone(&schema),
                     )?,

native/proto/src/proto/expr.proto

@@ -141,6 +141,10 @@ message AggExpr {
     BloomFilterAgg bloomFilterAgg = 16;
   }
 
+  // Optional filter expression for SQL FILTER (WHERE ...) clause.
+  // Only set in Partial aggregation mode; absent in Final/PartialMerge.
+  optional Expr filter = 89;
+
   // Optional QueryContext for error reporting (contains SQL text and position)
   optional QueryContext query_context = 90;
 

spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala

@@ -517,13 +517,25 @@ object QueryPlanSerde extends Logging with CometExprShim {
     }
 
     // Attach QueryContext and expr_id to the aggregate expression
-    protoAggExprOpt.map { protoAggExpr =>
+    protoAggExprOpt.flatMap { protoAggExpr =>
       val builder = protoAggExpr.toBuilder
       builder.setExprId(nextExprId())
+
+      // Serialize FILTER (WHERE ...) clause if present.
+      // The filter is only meaningful in Partial mode; Final/PartialMerge never set it.
+      if (aggExpr.filter.isDefined && aggExpr.mode == Partial) {
+        val filterProto = exprToProto(aggExpr.filter.get, inputs, binding)
+        if (filterProto.isEmpty) {
+          withInfo(aggExpr, aggExpr.filter.get)
+          return None
+        }
+        builder.setFilter(filterProto.get)
+      }
+
       extractQueryContext(fn).foreach { ctx =>
         builder.setQueryContext(ctx)
       }
-      builder.build()
+      Some(builder.build())
     }
   }
 

spark/src/main/scala/org/apache/spark/sql/comet/operators.scala

@@ -1362,12 +1362,6 @@ trait CometBaseAggregate {
       return None
     }
 
-    // Aggregate expressions with filter are not supported yet.
-    if (aggregateExpressions.exists(_.filter.isDefined)) {
-      withInfo(aggregate, "Aggregate expression with filter is not supported")
-      return None
-    }
-
     if (groupingExpressions.exists(expr =>
         expr.dataType match {
           case _: MapType => true

spark/src/test/resources/sql-tests/expressions/aggregate/aggregate_filter.sql

@@ -37,33 +37,33 @@ INSERT INTO test_agg_filter VALUES
   ('b', NULL, NULL, NULL, true)
 
 -- Basic FILTER on SUM(int)
-query expect_fallback(Aggregate expression with filter is not supported)
+query
 SELECT SUM(i) FILTER (WHERE flag = true) FROM test_agg_filter
 
 -- FILTER on SUM with GROUP BY
-query expect_fallback(Aggregate expression with filter is not supported)
+query
 SELECT grp, SUM(i) FILTER (WHERE flag = true) FROM test_agg_filter GROUP BY grp ORDER BY grp
 
 -- FILTER on SUM(long)
-query expect_fallback(Aggregate expression with filter is not supported)
+query
 SELECT SUM(l) FILTER (WHERE flag = true) FROM test_agg_filter
 
 -- FILTER on SUM(decimal)
-query expect_fallback(Aggregate expression with filter is not supported)
+query
 SELECT SUM(d) FILTER (WHERE flag = true) FROM test_agg_filter
 
 -- Multiple aggregates: one with filter, one without
-query expect_fallback(Aggregate expression with filter is not supported)
+query
 SELECT SUM(i), SUM(i) FILTER (WHERE flag = true) FROM test_agg_filter
 
 -- FILTER with NULL rows: NULLs should not be included even when filter passes
-query expect_fallback(Aggregate expression with filter is not supported)
+query
 SELECT grp, SUM(i) FILTER (WHERE flag = true) FROM test_agg_filter GROUP BY grp ORDER BY grp
 
 -- FILTER with COUNT
-query expect_fallback(Aggregate expression with filter is not supported)
+query
 SELECT COUNT(*) FILTER (WHERE flag = true) FROM test_agg_filter
 
 -- FILTER with COUNT GROUP BY
-query expect_fallback(Aggregate expression with filter is not supported)
+query
 SELECT grp, COUNT(*) FILTER (WHERE flag = true) FROM test_agg_filter GROUP BY grp ORDER BY grp