diff --git a/.github/workflows/pr_build_linux.yml b/.github/workflows/pr_build_linux.yml index 35709c7274e..99e787cb65d 100644 --- a/.github/workflows/pr_build_linux.yml +++ b/.github/workflows/pr_build_linux.yml @@ -346,6 +346,7 @@ jobs: - name: "scans" value: | org.apache.comet.parquet.CometParquetWriterSuite + org.apache.comet.parquet.CometEmptyRelationParquetWriterSuite org.apache.comet.parquet.ParquetReadV1Suite org.apache.comet.parquet.ParquetReadV2Suite org.apache.comet.parquet.ParquetReadFromFakeHadoopFsSuite @@ -395,6 +396,7 @@ jobs: org.apache.comet.exec.CometAggregateSuite org.apache.comet.exec.CometExec3_4PlusSuite org.apache.comet.exec.CometExecSuite + org.apache.comet.exec.CometEmptyRelationExecSuite org.apache.comet.exec.CometInMemoryCacheSuite org.apache.comet.exec.CometInMemoryCacheKryoSuite org.apache.comet.exec.CometGenerateExecSuite @@ -419,6 +421,7 @@ jobs: org.apache.comet.rules.CometScanContribSuite org.apache.comet.rules.CometScanSchemeFallbackSuite org.apache.comet.rules.CometExecRuleSuite + org.apache.comet.rules.CometEmptyRelationExecRuleSuite org.apache.comet.rules.RevertNativeForTransitionHeavyStagesSuite org.apache.spark.sql.CometTPCDSQuerySuite org.apache.spark.sql.CometTPCDSQueryTestSuite diff --git a/.github/workflows/pr_build_macos.yml b/.github/workflows/pr_build_macos.yml index 52e1e277642..aeb8919b0d6 100644 --- a/.github/workflows/pr_build_macos.yml +++ b/.github/workflows/pr_build_macos.yml @@ -119,6 +119,7 @@ jobs: - name: "scans" value: | org.apache.comet.parquet.CometParquetWriterSuite + org.apache.comet.parquet.CometEmptyRelationParquetWriterSuite org.apache.comet.parquet.ParquetReadV1Suite org.apache.comet.parquet.ParquetReadV2Suite org.apache.comet.parquet.ParquetReadFromFakeHadoopFsSuite @@ -168,6 +169,7 @@ jobs: org.apache.comet.exec.CometAggregateSuite org.apache.comet.exec.CometExec3_4PlusSuite org.apache.comet.exec.CometExecSuite + org.apache.comet.exec.CometEmptyRelationExecSuite org.apache.comet.exec.CometInMemoryCacheSuite org.apache.comet.exec.CometInMemoryCacheKryoSuite org.apache.comet.exec.CometGenerateExecSuite @@ -192,6 +194,7 @@ jobs: org.apache.comet.rules.CometScanContribSuite org.apache.comet.rules.CometScanSchemeFallbackSuite org.apache.comet.rules.CometExecRuleSuite + org.apache.comet.rules.CometEmptyRelationExecRuleSuite org.apache.comet.rules.RevertNativeForTransitionHeavyStagesSuite org.apache.spark.sql.CometTPCDSQuerySuite org.apache.spark.sql.CometTPCDSQueryTestSuite diff --git a/dev/diffs/4.0.4.diff b/dev/diffs/4.0.4.diff index db960a52b20..8c2bd69c7b3 100644 --- a/dev/diffs/4.0.4.diff +++ b/dev/diffs/4.0.4.diff @@ -2199,7 +2199,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 fb8fab6a80f..9168a1115c6 100644 +index fb8fab6a80f..403eb411920 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 @@ -2226,7 +2226,7 @@ index fb8fab6a80f..9168a1115c6 100644 } } -@@ -131,36 +134,46 @@ class AdaptiveQueryExecSuite +@@ -131,36 +134,55 @@ class AdaptiveQueryExecSuite private def findTopLevelSortMergeJoin(plan: SparkPlan): Seq[SortMergeJoinExec] = { collect(plan) { case j: SortMergeJoinExec => j @@ -2252,6 +2252,15 @@ index fb8fab6a80f..9168a1115c6 100644 } } ++ private def isEmptyRelation(plan: SparkPlan): Boolean = stripAQEPlan(plan) match { ++ case _: EmptyRelationExec | _: CometEmptyRelationExec => true ++ case CometScanWrapper(_, _: CometEmptyRelationExec) => true ++ case p: WholeStageCodegenExec => isEmptyRelation(p.child) ++ case p: InputAdapter => isEmptyRelation(p.child) ++ case p: ColumnarToRowTransition => isEmptyRelation(p.child) ++ case _ => false ++ } ++ private def findTopLevelSort(plan: SparkPlan): Seq[SortExec] = { collect(plan) { case s: SortExec => s @@ -2273,7 +2282,7 @@ index fb8fab6a80f..9168a1115c6 100644 } } -@@ -204,6 +217,7 @@ class AdaptiveQueryExecSuite +@@ -204,6 +226,7 @@ class AdaptiveQueryExecSuite val parts = rdd.partitions assert(parts.forall(rdd.preferredLocations(_).nonEmpty)) } @@ -2281,7 +2290,7 @@ index fb8fab6a80f..9168a1115c6 100644 assert(numShuffles === (numLocalReads.length + numShufflesWithoutLocalRead)) } -@@ -212,7 +226,7 @@ class AdaptiveQueryExecSuite +@@ -212,7 +235,7 @@ class AdaptiveQueryExecSuite val plan = df.queryExecution.executedPlan assert(plan.isInstanceOf[AdaptiveSparkPlanExec]) val shuffle = plan.asInstanceOf[AdaptiveSparkPlanExec].executedPlan.collect { @@ -2290,7 +2299,7 @@ index fb8fab6a80f..9168a1115c6 100644 } assert(shuffle.size == 1) assert(shuffle(0).outputPartitioning.numPartitions == numPartition) -@@ -228,7 +242,8 @@ class AdaptiveQueryExecSuite +@@ -228,7 +251,8 @@ class AdaptiveQueryExecSuite assert(smj.size == 1) val bhj = findTopLevelBroadcastHashJoin(adaptivePlan) assert(bhj.size == 1) @@ -2300,7 +2309,7 @@ index fb8fab6a80f..9168a1115c6 100644 } } -@@ -255,7 +270,8 @@ class AdaptiveQueryExecSuite +@@ -255,7 +279,8 @@ class AdaptiveQueryExecSuite } } @@ -2310,7 +2319,7 @@ index fb8fab6a80f..9168a1115c6 100644 withSQLConf( SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80", -@@ -287,7 +303,8 @@ class AdaptiveQueryExecSuite +@@ -287,7 +312,8 @@ class AdaptiveQueryExecSuite } } @@ -2320,7 +2329,7 @@ index fb8fab6a80f..9168a1115c6 100644 withSQLConf( SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80", -@@ -301,7 +318,8 @@ class AdaptiveQueryExecSuite +@@ -301,7 +327,8 @@ class AdaptiveQueryExecSuite val localReads = collect(adaptivePlan) { case read: AQEShuffleReadExec if read.isLocalRead => read } @@ -2330,7 +2339,7 @@ index fb8fab6a80f..9168a1115c6 100644 val localShuffleRDD0 = localReads(0).execute().asInstanceOf[ShuffledRowRDD] val localShuffleRDD1 = localReads(1).execute().asInstanceOf[ShuffledRowRDD] // the final parallelism is math.max(1, numReduces / numMappers): math.max(1, 5/2) = 2 -@@ -326,7 +344,9 @@ class AdaptiveQueryExecSuite +@@ -326,7 +353,9 @@ class AdaptiveQueryExecSuite .groupBy($"a").count() checkAnswer(testDf, Seq()) val plan = testDf.queryExecution.executedPlan @@ -2341,7 +2350,7 @@ index fb8fab6a80f..9168a1115c6 100644 val coalescedReads = collect(plan) { case r: AQEShuffleReadExec => r } -@@ -340,7 +360,9 @@ class AdaptiveQueryExecSuite +@@ -340,7 +369,9 @@ class AdaptiveQueryExecSuite .groupBy($"a").count() checkAnswer(testDf, Seq()) val plan = testDf.queryExecution.executedPlan @@ -2352,7 +2361,7 @@ index fb8fab6a80f..9168a1115c6 100644 val coalescedReads = collect(plan) { case r: AQEShuffleReadExec => r } -@@ -350,7 +372,7 @@ class AdaptiveQueryExecSuite +@@ -350,7 +381,7 @@ class AdaptiveQueryExecSuite } } @@ -2361,7 +2370,7 @@ index fb8fab6a80f..9168a1115c6 100644 withSQLConf( SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") { -@@ -365,7 +387,7 @@ class AdaptiveQueryExecSuite +@@ -365,7 +396,7 @@ class AdaptiveQueryExecSuite } } @@ -2370,7 +2379,7 @@ index fb8fab6a80f..9168a1115c6 100644 withSQLConf( SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") { -@@ -381,7 +403,7 @@ class AdaptiveQueryExecSuite +@@ -381,7 +412,7 @@ class AdaptiveQueryExecSuite } } @@ -2379,7 +2388,7 @@ index fb8fab6a80f..9168a1115c6 100644 withSQLConf( SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") { -@@ -426,7 +448,7 @@ class AdaptiveQueryExecSuite +@@ -426,7 +457,7 @@ class AdaptiveQueryExecSuite } } @@ -2388,7 +2397,7 @@ index fb8fab6a80f..9168a1115c6 100644 withSQLConf( SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") { -@@ -471,7 +493,7 @@ class AdaptiveQueryExecSuite +@@ -471,7 +502,7 @@ class AdaptiveQueryExecSuite } } @@ -2397,7 +2406,7 @@ index fb8fab6a80f..9168a1115c6 100644 withSQLConf( SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "500") { -@@ -517,7 +539,7 @@ class AdaptiveQueryExecSuite +@@ -517,7 +548,7 @@ class AdaptiveQueryExecSuite } } @@ -2406,7 +2415,7 @@ index fb8fab6a80f..9168a1115c6 100644 withSQLConf( SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") { -@@ -536,7 +558,7 @@ class AdaptiveQueryExecSuite +@@ -536,7 +567,7 @@ class AdaptiveQueryExecSuite } } @@ -2415,7 +2424,7 @@ index fb8fab6a80f..9168a1115c6 100644 withSQLConf( SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") { -@@ -567,7 +589,9 @@ class AdaptiveQueryExecSuite +@@ -567,7 +598,9 @@ class AdaptiveQueryExecSuite assert(smj.size == 1) val bhj = findTopLevelBroadcastHashJoin(adaptivePlan) assert(bhj.size == 1) @@ -2426,7 +2435,7 @@ index fb8fab6a80f..9168a1115c6 100644 // Even with local shuffle read, the query stage reuse can also work. val ex = findReusedExchange(adaptivePlan) assert(ex.nonEmpty) -@@ -588,7 +612,9 @@ class AdaptiveQueryExecSuite +@@ -588,7 +621,9 @@ class AdaptiveQueryExecSuite assert(smj.size == 1) val bhj = findTopLevelBroadcastHashJoin(adaptivePlan) assert(bhj.size == 1) @@ -2437,7 +2446,7 @@ index fb8fab6a80f..9168a1115c6 100644 // Even with local shuffle read, the query stage reuse can also work. val ex = findReusedExchange(adaptivePlan) assert(ex.isEmpty) -@@ -597,7 +623,8 @@ class AdaptiveQueryExecSuite +@@ -597,7 +632,8 @@ class AdaptiveQueryExecSuite } } @@ -2447,7 +2456,7 @@ index fb8fab6a80f..9168a1115c6 100644 withSQLConf( SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "20000000", -@@ -692,7 +719,8 @@ class AdaptiveQueryExecSuite +@@ -692,7 +728,8 @@ class AdaptiveQueryExecSuite val bhj = findTopLevelBroadcastHashJoin(adaptivePlan) assert(bhj.size == 1) // There is still a SMJ, and its two shuffles can't apply local read. @@ -2457,7 +2466,7 @@ index fb8fab6a80f..9168a1115c6 100644 } } -@@ -814,7 +842,8 @@ class AdaptiveQueryExecSuite +@@ -814,7 +851,8 @@ class AdaptiveQueryExecSuite } } @@ -2467,7 +2476,7 @@ index fb8fab6a80f..9168a1115c6 100644 Seq("SHUFFLE_MERGE", "SHUFFLE_HASH").foreach { joinHint => def getJoinNode(plan: SparkPlan): Seq[ShuffledJoin] = if (joinHint == "SHUFFLE_MERGE") { findTopLevelSortMergeJoin(plan) -@@ -1096,7 +1125,8 @@ class AdaptiveQueryExecSuite +@@ -1096,7 +1134,8 @@ class AdaptiveQueryExecSuite } } @@ -2477,7 +2486,66 @@ index fb8fab6a80f..9168a1115c6 100644 withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true") { val (_, adaptivePlan) = runAdaptiveAndVerifyResult( "SELECT key FROM testData GROUP BY key") -@@ -1730,7 +1760,7 @@ class AdaptiveQueryExecSuite +@@ -1662,13 +1701,13 @@ class AdaptiveQueryExecSuite + val (plan1, adaptivePlan1) = runAdaptiveAndVerifyResult( + "SELECT key FROM testData WHERE key = 0 ORDER BY key, value") + assert(findTopLevelSort(plan1).size == 1) +- assert(stripAQEPlan(adaptivePlan1).isInstanceOf[EmptyRelationExec]) ++ assert(isEmptyRelation(adaptivePlan1), adaptivePlan1) + + val (plan2, adaptivePlan2) = runAdaptiveAndVerifyResult( + "SELECT key FROM (SELECT * FROM testData WHERE value = 'no_match' ORDER BY key)" + + " WHERE key > rand()") + assert(findTopLevelSort(plan2).size == 1) +- assert(stripAQEPlan(adaptivePlan2).isInstanceOf[EmptyRelationExec]) ++ assert(isEmptyRelation(adaptivePlan2), adaptivePlan2) + } + } + +@@ -1676,18 +1715,18 @@ class AdaptiveQueryExecSuite + withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true") { + val (plan1, adaptivePlan1) = runAdaptiveAndVerifyResult( + "SELECT key, count(*) FROM testData WHERE value = 'no_match' GROUP BY key") +- assert(!plan1.isInstanceOf[EmptyRelationExec]) +- assert(stripAQEPlan(adaptivePlan1).isInstanceOf[EmptyRelationExec]) ++ assert(!isEmptyRelation(plan1), plan1) ++ assert(isEmptyRelation(adaptivePlan1), adaptivePlan1) + + val (plan2, adaptivePlan2) = runAdaptiveAndVerifyResult( + "SELECT key, count(*) FROM testData WHERE value = 'no_match' GROUP BY key limit 1") +- assert(!plan2.isInstanceOf[EmptyRelationExec]) +- assert(stripAQEPlan(adaptivePlan2).isInstanceOf[EmptyRelationExec]) ++ assert(!isEmptyRelation(plan2), plan2) ++ assert(isEmptyRelation(adaptivePlan2), adaptivePlan2) + + val (plan3, adaptivePlan3) = runAdaptiveAndVerifyResult( + "SELECT count(*) FROM testData WHERE value = 'no_match'") +- assert(!plan3.isInstanceOf[EmptyRelationExec]) +- assert(!stripAQEPlan(adaptivePlan3).isInstanceOf[EmptyRelationExec]) ++ assert(!isEmptyRelation(plan3), plan3) ++ assert(!isEmptyRelation(adaptivePlan3), adaptivePlan3) + } + } + +@@ -1708,7 +1747,7 @@ class AdaptiveQueryExecSuite + |""".stripMargin) + checkNumUnion(plan1, 1) + checkNumUnion(adaptivePlan1, 0) +- assert(!stripAQEPlan(adaptivePlan1).isInstanceOf[EmptyRelationExec]) ++ assert(!isEmptyRelation(adaptivePlan1), adaptivePlan1) + + val (plan2, adaptivePlan2) = runAdaptiveAndVerifyResult( + """ +@@ -1718,7 +1757,7 @@ class AdaptiveQueryExecSuite + |""".stripMargin) + checkNumUnion(plan2, 1) + checkNumUnion(adaptivePlan2, 0) +- assert(stripAQEPlan(adaptivePlan2).isInstanceOf[EmptyRelationExec]) ++ assert(isEmptyRelation(adaptivePlan2), adaptivePlan2) + } + } + +@@ -1730,7 +1769,7 @@ class AdaptiveQueryExecSuite val (_, adaptivePlan) = runAdaptiveAndVerifyResult( "SELECT id FROM v1 GROUP BY id DISTRIBUTE BY id") assert(collect(adaptivePlan) { @@ -2486,7 +2554,7 @@ index fb8fab6a80f..9168a1115c6 100644 }.length == 1) } } -@@ -1810,7 +1840,8 @@ class AdaptiveQueryExecSuite +@@ -1810,7 +1849,8 @@ class AdaptiveQueryExecSuite } } @@ -2496,7 +2564,7 @@ index fb8fab6a80f..9168a1115c6 100644 def hasRepartitionShuffle(plan: SparkPlan): Boolean = { find(plan) { case s: ShuffleExchangeLike => -@@ -1995,6 +2026,9 @@ class AdaptiveQueryExecSuite +@@ -1995,6 +2035,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 @@ -2506,7 +2574,7 @@ index fb8fab6a80f..9168a1115c6 100644 }.size == 1) ds.collect() val plan = ds.queryExecution.executedPlan -@@ -2003,6 +2037,9 @@ class AdaptiveQueryExecSuite +@@ -2003,6 +2046,9 @@ class AdaptiveQueryExecSuite }.isEmpty) assert(collect(plan) { case s: ShuffleExchangeExec if s.shuffleOrigin == origin && s.numPartitions == 2 => s @@ -2516,7 +2584,7 @@ index fb8fab6a80f..9168a1115c6 100644 }.size == 1) checkAnswer(ds, testData) } -@@ -2159,7 +2196,8 @@ class AdaptiveQueryExecSuite +@@ -2159,7 +2205,8 @@ class AdaptiveQueryExecSuite } } @@ -2526,7 +2594,7 @@ index fb8fab6a80f..9168a1115c6 100644 withTempView("t1", "t2") { def checkJoinStrategy(shouldShuffleHashJoin: Boolean): Unit = { Seq("100", "100000").foreach { size => -@@ -2245,7 +2283,8 @@ class AdaptiveQueryExecSuite +@@ -2245,7 +2292,8 @@ class AdaptiveQueryExecSuite } } @@ -2536,7 +2604,7 @@ index fb8fab6a80f..9168a1115c6 100644 withTempView("v") { withSQLConf( SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", -@@ -2344,7 +2383,7 @@ class AdaptiveQueryExecSuite +@@ -2344,7 +2392,7 @@ class AdaptiveQueryExecSuite runAdaptiveAndVerifyResult(s"SELECT $repartition key1 FROM skewData1 " + s"JOIN skewData2 ON key1 = key2 GROUP BY key1") val shuffles1 = collect(adaptive1) { @@ -2545,7 +2613,7 @@ index fb8fab6a80f..9168a1115c6 100644 } assert(shuffles1.size == 3) // shuffles1.head is the top-level shuffle under the Aggregate operator -@@ -2357,7 +2396,7 @@ class AdaptiveQueryExecSuite +@@ -2357,7 +2405,7 @@ class AdaptiveQueryExecSuite runAdaptiveAndVerifyResult(s"SELECT $repartition key1 FROM skewData1 " + s"JOIN skewData2 ON key1 = key2") val shuffles2 = collect(adaptive2) { @@ -2554,7 +2622,7 @@ index fb8fab6a80f..9168a1115c6 100644 } if (hasRequiredDistribution) { assert(shuffles2.size == 3) -@@ -2391,7 +2430,8 @@ class AdaptiveQueryExecSuite +@@ -2391,7 +2439,8 @@ class AdaptiveQueryExecSuite } } @@ -2564,7 +2632,7 @@ index fb8fab6a80f..9168a1115c6 100644 CostEvaluator.instantiate( classOf[SimpleShuffleSortCostEvaluator].getCanonicalName, spark.sparkContext.getConf) intercept[IllegalArgumentException] { -@@ -2557,6 +2597,7 @@ class AdaptiveQueryExecSuite +@@ -2557,6 +2606,7 @@ class AdaptiveQueryExecSuite val (_, adaptive) = runAdaptiveAndVerifyResult(query) assert(adaptive.collect { case sort: SortExec => sort @@ -2572,7 +2640,7 @@ index fb8fab6a80f..9168a1115c6 100644 }.size == 1) val read = collect(adaptive) { case read: AQEShuffleReadExec => read -@@ -2574,7 +2615,8 @@ class AdaptiveQueryExecSuite +@@ -2574,7 +2624,8 @@ class AdaptiveQueryExecSuite } } @@ -2582,7 +2650,7 @@ index fb8fab6a80f..9168a1115c6 100644 withTempView("v") { withSQLConf( SQLConf.ADAPTIVE_OPTIMIZE_SKEWS_IN_REBALANCE_PARTITIONS_ENABLED.key -> "true", -@@ -2686,7 +2728,7 @@ class AdaptiveQueryExecSuite +@@ -2686,7 +2737,7 @@ class AdaptiveQueryExecSuite runAdaptiveAndVerifyResult("SELECT key1 FROM skewData1 JOIN skewData2 ON key1 = key2 " + "JOIN skewData3 ON value2 = value3") val shuffles1 = collect(adaptive1) { @@ -2591,7 +2659,7 @@ index fb8fab6a80f..9168a1115c6 100644 } assert(shuffles1.size == 4) val smj1 = findTopLevelSortMergeJoin(adaptive1) -@@ -2697,7 +2739,7 @@ class AdaptiveQueryExecSuite +@@ -2697,7 +2748,7 @@ class AdaptiveQueryExecSuite runAdaptiveAndVerifyResult("SELECT key1 FROM skewData1 JOIN skewData2 ON key1 = key2 " + "JOIN skewData3 ON value1 = value3") val shuffles2 = collect(adaptive2) { @@ -2600,7 +2668,7 @@ index fb8fab6a80f..9168a1115c6 100644 } assert(shuffles2.size == 4) val smj2 = findTopLevelSortMergeJoin(adaptive2) -@@ -2955,6 +2997,7 @@ class AdaptiveQueryExecSuite +@@ -2955,6 +3006,7 @@ class AdaptiveQueryExecSuite }.size == (if (firstAccess) 1 else 0)) assert(collect(initialExecutedPlan) { case s: SortExec => s @@ -2608,7 +2676,7 @@ index fb8fab6a80f..9168a1115c6 100644 }.size == (if (firstAccess) 2 else 0)) assert(collect(initialExecutedPlan) { case i: InMemoryTableScanLike => i -@@ -2967,6 +3010,7 @@ class AdaptiveQueryExecSuite +@@ -2967,6 +3019,7 @@ class AdaptiveQueryExecSuite }.isEmpty) assert(collect(finalExecutedPlan) { case s: SortExec => s @@ -2616,7 +2684,7 @@ index fb8fab6a80f..9168a1115c6 100644 }.isEmpty) assert(collect(initialExecutedPlan) { case i: InMemoryTableScanLike => i -@@ -3048,7 +3092,8 @@ class AdaptiveQueryExecSuite +@@ -3048,7 +3101,8 @@ class AdaptiveQueryExecSuite } } @@ -2626,7 +2694,7 @@ index fb8fab6a80f..9168a1115c6 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")) -@@ -3138,7 +3183,8 @@ class AdaptiveQueryExecSuite +@@ -3138,7 +3192,8 @@ class AdaptiveQueryExecSuite val plan = df.queryExecution.executedPlan.asInstanceOf[AdaptiveSparkPlanExec] assert(plan.inputPlan.isInstanceOf[TakeOrderedAndProjectExec]) diff --git a/dev/diffs/4.1.3.diff b/dev/diffs/4.1.3.diff index d0289d0de06..99cf86432f2 100644 --- a/dev/diffs/4.1.3.diff +++ b/dev/diffs/4.1.3.diff @@ -2295,7 +2295,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 188a28ff1c0..3cacc4efbdf 100644 +index 188a28ff1c0..8fdccf31749 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 @@ -2322,7 +2322,7 @@ index 188a28ff1c0..3cacc4efbdf 100644 } } -@@ -133,36 +136,46 @@ class AdaptiveQueryExecSuite +@@ -133,36 +136,55 @@ class AdaptiveQueryExecSuite private def findTopLevelSortMergeJoin(plan: SparkPlan): Seq[SortMergeJoinExec] = { collect(plan) { case j: SortMergeJoinExec => j @@ -2348,6 +2348,15 @@ index 188a28ff1c0..3cacc4efbdf 100644 } } ++ private def isEmptyRelation(plan: SparkPlan): Boolean = stripAQEPlan(plan) match { ++ case _: EmptyRelationExec | _: CometEmptyRelationExec => true ++ case CometScanWrapper(_, _: CometEmptyRelationExec) => true ++ case p: WholeStageCodegenExec => isEmptyRelation(p.child) ++ case p: InputAdapter => isEmptyRelation(p.child) ++ case p: ColumnarToRowTransition => isEmptyRelation(p.child) ++ case _ => false ++ } ++ private def findTopLevelSort(plan: SparkPlan): Seq[SortExec] = { collect(plan) { case s: SortExec => s @@ -2369,7 +2378,7 @@ index 188a28ff1c0..3cacc4efbdf 100644 } } -@@ -206,6 +219,7 @@ class AdaptiveQueryExecSuite +@@ -206,6 +228,7 @@ class AdaptiveQueryExecSuite val parts = rdd.partitions assert(parts.forall(rdd.preferredLocations(_).nonEmpty)) } @@ -2377,7 +2386,7 @@ index 188a28ff1c0..3cacc4efbdf 100644 assert(numShuffles === (numLocalReads.length + numShufflesWithoutLocalRead)) } -@@ -214,7 +228,7 @@ class AdaptiveQueryExecSuite +@@ -214,7 +237,7 @@ class AdaptiveQueryExecSuite val plan = df.queryExecution.executedPlan assert(plan.isInstanceOf[AdaptiveSparkPlanExec]) val shuffle = plan.asInstanceOf[AdaptiveSparkPlanExec].executedPlan.collect { @@ -2386,7 +2395,7 @@ index 188a28ff1c0..3cacc4efbdf 100644 } assert(shuffle.size == 1) assert(shuffle(0).outputPartitioning.numPartitions == numPartition) -@@ -230,7 +244,8 @@ class AdaptiveQueryExecSuite +@@ -230,7 +253,8 @@ class AdaptiveQueryExecSuite assert(smj.size == 1) val bhj = findTopLevelBroadcastHashJoin(adaptivePlan) assert(bhj.size == 1) @@ -2396,7 +2405,7 @@ index 188a28ff1c0..3cacc4efbdf 100644 } } -@@ -257,7 +272,8 @@ class AdaptiveQueryExecSuite +@@ -257,7 +281,8 @@ class AdaptiveQueryExecSuite } } @@ -2406,7 +2415,7 @@ index 188a28ff1c0..3cacc4efbdf 100644 withSQLConf( SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80", -@@ -289,7 +305,8 @@ class AdaptiveQueryExecSuite +@@ -289,7 +314,8 @@ class AdaptiveQueryExecSuite } } @@ -2416,7 +2425,7 @@ index 188a28ff1c0..3cacc4efbdf 100644 withSQLConf( SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80", -@@ -303,7 +320,8 @@ class AdaptiveQueryExecSuite +@@ -303,7 +329,8 @@ class AdaptiveQueryExecSuite val localReads = collect(adaptivePlan) { case read: AQEShuffleReadExec if read.isLocalRead => read } @@ -2426,7 +2435,7 @@ index 188a28ff1c0..3cacc4efbdf 100644 val localShuffleRDD0 = localReads(0).execute().asInstanceOf[ShuffledRowRDD] val localShuffleRDD1 = localReads(1).execute().asInstanceOf[ShuffledRowRDD] // the final parallelism is math.max(1, numReduces / numMappers): math.max(1, 5/2) = 2 -@@ -328,7 +346,9 @@ class AdaptiveQueryExecSuite +@@ -328,7 +355,9 @@ class AdaptiveQueryExecSuite .groupBy($"a").count() checkAnswer(testDf, Seq()) val plan = testDf.queryExecution.executedPlan @@ -2437,7 +2446,7 @@ index 188a28ff1c0..3cacc4efbdf 100644 val coalescedReads = collect(plan) { case r: AQEShuffleReadExec => r } -@@ -342,7 +362,9 @@ class AdaptiveQueryExecSuite +@@ -342,7 +371,9 @@ class AdaptiveQueryExecSuite .groupBy($"a").count() checkAnswer(testDf, Seq()) val plan = testDf.queryExecution.executedPlan @@ -2448,7 +2457,7 @@ index 188a28ff1c0..3cacc4efbdf 100644 val coalescedReads = collect(plan) { case r: AQEShuffleReadExec => r } -@@ -352,7 +374,7 @@ class AdaptiveQueryExecSuite +@@ -352,7 +383,7 @@ class AdaptiveQueryExecSuite } } @@ -2457,7 +2466,7 @@ index 188a28ff1c0..3cacc4efbdf 100644 withSQLConf( SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") { -@@ -367,7 +389,7 @@ class AdaptiveQueryExecSuite +@@ -367,7 +398,7 @@ class AdaptiveQueryExecSuite } } @@ -2466,7 +2475,7 @@ index 188a28ff1c0..3cacc4efbdf 100644 withSQLConf( SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") { -@@ -383,7 +405,7 @@ class AdaptiveQueryExecSuite +@@ -383,7 +414,7 @@ class AdaptiveQueryExecSuite } } @@ -2475,7 +2484,7 @@ index 188a28ff1c0..3cacc4efbdf 100644 withSQLConf( SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") { -@@ -428,7 +450,7 @@ class AdaptiveQueryExecSuite +@@ -428,7 +459,7 @@ class AdaptiveQueryExecSuite } } @@ -2484,7 +2493,7 @@ index 188a28ff1c0..3cacc4efbdf 100644 withSQLConf( SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") { -@@ -473,7 +495,7 @@ class AdaptiveQueryExecSuite +@@ -473,7 +504,7 @@ class AdaptiveQueryExecSuite } } @@ -2493,7 +2502,7 @@ index 188a28ff1c0..3cacc4efbdf 100644 withSQLConf( SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "500") { -@@ -519,7 +541,7 @@ class AdaptiveQueryExecSuite +@@ -519,7 +550,7 @@ class AdaptiveQueryExecSuite } } @@ -2502,7 +2511,7 @@ index 188a28ff1c0..3cacc4efbdf 100644 withSQLConf( SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") { -@@ -538,7 +560,7 @@ class AdaptiveQueryExecSuite +@@ -538,7 +569,7 @@ class AdaptiveQueryExecSuite } } @@ -2511,7 +2520,7 @@ index 188a28ff1c0..3cacc4efbdf 100644 withSQLConf( SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") { -@@ -569,7 +591,9 @@ class AdaptiveQueryExecSuite +@@ -569,7 +600,9 @@ class AdaptiveQueryExecSuite assert(smj.size == 1) val bhj = findTopLevelBroadcastHashJoin(adaptivePlan) assert(bhj.size == 1) @@ -2522,7 +2531,7 @@ index 188a28ff1c0..3cacc4efbdf 100644 // Even with local shuffle read, the query stage reuse can also work. val ex = findReusedExchange(adaptivePlan) assert(ex.nonEmpty) -@@ -590,7 +614,9 @@ class AdaptiveQueryExecSuite +@@ -590,7 +623,9 @@ class AdaptiveQueryExecSuite assert(smj.size == 1) val bhj = findTopLevelBroadcastHashJoin(adaptivePlan) assert(bhj.size == 1) @@ -2533,7 +2542,7 @@ index 188a28ff1c0..3cacc4efbdf 100644 // Even with local shuffle read, the query stage reuse can also work. val ex = findReusedExchange(adaptivePlan) assert(ex.isEmpty) -@@ -599,7 +625,8 @@ class AdaptiveQueryExecSuite +@@ -599,7 +634,8 @@ class AdaptiveQueryExecSuite } } @@ -2543,7 +2552,7 @@ index 188a28ff1c0..3cacc4efbdf 100644 withSQLConf( SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "20000000", -@@ -694,7 +721,8 @@ class AdaptiveQueryExecSuite +@@ -694,7 +730,8 @@ class AdaptiveQueryExecSuite val bhj = findTopLevelBroadcastHashJoin(adaptivePlan) assert(bhj.size == 1) // There is still a SMJ, and its two shuffles can't apply local read. @@ -2553,7 +2562,7 @@ index 188a28ff1c0..3cacc4efbdf 100644 } } -@@ -822,7 +850,8 @@ class AdaptiveQueryExecSuite +@@ -822,7 +859,8 @@ class AdaptiveQueryExecSuite } } @@ -2563,7 +2572,7 @@ index 188a28ff1c0..3cacc4efbdf 100644 Seq("SHUFFLE_MERGE", "SHUFFLE_HASH").foreach { joinHint => def getJoinNode(plan: SparkPlan): Seq[ShuffledJoin] = if (joinHint == "SHUFFLE_MERGE") { findTopLevelSortMergeJoin(plan) -@@ -1266,7 +1295,8 @@ class AdaptiveQueryExecSuite +@@ -1266,7 +1304,8 @@ class AdaptiveQueryExecSuite } } @@ -2573,7 +2582,66 @@ index 188a28ff1c0..3cacc4efbdf 100644 withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true") { val (_, adaptivePlan) = runAdaptiveAndVerifyResult( "SELECT key FROM testData GROUP BY key") -@@ -1900,7 +1930,7 @@ class AdaptiveQueryExecSuite +@@ -1832,13 +1871,13 @@ class AdaptiveQueryExecSuite + val (plan1, adaptivePlan1) = runAdaptiveAndVerifyResult( + "SELECT key FROM testData WHERE key = 0 ORDER BY key, value") + assert(findTopLevelSort(plan1).size == 1) +- assert(stripAQEPlan(adaptivePlan1).isInstanceOf[EmptyRelationExec]) ++ assert(isEmptyRelation(adaptivePlan1), adaptivePlan1) + + val (plan2, adaptivePlan2) = runAdaptiveAndVerifyResult( + "SELECT key FROM (SELECT * FROM testData WHERE value = 'no_match' ORDER BY key)" + + " WHERE key > rand()") + assert(findTopLevelSort(plan2).size == 1) +- assert(stripAQEPlan(adaptivePlan2).isInstanceOf[EmptyRelationExec]) ++ assert(isEmptyRelation(adaptivePlan2), adaptivePlan2) + } + } + +@@ -1846,18 +1885,18 @@ class AdaptiveQueryExecSuite + withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true") { + val (plan1, adaptivePlan1) = runAdaptiveAndVerifyResult( + "SELECT key, count(*) FROM testData WHERE value = 'no_match' GROUP BY key") +- assert(!plan1.isInstanceOf[EmptyRelationExec]) +- assert(stripAQEPlan(adaptivePlan1).isInstanceOf[EmptyRelationExec]) ++ assert(!isEmptyRelation(plan1), plan1) ++ assert(isEmptyRelation(adaptivePlan1), adaptivePlan1) + + val (plan2, adaptivePlan2) = runAdaptiveAndVerifyResult( + "SELECT key, count(*) FROM testData WHERE value = 'no_match' GROUP BY key limit 1") +- assert(!plan2.isInstanceOf[EmptyRelationExec]) +- assert(stripAQEPlan(adaptivePlan2).isInstanceOf[EmptyRelationExec]) ++ assert(!isEmptyRelation(plan2), plan2) ++ assert(isEmptyRelation(adaptivePlan2), adaptivePlan2) + + val (plan3, adaptivePlan3) = runAdaptiveAndVerifyResult( + "SELECT count(*) FROM testData WHERE value = 'no_match'") +- assert(!plan3.isInstanceOf[EmptyRelationExec]) +- assert(!stripAQEPlan(adaptivePlan3).isInstanceOf[EmptyRelationExec]) ++ assert(!isEmptyRelation(plan3), plan3) ++ assert(!isEmptyRelation(adaptivePlan3), adaptivePlan3) + } + } + +@@ -1878,7 +1917,7 @@ class AdaptiveQueryExecSuite + |""".stripMargin) + checkNumUnion(plan1, 1) + checkNumUnion(adaptivePlan1, 0) +- assert(!stripAQEPlan(adaptivePlan1).isInstanceOf[EmptyRelationExec]) ++ assert(!isEmptyRelation(adaptivePlan1), adaptivePlan1) + + val (plan2, adaptivePlan2) = runAdaptiveAndVerifyResult( + """ +@@ -1888,7 +1927,7 @@ class AdaptiveQueryExecSuite + |""".stripMargin) + checkNumUnion(plan2, 1) + checkNumUnion(adaptivePlan2, 0) +- assert(stripAQEPlan(adaptivePlan2).isInstanceOf[EmptyRelationExec]) ++ assert(isEmptyRelation(adaptivePlan2), adaptivePlan2) + } + } + +@@ -1900,7 +1939,7 @@ class AdaptiveQueryExecSuite val (_, adaptivePlan) = runAdaptiveAndVerifyResult( "SELECT id FROM v1 GROUP BY id DISTRIBUTE BY id") assert(collect(adaptivePlan) { @@ -2582,7 +2650,7 @@ index 188a28ff1c0..3cacc4efbdf 100644 }.length == 1) } } -@@ -1981,7 +2011,8 @@ class AdaptiveQueryExecSuite +@@ -1981,7 +2020,8 @@ class AdaptiveQueryExecSuite } } @@ -2592,7 +2660,7 @@ index 188a28ff1c0..3cacc4efbdf 100644 def hasRepartitionShuffle(plan: SparkPlan): Boolean = { find(plan) { case s: ShuffleExchangeLike => -@@ -2168,6 +2199,9 @@ class AdaptiveQueryExecSuite +@@ -2168,6 +2208,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 @@ -2602,7 +2670,7 @@ index 188a28ff1c0..3cacc4efbdf 100644 }.size == 1) ds.collect() val plan = ds.queryExecution.executedPlan -@@ -2176,6 +2210,9 @@ class AdaptiveQueryExecSuite +@@ -2176,6 +2219,9 @@ class AdaptiveQueryExecSuite }.isEmpty) assert(collect(plan) { case s: ShuffleExchangeExec if s.shuffleOrigin == origin && s.numPartitions == 2 => s @@ -2612,7 +2680,7 @@ index 188a28ff1c0..3cacc4efbdf 100644 }.size == 1) checkAnswer(ds, testData) } -@@ -2340,7 +2377,8 @@ class AdaptiveQueryExecSuite +@@ -2340,7 +2386,8 @@ class AdaptiveQueryExecSuite } } @@ -2622,7 +2690,7 @@ index 188a28ff1c0..3cacc4efbdf 100644 withTempView("t1", "t2") { def checkJoinStrategy(shouldShuffleHashJoin: Boolean): Unit = { Seq("100", "100000").foreach { size => -@@ -2426,7 +2464,8 @@ class AdaptiveQueryExecSuite +@@ -2426,7 +2473,8 @@ class AdaptiveQueryExecSuite } } @@ -2632,7 +2700,7 @@ index 188a28ff1c0..3cacc4efbdf 100644 withTempView("v") { withSQLConf( SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", -@@ -2525,7 +2564,7 @@ class AdaptiveQueryExecSuite +@@ -2525,7 +2573,7 @@ class AdaptiveQueryExecSuite runAdaptiveAndVerifyResult(s"SELECT $repartition key1 FROM skewData1 " + s"JOIN skewData2 ON key1 = key2 GROUP BY key1") val shuffles1 = collect(adaptive1) { @@ -2641,7 +2709,7 @@ index 188a28ff1c0..3cacc4efbdf 100644 } assert(shuffles1.size == 3) // shuffles1.head is the top-level shuffle under the Aggregate operator -@@ -2538,7 +2577,7 @@ class AdaptiveQueryExecSuite +@@ -2538,7 +2586,7 @@ class AdaptiveQueryExecSuite runAdaptiveAndVerifyResult(s"SELECT $repartition key1 FROM skewData1 " + s"JOIN skewData2 ON key1 = key2") val shuffles2 = collect(adaptive2) { @@ -2650,7 +2718,7 @@ index 188a28ff1c0..3cacc4efbdf 100644 } if (hasRequiredDistribution) { assert(shuffles2.size == 3) -@@ -2572,7 +2611,8 @@ class AdaptiveQueryExecSuite +@@ -2572,7 +2620,8 @@ class AdaptiveQueryExecSuite } } @@ -2660,7 +2728,7 @@ index 188a28ff1c0..3cacc4efbdf 100644 CostEvaluator.instantiate( classOf[SimpleShuffleSortCostEvaluator].getCanonicalName, spark.sparkContext.getConf) intercept[IllegalArgumentException] { -@@ -2738,6 +2778,7 @@ class AdaptiveQueryExecSuite +@@ -2738,6 +2787,7 @@ class AdaptiveQueryExecSuite val (_, adaptive) = runAdaptiveAndVerifyResult(query) assert(adaptive.collect { case sort: SortExec => sort @@ -2668,7 +2736,7 @@ index 188a28ff1c0..3cacc4efbdf 100644 }.size == 1) val read = collect(adaptive) { case read: AQEShuffleReadExec => read -@@ -2755,7 +2796,8 @@ class AdaptiveQueryExecSuite +@@ -2755,7 +2805,8 @@ class AdaptiveQueryExecSuite } } @@ -2678,7 +2746,7 @@ index 188a28ff1c0..3cacc4efbdf 100644 withTempView("v") { withSQLConf( SQLConf.ADAPTIVE_OPTIMIZE_SKEWS_IN_REBALANCE_PARTITIONS_ENABLED.key -> "true", -@@ -2867,7 +2909,7 @@ class AdaptiveQueryExecSuite +@@ -2867,7 +2918,7 @@ class AdaptiveQueryExecSuite runAdaptiveAndVerifyResult("SELECT key1 FROM skewData1 JOIN skewData2 ON key1 = key2 " + "JOIN skewData3 ON value2 = value3") val shuffles1 = collect(adaptive1) { @@ -2687,7 +2755,7 @@ index 188a28ff1c0..3cacc4efbdf 100644 } assert(shuffles1.size == 4) val smj1 = findTopLevelSortMergeJoin(adaptive1) -@@ -2878,7 +2920,7 @@ class AdaptiveQueryExecSuite +@@ -2878,7 +2929,7 @@ class AdaptiveQueryExecSuite runAdaptiveAndVerifyResult("SELECT key1 FROM skewData1 JOIN skewData2 ON key1 = key2 " + "JOIN skewData3 ON value1 = value3") val shuffles2 = collect(adaptive2) { @@ -2696,7 +2764,7 @@ index 188a28ff1c0..3cacc4efbdf 100644 } assert(shuffles2.size == 4) val smj2 = findTopLevelSortMergeJoin(adaptive2) -@@ -3136,6 +3178,7 @@ class AdaptiveQueryExecSuite +@@ -3136,6 +3187,7 @@ class AdaptiveQueryExecSuite }.size == (if (firstAccess) 1 else 0)) assert(collect(initialExecutedPlan) { case s: SortExec => s @@ -2704,7 +2772,7 @@ index 188a28ff1c0..3cacc4efbdf 100644 }.size == (if (firstAccess) 2 else 0)) assert(collect(initialExecutedPlan) { case i: InMemoryTableScanLike => i -@@ -3148,6 +3191,7 @@ class AdaptiveQueryExecSuite +@@ -3148,6 +3200,7 @@ class AdaptiveQueryExecSuite }.isEmpty) assert(collect(finalExecutedPlan) { case s: SortExec => s @@ -2712,7 +2780,7 @@ index 188a28ff1c0..3cacc4efbdf 100644 }.isEmpty) assert(collect(initialExecutedPlan) { case i: InMemoryTableScanLike => i -@@ -3229,7 +3273,8 @@ class AdaptiveQueryExecSuite +@@ -3229,7 +3282,8 @@ class AdaptiveQueryExecSuite } } @@ -2722,7 +2790,7 @@ index 188a28ff1c0..3cacc4efbdf 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")) -@@ -3319,7 +3364,8 @@ class AdaptiveQueryExecSuite +@@ -3319,7 +3373,8 @@ class AdaptiveQueryExecSuite val plan = df.queryExecution.executedPlan.asInstanceOf[AdaptiveSparkPlanExec] assert(plan.inputPlan.isInstanceOf[TakeOrderedAndProjectExec]) diff --git a/docs/source/user-guide/latest/compatibility/operators.md b/docs/source/user-guide/latest/compatibility/operators.md index e039df23b76..1e1103955b7 100644 --- a/docs/source/user-guide/latest/compatibility/operators.md +++ b/docs/source/user-guide/latest/compatibility/operators.md @@ -19,6 +19,19 @@ under the License. # Operator Compatibility +## Empty Relations + +On Spark 4.0 and later, Comet supports `EmptyRelationExec` as a native input. It is enabled by +default and can be disabled with `spark.comet.exec.emptyRelation.enabled=false`. The operator +preserves Spark's output attributes and zero partitions; the eliminated logical subtree is not +executed. + +Supported parent joins and aggregates remain eligible for native execution. Global aggregates +still return one row (`COUNT = 0`, `SUM = NULL`), and grouped aggregates return no rows. Independent +operator restrictions and aggregate buffer compatibility checks still apply. +Parquet writes whose input plans contain an empty relation use Spark's writer to preserve +readable empty output files and their schema metadata. + ## Sampling Comet runs `SampleExec` natively when sampling is performed without replacement, which covers diff --git a/docs/source/user-guide/latest/operators.md b/docs/source/user-guide/latest/operators.md index 3a18d86606d..5f35e262d71 100644 --- a/docs/source/user-guide/latest/operators.md +++ b/docs/source/user-guide/latest/operators.md @@ -54,6 +54,7 @@ omitted from the tables below and may be reconsidered based on demand: | `FileSourceScanExec` | ✅ | Parquet only. Some types and configurations fall back. See [Parquet Scan Compatibility](compatibility/scans.md). | | `BatchScanExec` | ✅ | Parquet, Apache Iceberg Parquet, and CSV (native) scans. See [Parquet Scan Compatibility](compatibility/scans.md) and the [Iceberg Guide](iceberg.md). | | `LocalTableScanExec` | ⚠️ | Disabled by default; there is no acceleration advantage and this operator is typically only used in test code. Can be opted into via config ([#4393](https://github.com/apache/datafusion-comet/pull/4393)). | +| `EmptyRelationExec` | ✅ | Spark 4.0 and later. See [Empty Relations](compatibility/operators.md#empty-relations) for native-input support and writer fallback. | | `InMemoryTableScanExec` | 🔜 | Cached / in-memory table scans fall back today. | ## Projection and filtering diff --git a/spark/src/main/scala/org/apache/comet/CometConf.scala b/spark/src/main/scala/org/apache/comet/CometConf.scala index 43f030a7d9d..6e7401c8568 100644 --- a/spark/src/main/scala/org/apache/comet/CometConf.scala +++ b/spark/src/main/scala/org/apache/comet/CometConf.scala @@ -255,6 +255,8 @@ object CometConf extends ShimCometConf { createExecEnabledConfig("takeOrderedAndProject", defaultValue = true) val COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED: ConfigEntry[Boolean] = createExecEnabledConfig("localTableScan", defaultValue = false) + val COMET_EXEC_EMPTY_RELATION_ENABLED: ConfigEntry[Boolean] = + createExecEnabledConfig("emptyRelation", defaultValue = true) val COMET_EXEC_SAMPLE_ENABLED: ConfigEntry[Boolean] = createExecEnabledConfig("sample", defaultValue = true) diff --git a/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala b/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala index 58eab77e3b5..099722d93e9 100644 --- a/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala +++ b/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala @@ -31,6 +31,7 @@ import org.apache.spark.sql.catalyst.util.sideBySide import org.apache.spark.sql.comet._ import org.apache.spark.sql.comet.execution.arrow.ArrowCachedBatchSerializer import org.apache.spark.sql.comet.execution.shuffle.{CometColumnarShuffle, CometNativeShuffle, CometShuffleExchangeExec} +import org.apache.spark.sql.comet.shims.ShimCometEmptyRelation import org.apache.spark.sql.comet.util.Utils import org.apache.spark.sql.execution._ import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanExec, AQEShuffleReadExec, BroadcastQueryStageExec, QueryStageExec, ShuffleQueryStageExec} @@ -91,6 +92,8 @@ object CometExecRule { classOf[InMemoryTableScanExec] -> CometInMemoryTableScanExec, classOf[SampleExec] -> CometSampleExec, classOf[WindowExec] -> CometWindowExec) ++ + // EmptyRelationExec was introduced in Spark 4.0. + ShimCometEmptyRelation.emptyRelationClass.map(_ -> CometEmptyRelationExec) ++ // WindowGroupLimitExec exists only on Spark 3.5+; the shim returns None on 3.4. ShimCometWindowGroupLimit.windowGroupLimitClass.map(_ -> CometWindowGroupLimitExec) diff --git a/spark/src/main/scala/org/apache/comet/serde/operator/CometDataWritingCommand.scala b/spark/src/main/scala/org/apache/comet/serde/operator/CometDataWritingCommand.scala index aec4d71198e..c717e28d86c 100644 --- a/spark/src/main/scala/org/apache/comet/serde/operator/CometDataWritingCommand.scala +++ b/spark/src/main/scala/org/apache/comet/serde/operator/CometDataWritingCommand.scala @@ -26,10 +26,13 @@ import scala.jdk.CollectionConverters._ import org.apache.parquet.hadoop.ParquetOutputFormat import org.apache.spark.SparkException -import org.apache.spark.sql.comet.{CometNativeExec, CometNativeWriteExec} +import org.apache.spark.sql.comet.{CometEmptyRelationExec, CometNativeExec, CometNativeWriteExec, CometScanWrapper} +import org.apache.spark.sql.execution.SparkPlan +import org.apache.spark.sql.execution.adaptive.QueryStageExec import org.apache.spark.sql.execution.command.DataWritingCommandExec import org.apache.spark.sql.execution.datasources.{InsertIntoHadoopFsRelationCommand, WriteFilesExec} import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat +import org.apache.spark.sql.execution.exchange.ReusedExchangeExec import org.apache.spark.sql.internal.SQLConf import org.apache.comet.{CometConf, ConfigEntry} @@ -59,6 +62,16 @@ object CometDataWritingCommand extends CometOperatorSerde[DataWritingCommandExec case cmd: InsertIntoHadoopFsRelationCommand => cmd.fileFormat match { case _: ParquetFileFormat => + // AQE can replace the write input with a zero-partition empty relation. Keep + // Spark's writer, which creates an empty task to preserve the output file schema. + // The native writer only maps existing partitions; see #5303. This guard is + // conservative: an empty relation below an exchange can have nonzero partitions + // at the write input. Revisit the guard when native empty-file handling is fixed. + if (hasEmptyRelationInput(op.child)) { + return Unsupported(Some( + "Parquet writes with empty-relation inputs require Spark's empty-file handling")) + } + if (!cmd.outputPath.toString.startsWith("file:") && !cmd.outputPath.toString .startsWith("hdfs:")) { return Unsupported(Some("Supported output filesystems: local, HDFS")) @@ -86,6 +99,14 @@ object CometDataWritingCommand extends CometOperatorSerde[DataWritingCommandExec } } + private def hasEmptyRelationInput(plan: SparkPlan): Boolean = plan match { + case _: CometEmptyRelationExec => true + case wrapper: CometScanWrapper => hasEmptyRelationInput(wrapper.originalPlan) + case stage: QueryStageExec => hasEmptyRelationInput(stage.plan) + case reused: ReusedExchangeExec => hasEmptyRelationInput(reused.child) + case _ => plan.children.exists(hasEmptyRelationInput) + } + override def convert( op: DataWritingCommandExec, builder: Operator.Builder, diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/CometEmptyRelationExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/CometEmptyRelationExec.scala new file mode 100644 index 00000000000..3a8610d146e --- /dev/null +++ b/spark/src/main/scala/org/apache/spark/sql/comet/CometEmptyRelationExec.scala @@ -0,0 +1,66 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.spark.sql.comet + +import scala.reflect.ClassTag + +import org.apache.arrow.memory.BufferAllocator +import org.apache.arrow.vector.ipc.ArrowReader +import org.apache.spark.rdd.RDD +import org.apache.spark.sql.catalyst.expressions.Attribute +import org.apache.spark.sql.catalyst.plans.QueryPlan +import org.apache.spark.sql.comet.execution.arrow.CometNativeArrowSource +import org.apache.spark.sql.execution.{LeafExecNode, SparkPlan} + +import org.apache.comet.{CometConf, ConfigEntry} +import org.apache.comet.serde.OperatorOuterClass.Operator +import org.apache.comet.serde.operator.CometSink + +/** + * An empty native input. Spark's eliminated logical subtree is explanation data only; neither it + * nor an Arrow reader needs to run. Preserve the zero partitions of EmptyRelationExec so Spark's + * exchanges continue to control aggregate and join partitioning. + */ +case class CometEmptyRelationExec(originalPlan: SparkPlan, override val output: Seq[Attribute]) + extends CometExec + with LeafExecNode + with CometNativeArrowSource { + + // Render Spark's preserved logical subtree without adding an executable child. + override def innerChildren: Seq[QueryPlan[_]] = Seq(originalPlan) + + override protected def mapToReaders[T: ClassTag]( + consume: (String, BufferAllocator => ArrowReader) => Iterator[T]): RDD[T] = + sparkContext.emptyRDD[T] + + override def doCanonicalize(): SparkPlan = { + val canonical = originalPlan.canonicalized + copy(originalPlan = canonical, output = canonical.output) + } +} + +object CometEmptyRelationExec extends CometSink[SparkPlan] { + + override def enabledConfig: Option[ConfigEntry[Boolean]] = + Some(CometConf.COMET_EXEC_EMPTY_RELATION_ENABLED) + + override def createExec(nativeOp: Operator, op: SparkPlan): CometNativeExec = + CometScanWrapper(nativeOp, CometEmptyRelationExec(op, op.output)) +} 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 e71603d4809..93d942a3af0 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 @@ -861,8 +861,7 @@ abstract class CometNativeExec extends CometExec { case _: CometScanExec | _: CometBatchScanExec | _: QueryStageExec | _: AQEShuffleReadExec | _: CometShuffleExchangeExec | _: CometUnionExec | _: CometTakeOrderedAndProjectExec | _: CometCoalesceExec | _: ReusedExchangeExec | _: CometBroadcastExchangeExec | - _: CometSparkToColumnarExec | _: CometLocalTableScanExec | - _: CometInMemoryTableScanExec => + _: CometNativeArrowSource | _: CometInMemoryTableScanExec => func(plan) case _: CometPlan => // Other Comet operators, continue to traverse the tree. diff --git a/spark/src/main/spark-3.x/org/apache/spark/sql/comet/shims/ShimCometEmptyRelation.scala b/spark/src/main/spark-3.x/org/apache/spark/sql/comet/shims/ShimCometEmptyRelation.scala new file mode 100644 index 00000000000..e6a132d984a --- /dev/null +++ b/spark/src/main/spark-3.x/org/apache/spark/sql/comet/shims/ShimCometEmptyRelation.scala @@ -0,0 +1,28 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.spark.sql.comet.shims + +import org.apache.spark.sql.execution.SparkPlan + +/** EmptyRelationExec is available starting in Spark 4.0. */ +object ShimCometEmptyRelation { + + def emptyRelationClass: Option[Class[_ <: SparkPlan]] = None +} diff --git a/spark/src/main/spark-4.x/org/apache/spark/sql/comet/shims/ShimCometEmptyRelation.scala b/spark/src/main/spark-4.x/org/apache/spark/sql/comet/shims/ShimCometEmptyRelation.scala new file mode 100644 index 00000000000..43e7ff7a833 --- /dev/null +++ b/spark/src/main/spark-4.x/org/apache/spark/sql/comet/shims/ShimCometEmptyRelation.scala @@ -0,0 +1,28 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.spark.sql.comet.shims + +import org.apache.spark.sql.execution.{EmptyRelationExec, SparkPlan} + +/** EmptyRelationExec is available starting in Spark 4.0. */ +object ShimCometEmptyRelation { + + def emptyRelationClass: Option[Class[_ <: SparkPlan]] = Some(classOf[EmptyRelationExec]) +} diff --git a/spark/src/test/scala/org/apache/comet/parquet/CometParquetWriterSuite.scala b/spark/src/test/scala/org/apache/comet/parquet/CometParquetWriterSuite.scala index eef77d88246..75520a035d9 100644 --- a/spark/src/test/scala/org/apache/comet/parquet/CometParquetWriterSuite.scala +++ b/spark/src/test/scala/org/apache/comet/parquet/CometParquetWriterSuite.scala @@ -29,9 +29,9 @@ import org.apache.parquet.hadoop.ParquetFileReader import org.apache.parquet.hadoop.metadata.CompressionCodecName import org.apache.parquet.hadoop.util.HadoopInputFile import org.apache.parquet.schema.{MessageType, Type} -import org.apache.spark.sql.{AnalysisException, CometTestBase, DataFrame, Row, SaveMode} +import org.apache.spark.sql.{AnalysisException, DataFrame, Row, SaveMode} import org.apache.spark.sql.comet.{CometBatchScanExec, CometNativeScanExec, CometNativeWriteExec, CometScanExec} -import org.apache.spark.sql.execution.{FileSourceScanExec, QueryExecution, SparkPlan} +import org.apache.spark.sql.execution.{FileSourceScanExec, SparkPlan} import org.apache.spark.sql.execution.command.DataWritingCommandExec import org.apache.spark.sql.functions.{array, map, struct, when} import org.apache.spark.sql.internal.SQLConf @@ -41,7 +41,7 @@ import org.apache.comet.CometConf import org.apache.comet.CometSparkSessionExtensions.isSpark35Plus import org.apache.comet.testing.{DataGenOptions, FuzzDataGenerator, SchemaGenOptions} -class CometParquetWriterSuite extends CometTestBase { +class CometParquetWriterSuite extends CometParquetWriterTestBase { import testImplicits._ @@ -881,14 +881,6 @@ class CometParquetWriterSuite extends CometTestBase { inputPath } - private def withNativeWriter(f: => Unit): Unit = { - withSQLConf( - CometConf.COMET_NATIVE_PARQUET_WRITE_ENABLED.key -> "true", - CometConf.COMET_OPERATOR_DATA_WRITING_COMMAND_ALLOW_INCOMPAT.key -> "true", - CometConf.COMET_EXEC_ENABLED.key -> "true", - SQLConf.SESSION_LOCAL_TIMEZONE.key -> "America/Halifax")(f) - } - // Persist `df` to `sourcePath` with Comet disabled and return a DataFrame that reads it back, // so the source plan is a Comet scan (satisfying CometExecRule.requiresNativeChildren). private def materializeAsCometSource(df: DataFrame, sourcePath: String): DataFrame = { @@ -911,58 +903,6 @@ class CometParquetWriterSuite extends CometTestBase { } } - /** - * Captures the execution plan during a write operation. - * - * @param writeOp - * The write operation to execute (takes output path as parameter) - * @param outputPath - * The path to write to - * @return - * The captured execution plan - */ - private def captureWritePlan(writeOp: String => Unit, outputPath: String): SparkPlan = { - var capturedPlan: Option[QueryExecution] = None - - val listener = new org.apache.spark.sql.util.QueryExecutionListener { - override def onSuccess(funcName: String, qe: QueryExecution, durationNs: Long): Unit = { - if (funcName == "save" || funcName.contains("command")) { - capturedPlan = Some(qe) - } - } - - override def onFailure( - funcName: String, - qe: QueryExecution, - exception: Exception): Unit = {} - } - - spark.listenerManager.register(listener) - - try { - writeOp(outputPath) - - // Wait for listener to be called with timeout - val maxWaitTimeMs = 15000 - val checkIntervalMs = 100 - val maxIterations = maxWaitTimeMs / checkIntervalMs - var iterations = 0 - - while (capturedPlan.isEmpty && iterations < maxIterations) { - Thread.sleep(checkIntervalMs) - iterations += 1 - } - - assert( - capturedPlan.isDefined, - s"Listener was not called within ${maxWaitTimeMs}ms - no execution plan captured") - - stripAQEPlan(capturedPlan.get.executedPlan) - } finally { - spark.listenerManager.unregister(listener) - } - } - private def assertHasCometNativeWriteExec(plan: SparkPlan): Unit = { var nativeWriteCount = 0 plan.foreach { @@ -982,22 +922,6 @@ class CometParquetWriterSuite extends CometTestBase { s"Expected exactly one CometNativeWriteExec in the plan, but found $nativeWriteCount:\n${plan.treeString}") } - private def assertNoCometNativeWriteExec(plan: SparkPlan): Unit = { - val hasNativeWrite = plan.exists { - case _: CometNativeWriteExec => true - case d: DataWritingCommandExec => - d.child.exists { - case _: CometNativeWriteExec => true - case _ => false - } - case _ => false - } - - assert( - !hasNativeWrite, - s"Expected no CometNativeWriteExec in the plan, but found one:\n${plan.treeString}") - } - private def writeWithCometNativeWriteExec( inputPath: String, outputPath: String, diff --git a/spark/src/test/scala/org/apache/comet/parquet/CometParquetWriterTestBase.scala b/spark/src/test/scala/org/apache/comet/parquet/CometParquetWriterTestBase.scala new file mode 100644 index 00000000000..96b543e3761 --- /dev/null +++ b/spark/src/test/scala/org/apache/comet/parquet/CometParquetWriterTestBase.scala @@ -0,0 +1,107 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.comet.parquet + +import org.apache.spark.sql.CometTestBase +import org.apache.spark.sql.comet.CometNativeWriteExec +import org.apache.spark.sql.execution.{QueryExecution, SparkPlan} +import org.apache.spark.sql.execution.command.DataWritingCommandExec +import org.apache.spark.sql.internal.SQLConf + +import org.apache.comet.CometConf + +abstract class CometParquetWriterTestBase extends CometTestBase { + + protected def withNativeWriter(f: => Unit): Unit = { + withSQLConf( + CometConf.COMET_NATIVE_PARQUET_WRITE_ENABLED.key -> "true", + CometConf.COMET_OPERATOR_DATA_WRITING_COMMAND_ALLOW_INCOMPAT.key -> "true", + CometConf.COMET_EXEC_ENABLED.key -> "true", + SQLConf.SESSION_LOCAL_TIMEZONE.key -> "America/Halifax")(f) + } + + /** + * Captures the execution plan during a write operation. + * + * @param writeOp + * The write operation to execute (takes output path as parameter) + * @param outputPath + * The path to write to + * @return + * The captured execution plan + */ + protected def captureWritePlan(writeOp: String => Unit, outputPath: String): SparkPlan = { + var capturedPlan: Option[QueryExecution] = None + + val listener = new org.apache.spark.sql.util.QueryExecutionListener { + override def onSuccess(funcName: String, qe: QueryExecution, durationNs: Long): Unit = { + if (funcName == "save" || funcName.contains("command")) { + capturedPlan = Some(qe) + } + } + + override def onFailure( + funcName: String, + qe: QueryExecution, + exception: Exception): Unit = {} + } + + spark.listenerManager.register(listener) + + try { + writeOp(outputPath) + + // Wait for listener to be called with timeout + val maxWaitTimeMs = 15000 + val checkIntervalMs = 100 + val maxIterations = maxWaitTimeMs / checkIntervalMs + var iterations = 0 + + while (capturedPlan.isEmpty && iterations < maxIterations) { + Thread.sleep(checkIntervalMs) + iterations += 1 + } + + assert( + capturedPlan.isDefined, + s"Listener was not called within ${maxWaitTimeMs}ms - no execution plan captured") + + stripAQEPlan(capturedPlan.get.executedPlan) + } finally { + spark.listenerManager.unregister(listener) + } + } + + protected def assertNoCometNativeWriteExec(plan: SparkPlan): Unit = { + val hasNativeWrite = plan.exists { + case _: CometNativeWriteExec => true + case d: DataWritingCommandExec => + d.child.exists { + case _: CometNativeWriteExec => true + case _ => false + } + case _ => false + } + + assert( + !hasNativeWrite, + s"Expected no CometNativeWriteExec in the plan, but found one:\n${plan.treeString}") + } +} diff --git a/spark/src/test/spark-4.x/org/apache/comet/exec/CometEmptyRelationExecSuite.scala b/spark/src/test/spark-4.x/org/apache/comet/exec/CometEmptyRelationExecSuite.scala new file mode 100644 index 00000000000..c86addc09c8 --- /dev/null +++ b/spark/src/test/spark-4.x/org/apache/comet/exec/CometEmptyRelationExecSuite.scala @@ -0,0 +1,67 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.comet.exec + +import org.apache.spark.sql.{CometTestBase, Row} +import org.apache.spark.sql.comet.{CometBroadcastHashJoinExec, CometEmptyRelationExec, CometHashAggregateExec} +import org.apache.spark.sql.execution.joins.BroadcastHashJoinExec +import org.apache.spark.sql.internal.SQLConf + +import org.apache.comet.CometConf + +class CometEmptyRelationExecSuite extends CometTestBase { + + test("EmptyRelationExec discovered by AQE feeds native aggregates and Spark existence joins") { + withSQLConf( + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", + SQLConf.SHUFFLE_PARTITIONS.key -> "2", + CometConf.COMET_SPARK_TO_ARROW_ENABLED.key -> "false", + CometConf.COMET_SHUFFLE_CONVERT_FROM_SPARK_PLAN_ENABLED.key -> "false") { + withParquetTable(Seq((1, 2), (2, 3)), "aqe_empty_input") { + // Retain Spark's Range input so AQE can infer emptiness from its completed shuffle. + // After replacement, the supported parents must start native execution at the new leaf. + val empty = "(SELECT CAST(id % 2 AS INT) AS k, sum(id) AS v FROM range(0, 10, 1, 2) " + + "WHERE id < 0 GROUP BY id % 2)" + val aggregate = s"SELECT count(*), sum(v) FROM $empty" + val (_, aggregatePlan) = checkSparkAnswer(aggregate) + checkAnswer(sql(aggregate), Seq(Row(0L, null))) + assert( + collect(aggregatePlan) { case e: CometEmptyRelationExec => e }.nonEmpty, + aggregatePlan.toString) + assert( + collect(aggregatePlan) { case a: CometHashAggregateExec => a }.nonEmpty, + aggregatePlan.toString) + + // Existence joins retain Spark's fallback and preserve probe rows with false markers. + val existence = "SELECT l._1, EXISTS (SELECT /*+ BROADCAST(r) */ 1 FROM " + + s"$empty r WHERE r.k = l._1) AS matched FROM aqe_empty_input l" + val (_, joinPlan) = checkSparkAnswer(existence) + checkAnswer(sql(existence), Seq(Row(1, false), Row(2, false))) + assert( + collect(joinPlan) { case e: CometEmptyRelationExec => e }.nonEmpty, + joinPlan.toString) + assert( + collect(joinPlan) { case j: BroadcastHashJoinExec => j }.nonEmpty, + joinPlan.toString) + assert(collect(joinPlan) { case j: CometBroadcastHashJoinExec => j }.isEmpty) + } + } + } +} diff --git a/spark/src/test/spark-4.x/org/apache/comet/parquet/CometEmptyRelationParquetWriterSuite.scala b/spark/src/test/spark-4.x/org/apache/comet/parquet/CometEmptyRelationParquetWriterSuite.scala new file mode 100644 index 00000000000..b5580f80ed0 --- /dev/null +++ b/spark/src/test/spark-4.x/org/apache/comet/parquet/CometEmptyRelationParquetWriterSuite.scala @@ -0,0 +1,76 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.comet.parquet + +import java.io.File + +import org.apache.spark.sql.SaveMode +import org.apache.spark.sql.comet.CometEmptyRelationExec +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.types.StructType + +import org.apache.comet.CometConf + +class CometEmptyRelationParquetWriterSuite extends CometParquetWriterTestBase { + + test("EmptyRelationExec keeps AQE empty parquet output readable") { + withSQLConf( + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", + SQLConf.SHUFFLE_PARTITIONS.key -> "2", + "spark.sql.optimizer.plannedWrite.enabled" -> "true", + CometConf.COMET_SPARK_TO_ARROW_ENABLED.key -> "false", + CometConf.COMET_SHUFFLE_CONVERT_FROM_SPARK_PLAN_ENABLED.key -> "false") { + withNativeWriter { + withTempPath { dir => + for (nativeEmpty <- Seq(false, true)) { + withSQLConf(CometConf.COMET_EXEC_EMPTY_RELATION_ENABLED.key -> nativeEmpty.toString) { + val outputPath = new File(dir, s"empty_$nativeEmpty.parquet").getAbsolutePath + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + sql("SELECT CAST(1 AS INT) AS k, CAST(2 AS BIGINT) AS v").write.parquet( + outputPath) + } + // AQE replaces the completed empty shuffle with EmptyRelationExec and reruns + // preparation of the whole write command, including native-writer eligibility. + val empty = sql( + "SELECT CAST(id % 2 AS INT) AS k, sum(id) AS v " + + "FROM range(0, 10, 1, 2) WHERE id < 0 GROUP BY id % 2") + val plan = captureWritePlan( + path => empty.write.mode(SaveMode.Overwrite).parquet(path), + outputPath) + withClue(s"nativeEmpty=$nativeEmpty\n$plan\n") { + // Infer the schema from the output files, as a normal downstream reader does. + // Supplying an explicit schema would hide the absence of a Parquet data file. + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + val readback = spark.read.parquet(outputPath) + assert(readback.schema == StructType(empty.schema.map(_.copy(nullable = true)))) + assert(readback.collect().isEmpty) + } + assertNoCometNativeWriteExec(plan) + if (nativeEmpty) { + assert(collect(plan) { case e: CometEmptyRelationExec => e }.nonEmpty) + } + } + } + } + } + } + } + } +} diff --git a/spark/src/test/spark-4.x/org/apache/comet/rules/CometEmptyRelationExecRuleSuite.scala b/spark/src/test/spark-4.x/org/apache/comet/rules/CometEmptyRelationExecRuleSuite.scala new file mode 100644 index 00000000000..661792b0d6b --- /dev/null +++ b/spark/src/test/spark-4.x/org/apache/comet/rules/CometEmptyRelationExecRuleSuite.scala @@ -0,0 +1,290 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.comet.rules + +import org.apache.spark.sql.{CometTestBase, Row} +import org.apache.spark.sql.catalyst.{CatalystTypeConverters, InternalRow} +import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference} +import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight} +import org.apache.spark.sql.catalyst.plans.{ExistenceJoin, FullOuter, Inner, LeftAnti, LeftOuter, LeftSemi, RightOuter} +import org.apache.spark.sql.catalyst.plans.logical.{LocalRelation, Project} +import org.apache.spark.sql.comet._ +import org.apache.spark.sql.execution.{ApplyColumnarRulesAndInsertTransitions, CoalesceExec, EmptyRelationExec, LocalTableScanExec, SparkPlan} +import org.apache.spark.sql.execution.adaptive.BroadcastQueryStageExec +import org.apache.spark.sql.execution.exchange.{EnsureRequirements, ReusedExchangeExec} +import org.apache.spark.sql.execution.joins.{BroadcastHashJoinExec, ShuffledHashJoinExec, SortMergeJoinExec} +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.types.{ArrayType, BooleanType, DecimalType, IntegerType, StructType} + +import org.apache.comet.CometConf + +class CometEmptyRelationExecRuleSuite extends CometTestBase { + + import testImplicits._ + + private def emptyRelation(output: Seq[Attribute]): SparkPlan = { + EmptyRelationExec(LocalRelation(output)) + } + + private def localTableScan(output: Seq[Attribute]): LocalTableScanExec = + spark.sessionState.planner.plan(LocalRelation(output)).next().asInstanceOf[LocalTableScanExec] + + private def prepareEmptyInputPlan(plan: SparkPlan, native: Boolean): SparkPlan = { + val required = EnsureRequirements().apply(plan) + val converted = if (native) CometExecRule(spark).apply(required) else required + ApplyColumnarRulesAndInsertTransitions(Seq.empty, false).apply(converted) + } + + private def collectEmptyInputPlan(plan: SparkPlan): Seq[Row] = { + val toRow = CatalystTypeConverters.createToScalaConverter(plan.schema) + plan.executeCollect().map(row => toRow(row).asInstanceOf[Row]).toSeq + } + + private def withCometDisabled(plan: => SparkPlan): SparkPlan = { + var result: SparkPlan = null + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + result = plan + } + result + } + + test("EmptyRelationExec preserves attributes and zero partitions") { + val attributes = Seq( + AttributeReference("id", IntegerType, nullable = false)(), + AttributeReference("amount", DecimalType(18, 4), nullable = true)(), + AttributeReference("nested", ArrayType(IntegerType, containsNull = false))()) + val original = emptyRelation(attributes) + val converted = CometExecRule(spark).apply(original).asInstanceOf[CometEmptyRelationExec] + assert(converted.output == original.output) + assert(converted.schema == original.schema) + assert(converted.children.isEmpty) + assert(converted.executeColumnar().getNumPartitions == 0) + assert(converted.doExecuteAsArrowStream().getNumPartitions == 0) + assert(converted.executeCollect().isEmpty) + assert( + converted.canonicalized == + CometExecRule(spark).apply(emptyRelation(attributes.map(_.newInstance()))).canonicalized) + + withSQLConf(CometConf.COMET_EXEC_EMPTY_RELATION_ENABLED.key -> "false") { + assert(CometExecRule(spark).apply(emptyRelation(attributes)).getClass == original.getClass) + } + val unsupported = emptyRelation(Seq(AttributeReference("empty_struct", StructType(Nil))())) + assert(CometExecRule(spark).apply(unsupported).getClass == unsupported.getClass) + } + + test("EmptyRelationExec preserves the eliminated subtree in explain output") { + val attributes = Seq(AttributeReference("id", IntegerType, nullable = false)()) + val eliminated = Project(attributes, LocalRelation(attributes, Seq(InternalRow(1)))) + val original = EmptyRelationExec(eliminated) + val converted = CometExecRule(spark).apply(original).asInstanceOf[CometEmptyRelationExec] + + val explained = converted.treeString + assert(explained.contains("CometEmptyRelation"), explained) + assert(explained.contains("Project"), explained) + assert(explained.contains("LocalRelation"), explained) + assert(converted.children.isEmpty) + assert(converted.executeCollect().isEmpty) + } + + test("EmptyRelationExec supports global and grouped COUNT and SUM") { + withSQLConf( + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false", + SQLConf.SHUFFLE_PARTITIONS.key -> "2") { + withTempView("empty_aggregate_input") { + Seq((1, 2)).toDF("k", "v").createOrReplaceTempView("empty_aggregate_input") + val queries = Seq( + "SELECT count(*), count(v), sum(v) FROM empty_aggregate_input" -> Seq( + Row(0L, 0L, null)), + "SELECT sum(v) FROM empty_aggregate_input" -> Seq(Row(null)), + "SELECT count(*) FROM empty_aggregate_input" -> Seq(Row(0L)), + "SELECT k, count(*), sum(v) FROM empty_aggregate_input GROUP BY k" -> Seq.empty) + for { + (query, expected) <- queries + singleEmptyPartition <- Seq(false, true) + } { + def original: SparkPlan = withCometDisabled { + sql(query).queryExecution.sparkPlan.transformUp { case leaf: LocalTableScanExec => + val empty = emptyRelation(leaf.output) + // COALESCE(1) turns zero partitions into one empty columnar input stream. + if (singleEmptyPartition) CoalesceExec(1, empty) else empty + } + } + val native = prepareEmptyInputPlan(original, native = true) + withClue(s"$query\n$native") { + assert( + collectEmptyInputPlan(prepareEmptyInputPlan(original, native = false)) == expected) + assert(collectEmptyInputPlan(native) == expected) + assert(native.collect { case e: CometEmptyRelationExec => e }.size == 1) + assert(native.collect { case a: CometHashAggregateExec => a }.size == 2) + assert(native.collect { + case a: org.apache.spark.sql.execution.aggregate.HashAggregateExec => a + }.isEmpty) + } + } + } + } + } + + test("EmptyRelationExec supports empty build and probe hash joins") { + withSQLConf( + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false", + SQLConf.SHUFFLE_PARTITIONS.key -> "2", + CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true") { + for { + broadcast <- Seq(false, true) + joinType <- Seq(Inner, LeftOuter, RightOuter, FullOuter, LeftSemi, LeftAnti) + if !(broadcast && joinType == FullOuter) + (emptyLeft, emptyRight) <- Seq((false, true), (true, false), (true, true)) + } { + def original: SparkPlan = { + val leftKey = AttributeReference("l", IntegerType, nullable = true)() + val rightKey = AttributeReference("r", IntegerType, nullable = true)() + val left = + if (emptyLeft) emptyRelation(Seq(leftKey)) + else + localTableScan(Seq(leftKey)) + .copy(rows = Seq(InternalRow(1), InternalRow(1), InternalRow(null))) + val right = + if (emptyRight) emptyRelation(Seq(rightKey)) + else localTableScan(Seq(rightKey)).copy(rows = Seq(InternalRow(1), InternalRow(null))) + val buildSide = if (joinType == RightOuter) BuildLeft else BuildRight + if (broadcast) { + BroadcastHashJoinExec( + Seq(leftKey), + Seq(rightKey), + joinType, + buildSide, + None, + left, + right) + } else { + ShuffledHashJoinExec( + Seq(leftKey), + Seq(rightKey), + joinType, + buildSide, + None, + left, + right) + } + } + val native = prepareEmptyInputPlan(original, native = true) + withClue(s"$joinType broadcast=$broadcast empty=($emptyLeft,$emptyRight)\n$native") { + val expected = collectEmptyInputPlan(prepareEmptyInputPlan(original, native = false)) + assert( + collectEmptyInputPlan(native).groupBy(identity).map { case (r, rs) => + r -> rs.size + } == + expected.groupBy(identity).map { case (r, rs) => r -> rs.size }) + assert(native.collect { case e: CometEmptyRelationExec => e }.nonEmpty) + val joins = native.collect { + case j: CometBroadcastHashJoinExec => j + case j: CometHashJoinExec => j + } + assert(joins.size == 1) + } + } + } + } + + test("EmptyRelationExec retains incompatible aggregate buffer fallback") { + withSQLConf( + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false", + CometConf.COMET_ENABLE_FINAL_HASH_AGGREGATE.key -> "false") { + def original: SparkPlan = withCometDisabled { + sql("SELECT avg(v) FROM VALUES (CAST(1 AS DECIMAL(38, 0))) AS t(v)").queryExecution.sparkPlan + .transformUp { case leaf: LocalTableScanExec => + emptyRelation(leaf.output) + } + } + val native = prepareEmptyInputPlan(original, native = true) + assert(collectEmptyInputPlan(native) == Seq(Row(null))) + assert(native.collect { case e: CometEmptyRelationExec => e }.size == 1) + assert(native.collect { case a: CometHashAggregateExec => a }.isEmpty) + assert(native.collect { + case a: org.apache.spark.sql.execution.aggregate.HashAggregateExec => a + }.size == 2) + } + } + + test("EmptyRelationExec retains existence sort-merge join fallback") { + withSQLConf( + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false", + SQLConf.SHUFFLE_PARTITIONS.key -> "2", + CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true", + CometConf.COMET_EXEC_SORT_MERGE_JOIN_ENABLED.key -> "true") { + val leftKey = AttributeReference("l", IntegerType, nullable = true)() + val rightKey = AttributeReference("r", IntegerType, nullable = true)() + val left = localTableScan(Seq(leftKey)).copy(rows = Seq(InternalRow(1), InternalRow(null))) + val join = SortMergeJoinExec( + Seq(leftKey), + Seq(rightKey), + ExistenceJoin(AttributeReference("exists", BooleanType, nullable = false)()), + None, + left, + emptyRelation(Seq(rightKey))) + val native = prepareEmptyInputPlan(join, native = true) + assert( + collectEmptyInputPlan(native).sortBy(_.toString) == + Seq(Row(1, false), Row(null, false)).sortBy(_.toString)) + assert(native.collect { case e: CometEmptyRelationExec => e }.size == 1) + assert(native.collect { case j: SortMergeJoinExec => j }.size == 1) + assert(native.collect { case j: CometSortMergeJoinExec => j }.isEmpty) + } + } + + test("EmptyRelationExec supports reused broadcast and AQE broadcast stages") { + withSQLConf( + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false", + CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true") { + val leftKey = AttributeReference("l", IntegerType, nullable = true)() + val rightKey = AttributeReference("r", IntegerType, nullable = true)() + val left = localTableScan(Seq(leftKey)).copy(rows = Seq(InternalRow(1), InternalRow(null))) + def join(right: SparkPlan): SparkPlan = BroadcastHashJoinExec( + Seq(leftKey), + Seq(rightKey), + LeftOuter, + BuildRight, + None, + left, + right) + val first = prepareEmptyInputPlan(join(emptyRelation(Seq(rightKey))), native = true) + val exchange = first.collectFirst { case e: CometBroadcastExchangeExec => e }.get + assert(collectEmptyInputPlan(first) == Seq(Row(1, null), Row(null, null))) + + for (adaptive <- Seq(false, true)) { + val reused = ReusedExchangeExec(exchange.output, exchange) + val native = if (adaptive) { + val input = BroadcastQueryStageExec(0, reused, reused.canonicalized) + prepareEmptyInputPlan(join(input), native = true) + } else { + // Non-AQE reuse runs after native conversion; its consumers already have native plans. + first.transformUp { case _: CometBroadcastExchangeExec => reused } + } + withClue(s"AQE=$adaptive\n$native") { + assert(collectEmptyInputPlan(native) == Seq(Row(1, null), Row(null, null))) + assert(native.collect { case j: CometBroadcastHashJoinExec => j }.size == 1) + assert(collect(native) { case e: ReusedExchangeExec => e }.nonEmpty) + } + } + assert(exchange.metrics("numOutputRows").value == 0) + } + } +}