From a94862eddd2b5cfb98d86a4031efe5f52fed17c8 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Wed, 8 Jul 2026 08:09:07 -0600 Subject: [PATCH 1/8] refactor: make supportsMixedPartialFinal aggregate-function aware --- .../comet/serde/CometAggregateExpressionSerde.scala | 13 +++++++------ .../org/apache/comet/serde/QueryPlanSerde.scala | 2 +- .../scala/org/apache/comet/serde/aggregates.scala | 12 ++++++------ 3 files changed, 14 insertions(+), 13 deletions(-) diff --git a/spark/src/main/scala/org/apache/comet/serde/CometAggregateExpressionSerde.scala b/spark/src/main/scala/org/apache/comet/serde/CometAggregateExpressionSerde.scala index 9a83152168..a52d600821 100644 --- a/spark/src/main/scala/org/apache/comet/serde/CometAggregateExpressionSerde.scala +++ b/spark/src/main/scala/org/apache/comet/serde/CometAggregateExpressionSerde.scala @@ -82,15 +82,16 @@ trait CometAggregateExpressionSerde[T <: AggregateFunction] { def getSupportLevel(expr: T): SupportLevel = Compatible(None) /** - * Whether this aggregate's intermediate buffer format is compatible between Spark and Comet, - * making it safe to run the Partial in one engine and the Final in the other. Aggregates with - * simple single-value buffers (MIN, MAX, bitwise) are safe; those with complex or - * differently-encoded buffers (AVG, SUM with decimals, CollectSet, Variance) are not. COUNT is - * intentionally excluded: mixed COUNT partial/final regressed AQE's + * Whether this aggregate's intermediate buffer format is compatible between Spark and Comet for + * the given function instance, making it safe to run the Partial in one engine and the Final in + * the other. Aggregates with simple single-value buffers (MIN, MAX, bitwise) are always safe; + * SUM and non-decimal AVG match Spark's buffer and are safe except where noted per instance + * (e.g. TRY-mode SUM uses a Comet-internal flag column). COUNT is intentionally excluded + * despite a matching buffer: mixed COUNT partial/final regressed AQE's * PropagateEmptyRelationAfterAQE pattern (which matches BaseAggregateExec only) and the Spark * 4.0 count-bug decorrelation for correlated IN subqueries. */ - def supportsMixedPartialFinal: Boolean = false + def supportsMixedPartialFinal(fn: T): Boolean = false /** * Convert a Spark expression into a protocol buffer representation that can be passed into 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 91a93d633b..4d6126abce 100644 --- a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala +++ b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala @@ -432,7 +432,7 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim { case Some(handler) => handler .asInstanceOf[CometAggregateExpressionSerde[AggregateFunction]] - .supportsMixedPartialFinal + .supportsMixedPartialFinal(fn) case None => false } } diff --git a/spark/src/main/scala/org/apache/comet/serde/aggregates.scala b/spark/src/main/scala/org/apache/comet/serde/aggregates.scala index 5710232cb4..dcc85f4a6a 100644 --- a/spark/src/main/scala/org/apache/comet/serde/aggregates.scala +++ b/spark/src/main/scala/org/apache/comet/serde/aggregates.scala @@ -33,7 +33,7 @@ import org.apache.comet.shims.CometEvalModeUtil object CometMin extends CometAggregateExpressionSerde[Min] { - override def supportsMixedPartialFinal: Boolean = true + override def supportsMixedPartialFinal(fn: Min): Boolean = true override def getSupportLevel(expr: Min): SupportLevel = AggSerde.minMaxSupportLevel(expr.dataType) @@ -70,7 +70,7 @@ object CometMin extends CometAggregateExpressionSerde[Min] { object CometMax extends CometAggregateExpressionSerde[Max] { - override def supportsMixedPartialFinal: Boolean = true + override def supportsMixedPartialFinal(fn: Max): Boolean = true override def getSupportLevel(expr: Max): SupportLevel = AggSerde.minMaxSupportLevel(expr.dataType) @@ -300,7 +300,7 @@ object CometLast extends CometAggregateExpressionSerde[Last] { } object CometBitAndAgg extends CometAggregateExpressionSerde[BitAndAgg] { - override def supportsMixedPartialFinal: Boolean = true + override def supportsMixedPartialFinal(fn: BitAndAgg): Boolean = true override def getSupportLevel(expr: BitAndAgg): SupportLevel = if (AggSerde.bitwiseAggTypeSupported(expr.dataType)) { @@ -339,7 +339,7 @@ object CometBitAndAgg extends CometAggregateExpressionSerde[BitAndAgg] { } object CometBitOrAgg extends CometAggregateExpressionSerde[BitOrAgg] { - override def supportsMixedPartialFinal: Boolean = true + override def supportsMixedPartialFinal(fn: BitOrAgg): Boolean = true override def getSupportLevel(expr: BitOrAgg): SupportLevel = if (AggSerde.bitwiseAggTypeSupported(expr.dataType)) { @@ -378,7 +378,7 @@ object CometBitOrAgg extends CometAggregateExpressionSerde[BitOrAgg] { } object CometBitXOrAgg extends CometAggregateExpressionSerde[BitXorAgg] { - override def supportsMixedPartialFinal: Boolean = true + override def supportsMixedPartialFinal(fn: BitXorAgg): Boolean = true override def getSupportLevel(expr: BitXorAgg): SupportLevel = if (AggSerde.bitwiseAggTypeSupported(expr.dataType)) { @@ -707,7 +707,7 @@ object CometCorr extends CometAggregateExpressionSerde[Corr] { object CometBloomFilterAggregate extends CometAggregateExpressionSerde[BloomFilterAggregate] { - override def supportsMixedPartialFinal: Boolean = true + override def supportsMixedPartialFinal(fn: BloomFilterAggregate): Boolean = true override def getSupportLevel(expr: BloomFilterAggregate): SupportLevel = expr.child.dataType match { From 7bb3981144dd2d68c53d9eb370976fe2d3e42caa Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Wed, 8 Jul 2026 08:18:37 -0600 Subject: [PATCH 2/8] feat: enable mixed partial/final execution for sum and non-decimal avg --- .../org/apache/comet/serde/aggregates.scala | 11 +++ .../comet/rules/CometExecRuleSuite.scala | 77 ++++++++++++++++++- 2 files changed, 84 insertions(+), 4 deletions(-) diff --git a/spark/src/main/scala/org/apache/comet/serde/aggregates.scala b/spark/src/main/scala/org/apache/comet/serde/aggregates.scala index dcc85f4a6a..6c6544e93c 100644 --- a/spark/src/main/scala/org/apache/comet/serde/aggregates.scala +++ b/spark/src/main/scala/org/apache/comet/serde/aggregates.scala @@ -28,6 +28,7 @@ import org.apache.spark.sql.types.{ByteType, DecimalType, DoubleType, IntegerTyp import org.apache.comet.CometConf.COMET_EXEC_STRICT_FLOATING_POINT import org.apache.comet.CometSparkSessionExtensions.{isSpark41Plus, withFallbackReason} +import org.apache.comet.expressions.CometEvalMode import org.apache.comet.serde.QueryPlanSerde.{evalModeToProto, exprToProto, serializeDataType} import org.apache.comet.shims.CometEvalModeUtil @@ -130,6 +131,11 @@ object CometCount extends CometAggregateExpressionSerde[Count] { object CometAverage extends CometAggregateExpressionSerde[Average] { + override def supportsMixedPartialFinal(fn: Average): Boolean = + // Non-decimal AVG has a (sum: double, count: long) buffer matching Spark. Decimal AVG is + // deferred (overflow nulls count differently) and stays unsafe for mixed execution. + !fn.child.dataType.isInstanceOf[DecimalType] + override def getUnsupportedReasons(): Seq[String] = Seq( "YearMonthIntervalType and DayTimeIntervalType inputs are not supported") @@ -186,6 +192,11 @@ object CometAverage extends CometAggregateExpressionSerde[Average] { object CometSum extends CometAggregateExpressionSerde[Sum] { + override def supportsMixedPartialFinal(fn: Sum): Boolean = + // SUM's buffer matches Spark for Legacy/Ansi (decimal adds is_empty, also matching), but + // TRY-mode integer SUM carries a Comet-internal has_all_nulls column that Spark cannot read. + CometEvalModeUtil.fromSparkEvalMode(CometEvalModeUtil.sumEvalMode(fn)) != CometEvalMode.TRY + override def getSupportLevel(expr: Sum): SupportLevel = if (AggSerde.sumDataTypeSupported(expr.dataType)) { Compatible() diff --git a/spark/src/test/scala/org/apache/comet/rules/CometExecRuleSuite.scala b/spark/src/test/scala/org/apache/comet/rules/CometExecRuleSuite.scala index 7fa06a26cc..1f76361ad7 100644 --- a/spark/src/test/scala/org/apache/comet/rules/CometExecRuleSuite.scala +++ b/spark/src/test/scala/org/apache/comet/rules/CometExecRuleSuite.scala @@ -34,7 +34,7 @@ import org.apache.spark.sql.execution.exchange.{BroadcastExchangeExec, ShuffleEx import org.apache.spark.sql.types.{DataTypes, StructField, StructType} import org.apache.comet.CometConf -import org.apache.comet.CometSparkSessionExtensions.{isSpark40Plus, isSpark42Plus} +import org.apache.comet.CometSparkSessionExtensions.{isSpark35Plus, isSpark40Plus, isSpark42Plus} import org.apache.comet.testing.{DataGenOptions, FuzzDataGenerator} /** @@ -152,8 +152,8 @@ class CometExecRuleSuite extends CometTestBase { CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true") { val transformedPlan = applyCometExecRule(sparkPlan) - // SUM has incompatible intermediate buffers, so if the final aggregate cannot - // be converted to Comet, neither should be + // COUNT is intentionally excluded from mixed execution (AQE / count-bug reasons), so if + // the final aggregate cannot be converted to Comet, neither should the partial. assert( countOperators(transformedPlan, classOf[HashAggregateExec]) == originalHashAggCount) assert(countOperators(transformedPlan, classOf[CometHashAggregateExec]) == 0) @@ -177,7 +177,8 @@ class CometExecRuleSuite extends CometTestBase { CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true") { val transformedPlan = applyCometExecRule(sparkPlan) - // if the partial aggregate cannot be converted to Comet, then neither should be + // COUNT blocks mixed execution, so if the partial cannot be converted, neither should + // the final. assert( countOperators(transformedPlan, classOf[HashAggregateExec]) == originalHashAggCount) assert(countOperators(transformedPlan, classOf[CometHashAggregateExec]) == 0) @@ -231,6 +232,74 @@ class CometExecRuleSuite extends CometTestBase { } } + test("CometExecRule should allow SUM mixed Comet partial and Spark final") { + withTempView("test_data") { + createTestDataFrame.createOrReplaceTempView("test_data") + val sparkPlan = + createSparkPlan(spark, "SELECT SUM(id) FROM test_data GROUP BY (id % 3)") + assert(countOperators(sparkPlan, classOf[HashAggregateExec]) == 2) + withSQLConf( + CometConf.COMET_ENABLE_FINAL_HASH_AGGREGATE.key -> "false", + CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true") { + val transformedPlan = applyCometExecRule(sparkPlan) + // SUM buffer matches Spark: partial converts to Comet, final stays Spark. + assert(countOperators(transformedPlan, classOf[HashAggregateExec]) == 1) // final + assert(countOperators(transformedPlan, classOf[CometHashAggregateExec]) == 1) // partial + } + } + } + + test("CometExecRule should allow SUM mixed Spark partial and Comet final") { + withTempView("test_data") { + createTestDataFrame.createOrReplaceTempView("test_data") + val sparkPlan = + createSparkPlan(spark, "SELECT SUM(id) FROM test_data GROUP BY (id % 3)") + assert(countOperators(sparkPlan, classOf[HashAggregateExec]) == 2) + withSQLConf( + CometConf.COMET_ENABLE_PARTIAL_HASH_AGGREGATE.key -> "false", + CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true") { + val transformedPlan = applyCometExecRule(sparkPlan) + assert(countOperators(transformedPlan, classOf[HashAggregateExec]) == 1) // partial + assert(countOperators(transformedPlan, classOf[CometHashAggregateExec]) == 1) // final + } + } + } + + test("CometExecRule should allow AVG mixed Comet partial and Spark final") { + withTempView("test_data") { + createTestDataFrame.createOrReplaceTempView("test_data") + val sparkPlan = + createSparkPlan(spark, "SELECT AVG(id) FROM test_data GROUP BY (id % 3)") + assert(countOperators(sparkPlan, classOf[HashAggregateExec]) == 2) + withSQLConf( + CometConf.COMET_ENABLE_FINAL_HASH_AGGREGATE.key -> "false", + CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true") { + val transformedPlan = applyCometExecRule(sparkPlan) + assert(countOperators(transformedPlan, classOf[HashAggregateExec]) == 1) // final + assert(countOperators(transformedPlan, classOf[CometHashAggregateExec]) == 1) // partial + } + } + } + + test("CometExecRule should not allow try_sum mixed execution") { + assume(isSpark35Plus, "try_sum was added in Spark 3.5") + withTempView("test_data") { + createTestDataFrame.createOrReplaceTempView("test_data") + val sparkPlan = + createSparkPlan(spark, "SELECT try_sum(id) FROM test_data GROUP BY (id % 3)") + assert(countOperators(sparkPlan, classOf[HashAggregateExec]) == 2) + withSQLConf( + CometConf.COMET_ENABLE_FINAL_HASH_AGGREGATE.key -> "false", + CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true") { + val transformedPlan = applyCometExecRule(sparkPlan) + // TRY-mode SUM uses a Comet-internal buffer column, so mixing is unsafe: + // the partial must also fall back to Spark. + assert(countOperators(transformedPlan, classOf[HashAggregateExec]) == 2) + assert(countOperators(transformedPlan, classOf[CometHashAggregateExec]) == 0) + } + } + } + test("CometExecRule should allow BloomFilter mixed Comet partial and Spark final") { assume(!isSpark42Plus, "https://github.com/apache/datafusion-comet/issues/4142") val funcId = new FunctionIdentifier("bloom_filter_agg") From 2cd888cf1b7448013acd365e2092e18134285cf9 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Wed, 8 Jul 2026 08:30:03 -0600 Subject: [PATCH 3/8] test: verify mixed-engine sum and avg match Spark --- .../comet/exec/CometAggregateSuite.scala | 42 +++++++++++++++++++ 1 file changed, 42 insertions(+) diff --git a/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala index ae14c68207..f8c990fd09 100644 --- a/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala +++ b/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala @@ -183,6 +183,48 @@ class CometAggregateSuite extends CometTestBase with AdaptiveSparkPlanHelper { } } + test("mixed engine sum/avg: Comet partial + Spark final matches Spark") { + val data = (0 until 100).map(i => (i, i.toLong, i.toDouble, i % 7)) + withParquetTable(data, "tbl") { + withSQLConf( + CometConf.COMET_ENABLE_FINAL_HASH_AGGREGATE.key -> "false", + CometConf.COMET_EXEC_SHUFFLE_ENABLED.key -> "true", + CometConf.COMET_SHUFFLE_MODE.key -> "jvm") { + checkSparkAnswer( + "SELECT _4, SUM(_1), SUM(_2), SUM(_3), AVG(_1), AVG(_3) FROM tbl GROUP BY _4") + } + } + } + + test("mixed engine sum/avg: Spark partial + Comet final matches Spark") { + val data = (0 until 100).map(i => (i, i.toLong, i.toDouble, i % 7)) + withParquetTable(data, "tbl") { + withSQLConf( + CometConf.COMET_ENABLE_PARTIAL_HASH_AGGREGATE.key -> "false", + CometConf.COMET_EXEC_SHUFFLE_ENABLED.key -> "true", + CometConf.COMET_SHUFFLE_MODE.key -> "jvm") { + checkSparkAnswer( + "SELECT _4, SUM(_1), SUM(_2), SUM(_3), AVG(_1), AVG(_3) FROM tbl GROUP BY _4") + } + } + } + + test("mixed engine decimal sum: both split directions match Spark") { + val data = (0 until 100).map(i => (BigDecimal(i), i % 7)) + withParquetTable(data, "tbl") { + Seq( + CometConf.COMET_ENABLE_FINAL_HASH_AGGREGATE.key, + CometConf.COMET_ENABLE_PARTIAL_HASH_AGGREGATE.key).foreach { disabledConf => + withSQLConf( + disabledConf -> "false", + CometConf.COMET_EXEC_SHUFFLE_ENABLED.key -> "true", + CometConf.COMET_SHUFFLE_MODE.key -> "jvm") { + checkSparkAnswer("SELECT _2, SUM(_1) FROM tbl GROUP BY _2") + } + } + } + } + test("Aggregation without aggregate expressions should use correct result expressions") { withSQLConf( CometConf.COMET_ENABLED.key -> "true", From 7c295ee7d5d720ec6e0264e594fbb4a64a83b0fe Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Wed, 8 Jul 2026 08:48:20 -0600 Subject: [PATCH 4/8] test: regenerate tpcds golden files for mixed sum/avg execution --- .../q70/extended.txt | 8 +- .../approved-plans-v1_4/q35/extended.txt | 2 +- .../approved-plans-v1_4/q45/extended.txt | 89 ++++++++++--------- .../q70a/extended.txt | 32 +++---- .../approved-plans-v2_7/q35/extended.txt | 2 +- 5 files changed, 67 insertions(+), 66 deletions(-) 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 08df1c04a0..f9fb64d9e4 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 @@ -3,9 +3,9 @@ CometNativeColumnarToRow +- CometProject +- CometWindowExec +- CometSort - +- CometColumnarExchange - +- HashAggregate [COMET: Spark Final aggregate without Comet Partial requires compatible intermediate buffer formats, but the following aggregate function(s) have incompatible buffers: sum] - +- Exchange + +- CometExchange + +- CometHashAggregate + +- CometColumnarExchange +- HashAggregate +- Expand +- Project @@ -56,4 +56,4 @@ CometNativeColumnarToRow +- CometFilter +- CometNativeScan parquet spark_catalog.default.date_dim -Comet accelerated 37 out of 53 eligible operators (69%). Final plan contains 4 transitions between Spark and Comet. \ No newline at end of file +Comet accelerated 39 out of 53 eligible operators (73%). Final plan contains 4 transitions between Spark and Comet. \ No newline at end of file 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 50e9ebb965..fcb5e0003d 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: avg, count] ++- 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/q45/extended.txt b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q45/extended.txt index 2d463ef986..6f73e28782 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,45 +1,46 @@ -TakeOrderedAndProject -+- 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 - +- Filter - +- BroadcastHashJoin [COMET: Unsupported join type ExistenceJoin(exists#1)] - :- CometNativeColumnarToRow - : +- CometProject - : +- CometBroadcastHashJoin - : :- CometProject - : : +- CometBroadcastHashJoin - : : :- CometProject - : : : +- CometBroadcastHashJoin - : : : :- CometProject - : : : : +- CometBroadcastHashJoin - : : : : :- CometFilter - : : : : : +- CometNativeScan parquet spark_catalog.default.web_sales - : : : : : +- CometSubqueryBroadcast - : : : : : +- CometBroadcastExchange - : : : : : +- CometProject - : : : : : +- CometFilter - : : : : : +- CometNativeScan parquet spark_catalog.default.date_dim - : : : : +- CometBroadcastExchange - : : : : +- CometFilter - : : : : +- CometNativeScan parquet spark_catalog.default.customer - : : : +- CometBroadcastExchange - : : : +- CometProject - : : : +- CometFilter - : : : +- CometNativeScan parquet spark_catalog.default.customer_address - : : +- CometBroadcastExchange - : : +- CometProject - : : +- CometFilter - : : +- CometNativeScan parquet spark_catalog.default.date_dim - : +- CometBroadcastExchange - : +- CometProject - : +- CometFilter - : +- CometNativeScan parquet spark_catalog.default.item - +- BroadcastExchange - +- CometNativeColumnarToRow - +- CometProject - +- CometFilter - +- CometNativeScan parquet spark_catalog.default.item +CometNativeColumnarToRow ++- CometTakeOrderedAndProject + +- CometHashAggregate + +- CometColumnarExchange + +- HashAggregate + +- Project + +- Filter + +- BroadcastHashJoin [COMET: Unsupported join type ExistenceJoin(exists#1)] + :- CometNativeColumnarToRow + : +- CometProject + : +- CometBroadcastHashJoin + : :- CometProject + : : +- CometBroadcastHashJoin + : : :- CometProject + : : : +- CometBroadcastHashJoin + : : : :- CometProject + : : : : +- CometBroadcastHashJoin + : : : : :- CometFilter + : : : : : +- CometNativeScan parquet spark_catalog.default.web_sales + : : : : : +- CometSubqueryBroadcast + : : : : : +- CometBroadcastExchange + : : : : : +- CometProject + : : : : : +- CometFilter + : : : : : +- CometNativeScan parquet spark_catalog.default.date_dim + : : : : +- CometBroadcastExchange + : : : : +- CometFilter + : : : : +- CometNativeScan parquet spark_catalog.default.customer + : : : +- CometBroadcastExchange + : : : +- CometProject + : : : +- CometFilter + : : : +- CometNativeScan parquet spark_catalog.default.customer_address + : : +- CometBroadcastExchange + : : +- CometProject + : : +- CometFilter + : : +- CometNativeScan parquet spark_catalog.default.date_dim + : +- CometBroadcastExchange + : +- CometProject + : +- CometFilter + : +- CometNativeScan parquet spark_catalog.default.item + +- BroadcastExchange + +- CometNativeColumnarToRow + +- CometProject + +- CometFilter + +- CometNativeScan parquet spark_catalog.default.item -Comet accelerated 32 out of 41 eligible operators (78%). Final plan contains 2 transitions between Spark and Comet. \ No newline at end of file +Comet accelerated 35 out of 41 eligible operators (85%). Final plan contains 3 transitions between Spark and Comet. \ No newline at end of file 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 e527d95f46..2aac2972c3 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 @@ -5,11 +5,11 @@ CometNativeColumnarToRow +- CometSort +- CometExchange +- CometHashAggregate - +- CometColumnarExchange - +- HashAggregate - +- Union - :- HashAggregate [COMET: Spark Final aggregate without Comet Partial requires compatible intermediate buffer formats, but the following aggregate function(s) have incompatible buffers: sum] - : +- Exchange + +- CometExchange + +- CometHashAggregate + +- CometUnion + :- CometHashAggregate + : +- CometColumnarExchange : +- HashAggregate : +- Project : +- BroadcastHashJoin @@ -58,11 +58,11 @@ CometNativeColumnarToRow : +- CometProject : +- CometFilter : +- CometNativeScan parquet spark_catalog.default.date_dim - :- 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, but the following aggregate function(s) have incompatible buffers: sum] - : +- Exchange + :- CometHashAggregate + : +- CometExchange + : +- CometHashAggregate + : +- CometHashAggregate + : +- CometColumnarExchange : +- HashAggregate : +- Project : +- BroadcastHashJoin @@ -111,11 +111,11 @@ CometNativeColumnarToRow : +- CometProject : +- CometFilter : +- CometNativeScan parquet spark_catalog.default.date_dim - +- 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, but the following aggregate function(s) have incompatible buffers: sum] - +- Exchange + +- CometHashAggregate + +- CometExchange + +- CometHashAggregate + +- CometHashAggregate + +- CometColumnarExchange +- HashAggregate +- Project +- BroadcastHashJoin @@ -165,4 +165,4 @@ CometNativeColumnarToRow +- CometFilter +- CometNativeScan parquet spark_catalog.default.date_dim -Comet accelerated 103 out of 156 eligible operators (66%). Final plan contains 10 transitions between Spark and Comet. \ No newline at end of file +Comet accelerated 117 out of 156 eligible operators (75%). Final plan contains 10 transitions between Spark and Comet. \ No newline at end of file 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 c340d4a42c..fcb5e0003d 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: avg, count, sum] ++- 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 From 8926d4e0340a8e0622fd363d54487b381272907e Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Wed, 8 Jul 2026 09:03:39 -0600 Subject: [PATCH 5/8] test: add decimal-AVG-blocked guard and AVG parity/coverage tests --- .../comet/exec/CometAggregateSuite.scala | 4 +- .../comet/rules/CometExecRuleSuite.scala | 39 +++++++++++++++++++ 2 files changed, 41 insertions(+), 2 deletions(-) diff --git a/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala index f8c990fd09..7e5eca4e99 100644 --- a/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala +++ b/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala @@ -191,7 +191,7 @@ class CometAggregateSuite extends CometTestBase with AdaptiveSparkPlanHelper { CometConf.COMET_EXEC_SHUFFLE_ENABLED.key -> "true", CometConf.COMET_SHUFFLE_MODE.key -> "jvm") { checkSparkAnswer( - "SELECT _4, SUM(_1), SUM(_2), SUM(_3), AVG(_1), AVG(_3) FROM tbl GROUP BY _4") + "SELECT _4, SUM(_1), SUM(_2), SUM(_3), AVG(_1), AVG(_2), AVG(_3) FROM tbl GROUP BY _4") } } } @@ -204,7 +204,7 @@ class CometAggregateSuite extends CometTestBase with AdaptiveSparkPlanHelper { CometConf.COMET_EXEC_SHUFFLE_ENABLED.key -> "true", CometConf.COMET_SHUFFLE_MODE.key -> "jvm") { checkSparkAnswer( - "SELECT _4, SUM(_1), SUM(_2), SUM(_3), AVG(_1), AVG(_3) FROM tbl GROUP BY _4") + "SELECT _4, SUM(_1), SUM(_2), SUM(_3), AVG(_1), AVG(_2), AVG(_3) FROM tbl GROUP BY _4") } } } diff --git a/spark/src/test/scala/org/apache/comet/rules/CometExecRuleSuite.scala b/spark/src/test/scala/org/apache/comet/rules/CometExecRuleSuite.scala index 1f76361ad7..893405ff66 100644 --- a/spark/src/test/scala/org/apache/comet/rules/CometExecRuleSuite.scala +++ b/spark/src/test/scala/org/apache/comet/rules/CometExecRuleSuite.scala @@ -300,6 +300,45 @@ class CometExecRuleSuite extends CometTestBase { } } + test("CometExecRule should not allow decimal AVG mixed execution") { + withTempView("test_data") { + createTestDataFrame.createOrReplaceTempView("test_data") + // Precision must be large enough (prec + 4 > 15) that Spark's own DecimalAggregates + // optimizer rule does not rewrite AVG to operate on the unscaled Long value, which would + // sidestep the decimal buffer path this test is meant to exercise. + val sparkPlan = + createSparkPlan( + spark, + "SELECT AVG(CAST(id AS DECIMAL(20, 2))) FROM test_data GROUP BY (id % 3)") + assert(countOperators(sparkPlan, classOf[HashAggregateExec]) == 2) + withSQLConf( + CometConf.COMET_ENABLE_FINAL_HASH_AGGREGATE.key -> "false", + CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true") { + val transformedPlan = applyCometExecRule(sparkPlan) + // Decimal AVG is deferred (its overflow path nulls count differently from Spark), so + // mixed execution is unsafe and the partial must also fall back to Spark. + assert(countOperators(transformedPlan, classOf[HashAggregateExec]) == 2) + assert(countOperators(transformedPlan, classOf[CometHashAggregateExec]) == 0) + } + } + } + + test("CometExecRule should allow AVG mixed Spark partial and Comet final") { + withTempView("test_data") { + createTestDataFrame.createOrReplaceTempView("test_data") + val sparkPlan = + createSparkPlan(spark, "SELECT AVG(id) FROM test_data GROUP BY (id % 3)") + assert(countOperators(sparkPlan, classOf[HashAggregateExec]) == 2) + withSQLConf( + CometConf.COMET_ENABLE_PARTIAL_HASH_AGGREGATE.key -> "false", + CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true") { + val transformedPlan = applyCometExecRule(sparkPlan) + assert(countOperators(transformedPlan, classOf[HashAggregateExec]) == 1) // partial + assert(countOperators(transformedPlan, classOf[CometHashAggregateExec]) == 1) // final + } + } + } + test("CometExecRule should allow BloomFilter mixed Comet partial and Spark final") { assume(!isSpark42Plus, "https://github.com/apache/datafusion-comet/issues/4142") val funcId = new FunctionIdentifier("bloom_filter_agg") From 9c8e7456a81b07dfb900e93acf86180291dcaa0c Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Wed, 8 Jul 2026 10:35:53 -0600 Subject: [PATCH 6/8] fix: include resultExpressions in CometHashAggregateExec identity Add resultExpressions to CometHashAggregateExec equals and hashCode. It carries any post-aggregate projection fused into the Final aggregate (for example count+1 versus count-1 in count-bug decorrelation). Without it, two aggregates differing only in that projection canonicalize identically and an enclosing broadcast exchange is wrongly deduplicated by exchange reuse. Fixes #4242 --- .../org/apache/spark/sql/comet/operators.scala | 15 ++++++++++++++- 1 file changed, 14 insertions(+), 1 deletion(-) 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 2db1873aa0..a9f7b54eaa 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 @@ -1980,6 +1980,12 @@ case class CometHashAggregateExec( this.output == other.output && this.groupingExpressions == other.groupingExpressions && this.aggregateExpressions == other.aggregateExpressions && + // resultExpressions carries any post-aggregate projection fused into the Final + // aggregate (e.g. count+1 vs count-1 in count-bug decorrelation). It must + // participate in identity, otherwise two aggregates differing only in that + // projection canonicalize identically and an enclosing broadcast exchange is + // wrongly deduplicated by exchange reuse. See issue #4242. + this.resultExpressions == other.resultExpressions && this.input == other.input && this.modes == other.modes && this.child == other.child && @@ -1990,7 +1996,14 @@ case class CometHashAggregateExec( } override def hashCode(): Int = - Objects.hashCode(output, groupingExpressions, aggregateExpressions, input, modes, child) + Objects.hashCode( + output, + groupingExpressions, + aggregateExpressions, + resultExpressions, + input, + modes, + child) override protected def outputExpressions: Seq[NamedExpression] = resultExpressions } From fda26d8e0d6cc66524a5003a2ed2bf366e8e9d38 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Wed, 8 Jul 2026 10:35:53 -0600 Subject: [PATCH 7/8] feat: enable mixed partial/final execution for COUNT Set supportsMixedPartialFinal for CometCount so a Spark partial COUNT can feed a Comet final COUNT. Add regression tests covering AQE PropagateEmptyRelation and count-bug decorrelation with correlated IN. Annotate the Spark SQL tests whose plan-shape or exchange-reuse assertions change under Comet, and regenerate the TPC-DS plan-stability golden files. --- dev/diffs/3.4.3.diff | 120 ++++++++++++++--- dev/diffs/3.5.8.diff | 124 ++++++++++++++--- dev/diffs/4.0.2.diff | 126 +++++++++++++++--- dev/diffs/4.1.2.diff | 126 +++++++++++++++--- .../org/apache/comet/serde/aggregates.scala | 3 + .../approved-plans-v1_4/q10/extended.txt | 121 ++++++++--------- .../approved-plans-v1_4/q35/extended.txt | 121 ++++++++--------- .../approved-plans-v1_4/q44/extended.txt | 11 +- .../approved-plans-v2_7/q35/extended.txt | 121 ++++++++--------- .../comet/exec/CometAggregateSuite.scala | 56 ++++++++ 10 files changed, 664 insertions(+), 265 deletions(-) diff --git a/dev/diffs/3.4.3.diff b/dev/diffs/3.4.3.diff index 6faf9dab78..5f63429d43 100644 --- a/dev/diffs/3.4.3.diff +++ b/dev/diffs/3.4.3.diff @@ -1,5 +1,5 @@ diff --git a/pom.xml b/pom.xml -index d3544881af1..aae0ae3b27b 100644 +index d3544881af1..ff963395ec3 100644 --- a/pom.xml +++ b/pom.xml @@ -148,6 +148,8 @@ @@ -236,6 +236,30 @@ index 0efe0877e9b..423d3b3d76d 100644 -- -- SELECT_HAVING -- https://github.com/postgres/postgres/blob/REL_12_BETA2/src/test/regress/sql/select_having.sql +diff --git a/sql/core/src/test/scala/org/apache/spark/sql/CTEInlineSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/CTEInlineSuite.scala +index c7c09bf7c79..5eaa5222142 100644 +--- a/sql/core/src/test/scala/org/apache/spark/sql/CTEInlineSuite.scala ++++ b/sql/core/src/test/scala/org/apache/spark/sql/CTEInlineSuite.scala +@@ -280,7 +280,8 @@ abstract class CTEInlineSuiteBase + } + } + +- test("CTE Predicate push-down and column pruning") { ++ test("CTE Predicate push-down and column pruning", ++ IgnoreComet("Comet changes the exchange reuse count asserted by this test")) { + withTempView("t") { + Seq((0, 1), (1, 2)).toDF("c1", "c2").createOrReplaceTempView("t") + val df = sql( +@@ -330,7 +331,8 @@ abstract class CTEInlineSuiteBase + } + } + +- test("CTE Predicate push-down and column pruning - combined predicate") { ++ test("CTE Predicate push-down and column pruning - combined predicate", ++ IgnoreComet("Comet changes the exchange reuse count asserted by this test")) { + withTempView("t") { + Seq((0, 1, 2), (1, 2, 3)).toDF("c1", "c2", "c3").createOrReplaceTempView("t") + val df = sql( diff --git a/sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala index cf40e944c09..bdd5be4f462 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala @@ -1532,7 +1556,7 @@ index ac710c32296..2854b433dd3 100644 import testImplicits._ diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala -index 593bd7bb4ba..32af28b0238 100644 +index 593bd7bb4ba..db92514eb2d 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala @@ -26,9 +26,11 @@ import org.scalatest.time.SpanSugar._ @@ -1800,7 +1824,27 @@ index 593bd7bb4ba..32af28b0238 100644 withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true") { val (_, adaptivePlan) = runAdaptiveAndVerifyResult( "SELECT key FROM testData GROUP BY key") -@@ -1599,7 +1628,7 @@ class AdaptiveQueryExecSuite +@@ -1541,7 +1570,8 @@ class AdaptiveQueryExecSuite + } + } + +- test("SPARK-35442: Support propagate empty relation through aggregate") { ++ test("SPARK-35442: Support propagate empty relation through aggregate", ++ IgnoreComet("https://github.com/apache/datafusion-comet/issues/4412")) { + withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true") { + val (plan1, adaptivePlan1) = runAdaptiveAndVerifyResult( + "SELECT key, count(*) FROM testData WHERE value = 'no_match' GROUP BY key") +@@ -1560,7 +1590,8 @@ class AdaptiveQueryExecSuite + } + } + +- test("SPARK-35442: Support propagate empty relation through union") { ++ test("SPARK-35442: Support propagate empty relation through union", ++ IgnoreComet("https://github.com/apache/datafusion-comet/issues/4412")) { + def checkNumUnion(plan: SparkPlan, numUnion: Int): Unit = { + assert( + collect(plan) { +@@ -1599,7 +1630,7 @@ class AdaptiveQueryExecSuite val (_, adaptivePlan) = runAdaptiveAndVerifyResult( "SELECT id FROM v1 GROUP BY id DISTRIBUTE BY id") assert(collect(adaptivePlan) { @@ -1809,7 +1853,7 @@ index 593bd7bb4ba..32af28b0238 100644 }.length == 1) } } -@@ -1679,7 +1708,8 @@ class AdaptiveQueryExecSuite +@@ -1679,7 +1710,8 @@ class AdaptiveQueryExecSuite } } @@ -1819,7 +1863,7 @@ index 593bd7bb4ba..32af28b0238 100644 def hasRepartitionShuffle(plan: SparkPlan): Boolean = { find(plan) { case s: ShuffleExchangeLike => -@@ -1864,6 +1894,9 @@ class AdaptiveQueryExecSuite +@@ -1864,6 +1896,9 @@ class AdaptiveQueryExecSuite def checkNoCoalescePartitions(ds: Dataset[Row], origin: ShuffleOrigin): Unit = { assert(collect(ds.queryExecution.executedPlan) { case s: ShuffleExchangeExec if s.shuffleOrigin == origin && s.numPartitions == 2 => s @@ -1829,7 +1873,7 @@ index 593bd7bb4ba..32af28b0238 100644 }.size == 1) ds.collect() val plan = ds.queryExecution.executedPlan -@@ -1872,6 +1905,9 @@ class AdaptiveQueryExecSuite +@@ -1872,6 +1907,9 @@ class AdaptiveQueryExecSuite }.isEmpty) assert(collect(plan) { case s: ShuffleExchangeExec if s.shuffleOrigin == origin && s.numPartitions == 2 => s @@ -1839,7 +1883,7 @@ index 593bd7bb4ba..32af28b0238 100644 }.size == 1) checkAnswer(ds, testData) } -@@ -2028,7 +2064,8 @@ class AdaptiveQueryExecSuite +@@ -2028,7 +2066,8 @@ class AdaptiveQueryExecSuite } } @@ -1849,7 +1893,7 @@ index 593bd7bb4ba..32af28b0238 100644 withTempView("t1", "t2") { def checkJoinStrategy(shouldShuffleHashJoin: Boolean): Unit = { Seq("100", "100000").foreach { size => -@@ -2114,7 +2151,8 @@ class AdaptiveQueryExecSuite +@@ -2114,7 +2153,8 @@ class AdaptiveQueryExecSuite } } @@ -1859,7 +1903,7 @@ index 593bd7bb4ba..32af28b0238 100644 withTempView("v") { withSQLConf( SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", -@@ -2213,7 +2251,7 @@ class AdaptiveQueryExecSuite +@@ -2213,7 +2253,7 @@ class AdaptiveQueryExecSuite runAdaptiveAndVerifyResult(s"SELECT $repartition key1 FROM skewData1 " + s"JOIN skewData2 ON key1 = key2 GROUP BY key1") val shuffles1 = collect(adaptive1) { @@ -1868,7 +1912,7 @@ index 593bd7bb4ba..32af28b0238 100644 } assert(shuffles1.size == 3) // shuffles1.head is the top-level shuffle under the Aggregate operator -@@ -2226,7 +2264,7 @@ class AdaptiveQueryExecSuite +@@ -2226,7 +2266,7 @@ class AdaptiveQueryExecSuite runAdaptiveAndVerifyResult(s"SELECT $repartition key1 FROM skewData1 " + s"JOIN skewData2 ON key1 = key2") val shuffles2 = collect(adaptive2) { @@ -1877,7 +1921,7 @@ index 593bd7bb4ba..32af28b0238 100644 } if (hasRequiredDistribution) { assert(shuffles2.size == 3) -@@ -2260,7 +2298,8 @@ class AdaptiveQueryExecSuite +@@ -2260,7 +2300,8 @@ class AdaptiveQueryExecSuite } } @@ -1887,7 +1931,7 @@ index 593bd7bb4ba..32af28b0238 100644 CostEvaluator.instantiate( classOf[SimpleShuffleSortCostEvaluator].getCanonicalName, spark.sparkContext.getConf) intercept[IllegalArgumentException] { -@@ -2404,6 +2443,7 @@ class AdaptiveQueryExecSuite +@@ -2404,6 +2445,7 @@ class AdaptiveQueryExecSuite val (_, adaptive) = runAdaptiveAndVerifyResult(query) assert(adaptive.collect { case sort: SortExec => sort @@ -1895,7 +1939,7 @@ index 593bd7bb4ba..32af28b0238 100644 }.size == 1) val read = collect(adaptive) { case read: AQEShuffleReadExec => read -@@ -2421,7 +2461,8 @@ class AdaptiveQueryExecSuite +@@ -2421,7 +2463,8 @@ class AdaptiveQueryExecSuite } } @@ -1905,7 +1949,7 @@ index 593bd7bb4ba..32af28b0238 100644 withTempView("v") { withSQLConf( SQLConf.ADAPTIVE_OPTIMIZE_SKEWS_IN_REBALANCE_PARTITIONS_ENABLED.key -> "true", -@@ -2533,7 +2574,7 @@ class AdaptiveQueryExecSuite +@@ -2533,7 +2576,7 @@ class AdaptiveQueryExecSuite runAdaptiveAndVerifyResult("SELECT key1 FROM skewData1 JOIN skewData2 ON key1 = key2 " + "JOIN skewData3 ON value2 = value3") val shuffles1 = collect(adaptive1) { @@ -1914,7 +1958,7 @@ index 593bd7bb4ba..32af28b0238 100644 } assert(shuffles1.size == 4) val smj1 = findTopLevelSortMergeJoin(adaptive1) -@@ -2544,7 +2585,7 @@ class AdaptiveQueryExecSuite +@@ -2544,7 +2587,7 @@ class AdaptiveQueryExecSuite runAdaptiveAndVerifyResult("SELECT key1 FROM skewData1 JOIN skewData2 ON key1 = key2 " + "JOIN skewData3 ON value1 = value3") val shuffles2 = collect(adaptive2) { @@ -2385,7 +2429,7 @@ index bf5c51b89bb..f7402b7d883 100644 val e = testSchemaMismatch(dir.getCanonicalPath, vectorizedReaderEnabled = true) assert(e.getCause.isInstanceOf[SparkException]) diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/debug/DebuggingSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/debug/DebuggingSuite.scala -index 3a0bd35cb70..b28f06a757f 100644 +index 3a0bd35cb70..99b70606261 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/debug/DebuggingSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/debug/DebuggingSuite.scala @@ -20,6 +20,7 @@ package org.apache.spark.sql.execution.debug @@ -2396,7 +2440,27 @@ index 3a0bd35cb70..b28f06a757f 100644 import org.apache.spark.sql.catalyst.InternalRow import org.apache.spark.sql.catalyst.expressions.Attribute import org.apache.spark.sql.catalyst.expressions.codegen.CodegenContext -@@ -124,7 +125,8 @@ class DebuggingSuite extends DebuggingSuiteBase with DisableAdaptiveExecutionSui +@@ -41,7 +42,8 @@ abstract class DebuggingSuiteBase extends SharedSparkSession { + testData.as[TestData].debug() + } + +- test("debugCodegen") { ++ test("debugCodegen", ++ IgnoreComet("Comet changes the WholeStageCodegen subtree count")) { + val df = spark.range(10).groupBy(col("id") * 2).count() + df.collect() + val res = codegenString(df.queryExecution.executedPlan) +@@ -50,7 +52,8 @@ abstract class DebuggingSuiteBase extends SharedSparkSession { + assert(res.contains("Object[]")) + } + +- test("debugCodegenStringSeq") { ++ test("debugCodegenStringSeq", ++ IgnoreComet("Comet changes the WholeStageCodegen subtree count")) { + val df = spark.range(10).groupBy(col("id") * 2).count() + df.collect() + val res = codegenStringSeq(df.queryExecution.executedPlan) +@@ -124,7 +127,8 @@ class DebuggingSuite extends DebuggingSuiteBase with DisableAdaptiveExecutionSui | id LongType: {}""".stripMargin)) } @@ -2506,6 +2570,28 @@ index d083cac48ff..3c11bcde807 100644 import testImplicits._ +diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/ui/SQLAppStatusListenerSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/ui/SQLAppStatusListenerSuite.scala +index fdc633f3556..f1e1751171a 100644 +--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/ui/SQLAppStatusListenerSuite.scala ++++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/ui/SQLAppStatusListenerSuite.scala +@@ -36,6 +36,7 @@ import org.apache.spark.internal.config.Status._ + import org.apache.spark.rdd.RDD + import org.apache.spark.resource.ResourceProfile + import org.apache.spark.scheduler._ ++import org.apache.spark.sql.IgnoreComet + import org.apache.spark.sql.{DataFrame, SparkSession} + import org.apache.spark.sql.catalyst.InternalRow + import org.apache.spark.sql.catalyst.expressions.Attribute +@@ -699,7 +700,8 @@ abstract class SQLAppStatusListenerSuite extends SharedSparkSession with JsonTes + } + + test("SPARK-29894 test Codegen Stage Id in SparkPlanInfo", +- DisableAdaptiveExecution("WSCG rule is applied later in AQE")) { ++ DisableAdaptiveExecution("WSCG rule is applied later in AQE"), ++ IgnoreComet("Comet changes the WholeStageCodegen subtree count")) { + // with AQE on, the WholeStageCodegen rule is applied when running QueryStageExec. + val df = createTestDataFrame.select(count("*")) + val sparkPlanInfo = SparkPlanInfo.fromSparkPlan(df.queryExecution.executedPlan) diff --git a/sql/core/src/test/scala/org/apache/spark/sql/sources/BucketedReadSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/sources/BucketedReadSuite.scala index 266bb343526..f8ad838e2b2 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/sources/BucketedReadSuite.scala diff --git a/dev/diffs/3.5.8.diff b/dev/diffs/3.5.8.diff index 7155e9a9b4..cfe8f5f0cc 100644 --- a/dev/diffs/3.5.8.diff +++ b/dev/diffs/3.5.8.diff @@ -1,5 +1,5 @@ diff --git a/pom.xml b/pom.xml -index edd2ad57880..a47b7dec672 100644 +index edd2ad57880..45f8fd01538 100644 --- a/pom.xml +++ b/pom.xml @@ -152,6 +152,8 @@ @@ -217,6 +217,30 @@ index 0efe0877e9b..423d3b3d76d 100644 -- -- SELECT_HAVING -- https://github.com/postgres/postgres/blob/REL_12_BETA2/src/test/regress/sql/select_having.sql +diff --git a/sql/core/src/test/scala/org/apache/spark/sql/CTEInlineSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/CTEInlineSuite.scala +index 73f5b742715..5de6754c70d 100644 +--- a/sql/core/src/test/scala/org/apache/spark/sql/CTEInlineSuite.scala ++++ b/sql/core/src/test/scala/org/apache/spark/sql/CTEInlineSuite.scala +@@ -280,7 +280,8 @@ abstract class CTEInlineSuiteBase + } + } + +- test("CTE Predicate push-down and column pruning") { ++ test("CTE Predicate push-down and column pruning", ++ IgnoreComet("Comet changes the exchange reuse count asserted by this test")) { + withTempView("t") { + Seq((0, 1), (1, 2)).toDF("c1", "c2").createOrReplaceTempView("t") + val df = sql( +@@ -330,7 +331,8 @@ abstract class CTEInlineSuiteBase + } + } + +- test("CTE Predicate push-down and column pruning - combined predicate") { ++ test("CTE Predicate push-down and column pruning - combined predicate", ++ IgnoreComet("Comet changes the exchange reuse count asserted by this test")) { + withTempView("t") { + Seq((0, 1, 2), (1, 2, 3)).toDF("c1", "c2", "c3").createOrReplaceTempView("t") + val df = sql( diff --git a/sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala index e5494726695..00937f025c2 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala @@ -1498,7 +1522,7 @@ index 5a413c77754..207b66e1d7b 100644 import testImplicits._ diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala -index 2f8e401e743..dbcf3171946 100644 +index 2f8e401e743..c6ed076e24e 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala @@ -27,9 +27,11 @@ import org.scalatest.time.SpanSugar._ @@ -1773,7 +1797,27 @@ index 2f8e401e743..dbcf3171946 100644 withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true") { val (_, adaptivePlan) = runAdaptiveAndVerifyResult( "SELECT key FROM testData GROUP BY key") -@@ -1625,7 +1655,7 @@ class AdaptiveQueryExecSuite +@@ -1567,7 +1597,8 @@ class AdaptiveQueryExecSuite + } + } + +- test("SPARK-35442: Support propagate empty relation through aggregate") { ++ test("SPARK-35442: Support propagate empty relation through aggregate", ++ IgnoreComet("https://github.com/apache/datafusion-comet/issues/4412")) { + withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true") { + val (plan1, adaptivePlan1) = runAdaptiveAndVerifyResult( + "SELECT key, count(*) FROM testData WHERE value = 'no_match' GROUP BY key") +@@ -1586,7 +1617,8 @@ class AdaptiveQueryExecSuite + } + } + +- test("SPARK-35442: Support propagate empty relation through union") { ++ test("SPARK-35442: Support propagate empty relation through union", ++ IgnoreComet("https://github.com/apache/datafusion-comet/issues/4412")) { + def checkNumUnion(plan: SparkPlan, numUnion: Int): Unit = { + assert( + collect(plan) { +@@ -1625,7 +1657,7 @@ class AdaptiveQueryExecSuite val (_, adaptivePlan) = runAdaptiveAndVerifyResult( "SELECT id FROM v1 GROUP BY id DISTRIBUTE BY id") assert(collect(adaptivePlan) { @@ -1782,7 +1826,7 @@ index 2f8e401e743..dbcf3171946 100644 }.length == 1) } } -@@ -1705,7 +1735,8 @@ class AdaptiveQueryExecSuite +@@ -1705,7 +1737,8 @@ class AdaptiveQueryExecSuite } } @@ -1792,7 +1836,7 @@ index 2f8e401e743..dbcf3171946 100644 def hasRepartitionShuffle(plan: SparkPlan): Boolean = { find(plan) { case s: ShuffleExchangeLike => -@@ -1890,6 +1921,9 @@ class AdaptiveQueryExecSuite +@@ -1890,6 +1923,9 @@ class AdaptiveQueryExecSuite def checkNoCoalescePartitions(ds: Dataset[Row], origin: ShuffleOrigin): Unit = { assert(collect(ds.queryExecution.executedPlan) { case s: ShuffleExchangeExec if s.shuffleOrigin == origin && s.numPartitions == 2 => s @@ -1802,7 +1846,7 @@ index 2f8e401e743..dbcf3171946 100644 }.size == 1) ds.collect() val plan = ds.queryExecution.executedPlan -@@ -1898,6 +1932,9 @@ class AdaptiveQueryExecSuite +@@ -1898,6 +1934,9 @@ class AdaptiveQueryExecSuite }.isEmpty) assert(collect(plan) { case s: ShuffleExchangeExec if s.shuffleOrigin == origin && s.numPartitions == 2 => s @@ -1812,7 +1856,7 @@ index 2f8e401e743..dbcf3171946 100644 }.size == 1) checkAnswer(ds, testData) } -@@ -2054,7 +2091,8 @@ class AdaptiveQueryExecSuite +@@ -2054,7 +2093,8 @@ class AdaptiveQueryExecSuite } } @@ -1822,7 +1866,7 @@ index 2f8e401e743..dbcf3171946 100644 withTempView("t1", "t2") { def checkJoinStrategy(shouldShuffleHashJoin: Boolean): Unit = { Seq("100", "100000").foreach { size => -@@ -2140,7 +2178,8 @@ class AdaptiveQueryExecSuite +@@ -2140,7 +2180,8 @@ class AdaptiveQueryExecSuite } } @@ -1832,7 +1876,7 @@ index 2f8e401e743..dbcf3171946 100644 withTempView("v") { withSQLConf( SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", -@@ -2239,7 +2278,7 @@ class AdaptiveQueryExecSuite +@@ -2239,7 +2280,7 @@ class AdaptiveQueryExecSuite runAdaptiveAndVerifyResult(s"SELECT $repartition key1 FROM skewData1 " + s"JOIN skewData2 ON key1 = key2 GROUP BY key1") val shuffles1 = collect(adaptive1) { @@ -1841,7 +1885,7 @@ index 2f8e401e743..dbcf3171946 100644 } assert(shuffles1.size == 3) // shuffles1.head is the top-level shuffle under the Aggregate operator -@@ -2252,7 +2291,7 @@ class AdaptiveQueryExecSuite +@@ -2252,7 +2293,7 @@ class AdaptiveQueryExecSuite runAdaptiveAndVerifyResult(s"SELECT $repartition key1 FROM skewData1 " + s"JOIN skewData2 ON key1 = key2") val shuffles2 = collect(adaptive2) { @@ -1850,7 +1894,7 @@ index 2f8e401e743..dbcf3171946 100644 } if (hasRequiredDistribution) { assert(shuffles2.size == 3) -@@ -2286,7 +2325,8 @@ class AdaptiveQueryExecSuite +@@ -2286,7 +2327,8 @@ class AdaptiveQueryExecSuite } } @@ -1860,7 +1904,7 @@ index 2f8e401e743..dbcf3171946 100644 CostEvaluator.instantiate( classOf[SimpleShuffleSortCostEvaluator].getCanonicalName, spark.sparkContext.getConf) intercept[IllegalArgumentException] { -@@ -2452,6 +2492,7 @@ class AdaptiveQueryExecSuite +@@ -2452,6 +2494,7 @@ class AdaptiveQueryExecSuite val (_, adaptive) = runAdaptiveAndVerifyResult(query) assert(adaptive.collect { case sort: SortExec => sort @@ -1868,7 +1912,7 @@ index 2f8e401e743..dbcf3171946 100644 }.size == 1) val read = collect(adaptive) { case read: AQEShuffleReadExec => read -@@ -2469,7 +2510,8 @@ class AdaptiveQueryExecSuite +@@ -2469,7 +2512,8 @@ class AdaptiveQueryExecSuite } } @@ -1878,7 +1922,7 @@ index 2f8e401e743..dbcf3171946 100644 withTempView("v") { withSQLConf( SQLConf.ADAPTIVE_OPTIMIZE_SKEWS_IN_REBALANCE_PARTITIONS_ENABLED.key -> "true", -@@ -2581,7 +2623,7 @@ class AdaptiveQueryExecSuite +@@ -2581,7 +2625,7 @@ class AdaptiveQueryExecSuite runAdaptiveAndVerifyResult("SELECT key1 FROM skewData1 JOIN skewData2 ON key1 = key2 " + "JOIN skewData3 ON value2 = value3") val shuffles1 = collect(adaptive1) { @@ -1887,7 +1931,7 @@ index 2f8e401e743..dbcf3171946 100644 } assert(shuffles1.size == 4) val smj1 = findTopLevelSortMergeJoin(adaptive1) -@@ -2592,7 +2634,7 @@ class AdaptiveQueryExecSuite +@@ -2592,7 +2636,7 @@ class AdaptiveQueryExecSuite runAdaptiveAndVerifyResult("SELECT key1 FROM skewData1 JOIN skewData2 ON key1 = key2 " + "JOIN skewData3 ON value1 = value3") val shuffles2 = collect(adaptive2) { @@ -1896,7 +1940,7 @@ index 2f8e401e743..dbcf3171946 100644 } assert(shuffles2.size == 4) val smj2 = findTopLevelSortMergeJoin(adaptive2) -@@ -2850,6 +2892,7 @@ class AdaptiveQueryExecSuite +@@ -2850,6 +2894,7 @@ class AdaptiveQueryExecSuite }.size == (if (firstAccess) 1 else 0)) assert(collect(initialExecutedPlan) { case s: SortExec => s @@ -1904,7 +1948,7 @@ index 2f8e401e743..dbcf3171946 100644 }.size == (if (firstAccess) 2 else 0)) assert(collect(initialExecutedPlan) { case i: InMemoryTableScanLike => i -@@ -2980,7 +3023,9 @@ class AdaptiveQueryExecSuite +@@ -2980,7 +3025,9 @@ class AdaptiveQueryExecSuite val plan = df.queryExecution.executedPlan.asInstanceOf[AdaptiveSparkPlanExec] assert(plan.inputPlan.isInstanceOf[TakeOrderedAndProjectExec]) @@ -2349,7 +2393,7 @@ index 3f47c5e506f..8e8d2a1634d 100644 val e = testSchemaMismatch(dir.getCanonicalPath, vectorizedReaderEnabled = true) assert(e.getCause.isInstanceOf[SparkException]) diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/debug/DebuggingSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/debug/DebuggingSuite.scala -index b8f3ea3c6f3..bbd44221288 100644 +index b8f3ea3c6f3..0d92198bd83 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/debug/DebuggingSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/debug/DebuggingSuite.scala @@ -20,6 +20,7 @@ package org.apache.spark.sql.execution.debug @@ -2360,7 +2404,27 @@ index b8f3ea3c6f3..bbd44221288 100644 import org.apache.spark.sql.catalyst.InternalRow import org.apache.spark.sql.catalyst.expressions.Attribute import org.apache.spark.sql.catalyst.expressions.codegen.CodegenContext -@@ -125,7 +126,8 @@ class DebuggingSuite extends DebuggingSuiteBase with DisableAdaptiveExecutionSui +@@ -42,7 +43,8 @@ abstract class DebuggingSuiteBase extends SharedSparkSession { + testData.as[TestData].debug() + } + +- test("debugCodegen") { ++ test("debugCodegen", ++ IgnoreComet("Comet changes the WholeStageCodegen subtree count")) { + val df = spark.range(10).groupBy(col("id") * 2).count() + df.collect() + val res = codegenString(df.queryExecution.executedPlan) +@@ -51,7 +53,8 @@ abstract class DebuggingSuiteBase extends SharedSparkSession { + assert(res.contains("Object[]")) + } + +- test("debugCodegenStringSeq") { ++ test("debugCodegenStringSeq", ++ IgnoreComet("Comet changes the WholeStageCodegen subtree count")) { + val df = spark.range(10).groupBy(col("id") * 2).count() + df.collect() + val res = codegenStringSeq(df.queryExecution.executedPlan) +@@ -125,7 +128,8 @@ class DebuggingSuite extends DebuggingSuiteBase with DisableAdaptiveExecutionSui | id LongType: {}""".stripMargin)) } @@ -2470,6 +2534,28 @@ index d083cac48ff..3c11bcde807 100644 import testImplicits._ +diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/ui/SQLAppStatusListenerSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/ui/SQLAppStatusListenerSuite.scala +index 67206e9c655..b87172d258e 100644 +--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/ui/SQLAppStatusListenerSuite.scala ++++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/ui/SQLAppStatusListenerSuite.scala +@@ -35,6 +35,7 @@ import org.apache.spark.internal.config.Status._ + import org.apache.spark.rdd.RDD + import org.apache.spark.resource.ResourceProfile + import org.apache.spark.scheduler._ ++import org.apache.spark.sql.IgnoreComet + import org.apache.spark.sql.{DataFrame, SparkSession} + import org.apache.spark.sql.catalyst.InternalRow + import org.apache.spark.sql.catalyst.expressions.Attribute +@@ -699,7 +700,8 @@ abstract class SQLAppStatusListenerSuite extends SharedSparkSession with JsonTes + } + + test("SPARK-29894 test Codegen Stage Id in SparkPlanInfo", +- DisableAdaptiveExecution("WSCG rule is applied later in AQE")) { ++ DisableAdaptiveExecution("WSCG rule is applied later in AQE"), ++ IgnoreComet("Comet changes the WholeStageCodegen subtree count")) { + // with AQE on, the WholeStageCodegen rule is applied when running QueryStageExec. + val df = createTestDataFrame.select(count("*")) + val sparkPlanInfo = SparkPlanInfo.fromSparkPlan(df.queryExecution.executedPlan) diff --git a/sql/core/src/test/scala/org/apache/spark/sql/sources/BucketedReadSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/sources/BucketedReadSuite.scala index 746f289c393..e5dc13b87d5 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/sources/BucketedReadSuite.scala diff --git a/dev/diffs/4.0.2.diff b/dev/diffs/4.0.2.diff index 28cac668c8..c7ddaacfc1 100644 --- a/dev/diffs/4.0.2.diff +++ b/dev/diffs/4.0.2.diff @@ -39,7 +39,7 @@ index 6c51bd4ff2e..e72ec1d26e2 100644 withSpark(sc) { sc => TestUtils.waitUntilExecutorsUp(sc, 2, 60000) diff --git a/pom.xml b/pom.xml -index 252cfdf9073..60cb9dcb7cf 100644 +index 252cfdf9073..50ec9d6314e 100644 --- a/pom.xml +++ b/pom.xml @@ -148,6 +148,8 @@ @@ -332,6 +332,30 @@ index 21a3ce1e122..f4762ab98f0 100644 SET spark.sql.ansi.enabled = false; -- In COMPENSATION views get invalidated if the type can't cast +diff --git a/sql/core/src/test/scala/org/apache/spark/sql/CTEInlineSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/CTEInlineSuite.scala +index 80450b79a9a..41d1bf57938 100644 +--- a/sql/core/src/test/scala/org/apache/spark/sql/CTEInlineSuite.scala ++++ b/sql/core/src/test/scala/org/apache/spark/sql/CTEInlineSuite.scala +@@ -282,7 +282,8 @@ abstract class CTEInlineSuiteBase + } + } + +- test("CTE Predicate push-down and column pruning") { ++ test("CTE Predicate push-down and column pruning", ++ IgnoreComet("Comet changes the exchange reuse count asserted by this test")) { + withTempView("t") { + Seq((0, 1), (1, 2)).toDF("c1", "c2").createOrReplaceTempView("t") + val df = sql( +@@ -332,7 +333,8 @@ abstract class CTEInlineSuiteBase + } + } + +- test("CTE Predicate push-down and column pruning - combined predicate") { ++ test("CTE Predicate push-down and column pruning - combined predicate", ++ IgnoreComet("Comet changes the exchange reuse count asserted by this test")) { + withTempView("t") { + Seq((0, 1, 2, 3), (1, 2, 3, 4)).toDF("c1", "c2", "c3", "c4").createOrReplaceTempView("t") + val df = sql( diff --git a/sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala index 0f42502f1d9..e9ff802141f 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala @@ -2122,7 +2146,7 @@ index a3cfdc5a240..3793b6191bf 100644 }) checkAnswer(distinctWithId, Seq(Row(1, 0), Row(1, 0))) diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala -index 272be70f9fe..12daa1f5932 100644 +index 272be70f9fe..747bcb6c0e0 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala @@ -28,12 +28,14 @@ import org.apache.spark.SparkException @@ -2400,7 +2424,27 @@ index 272be70f9fe..12daa1f5932 100644 withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true") { val (_, adaptivePlan) = runAdaptiveAndVerifyResult( "SELECT key FROM testData GROUP BY key") -@@ -1721,7 +1751,7 @@ class AdaptiveQueryExecSuite +@@ -1663,7 +1693,8 @@ class AdaptiveQueryExecSuite + } + } + +- test("SPARK-35442: Support propagate empty relation through aggregate") { ++ test("SPARK-35442: Support propagate empty relation through aggregate", ++ IgnoreComet("https://github.com/apache/datafusion-comet/issues/4412")) { + withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true") { + val (plan1, adaptivePlan1) = runAdaptiveAndVerifyResult( + "SELECT key, count(*) FROM testData WHERE value = 'no_match' GROUP BY key") +@@ -1682,7 +1713,8 @@ class AdaptiveQueryExecSuite + } + } + +- test("SPARK-35442: Support propagate empty relation through union") { ++ test("SPARK-35442: Support propagate empty relation through union", ++ IgnoreComet("https://github.com/apache/datafusion-comet/issues/4412")) { + def checkNumUnion(plan: SparkPlan, numUnion: Int): Unit = { + assert( + collect(plan) { +@@ -1721,7 +1753,7 @@ class AdaptiveQueryExecSuite val (_, adaptivePlan) = runAdaptiveAndVerifyResult( "SELECT id FROM v1 GROUP BY id DISTRIBUTE BY id") assert(collect(adaptivePlan) { @@ -2409,7 +2453,7 @@ index 272be70f9fe..12daa1f5932 100644 }.length == 1) } } -@@ -1801,7 +1831,8 @@ class AdaptiveQueryExecSuite +@@ -1801,7 +1833,8 @@ class AdaptiveQueryExecSuite } } @@ -2419,7 +2463,7 @@ index 272be70f9fe..12daa1f5932 100644 def hasRepartitionShuffle(plan: SparkPlan): Boolean = { find(plan) { case s: ShuffleExchangeLike => -@@ -1986,6 +2017,9 @@ class AdaptiveQueryExecSuite +@@ -1986,6 +2019,9 @@ class AdaptiveQueryExecSuite def checkNoCoalescePartitions(ds: Dataset[Row], origin: ShuffleOrigin): Unit = { assert(collect(ds.queryExecution.executedPlan) { case s: ShuffleExchangeExec if s.shuffleOrigin == origin && s.numPartitions == 2 => s @@ -2429,7 +2473,7 @@ index 272be70f9fe..12daa1f5932 100644 }.size == 1) ds.collect() val plan = ds.queryExecution.executedPlan -@@ -1994,6 +2028,9 @@ class AdaptiveQueryExecSuite +@@ -1994,6 +2030,9 @@ class AdaptiveQueryExecSuite }.isEmpty) assert(collect(plan) { case s: ShuffleExchangeExec if s.shuffleOrigin == origin && s.numPartitions == 2 => s @@ -2439,7 +2483,7 @@ index 272be70f9fe..12daa1f5932 100644 }.size == 1) checkAnswer(ds, testData) } -@@ -2150,7 +2187,8 @@ class AdaptiveQueryExecSuite +@@ -2150,7 +2189,8 @@ class AdaptiveQueryExecSuite } } @@ -2449,7 +2493,7 @@ index 272be70f9fe..12daa1f5932 100644 withTempView("t1", "t2") { def checkJoinStrategy(shouldShuffleHashJoin: Boolean): Unit = { Seq("100", "100000").foreach { size => -@@ -2236,7 +2274,8 @@ class AdaptiveQueryExecSuite +@@ -2236,7 +2276,8 @@ class AdaptiveQueryExecSuite } } @@ -2459,7 +2503,7 @@ index 272be70f9fe..12daa1f5932 100644 withTempView("v") { withSQLConf( SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", -@@ -2335,7 +2374,7 @@ class AdaptiveQueryExecSuite +@@ -2335,7 +2376,7 @@ class AdaptiveQueryExecSuite runAdaptiveAndVerifyResult(s"SELECT $repartition key1 FROM skewData1 " + s"JOIN skewData2 ON key1 = key2 GROUP BY key1") val shuffles1 = collect(adaptive1) { @@ -2468,7 +2512,7 @@ index 272be70f9fe..12daa1f5932 100644 } assert(shuffles1.size == 3) // shuffles1.head is the top-level shuffle under the Aggregate operator -@@ -2348,7 +2387,7 @@ class AdaptiveQueryExecSuite +@@ -2348,7 +2389,7 @@ class AdaptiveQueryExecSuite runAdaptiveAndVerifyResult(s"SELECT $repartition key1 FROM skewData1 " + s"JOIN skewData2 ON key1 = key2") val shuffles2 = collect(adaptive2) { @@ -2477,7 +2521,7 @@ index 272be70f9fe..12daa1f5932 100644 } if (hasRequiredDistribution) { assert(shuffles2.size == 3) -@@ -2382,7 +2421,8 @@ class AdaptiveQueryExecSuite +@@ -2382,7 +2423,8 @@ class AdaptiveQueryExecSuite } } @@ -2487,7 +2531,7 @@ index 272be70f9fe..12daa1f5932 100644 CostEvaluator.instantiate( classOf[SimpleShuffleSortCostEvaluator].getCanonicalName, spark.sparkContext.getConf) intercept[IllegalArgumentException] { -@@ -2548,6 +2588,7 @@ class AdaptiveQueryExecSuite +@@ -2548,6 +2590,7 @@ class AdaptiveQueryExecSuite val (_, adaptive) = runAdaptiveAndVerifyResult(query) assert(adaptive.collect { case sort: SortExec => sort @@ -2495,7 +2539,7 @@ index 272be70f9fe..12daa1f5932 100644 }.size == 1) val read = collect(adaptive) { case read: AQEShuffleReadExec => read -@@ -2565,7 +2606,8 @@ class AdaptiveQueryExecSuite +@@ -2565,7 +2608,8 @@ class AdaptiveQueryExecSuite } } @@ -2505,7 +2549,7 @@ index 272be70f9fe..12daa1f5932 100644 withTempView("v") { withSQLConf( SQLConf.ADAPTIVE_OPTIMIZE_SKEWS_IN_REBALANCE_PARTITIONS_ENABLED.key -> "true", -@@ -2677,7 +2719,7 @@ class AdaptiveQueryExecSuite +@@ -2677,7 +2721,7 @@ class AdaptiveQueryExecSuite runAdaptiveAndVerifyResult("SELECT key1 FROM skewData1 JOIN skewData2 ON key1 = key2 " + "JOIN skewData3 ON value2 = value3") val shuffles1 = collect(adaptive1) { @@ -2514,7 +2558,7 @@ index 272be70f9fe..12daa1f5932 100644 } assert(shuffles1.size == 4) val smj1 = findTopLevelSortMergeJoin(adaptive1) -@@ -2688,7 +2730,7 @@ class AdaptiveQueryExecSuite +@@ -2688,7 +2732,7 @@ class AdaptiveQueryExecSuite runAdaptiveAndVerifyResult("SELECT key1 FROM skewData1 JOIN skewData2 ON key1 = key2 " + "JOIN skewData3 ON value1 = value3") val shuffles2 = collect(adaptive2) { @@ -2523,7 +2567,7 @@ index 272be70f9fe..12daa1f5932 100644 } assert(shuffles2.size == 4) val smj2 = findTopLevelSortMergeJoin(adaptive2) -@@ -2946,6 +2988,7 @@ class AdaptiveQueryExecSuite +@@ -2946,6 +2990,7 @@ class AdaptiveQueryExecSuite }.size == (if (firstAccess) 1 else 0)) assert(collect(initialExecutedPlan) { case s: SortExec => s @@ -2531,7 +2575,7 @@ index 272be70f9fe..12daa1f5932 100644 }.size == (if (firstAccess) 2 else 0)) assert(collect(initialExecutedPlan) { case i: InMemoryTableScanLike => i -@@ -2958,6 +3001,7 @@ class AdaptiveQueryExecSuite +@@ -2958,6 +3003,7 @@ class AdaptiveQueryExecSuite }.isEmpty) assert(collect(finalExecutedPlan) { case s: SortExec => s @@ -2539,7 +2583,7 @@ index 272be70f9fe..12daa1f5932 100644 }.isEmpty) assert(collect(initialExecutedPlan) { case i: InMemoryTableScanLike => i -@@ -3129,7 +3173,8 @@ class AdaptiveQueryExecSuite +@@ -3129,7 +3175,8 @@ class AdaptiveQueryExecSuite val plan = df.queryExecution.executedPlan.asInstanceOf[AdaptiveSparkPlanExec] assert(plan.inputPlan.isInstanceOf[TakeOrderedAndProjectExec]) @@ -3013,7 +3057,7 @@ index 09ed6955a51..236a4e99824 100644 } test(s"parquet widening conversion $fromType -> $toType") { diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/debug/DebuggingSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/debug/DebuggingSuite.scala -index b8f3ea3c6f3..bbd44221288 100644 +index b8f3ea3c6f3..0d92198bd83 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/debug/DebuggingSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/debug/DebuggingSuite.scala @@ -20,6 +20,7 @@ package org.apache.spark.sql.execution.debug @@ -3024,7 +3068,27 @@ index b8f3ea3c6f3..bbd44221288 100644 import org.apache.spark.sql.catalyst.InternalRow import org.apache.spark.sql.catalyst.expressions.Attribute import org.apache.spark.sql.catalyst.expressions.codegen.CodegenContext -@@ -125,7 +126,8 @@ class DebuggingSuite extends DebuggingSuiteBase with DisableAdaptiveExecutionSui +@@ -42,7 +43,8 @@ abstract class DebuggingSuiteBase extends SharedSparkSession { + testData.as[TestData].debug() + } + +- test("debugCodegen") { ++ test("debugCodegen", ++ IgnoreComet("Comet changes the WholeStageCodegen subtree count")) { + val df = spark.range(10).groupBy(col("id") * 2).count() + df.collect() + val res = codegenString(df.queryExecution.executedPlan) +@@ -51,7 +53,8 @@ abstract class DebuggingSuiteBase extends SharedSparkSession { + assert(res.contains("Object[]")) + } + +- test("debugCodegenStringSeq") { ++ test("debugCodegenStringSeq", ++ IgnoreComet("Comet changes the WholeStageCodegen subtree count")) { + val df = spark.range(10).groupBy(col("id") * 2).count() + df.collect() + val res = codegenStringSeq(df.queryExecution.executedPlan) +@@ -125,7 +128,8 @@ class DebuggingSuite extends DebuggingSuiteBase with DisableAdaptiveExecutionSui | id LongType: {}""".stripMargin)) } @@ -3175,6 +3239,28 @@ index 89f22186f7e..425233f00b2 100644 before { StateStore.stop() require(!StateStore.isMaintenanceRunning) +diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/ui/SQLAppStatusListenerSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/ui/SQLAppStatusListenerSuite.scala +index 800a58f0c1d..22a3c49414d 100644 +--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/ui/SQLAppStatusListenerSuite.scala ++++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/ui/SQLAppStatusListenerSuite.scala +@@ -35,6 +35,7 @@ import org.apache.spark.internal.config.Status._ + import org.apache.spark.rdd.RDD + import org.apache.spark.resource.ResourceProfile + import org.apache.spark.scheduler._ ++import org.apache.spark.sql.IgnoreComet + import org.apache.spark.sql.catalyst.InternalRow + import org.apache.spark.sql.catalyst.expressions.Attribute + import org.apache.spark.sql.catalyst.plans.logical.LocalRelation +@@ -699,7 +700,8 @@ abstract class SQLAppStatusListenerSuite extends SharedSparkSession with JsonTes + } + + test("SPARK-29894 test Codegen Stage Id in SparkPlanInfo", +- DisableAdaptiveExecution("WSCG rule is applied later in AQE")) { ++ DisableAdaptiveExecution("WSCG rule is applied later in AQE"), ++ IgnoreComet("Comet changes the WholeStageCodegen subtree count")) { + // with AQE on, the WholeStageCodegen rule is applied when running QueryStageExec. + val df = createTestDataFrame.select(count("*")) + val sparkPlanInfo = SparkPlanInfo.fromSparkPlan(df.queryExecution.executedPlan) diff --git a/sql/core/src/test/scala/org/apache/spark/sql/sources/BucketedReadSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/sources/BucketedReadSuite.scala index c4b09c4b289..75c3437788e 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/sources/BucketedReadSuite.scala diff --git a/dev/diffs/4.1.2.diff b/dev/diffs/4.1.2.diff index 946dae3949..0e1551bca9 100644 --- a/dev/diffs/4.1.2.diff +++ b/dev/diffs/4.1.2.diff @@ -39,7 +39,7 @@ index 6df8bc85b51..dabb75e2b75 100644 withSpark(sc) { sc => TestUtils.waitUntilExecutorsUp(sc, 2, 60000) diff --git a/pom.xml b/pom.xml -index dc201151999..d5c08f11ded 100644 +index dc201151999..20ee0e7482a 100644 --- a/pom.xml +++ b/pom.xml @@ -152,6 +152,8 @@ @@ -346,6 +346,30 @@ index 21a3ce1e122..f4762ab98f0 100644 SET spark.sql.ansi.enabled = false; -- In COMPENSATION views get invalidated if the type can't cast +diff --git a/sql/core/src/test/scala/org/apache/spark/sql/CTEInlineSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/CTEInlineSuite.scala +index c1a001117be..4545ba2fc20 100644 +--- a/sql/core/src/test/scala/org/apache/spark/sql/CTEInlineSuite.scala ++++ b/sql/core/src/test/scala/org/apache/spark/sql/CTEInlineSuite.scala +@@ -282,7 +282,8 @@ abstract class CTEInlineSuiteBase + } + } + +- test("CTE Predicate push-down and column pruning") { ++ test("CTE Predicate push-down and column pruning", ++ IgnoreComet("Comet changes the exchange reuse count asserted by this test")) { + withTempView("t") { + Seq((0, 1), (1, 2)).toDF("c1", "c2").createOrReplaceTempView("t") + val df = sql( +@@ -332,7 +333,8 @@ abstract class CTEInlineSuiteBase + } + } + +- test("CTE Predicate push-down and column pruning - combined predicate") { ++ test("CTE Predicate push-down and column pruning - combined predicate", ++ IgnoreComet("Comet changes the exchange reuse count asserted by this test")) { + withTempView("t") { + Seq((0, 1, 2, 3), (1, 2, 3, 4)).toDF("c1", "c2", "c3", "c4").createOrReplaceTempView("t") + val df = sql( diff --git a/sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala index 0d807aeae4d..6d7744e771b 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala @@ -2245,7 +2269,7 @@ index a3cfdc5a240..3793b6191bf 100644 }) checkAnswer(distinctWithId, Seq(Row(1, 0), Row(1, 0))) diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala -index 3e7d26f74bd..7e70e72fa3e 100644 +index 3e7d26f74bd..952112cfa9a 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala @@ -27,12 +27,14 @@ import org.apache.spark.SparkException @@ -2523,7 +2547,27 @@ index 3e7d26f74bd..7e70e72fa3e 100644 withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true") { val (_, adaptivePlan) = runAdaptiveAndVerifyResult( "SELECT key FROM testData GROUP BY key") -@@ -1891,7 +1921,7 @@ class AdaptiveQueryExecSuite +@@ -1833,7 +1863,8 @@ class AdaptiveQueryExecSuite + } + } + +- test("SPARK-35442: Support propagate empty relation through aggregate") { ++ test("SPARK-35442: Support propagate empty relation through aggregate", ++ IgnoreComet("https://github.com/apache/datafusion-comet/issues/4412")) { + withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true") { + val (plan1, adaptivePlan1) = runAdaptiveAndVerifyResult( + "SELECT key, count(*) FROM testData WHERE value = 'no_match' GROUP BY key") +@@ -1852,7 +1883,8 @@ class AdaptiveQueryExecSuite + } + } + +- test("SPARK-35442: Support propagate empty relation through union") { ++ test("SPARK-35442: Support propagate empty relation through union", ++ IgnoreComet("https://github.com/apache/datafusion-comet/issues/4412")) { + def checkNumUnion(plan: SparkPlan, numUnion: Int): Unit = { + assert( + collect(plan) { +@@ -1891,7 +1923,7 @@ class AdaptiveQueryExecSuite val (_, adaptivePlan) = runAdaptiveAndVerifyResult( "SELECT id FROM v1 GROUP BY id DISTRIBUTE BY id") assert(collect(adaptivePlan) { @@ -2532,7 +2576,7 @@ index 3e7d26f74bd..7e70e72fa3e 100644 }.length == 1) } } -@@ -1972,7 +2002,8 @@ class AdaptiveQueryExecSuite +@@ -1972,7 +2004,8 @@ class AdaptiveQueryExecSuite } } @@ -2542,7 +2586,7 @@ index 3e7d26f74bd..7e70e72fa3e 100644 def hasRepartitionShuffle(plan: SparkPlan): Boolean = { find(plan) { case s: ShuffleExchangeLike => -@@ -2159,6 +2190,9 @@ class AdaptiveQueryExecSuite +@@ -2159,6 +2192,9 @@ class AdaptiveQueryExecSuite def checkNoCoalescePartitions(ds: Dataset[Row], origin: ShuffleOrigin): Unit = { assert(collect(ds.queryExecution.executedPlan) { case s: ShuffleExchangeExec if s.shuffleOrigin == origin && s.numPartitions == 2 => s @@ -2552,7 +2596,7 @@ index 3e7d26f74bd..7e70e72fa3e 100644 }.size == 1) ds.collect() val plan = ds.queryExecution.executedPlan -@@ -2167,6 +2201,9 @@ class AdaptiveQueryExecSuite +@@ -2167,6 +2203,9 @@ class AdaptiveQueryExecSuite }.isEmpty) assert(collect(plan) { case s: ShuffleExchangeExec if s.shuffleOrigin == origin && s.numPartitions == 2 => s @@ -2562,7 +2606,7 @@ index 3e7d26f74bd..7e70e72fa3e 100644 }.size == 1) checkAnswer(ds, testData) } -@@ -2331,7 +2368,8 @@ class AdaptiveQueryExecSuite +@@ -2331,7 +2370,8 @@ class AdaptiveQueryExecSuite } } @@ -2572,7 +2616,7 @@ index 3e7d26f74bd..7e70e72fa3e 100644 withTempView("t1", "t2") { def checkJoinStrategy(shouldShuffleHashJoin: Boolean): Unit = { Seq("100", "100000").foreach { size => -@@ -2417,7 +2455,8 @@ class AdaptiveQueryExecSuite +@@ -2417,7 +2457,8 @@ class AdaptiveQueryExecSuite } } @@ -2582,7 +2626,7 @@ index 3e7d26f74bd..7e70e72fa3e 100644 withTempView("v") { withSQLConf( SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", -@@ -2516,7 +2555,7 @@ class AdaptiveQueryExecSuite +@@ -2516,7 +2557,7 @@ class AdaptiveQueryExecSuite runAdaptiveAndVerifyResult(s"SELECT $repartition key1 FROM skewData1 " + s"JOIN skewData2 ON key1 = key2 GROUP BY key1") val shuffles1 = collect(adaptive1) { @@ -2591,7 +2635,7 @@ index 3e7d26f74bd..7e70e72fa3e 100644 } assert(shuffles1.size == 3) // shuffles1.head is the top-level shuffle under the Aggregate operator -@@ -2529,7 +2568,7 @@ class AdaptiveQueryExecSuite +@@ -2529,7 +2570,7 @@ class AdaptiveQueryExecSuite runAdaptiveAndVerifyResult(s"SELECT $repartition key1 FROM skewData1 " + s"JOIN skewData2 ON key1 = key2") val shuffles2 = collect(adaptive2) { @@ -2600,7 +2644,7 @@ index 3e7d26f74bd..7e70e72fa3e 100644 } if (hasRequiredDistribution) { assert(shuffles2.size == 3) -@@ -2563,7 +2602,8 @@ class AdaptiveQueryExecSuite +@@ -2563,7 +2604,8 @@ class AdaptiveQueryExecSuite } } @@ -2610,7 +2654,7 @@ index 3e7d26f74bd..7e70e72fa3e 100644 CostEvaluator.instantiate( classOf[SimpleShuffleSortCostEvaluator].getCanonicalName, spark.sparkContext.getConf) intercept[IllegalArgumentException] { -@@ -2729,6 +2769,7 @@ class AdaptiveQueryExecSuite +@@ -2729,6 +2771,7 @@ class AdaptiveQueryExecSuite val (_, adaptive) = runAdaptiveAndVerifyResult(query) assert(adaptive.collect { case sort: SortExec => sort @@ -2618,7 +2662,7 @@ index 3e7d26f74bd..7e70e72fa3e 100644 }.size == 1) val read = collect(adaptive) { case read: AQEShuffleReadExec => read -@@ -2746,7 +2787,8 @@ class AdaptiveQueryExecSuite +@@ -2746,7 +2789,8 @@ class AdaptiveQueryExecSuite } } @@ -2628,7 +2672,7 @@ index 3e7d26f74bd..7e70e72fa3e 100644 withTempView("v") { withSQLConf( SQLConf.ADAPTIVE_OPTIMIZE_SKEWS_IN_REBALANCE_PARTITIONS_ENABLED.key -> "true", -@@ -2858,7 +2900,7 @@ class AdaptiveQueryExecSuite +@@ -2858,7 +2902,7 @@ class AdaptiveQueryExecSuite runAdaptiveAndVerifyResult("SELECT key1 FROM skewData1 JOIN skewData2 ON key1 = key2 " + "JOIN skewData3 ON value2 = value3") val shuffles1 = collect(adaptive1) { @@ -2637,7 +2681,7 @@ index 3e7d26f74bd..7e70e72fa3e 100644 } assert(shuffles1.size == 4) val smj1 = findTopLevelSortMergeJoin(adaptive1) -@@ -2869,7 +2911,7 @@ class AdaptiveQueryExecSuite +@@ -2869,7 +2913,7 @@ class AdaptiveQueryExecSuite runAdaptiveAndVerifyResult("SELECT key1 FROM skewData1 JOIN skewData2 ON key1 = key2 " + "JOIN skewData3 ON value1 = value3") val shuffles2 = collect(adaptive2) { @@ -2646,7 +2690,7 @@ index 3e7d26f74bd..7e70e72fa3e 100644 } assert(shuffles2.size == 4) val smj2 = findTopLevelSortMergeJoin(adaptive2) -@@ -3127,6 +3169,7 @@ class AdaptiveQueryExecSuite +@@ -3127,6 +3171,7 @@ class AdaptiveQueryExecSuite }.size == (if (firstAccess) 1 else 0)) assert(collect(initialExecutedPlan) { case s: SortExec => s @@ -2654,7 +2698,7 @@ index 3e7d26f74bd..7e70e72fa3e 100644 }.size == (if (firstAccess) 2 else 0)) assert(collect(initialExecutedPlan) { case i: InMemoryTableScanLike => i -@@ -3139,6 +3182,7 @@ class AdaptiveQueryExecSuite +@@ -3139,6 +3184,7 @@ class AdaptiveQueryExecSuite }.isEmpty) assert(collect(finalExecutedPlan) { case s: SortExec => s @@ -2662,7 +2706,7 @@ index 3e7d26f74bd..7e70e72fa3e 100644 }.isEmpty) assert(collect(initialExecutedPlan) { case i: InMemoryTableScanLike => i -@@ -3310,7 +3354,8 @@ class AdaptiveQueryExecSuite +@@ -3310,7 +3356,8 @@ class AdaptiveQueryExecSuite val plan = df.queryExecution.executedPlan.asInstanceOf[AdaptiveSparkPlanExec] assert(plan.inputPlan.isInstanceOf[TakeOrderedAndProjectExec]) @@ -3202,7 +3246,7 @@ index 1cc6d3afbee..8275727fbb4 100644 private def testWithTempDir(name: String)(block: File => Unit): Unit = test(name) { withTempDir { dir => diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/debug/DebuggingSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/debug/DebuggingSuite.scala -index b8f3ea3c6f3..bbd44221288 100644 +index b8f3ea3c6f3..0d92198bd83 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/debug/DebuggingSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/debug/DebuggingSuite.scala @@ -20,6 +20,7 @@ package org.apache.spark.sql.execution.debug @@ -3213,7 +3257,27 @@ index b8f3ea3c6f3..bbd44221288 100644 import org.apache.spark.sql.catalyst.InternalRow import org.apache.spark.sql.catalyst.expressions.Attribute import org.apache.spark.sql.catalyst.expressions.codegen.CodegenContext -@@ -125,7 +126,8 @@ class DebuggingSuite extends DebuggingSuiteBase with DisableAdaptiveExecutionSui +@@ -42,7 +43,8 @@ abstract class DebuggingSuiteBase extends SharedSparkSession { + testData.as[TestData].debug() + } + +- test("debugCodegen") { ++ test("debugCodegen", ++ IgnoreComet("Comet changes the WholeStageCodegen subtree count")) { + val df = spark.range(10).groupBy(col("id") * 2).count() + df.collect() + val res = codegenString(df.queryExecution.executedPlan) +@@ -51,7 +53,8 @@ abstract class DebuggingSuiteBase extends SharedSparkSession { + assert(res.contains("Object[]")) + } + +- test("debugCodegenStringSeq") { ++ test("debugCodegenStringSeq", ++ IgnoreComet("Comet changes the WholeStageCodegen subtree count")) { + val df = spark.range(10).groupBy(col("id") * 2).count() + df.collect() + val res = codegenStringSeq(df.queryExecution.executedPlan) +@@ -125,7 +128,8 @@ class DebuggingSuite extends DebuggingSuiteBase with DisableAdaptiveExecutionSui | id LongType: {}""".stripMargin)) } @@ -3385,6 +3449,28 @@ index 232332a6575..324afe9ebb7 100644 before { StateStore.stop() require(!StateStore.isMaintenanceRunning) +diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/ui/SQLAppStatusListenerSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/ui/SQLAppStatusListenerSuite.scala +index 800a58f0c1d..22a3c49414d 100644 +--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/ui/SQLAppStatusListenerSuite.scala ++++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/ui/SQLAppStatusListenerSuite.scala +@@ -35,6 +35,7 @@ import org.apache.spark.internal.config.Status._ + import org.apache.spark.rdd.RDD + import org.apache.spark.resource.ResourceProfile + import org.apache.spark.scheduler._ ++import org.apache.spark.sql.IgnoreComet + import org.apache.spark.sql.catalyst.InternalRow + import org.apache.spark.sql.catalyst.expressions.Attribute + import org.apache.spark.sql.catalyst.plans.logical.LocalRelation +@@ -699,7 +700,8 @@ abstract class SQLAppStatusListenerSuite extends SharedSparkSession with JsonTes + } + + test("SPARK-29894 test Codegen Stage Id in SparkPlanInfo", +- DisableAdaptiveExecution("WSCG rule is applied later in AQE")) { ++ DisableAdaptiveExecution("WSCG rule is applied later in AQE"), ++ IgnoreComet("Comet changes the WholeStageCodegen subtree count")) { + // with AQE on, the WholeStageCodegen rule is applied when running QueryStageExec. + val df = createTestDataFrame.select(count("*")) + val sparkPlanInfo = SparkPlanInfo.fromSparkPlan(df.queryExecution.executedPlan) diff --git a/sql/core/src/test/scala/org/apache/spark/sql/sources/BucketedReadSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/sources/BucketedReadSuite.scala index 83ebd24384c..32511091bb2 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/sources/BucketedReadSuite.scala diff --git a/spark/src/main/scala/org/apache/comet/serde/aggregates.scala b/spark/src/main/scala/org/apache/comet/serde/aggregates.scala index 6c6544e93c..53eb960b02 100644 --- a/spark/src/main/scala/org/apache/comet/serde/aggregates.scala +++ b/spark/src/main/scala/org/apache/comet/serde/aggregates.scala @@ -107,6 +107,9 @@ object CometMax extends CometAggregateExpressionSerde[Max] { } object CometCount extends CometAggregateExpressionSerde[Count] { + + override def supportsMixedPartialFinal(fn: Count): Boolean = true + override def convert( aggExpr: AggregateExpression, expr: Count, 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 fcb5e0003d..309ce2c012 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,61 +1,62 @@ -TakeOrderedAndProject -+- 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 - +- BroadcastHashJoin - :- Project - : +- BroadcastHashJoin - : :- Project - : : +- Filter - : : +- BroadcastHashJoin - : : :- BroadcastHashJoin [COMET: Unsupported join type ExistenceJoin(exists#1)] - : : : :- CometNativeColumnarToRow - : : : : +- CometBroadcastHashJoin - : : : : :- CometFilter - : : : : : +- CometNativeScan parquet spark_catalog.default.customer - : : : : +- CometBroadcastExchange - : : : : +- CometProject - : : : : +- CometBroadcastHashJoin - : : : : :- CometNativeScan parquet spark_catalog.default.store_sales - : : : : : +- CometSubqueryBroadcast - : : : : : +- CometBroadcastExchange - : : : : : +- CometProject - : : : : : +- CometFilter - : : : : : +- CometNativeScan parquet spark_catalog.default.date_dim - : : : : +- CometBroadcastExchange - : : : : +- CometProject - : : : : +- CometFilter - : : : : +- CometNativeScan parquet spark_catalog.default.date_dim - : : : +- BroadcastExchange - : : : +- CometNativeColumnarToRow - : : : +- CometProject - : : : +- CometBroadcastHashJoin - : : : :- CometNativeScan parquet spark_catalog.default.web_sales - : : : : +- ReusedSubquery - : : : +- CometBroadcastExchange - : : : +- CometProject - : : : +- CometFilter - : : : +- CometNativeScan parquet spark_catalog.default.date_dim - : : +- BroadcastExchange - : : +- CometNativeColumnarToRow - : : +- CometProject - : : +- CometBroadcastHashJoin - : : :- CometNativeScan parquet spark_catalog.default.catalog_sales - : : : +- ReusedSubquery - : : +- CometBroadcastExchange - : : +- CometProject - : : +- CometFilter - : : +- CometNativeScan parquet spark_catalog.default.date_dim - : +- BroadcastExchange - : +- CometNativeColumnarToRow - : +- CometProject - : +- CometFilter - : +- CometNativeScan parquet spark_catalog.default.customer_address - +- BroadcastExchange - +- CometNativeColumnarToRow - +- CometProject - +- CometFilter - +- CometNativeScan parquet spark_catalog.default.customer_demographics +CometNativeColumnarToRow ++- CometTakeOrderedAndProject + +- CometHashAggregate + +- CometColumnarExchange + +- HashAggregate + +- Project + +- BroadcastHashJoin + :- Project + : +- BroadcastHashJoin + : :- Project + : : +- Filter + : : +- BroadcastHashJoin + : : :- BroadcastHashJoin [COMET: Unsupported join type ExistenceJoin(exists#1)] + : : : :- CometNativeColumnarToRow + : : : : +- CometBroadcastHashJoin + : : : : :- CometFilter + : : : : : +- CometNativeScan parquet spark_catalog.default.customer + : : : : +- CometBroadcastExchange + : : : : +- CometProject + : : : : +- CometBroadcastHashJoin + : : : : :- CometNativeScan parquet spark_catalog.default.store_sales + : : : : : +- CometSubqueryBroadcast + : : : : : +- CometBroadcastExchange + : : : : : +- CometProject + : : : : : +- CometFilter + : : : : : +- CometNativeScan parquet spark_catalog.default.date_dim + : : : : +- CometBroadcastExchange + : : : : +- CometProject + : : : : +- CometFilter + : : : : +- CometNativeScan parquet spark_catalog.default.date_dim + : : : +- BroadcastExchange + : : : +- CometNativeColumnarToRow + : : : +- CometProject + : : : +- CometBroadcastHashJoin + : : : :- CometNativeScan parquet spark_catalog.default.web_sales + : : : : +- ReusedSubquery + : : : +- CometBroadcastExchange + : : : +- CometProject + : : : +- CometFilter + : : : +- CometNativeScan parquet spark_catalog.default.date_dim + : : +- BroadcastExchange + : : +- CometNativeColumnarToRow + : : +- CometProject + : : +- CometBroadcastHashJoin + : : :- CometNativeScan parquet spark_catalog.default.catalog_sales + : : : +- ReusedSubquery + : : +- CometBroadcastExchange + : : +- CometProject + : : +- CometFilter + : : +- CometNativeScan parquet spark_catalog.default.date_dim + : +- BroadcastExchange + : +- CometNativeColumnarToRow + : +- CometProject + : +- CometFilter + : +- CometNativeScan parquet spark_catalog.default.customer_address + +- BroadcastExchange + +- CometNativeColumnarToRow + +- CometProject + +- CometFilter + +- CometNativeScan parquet spark_catalog.default.customer_demographics -Comet accelerated 35 out of 54 eligible operators (64%). Final plan contains 5 transitions between Spark and Comet. \ No newline at end of file +Comet accelerated 38 out of 54 eligible operators (70%). Final plan contains 6 transitions between Spark and Comet. \ No newline at end of file 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 fcb5e0003d..309ce2c012 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,61 +1,62 @@ -TakeOrderedAndProject -+- 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 - +- BroadcastHashJoin - :- Project - : +- BroadcastHashJoin - : :- Project - : : +- Filter - : : +- BroadcastHashJoin - : : :- BroadcastHashJoin [COMET: Unsupported join type ExistenceJoin(exists#1)] - : : : :- CometNativeColumnarToRow - : : : : +- CometBroadcastHashJoin - : : : : :- CometFilter - : : : : : +- CometNativeScan parquet spark_catalog.default.customer - : : : : +- CometBroadcastExchange - : : : : +- CometProject - : : : : +- CometBroadcastHashJoin - : : : : :- CometNativeScan parquet spark_catalog.default.store_sales - : : : : : +- CometSubqueryBroadcast - : : : : : +- CometBroadcastExchange - : : : : : +- CometProject - : : : : : +- CometFilter - : : : : : +- CometNativeScan parquet spark_catalog.default.date_dim - : : : : +- CometBroadcastExchange - : : : : +- CometProject - : : : : +- CometFilter - : : : : +- CometNativeScan parquet spark_catalog.default.date_dim - : : : +- BroadcastExchange - : : : +- CometNativeColumnarToRow - : : : +- CometProject - : : : +- CometBroadcastHashJoin - : : : :- CometNativeScan parquet spark_catalog.default.web_sales - : : : : +- ReusedSubquery - : : : +- CometBroadcastExchange - : : : +- CometProject - : : : +- CometFilter - : : : +- CometNativeScan parquet spark_catalog.default.date_dim - : : +- BroadcastExchange - : : +- CometNativeColumnarToRow - : : +- CometProject - : : +- CometBroadcastHashJoin - : : :- CometNativeScan parquet spark_catalog.default.catalog_sales - : : : +- ReusedSubquery - : : +- CometBroadcastExchange - : : +- CometProject - : : +- CometFilter - : : +- CometNativeScan parquet spark_catalog.default.date_dim - : +- BroadcastExchange - : +- CometNativeColumnarToRow - : +- CometProject - : +- CometFilter - : +- CometNativeScan parquet spark_catalog.default.customer_address - +- BroadcastExchange - +- CometNativeColumnarToRow - +- CometProject - +- CometFilter - +- CometNativeScan parquet spark_catalog.default.customer_demographics +CometNativeColumnarToRow ++- CometTakeOrderedAndProject + +- CometHashAggregate + +- CometColumnarExchange + +- HashAggregate + +- Project + +- BroadcastHashJoin + :- Project + : +- BroadcastHashJoin + : :- Project + : : +- Filter + : : +- BroadcastHashJoin + : : :- BroadcastHashJoin [COMET: Unsupported join type ExistenceJoin(exists#1)] + : : : :- CometNativeColumnarToRow + : : : : +- CometBroadcastHashJoin + : : : : :- CometFilter + : : : : : +- CometNativeScan parquet spark_catalog.default.customer + : : : : +- CometBroadcastExchange + : : : : +- CometProject + : : : : +- CometBroadcastHashJoin + : : : : :- CometNativeScan parquet spark_catalog.default.store_sales + : : : : : +- CometSubqueryBroadcast + : : : : : +- CometBroadcastExchange + : : : : : +- CometProject + : : : : : +- CometFilter + : : : : : +- CometNativeScan parquet spark_catalog.default.date_dim + : : : : +- CometBroadcastExchange + : : : : +- CometProject + : : : : +- CometFilter + : : : : +- CometNativeScan parquet spark_catalog.default.date_dim + : : : +- BroadcastExchange + : : : +- CometNativeColumnarToRow + : : : +- CometProject + : : : +- CometBroadcastHashJoin + : : : :- CometNativeScan parquet spark_catalog.default.web_sales + : : : : +- ReusedSubquery + : : : +- CometBroadcastExchange + : : : +- CometProject + : : : +- CometFilter + : : : +- CometNativeScan parquet spark_catalog.default.date_dim + : : +- BroadcastExchange + : : +- CometNativeColumnarToRow + : : +- CometProject + : : +- CometBroadcastHashJoin + : : :- CometNativeScan parquet spark_catalog.default.catalog_sales + : : : +- ReusedSubquery + : : +- CometBroadcastExchange + : : +- CometProject + : : +- CometFilter + : : +- CometNativeScan parquet spark_catalog.default.date_dim + : +- BroadcastExchange + : +- CometNativeColumnarToRow + : +- CometProject + : +- CometFilter + : +- CometNativeScan parquet spark_catalog.default.customer_address + +- BroadcastExchange + +- CometNativeColumnarToRow + +- CometProject + +- CometFilter + +- CometNativeScan parquet spark_catalog.default.customer_demographics -Comet accelerated 35 out of 54 eligible operators (64%). Final plan contains 5 transitions between Spark and Comet. \ No newline at end of file +Comet accelerated 38 out of 54 eligible operators (70%). Final plan contains 6 transitions between Spark and Comet. \ No newline at end of file diff --git a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q44/extended.txt b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q44/extended.txt index b9a5a6e3eb..7c606a1466 100644 --- a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q44/extended.txt +++ b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q44/extended.txt @@ -34,14 +34,7 @@ CometNativeColumnarToRow : : +- CometSort : : +- CometExchange : : +- CometFilter - : : : +- Subquery - : : : +- CometNativeColumnarToRow - : : : +- CometHashAggregate - : : : +- CometExchange - : : : +- CometHashAggregate - : : : +- CometProject - : : : +- CometFilter - : : : +- CometNativeScan parquet spark_catalog.default.store_sales + : : : +- ReusedSubquery : : +- CometHashAggregate : : +- CometExchange : : +- CometHashAggregate @@ -57,4 +50,4 @@ CometNativeColumnarToRow +- CometFilter +- CometNativeScan parquet spark_catalog.default.item -Comet accelerated 53 out of 55 eligible operators (96%). Final plan contains 3 transitions between Spark and Comet. \ No newline at end of file +Comet accelerated 47 out of 49 eligible operators (95%). Final plan contains 2 transitions between Spark and Comet. \ No newline at end of file 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 fcb5e0003d..309ce2c012 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,61 +1,62 @@ -TakeOrderedAndProject -+- 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 - +- BroadcastHashJoin - :- Project - : +- BroadcastHashJoin - : :- Project - : : +- Filter - : : +- BroadcastHashJoin - : : :- BroadcastHashJoin [COMET: Unsupported join type ExistenceJoin(exists#1)] - : : : :- CometNativeColumnarToRow - : : : : +- CometBroadcastHashJoin - : : : : :- CometFilter - : : : : : +- CometNativeScan parquet spark_catalog.default.customer - : : : : +- CometBroadcastExchange - : : : : +- CometProject - : : : : +- CometBroadcastHashJoin - : : : : :- CometNativeScan parquet spark_catalog.default.store_sales - : : : : : +- CometSubqueryBroadcast - : : : : : +- CometBroadcastExchange - : : : : : +- CometProject - : : : : : +- CometFilter - : : : : : +- CometNativeScan parquet spark_catalog.default.date_dim - : : : : +- CometBroadcastExchange - : : : : +- CometProject - : : : : +- CometFilter - : : : : +- CometNativeScan parquet spark_catalog.default.date_dim - : : : +- BroadcastExchange - : : : +- CometNativeColumnarToRow - : : : +- CometProject - : : : +- CometBroadcastHashJoin - : : : :- CometNativeScan parquet spark_catalog.default.web_sales - : : : : +- ReusedSubquery - : : : +- CometBroadcastExchange - : : : +- CometProject - : : : +- CometFilter - : : : +- CometNativeScan parquet spark_catalog.default.date_dim - : : +- BroadcastExchange - : : +- CometNativeColumnarToRow - : : +- CometProject - : : +- CometBroadcastHashJoin - : : :- CometNativeScan parquet spark_catalog.default.catalog_sales - : : : +- ReusedSubquery - : : +- CometBroadcastExchange - : : +- CometProject - : : +- CometFilter - : : +- CometNativeScan parquet spark_catalog.default.date_dim - : +- BroadcastExchange - : +- CometNativeColumnarToRow - : +- CometProject - : +- CometFilter - : +- CometNativeScan parquet spark_catalog.default.customer_address - +- BroadcastExchange - +- CometNativeColumnarToRow - +- CometProject - +- CometFilter - +- CometNativeScan parquet spark_catalog.default.customer_demographics +CometNativeColumnarToRow ++- CometTakeOrderedAndProject + +- CometHashAggregate + +- CometColumnarExchange + +- HashAggregate + +- Project + +- BroadcastHashJoin + :- Project + : +- BroadcastHashJoin + : :- Project + : : +- Filter + : : +- BroadcastHashJoin + : : :- BroadcastHashJoin [COMET: Unsupported join type ExistenceJoin(exists#1)] + : : : :- CometNativeColumnarToRow + : : : : +- CometBroadcastHashJoin + : : : : :- CometFilter + : : : : : +- CometNativeScan parquet spark_catalog.default.customer + : : : : +- CometBroadcastExchange + : : : : +- CometProject + : : : : +- CometBroadcastHashJoin + : : : : :- CometNativeScan parquet spark_catalog.default.store_sales + : : : : : +- CometSubqueryBroadcast + : : : : : +- CometBroadcastExchange + : : : : : +- CometProject + : : : : : +- CometFilter + : : : : : +- CometNativeScan parquet spark_catalog.default.date_dim + : : : : +- CometBroadcastExchange + : : : : +- CometProject + : : : : +- CometFilter + : : : : +- CometNativeScan parquet spark_catalog.default.date_dim + : : : +- BroadcastExchange + : : : +- CometNativeColumnarToRow + : : : +- CometProject + : : : +- CometBroadcastHashJoin + : : : :- CometNativeScan parquet spark_catalog.default.web_sales + : : : : +- ReusedSubquery + : : : +- CometBroadcastExchange + : : : +- CometProject + : : : +- CometFilter + : : : +- CometNativeScan parquet spark_catalog.default.date_dim + : : +- BroadcastExchange + : : +- CometNativeColumnarToRow + : : +- CometProject + : : +- CometBroadcastHashJoin + : : :- CometNativeScan parquet spark_catalog.default.catalog_sales + : : : +- ReusedSubquery + : : +- CometBroadcastExchange + : : +- CometProject + : : +- CometFilter + : : +- CometNativeScan parquet spark_catalog.default.date_dim + : +- BroadcastExchange + : +- CometNativeColumnarToRow + : +- CometProject + : +- CometFilter + : +- CometNativeScan parquet spark_catalog.default.customer_address + +- BroadcastExchange + +- CometNativeColumnarToRow + +- CometProject + +- CometFilter + +- CometNativeScan parquet spark_catalog.default.customer_demographics -Comet accelerated 35 out of 54 eligible operators (64%). Final plan contains 5 transitions between Spark and Comet. \ No newline at end of file +Comet accelerated 38 out of 54 eligible operators (70%). Final plan contains 6 transitions between Spark and Comet. \ No newline at end of file diff --git a/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala index 7e5eca4e99..fd0d7f25e6 100644 --- a/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala +++ b/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala @@ -2109,6 +2109,62 @@ class CometAggregateSuite extends CometTestBase with AdaptiveSparkPlanHelper { } } + // https://github.com/apache/datafusion-comet/issues/4242 + // AQE's PropagateEmptyRelationAfterAQE rule pattern-matches BaseAggregateExec only, not + // CometHashAggregateExec. With mixed Spark-Partial / Comet-Final COUNT, the Final escapes + // the rule and propagation of empty intermediate results stops, changing downstream results + // in some queries. + test("issue #4242: AQE PropagateEmptyRelation with mixed Spark Partial / Comet Final COUNT") { + withSQLConf( + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", + SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") { + withTempDir { dir => + spark.range(10).toDF("a").write.parquet(dir.getCanonicalPath + "/t1") + spark.range(10).toDF("b").write.parquet(dir.getCanonicalPath + "/t2") + withTempView("t1", "t2") { + spark.read.parquet(dir.getCanonicalPath + "/t1").createOrReplaceTempView("t1") + spark.read.parquet(dir.getCanonicalPath + "/t2").createOrReplaceTempView("t2") + + // Empty left side joined with empty right side then grouped count. + val q1 = + """SELECT inner_a, COUNT(*) FROM ( + | SELECT t1.a AS inner_a FROM t1 LEFT OUTER JOIN t2 ON t1.a = t2.b + | WHERE t1.a > 100 AND t2.b > 100 + |) GROUP BY inner_a""".stripMargin + checkSparkAnswer(q1) + + // Global ungrouped count over an empty input should produce one row with COUNT = 0. + val q2 = "SELECT COUNT(*) FROM t1 WHERE a > 100" + checkSparkAnswer(q2) + } + } + } + } + + // https://github.com/apache/datafusion-comet/issues/4242 + // Mirrors the OR pattern in Spark's in-count-bug.sql. Decorrelating correlated IN with + // COUNT inside an OR is known to drop a row when partial/final aggregate stages are + // split between Spark and Comet. + test("issue #4242: count-bug decorrelation with correlated IN OR pattern") { + withTempView("t1", "t2") { + sql("CREATE TEMPORARY VIEW t1(c1, c2) AS VALUES (0, 1), (1, 2)") + sql("CREATE TEMPORARY VIEW t2(c1, c2) AS VALUES (0, 2), (0, 3)") + + val orQuery = + """SELECT * FROM t1 WHERE + | c1 IN (SELECT count(*) + 1 FROM t2 WHERE t2.c1 = t1.c1) OR + | c2 IN (SELECT count(*) - 1 FROM t2 WHERE t2.c1 = t1.c1)""".stripMargin + checkSparkAnswer(orQuery) + + val orAndNotInQuery = + """SELECT * FROM t1 WHERE + | (c1 IN (SELECT count(*) + 1 FROM t2 WHERE t2.c1 = t1.c1) OR + | c2 IN (SELECT count(*) - 1 FROM t2 WHERE t2.c1 = t1.c1)) AND + | c1 NOT IN (SELECT count(*) FROM t2 WHERE t2.c1 = t1.c2)""".stripMargin + checkSparkAnswer(orAndNotInQuery) + } + } + protected def checkSparkAnswerAndNumOfAggregates(query: String, numAggregates: Int): Unit = { val df = sql(query) checkSparkAnswer(df) From 153599520fa9199fbd6dc7f48a3505839e030b1e Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Thu, 9 Jul 2026 08:25:36 -0600 Subject: [PATCH 8/8] fix: remove redundant reference in fs-hdfs info! argument to satisfy clippy --- native/fs-hdfs/src/hdfs.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/native/fs-hdfs/src/hdfs.rs b/native/fs-hdfs/src/hdfs.rs index d0f4d63e1d..ba77f5bb82 100644 --- a/native/fs-hdfs/src/hdfs.rs +++ b/native/fs-hdfs/src/hdfs.rs @@ -100,7 +100,7 @@ impl HdfsManager { let hdfs_builder = hdfsNewBuilder(); let cstr_uri = CString::new(namenode_uri.as_bytes()).unwrap(); hdfsBuilderSetNameNode(hdfs_builder, cstr_uri.as_ptr()); - info!("Connecting to Namenode ({})", &namenode_uri); + info!("Connecting to Namenode ({})", namenode_uri); hdfsBuilderConnect(hdfs_builder) };