diff --git a/dev/diffs/3.4.3.diff b/dev/diffs/3.4.3.diff index 487c5d1482..c15eb2835c 100644 --- a/dev/diffs/3.4.3.diff +++ b/dev/diffs/3.4.3.diff @@ -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 @@ -1545,7 +1569,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..b327d84d5cc 100644 +index 593bd7bb4ba..3ec90d8a44c 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._ @@ -1813,7 +1837,27 @@ index 593bd7bb4ba..b327d84d5cc 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) { @@ -1822,7 +1866,7 @@ index 593bd7bb4ba..b327d84d5cc 100644 }.length == 1) } } -@@ -1679,7 +1708,8 @@ class AdaptiveQueryExecSuite +@@ -1679,7 +1710,8 @@ class AdaptiveQueryExecSuite } } @@ -1832,7 +1876,7 @@ index 593bd7bb4ba..b327d84d5cc 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 @@ -1842,7 +1886,7 @@ index 593bd7bb4ba..b327d84d5cc 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 @@ -1852,7 +1896,7 @@ index 593bd7bb4ba..b327d84d5cc 100644 }.size == 1) checkAnswer(ds, testData) } -@@ -2028,7 +2064,8 @@ class AdaptiveQueryExecSuite +@@ -2028,7 +2066,8 @@ class AdaptiveQueryExecSuite } } @@ -1862,7 +1906,7 @@ index 593bd7bb4ba..b327d84d5cc 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 } } @@ -1872,7 +1916,7 @@ index 593bd7bb4ba..b327d84d5cc 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) { @@ -1881,7 +1925,7 @@ index 593bd7bb4ba..b327d84d5cc 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) { @@ -1890,7 +1934,7 @@ index 593bd7bb4ba..b327d84d5cc 100644 } if (hasRequiredDistribution) { assert(shuffles2.size == 3) -@@ -2260,7 +2298,8 @@ class AdaptiveQueryExecSuite +@@ -2260,7 +2300,8 @@ class AdaptiveQueryExecSuite } } @@ -1900,7 +1944,7 @@ index 593bd7bb4ba..b327d84d5cc 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 @@ -1908,7 +1952,7 @@ index 593bd7bb4ba..b327d84d5cc 100644 }.size == 1) val read = collect(adaptive) { case read: AQEShuffleReadExec => read -@@ -2421,7 +2461,8 @@ class AdaptiveQueryExecSuite +@@ -2421,7 +2463,8 @@ class AdaptiveQueryExecSuite } } @@ -1918,7 +1962,7 @@ index 593bd7bb4ba..b327d84d5cc 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) { @@ -1927,7 +1971,7 @@ index 593bd7bb4ba..b327d84d5cc 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) { @@ -1936,7 +1980,7 @@ index 593bd7bb4ba..b327d84d5cc 100644 } assert(shuffles2.size == 4) val smj2 = findTopLevelSortMergeJoin(adaptive2) -@@ -2703,7 +2744,8 @@ class AdaptiveQueryExecSuite +@@ -2703,7 +2746,8 @@ class AdaptiveQueryExecSuite } } @@ -2408,7 +2452,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 @@ -2419,7 +2463,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)) } @@ -2529,6 +2593,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 fe87568b96..fd46d9162c 100644 --- a/dev/diffs/3.5.8.diff +++ b/dev/diffs/3.5.8.diff @@ -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 @@ -1511,7 +1535,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..7849c685b19 100644 +index 2f8e401e743..64ad87b578d 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._ @@ -1786,7 +1810,27 @@ index 2f8e401e743..7849c685b19 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) { @@ -1795,7 +1839,7 @@ index 2f8e401e743..7849c685b19 100644 }.length == 1) } } -@@ -1705,7 +1735,8 @@ class AdaptiveQueryExecSuite +@@ -1705,7 +1737,8 @@ class AdaptiveQueryExecSuite } } @@ -1805,7 +1849,7 @@ index 2f8e401e743..7849c685b19 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 @@ -1815,7 +1859,7 @@ index 2f8e401e743..7849c685b19 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 @@ -1825,7 +1869,7 @@ index 2f8e401e743..7849c685b19 100644 }.size == 1) checkAnswer(ds, testData) } -@@ -2054,7 +2091,8 @@ class AdaptiveQueryExecSuite +@@ -2054,7 +2093,8 @@ class AdaptiveQueryExecSuite } } @@ -1835,7 +1879,7 @@ index 2f8e401e743..7849c685b19 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 } } @@ -1845,7 +1889,7 @@ index 2f8e401e743..7849c685b19 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) { @@ -1854,7 +1898,7 @@ index 2f8e401e743..7849c685b19 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) { @@ -1863,7 +1907,7 @@ index 2f8e401e743..7849c685b19 100644 } if (hasRequiredDistribution) { assert(shuffles2.size == 3) -@@ -2286,7 +2325,8 @@ class AdaptiveQueryExecSuite +@@ -2286,7 +2327,8 @@ class AdaptiveQueryExecSuite } } @@ -1873,7 +1917,7 @@ index 2f8e401e743..7849c685b19 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 @@ -1881,7 +1925,7 @@ index 2f8e401e743..7849c685b19 100644 }.size == 1) val read = collect(adaptive) { case read: AQEShuffleReadExec => read -@@ -2469,7 +2510,8 @@ class AdaptiveQueryExecSuite +@@ -2469,7 +2512,8 @@ class AdaptiveQueryExecSuite } } @@ -1891,7 +1935,7 @@ index 2f8e401e743..7849c685b19 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) { @@ -1900,7 +1944,7 @@ index 2f8e401e743..7849c685b19 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) { @@ -1909,7 +1953,7 @@ index 2f8e401e743..7849c685b19 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 @@ -1917,7 +1961,7 @@ index 2f8e401e743..7849c685b19 100644 }.size == (if (firstAccess) 2 else 0)) assert(collect(initialExecutedPlan) { case i: InMemoryTableScanLike => i -@@ -2938,7 +2981,8 @@ class AdaptiveQueryExecSuite +@@ -2938,7 +2983,8 @@ class AdaptiveQueryExecSuite } } @@ -1927,7 +1971,7 @@ index 2f8e401e743..7849c685b19 100644 val emptyDf = spark.range(1).where("false") val aggDf1 = emptyDf.agg(sum("id").as("id")).withColumn("name", lit("df1")) val aggDf2 = emptyDf.agg(sum("id").as("id")).withColumn("name", lit("df2")) -@@ -2980,7 +3024,9 @@ class AdaptiveQueryExecSuite +@@ -2980,7 +3026,9 @@ class AdaptiveQueryExecSuite val plan = df.queryExecution.executedPlan.asInstanceOf[AdaptiveSparkPlanExec] assert(plan.inputPlan.isInstanceOf[TakeOrderedAndProjectExec]) @@ -2372,7 +2416,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 @@ -2383,7 +2427,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)) } @@ -2493,6 +2557,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 d8e506b631..972043fc46 100644 --- a/dev/diffs/4.0.2.diff +++ b/dev/diffs/4.0.2.diff @@ -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 @@ -2135,7 +2159,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..4a175083adf 100644 +index 272be70f9fe..f8429423721 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 @@ -2413,7 +2437,27 @@ index 272be70f9fe..4a175083adf 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) { @@ -2422,7 +2466,7 @@ index 272be70f9fe..4a175083adf 100644 }.length == 1) } } -@@ -1801,7 +1831,8 @@ class AdaptiveQueryExecSuite +@@ -1801,7 +1833,8 @@ class AdaptiveQueryExecSuite } } @@ -2432,7 +2476,7 @@ index 272be70f9fe..4a175083adf 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 @@ -2442,7 +2486,7 @@ index 272be70f9fe..4a175083adf 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 @@ -2452,7 +2496,7 @@ index 272be70f9fe..4a175083adf 100644 }.size == 1) checkAnswer(ds, testData) } -@@ -2150,7 +2187,8 @@ class AdaptiveQueryExecSuite +@@ -2150,7 +2189,8 @@ class AdaptiveQueryExecSuite } } @@ -2462,7 +2506,7 @@ index 272be70f9fe..4a175083adf 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 } } @@ -2472,7 +2516,7 @@ index 272be70f9fe..4a175083adf 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) { @@ -2481,7 +2525,7 @@ index 272be70f9fe..4a175083adf 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) { @@ -2490,7 +2534,7 @@ index 272be70f9fe..4a175083adf 100644 } if (hasRequiredDistribution) { assert(shuffles2.size == 3) -@@ -2382,7 +2421,8 @@ class AdaptiveQueryExecSuite +@@ -2382,7 +2423,8 @@ class AdaptiveQueryExecSuite } } @@ -2500,7 +2544,7 @@ index 272be70f9fe..4a175083adf 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 @@ -2508,7 +2552,7 @@ index 272be70f9fe..4a175083adf 100644 }.size == 1) val read = collect(adaptive) { case read: AQEShuffleReadExec => read -@@ -2565,7 +2606,8 @@ class AdaptiveQueryExecSuite +@@ -2565,7 +2608,8 @@ class AdaptiveQueryExecSuite } } @@ -2518,7 +2562,7 @@ index 272be70f9fe..4a175083adf 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) { @@ -2527,7 +2571,7 @@ index 272be70f9fe..4a175083adf 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) { @@ -2536,7 +2580,7 @@ index 272be70f9fe..4a175083adf 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 @@ -2544,7 +2588,7 @@ index 272be70f9fe..4a175083adf 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 @@ -2552,7 +2596,7 @@ index 272be70f9fe..4a175083adf 100644 }.isEmpty) assert(collect(initialExecutedPlan) { case i: InMemoryTableScanLike => i -@@ -3039,7 +3083,8 @@ class AdaptiveQueryExecSuite +@@ -3039,7 +3085,8 @@ class AdaptiveQueryExecSuite } } @@ -2562,7 +2606,7 @@ index 272be70f9fe..4a175083adf 100644 val emptyDf = spark.range(1).where("false") val aggDf1 = emptyDf.agg(sum("id").as("id")).withColumn("name", lit("df1")) val aggDf2 = emptyDf.agg(sum("id").as("id")).withColumn("name", lit("df2")) -@@ -3129,7 +3174,8 @@ class AdaptiveQueryExecSuite +@@ -3129,7 +3176,8 @@ class AdaptiveQueryExecSuite val plan = df.queryExecution.executedPlan.asInstanceOf[AdaptiveSparkPlanExec] assert(plan.inputPlan.isInstanceOf[TakeOrderedAndProjectExec]) @@ -3036,7 +3080,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 @@ -3047,7 +3091,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)) } @@ -3198,6 +3262,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 9a1b30b010..1751377882 100644 --- a/dev/diffs/4.1.2.diff +++ b/dev/diffs/4.1.2.diff @@ -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 @@ -2258,7 +2282,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..79232dc3664 100644 +index 3e7d26f74bd..5ff372814d9 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 @@ -2536,7 +2560,27 @@ index 3e7d26f74bd..79232dc3664 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) { @@ -2545,7 +2589,7 @@ index 3e7d26f74bd..79232dc3664 100644 }.length == 1) } } -@@ -1972,7 +2002,8 @@ class AdaptiveQueryExecSuite +@@ -1972,7 +2004,8 @@ class AdaptiveQueryExecSuite } } @@ -2555,7 +2599,7 @@ index 3e7d26f74bd..79232dc3664 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 @@ -2565,7 +2609,7 @@ index 3e7d26f74bd..79232dc3664 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 @@ -2575,7 +2619,7 @@ index 3e7d26f74bd..79232dc3664 100644 }.size == 1) checkAnswer(ds, testData) } -@@ -2331,7 +2368,8 @@ class AdaptiveQueryExecSuite +@@ -2331,7 +2370,8 @@ class AdaptiveQueryExecSuite } } @@ -2585,7 +2629,7 @@ index 3e7d26f74bd..79232dc3664 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 } } @@ -2595,7 +2639,7 @@ index 3e7d26f74bd..79232dc3664 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) { @@ -2604,7 +2648,7 @@ index 3e7d26f74bd..79232dc3664 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) { @@ -2613,7 +2657,7 @@ index 3e7d26f74bd..79232dc3664 100644 } if (hasRequiredDistribution) { assert(shuffles2.size == 3) -@@ -2563,7 +2602,8 @@ class AdaptiveQueryExecSuite +@@ -2563,7 +2604,8 @@ class AdaptiveQueryExecSuite } } @@ -2623,7 +2667,7 @@ index 3e7d26f74bd..79232dc3664 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 @@ -2631,7 +2675,7 @@ index 3e7d26f74bd..79232dc3664 100644 }.size == 1) val read = collect(adaptive) { case read: AQEShuffleReadExec => read -@@ -2746,7 +2787,8 @@ class AdaptiveQueryExecSuite +@@ -2746,7 +2789,8 @@ class AdaptiveQueryExecSuite } } @@ -2641,7 +2685,7 @@ index 3e7d26f74bd..79232dc3664 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) { @@ -2650,7 +2694,7 @@ index 3e7d26f74bd..79232dc3664 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) { @@ -2659,7 +2703,7 @@ index 3e7d26f74bd..79232dc3664 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 @@ -2667,7 +2711,7 @@ index 3e7d26f74bd..79232dc3664 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 @@ -2675,7 +2719,7 @@ index 3e7d26f74bd..79232dc3664 100644 }.isEmpty) assert(collect(initialExecutedPlan) { case i: InMemoryTableScanLike => i -@@ -3220,7 +3264,8 @@ class AdaptiveQueryExecSuite +@@ -3220,7 +3266,8 @@ class AdaptiveQueryExecSuite } } @@ -2685,7 +2729,7 @@ index 3e7d26f74bd..79232dc3664 100644 val emptyDf = spark.range(1).where("false") val aggDf1 = emptyDf.agg(sum("id").as("id")).withColumn("name", lit("df1")) val aggDf2 = emptyDf.agg(sum("id").as("id")).withColumn("name", lit("df2")) -@@ -3310,7 +3355,8 @@ class AdaptiveQueryExecSuite +@@ -3310,7 +3357,8 @@ class AdaptiveQueryExecSuite val plan = df.queryExecution.executedPlan.asInstanceOf[AdaptiveSparkPlanExec] assert(plan.inputPlan.isInstanceOf[TakeOrderedAndProjectExec]) @@ -3225,7 +3269,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 @@ -3236,7 +3280,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)) } @@ -3408,6 +3472,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/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) }; 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 b6cbecc3c3..ced5c4f52f 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/main/scala/org/apache/spark/sql/comet/operators.scala b/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala index 2ee742ec36..139d006309 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 @@ -1978,6 +1978,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 && @@ -1988,7 +1994,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 } 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 e597e6e6e4..25769fd18d 100644 --- a/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala +++ b/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala @@ -2093,6 +2093,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) 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 5135d8483c..1ad0a55206 100644 --- a/spark/src/test/scala/org/apache/comet/rules/CometExecRuleSuite.scala +++ b/spark/src/test/scala/org/apache/comet/rules/CometExecRuleSuite.scala @@ -136,7 +136,7 @@ class CometExecRuleSuite extends CometTestBase { } // Regression test for https://github.com/apache/datafusion-comet/issues/1389 - test("CometExecRule should not allow Comet partial and Spark final hash aggregate") { + test("CometExecRule should allow Comet partial and Spark final hash aggregate for COUNT") { withTempView("test_data") { createTestDataFrame.createOrReplaceTempView("test_data") @@ -152,16 +152,15 @@ class CometExecRuleSuite extends CometTestBase { CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true") { val transformedPlan = applyCometExecRule(sparkPlan) - // 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) + // COUNT and SUM both support mixed partial/final execution, so the partial converts to + // Comet while the final stays on Spark. + assert(countOperators(transformedPlan, classOf[HashAggregateExec]) == 1) + assert(countOperators(transformedPlan, classOf[CometHashAggregateExec]) == 1) } } } - test("CometExecRule should not allow Spark partial and Comet final hash aggregate") { + test("CometExecRule should allow Spark partial and Comet final hash aggregate for COUNT") { withTempView("test_data") { createTestDataFrame.createOrReplaceTempView("test_data") @@ -177,11 +176,10 @@ class CometExecRuleSuite extends CometTestBase { CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true") { val transformedPlan = applyCometExecRule(sparkPlan) - // 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) + // COUNT and SUM both support mixed partial/final execution, so the final converts to + // Comet while the partial stays on Spark. + assert(countOperators(transformedPlan, classOf[HashAggregateExec]) == 1) + assert(countOperators(transformedPlan, classOf[CometHashAggregateExec]) == 1) } } } @@ -518,8 +516,14 @@ class CometExecRuleSuite extends CometTestBase { withTempView("test_data") { createTestDataFrame.createOrReplaceTempView("test_data") + // Decimal SUM is excluded from mixed partial/final execution, so disabling the partial + // forces both aggregates back to Spark JVM (the scenario this test exercises). Precision + // must be large enough (prec + 4 > 15) that Spark's DecimalAggregates rule does not rewrite + // SUM onto the unscaled Long value, which would sidestep the decimal buffer path. val sparkPlan = - createSparkPlan(spark, "SELECT COUNT(*), SUM(id) FROM test_data GROUP BY (id % 3)") + createSparkPlan( + spark, + "SELECT SUM(CAST(id AS DECIMAL(20, 2))) FROM test_data GROUP BY (id % 3)") val originalShuffleExchangeCount = countOperators(sparkPlan, classOf[ShuffleExchangeExec]) assert(originalShuffleExchangeCount == 1) @@ -551,8 +555,14 @@ class CometExecRuleSuite extends CometTestBase { withTempView("test_data") { createTestDataFrame.createOrReplaceTempView("test_data") + // Decimal SUM is excluded from mixed partial/final execution, so disabling the partial + // forces both aggregates back to Spark JVM (see the prior test). Precision must be large + // enough (prec + 4 > 15) that Spark's DecimalAggregates rule does not rewrite SUM onto the + // unscaled Long value. val sparkPlan = - createSparkPlan(spark, "SELECT COUNT(*), SUM(id) FROM test_data GROUP BY (id % 3)") + createSparkPlan( + spark, + "SELECT SUM(CAST(id AS DECIMAL(20, 2))) FROM test_data GROUP BY (id % 3)") assert(countOperators(sparkPlan, classOf[ShuffleExchangeExec]) == 1) assert(countOperators(sparkPlan, classOf[HashAggregateExec]) == 2)