From 6de04e41e9da14659ac5e9d09d3a6c88762d4144 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Mon, 29 Jun 2026 09:36:55 -0600 Subject: [PATCH 1/2] feat: name incompatible aggregate functions in mixed-execution fallback reason When a Spark Final aggregate cannot run because the partial aggregate ran natively (or would have) and the intermediate buffer formats are incompatible between Spark and Comet, the fallback reason now names the specific aggregate function(s) responsible instead of only stating that compatible buffers are required. --- .../org/apache/comet/serde/QueryPlanSerde.scala | 14 ++++++++++++-- .../org/apache/spark/sql/comet/operators.scala | 9 ++++++--- .../approved-plans-v1_4-spark3_5/q70/extended.txt | 2 +- .../approved-plans-v1_4/q10/extended.txt | 2 +- .../approved-plans-v1_4/q35/extended.txt | 2 +- .../approved-plans-v1_4/q45/extended.txt | 2 +- .../approved-plans-v2_7-spark3_5/q70a/extended.txt | 10 +++++----- .../approved-plans-v2_7/q35/extended.txt | 2 +- 8 files changed, 28 insertions(+), 15 deletions(-) diff --git a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala index 40d8e39dbf..6d01dfe0a6 100644 --- a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala +++ b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala @@ -401,8 +401,18 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim { * in the other. */ def allAggsSupportMixedExecution(aggExprs: Seq[AggregateExpression]): Boolean = { - aggExprs.forall { aggExpr => - val fn = aggExpr.aggregateFunction + aggsNotSupportingMixedExecution(aggExprs).isEmpty + } + + /** + * Returns the aggregate functions in the list whose intermediate buffer formats are not known + * to be compatible between Spark and Comet. These are the functions that prevent a Spark Final + * aggregate (without a Comet Partial) from running, since the buffer produced by one engine + * cannot be safely consumed by the other. + */ + def aggsNotSupportingMixedExecution( + aggExprs: Seq[AggregateExpression]): Seq[AggregateFunction] = { + aggExprs.map(_.aggregateFunction).filterNot { fn => aggrSerdeMap.get(fn.getClass) match { case Some(handler) => handler diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala b/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala index af9e1df8a3..edeaaf7550 100644 --- a/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala +++ b/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala @@ -1538,12 +1538,15 @@ trait CometBaseAggregate { return None } - if (sparkFinalMode && - !QueryPlanSerde.allAggsSupportMixedExecution(aggregate.aggregateExpressions)) { + val incompatibleAggs = + QueryPlanSerde.aggsNotSupportingMixedExecution(aggregate.aggregateExpressions) + if (sparkFinalMode && incompatibleAggs.nonEmpty) { + val names = incompatibleAggs.map(_.prettyName).distinct.mkString(", ") withFallbackReason( aggregate, "Spark Final aggregate without Comet Partial requires compatible " + - "intermediate buffer formats") + "intermediate buffer formats, but the following aggregate function(s) " + + s"have incompatible buffers: $names") return None } diff --git a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark3_5/q70/extended.txt b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark3_5/q70/extended.txt index 14ad77dad4..08df1c04a0 100644 --- a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark3_5/q70/extended.txt +++ b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark3_5/q70/extended.txt @@ -4,7 +4,7 @@ CometNativeColumnarToRow +- CometWindowExec +- CometSort +- CometColumnarExchange - +- HashAggregate [COMET: Spark Final aggregate without Comet Partial requires compatible intermediate buffer formats] + +- HashAggregate [COMET: Spark Final aggregate without Comet Partial requires compatible intermediate buffer formats, but the following aggregate function(s) have incompatible buffers: sum] +- Exchange +- HashAggregate +- Expand diff --git a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q10/extended.txt b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q10/extended.txt index 07af300183..fcb5e0003d 100644 --- a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q10/extended.txt +++ b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q10/extended.txt @@ -1,5 +1,5 @@ TakeOrderedAndProject -+- HashAggregate [COMET: Spark Final aggregate without Comet Partial requires compatible intermediate buffer formats] ++- HashAggregate [COMET: Spark Final aggregate without Comet Partial requires compatible intermediate buffer formats, but the following aggregate function(s) have incompatible buffers: count] +- Exchange +- HashAggregate +- Project diff --git a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q35/extended.txt b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q35/extended.txt index 07af300183..bb513d51a9 100644 --- a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q35/extended.txt +++ b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q35/extended.txt @@ -1,5 +1,5 @@ TakeOrderedAndProject -+- HashAggregate [COMET: Spark Final aggregate without Comet Partial requires compatible intermediate buffer formats] ++- HashAggregate [COMET: Spark Final aggregate without Comet Partial requires compatible intermediate buffer formats, but the following aggregate function(s) have incompatible buffers: count, avg] +- Exchange +- HashAggregate +- Project diff --git a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q45/extended.txt b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q45/extended.txt index f95c69368f..2d463ef986 100644 --- a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q45/extended.txt +++ b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q45/extended.txt @@ -1,5 +1,5 @@ TakeOrderedAndProject -+- HashAggregate [COMET: Spark Final aggregate without Comet Partial requires compatible intermediate buffer formats] ++- HashAggregate [COMET: Spark Final aggregate without Comet Partial requires compatible intermediate buffer formats, but the following aggregate function(s) have incompatible buffers: sum] +- Exchange +- HashAggregate +- Project diff --git a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v2_7-spark3_5/q70a/extended.txt b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v2_7-spark3_5/q70a/extended.txt index 10ea854de0..e527d95f46 100644 --- a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v2_7-spark3_5/q70a/extended.txt +++ b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v2_7-spark3_5/q70a/extended.txt @@ -8,7 +8,7 @@ CometNativeColumnarToRow +- CometColumnarExchange +- HashAggregate +- Union - :- HashAggregate [COMET: Spark Final aggregate without Comet Partial requires compatible intermediate buffer formats] + :- HashAggregate [COMET: Spark Final aggregate without Comet Partial requires compatible intermediate buffer formats, but the following aggregate function(s) have incompatible buffers: sum] : +- Exchange : +- HashAggregate : +- Project @@ -58,10 +58,10 @@ CometNativeColumnarToRow : +- CometProject : +- CometFilter : +- CometNativeScan parquet spark_catalog.default.date_dim - :- HashAggregate [COMET: Spark Final aggregate without Comet Partial requires compatible intermediate buffer formats] + :- HashAggregate [COMET: Spark Final aggregate without Comet Partial requires compatible intermediate buffer formats, but the following aggregate function(s) have incompatible buffers: sum] : +- Exchange : +- HashAggregate - : +- HashAggregate [COMET: Spark Final aggregate without Comet Partial requires compatible intermediate buffer formats] + : +- HashAggregate [COMET: Spark Final aggregate without Comet Partial requires compatible intermediate buffer formats, but the following aggregate function(s) have incompatible buffers: sum] : +- Exchange : +- HashAggregate : +- Project @@ -111,10 +111,10 @@ CometNativeColumnarToRow : +- CometProject : +- CometFilter : +- CometNativeScan parquet spark_catalog.default.date_dim - +- HashAggregate [COMET: Spark Final aggregate without Comet Partial requires compatible intermediate buffer formats] + +- HashAggregate [COMET: Spark Final aggregate without Comet Partial requires compatible intermediate buffer formats, but the following aggregate function(s) have incompatible buffers: sum] +- Exchange +- HashAggregate - +- HashAggregate [COMET: Spark Final aggregate without Comet Partial requires compatible intermediate buffer formats] + +- HashAggregate [COMET: Spark Final aggregate without Comet Partial requires compatible intermediate buffer formats, but the following aggregate function(s) have incompatible buffers: sum] +- Exchange +- HashAggregate +- Project diff --git a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v2_7/q35/extended.txt b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v2_7/q35/extended.txt index 07af300183..7031d3afb6 100644 --- a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v2_7/q35/extended.txt +++ b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v2_7/q35/extended.txt @@ -1,5 +1,5 @@ TakeOrderedAndProject -+- HashAggregate [COMET: Spark Final aggregate without Comet Partial requires compatible intermediate buffer formats] ++- HashAggregate [COMET: Spark Final aggregate without Comet Partial requires compatible intermediate buffer formats, but the following aggregate function(s) have incompatible buffers: count, avg, sum] +- Exchange +- HashAggregate +- Project From 0e6fb72ee1118d013e37036a0b2d84b9a5bb1861 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Mon, 29 Jun 2026 09:56:36 -0600 Subject: [PATCH 2/2] refactor: sort incompatible aggregate names and tidy mixed-execution helpers Sort the offending aggregate function names alphabetically so the fallback reason is deterministic regardless of expression order. Extract a shared supportsMixedExecution predicate so allAggsSupportMixedExecution keeps its forall short-circuit and aggsNotSupportingMixedExecution reuses the same lookup, and only compute the incompatible list when in Spark final mode. --- .../apache/comet/serde/QueryPlanSerde.scala | 20 +++++++++-------- .../apache/spark/sql/comet/operators.scala | 22 ++++++++++--------- .../approved-plans-v1_4/q35/extended.txt | 2 +- .../approved-plans-v2_7/q35/extended.txt | 2 +- 4 files changed, 25 insertions(+), 21 deletions(-) diff --git a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala index 6d01dfe0a6..c429d05714 100644 --- a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala +++ b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala @@ -401,7 +401,7 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim { * in the other. */ def allAggsSupportMixedExecution(aggExprs: Seq[AggregateExpression]): Boolean = { - aggsNotSupportingMixedExecution(aggExprs).isEmpty + aggExprs.forall(aggExpr => supportsMixedExecution(aggExpr.aggregateFunction)) } /** @@ -412,14 +412,16 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim { */ def aggsNotSupportingMixedExecution( aggExprs: Seq[AggregateExpression]): Seq[AggregateFunction] = { - aggExprs.map(_.aggregateFunction).filterNot { fn => - aggrSerdeMap.get(fn.getClass) match { - case Some(handler) => - handler - .asInstanceOf[CometAggregateExpressionSerde[AggregateFunction]] - .supportsMixedPartialFinal - case None => false - } + aggExprs.map(_.aggregateFunction).filterNot(supportsMixedExecution) + } + + private def supportsMixedExecution(fn: AggregateFunction): Boolean = { + aggrSerdeMap.get(fn.getClass) match { + case Some(handler) => + handler + .asInstanceOf[CometAggregateExpressionSerde[AggregateFunction]] + .supportsMixedPartialFinal + case None => false } } diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala b/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala index edeaaf7550..83f5099607 100644 --- a/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala +++ b/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala @@ -1538,16 +1538,18 @@ trait CometBaseAggregate { return None } - val incompatibleAggs = - QueryPlanSerde.aggsNotSupportingMixedExecution(aggregate.aggregateExpressions) - if (sparkFinalMode && incompatibleAggs.nonEmpty) { - val names = incompatibleAggs.map(_.prettyName).distinct.mkString(", ") - withFallbackReason( - aggregate, - "Spark Final aggregate without Comet Partial requires compatible " + - "intermediate buffer formats, but the following aggregate function(s) " + - s"have incompatible buffers: $names") - return None + if (sparkFinalMode) { + val incompatibleAggs = + QueryPlanSerde.aggsNotSupportingMixedExecution(aggregate.aggregateExpressions) + if (incompatibleAggs.nonEmpty) { + val names = incompatibleAggs.map(_.prettyName).distinct.sorted.mkString(", ") + withFallbackReason( + aggregate, + "Spark Final aggregate without Comet Partial requires compatible " + + "intermediate buffer formats, but the following aggregate function(s) " + + s"have incompatible buffers: $names") + return None + } } // Check if this aggregate has been tagged as unsafe for mixed execution diff --git a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q35/extended.txt b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q35/extended.txt index bb513d51a9..50e9ebb965 100644 --- a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q35/extended.txt +++ b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q35/extended.txt @@ -1,5 +1,5 @@ TakeOrderedAndProject -+- HashAggregate [COMET: Spark Final aggregate without Comet Partial requires compatible intermediate buffer formats, but the following aggregate function(s) have incompatible buffers: count, avg] ++- HashAggregate [COMET: Spark Final aggregate without Comet Partial requires compatible intermediate buffer formats, but the following aggregate function(s) have incompatible buffers: avg, count] +- Exchange +- HashAggregate +- Project diff --git a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v2_7/q35/extended.txt b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v2_7/q35/extended.txt index 7031d3afb6..c340d4a42c 100644 --- a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v2_7/q35/extended.txt +++ b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v2_7/q35/extended.txt @@ -1,5 +1,5 @@ TakeOrderedAndProject -+- HashAggregate [COMET: Spark Final aggregate without Comet Partial requires compatible intermediate buffer formats, but the following aggregate function(s) have incompatible buffers: count, avg, sum] ++- HashAggregate [COMET: Spark Final aggregate without Comet Partial requires compatible intermediate buffer formats, but the following aggregate function(s) have incompatible buffers: avg, count, sum] +- Exchange +- HashAggregate +- Project