[GitHub] [spark] JkSelf commented on a change in pull request #31994: [SPARK-34899][SQL] Use origin plan if we can not coalesce shuffle partition
JkSelf commented on a change in pull request #31994: URL: https://github.com/apache/spark/pull/31994#discussion_r603823927 ## File path: sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala ## @@ -1543,4 +1543,27 @@ class AdaptiveQueryExecSuite assert(materializeLogs(0).startsWith("Materialize query stage BroadcastQueryStageExec")) assert(materializeLogs(1).startsWith("Materialize query stage ShuffleQueryStageExec")) } + + test("SPARK-34899: Use origin plan if we can not coalesce shuffle partition") { +def check(ds: Dataset[Row], origin: ShuffleOrigin): Unit = { + ds.collect() + val plan = ds.queryExecution.executedPlan.asInstanceOf[AdaptiveSparkPlanExec].executedPlan + assert(collect(plan) { +case c: CustomShuffleReaderExec => c + }.isEmpty) + assert(collect(plan) { +case s: ShuffleExchangeExec if s.shuffleOrigin == origin && s.numPartitions == 3 => s + }.size == 1) + checkAnswer(ds, testData) +} + +withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", + SQLConf.COALESCE_PARTITIONS_ENABLED.key -> "true", + SQLConf.ADVISORY_PARTITION_SIZE_IN_BYTES.key -> "10", + SQLConf.COALESCE_PARTITIONS_MIN_PARTITION_NUM.key -> "1", + SQLConf.SHUFFLE_PARTITIONS.key -> "3") { + check(testData.repartition(), REPARTITION) Review comment: `CoalesceShufflePartitions` rule depend on the map output statistics to coalesce the shuffle partitions. I think this size is more meaningful here. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] JkSelf commented on a change in pull request #31994: [SPARK-34899][SQL] Use origin plan if we can not coalesce shuffle partition
JkSelf commented on a change in pull request #31994: URL: https://github.com/apache/spark/pull/31994#discussion_r603817326 ## File path: sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala ## @@ -1543,4 +1543,27 @@ class AdaptiveQueryExecSuite assert(materializeLogs(0).startsWith("Materialize query stage BroadcastQueryStageExec")) assert(materializeLogs(1).startsWith("Materialize query stage ShuffleQueryStageExec")) } + + test("SPARK-34899: Use origin plan if we can not coalesce shuffle partition") { +def check(ds: Dataset[Row], origin: ShuffleOrigin): Unit = { + ds.collect() + val plan = ds.queryExecution.executedPlan.asInstanceOf[AdaptiveSparkPlanExec].executedPlan + assert(collect(plan) { +case c: CustomShuffleReaderExec => c + }.isEmpty) + assert(collect(plan) { +case s: ShuffleExchangeExec if s.shuffleOrigin == origin && s.numPartitions == 3 => s + }.size == 1) + checkAnswer(ds, testData) +} + +withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", + SQLConf.COALESCE_PARTITIONS_ENABLED.key -> "true", + SQLConf.ADVISORY_PARTITION_SIZE_IN_BYTES.key -> "10", + SQLConf.COALESCE_PARTITIONS_MIN_PARTITION_NUM.key -> "1", + SQLConf.SHUFFLE_PARTITIONS.key -> "3") { + check(testData.repartition(), REPARTITION) Review comment: The partition size is stored in `ShuffleQueryStageExec#mapStats#bytesByPartitionId`. You can traverse the `bytesByPartitionId `value of each `mapStats `in [validMetrics ](https://github.com/apache/spark/blob/1237124062a541082922c650d1f1766b691f47bb/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/CoalesceShufflePartitions.scala#L59). -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] JkSelf commented on a change in pull request #31994: [SPARK-34899][SQL] Use origin plan if we can not coalesce shuffle partition
JkSelf commented on a change in pull request #31994: URL: https://github.com/apache/spark/pull/31994#discussion_r603801610 ## File path: sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala ## @@ -1543,4 +1543,27 @@ class AdaptiveQueryExecSuite assert(materializeLogs(0).startsWith("Materialize query stage BroadcastQueryStageExec")) assert(materializeLogs(1).startsWith("Materialize query stage ShuffleQueryStageExec")) } + + test("SPARK-34899: Use origin plan if we can not coalesce shuffle partition") { +def check(ds: Dataset[Row], origin: ShuffleOrigin): Unit = { + ds.collect() + val plan = ds.queryExecution.executedPlan.asInstanceOf[AdaptiveSparkPlanExec].executedPlan + assert(collect(plan) { +case c: CustomShuffleReaderExec => c + }.isEmpty) + assert(collect(plan) { +case s: ShuffleExchangeExec if s.shuffleOrigin == origin && s.numPartitions == 3 => s + }.size == 1) + checkAnswer(ds, testData) +} + +withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", + SQLConf.COALESCE_PARTITIONS_ENABLED.key -> "true", + SQLConf.ADVISORY_PARTITION_SIZE_IN_BYTES.key -> "10", + SQLConf.COALESCE_PARTITIONS_MIN_PARTITION_NUM.key -> "1", + SQLConf.SHUFFLE_PARTITIONS.key -> "3") { + check(testData.repartition(), REPARTITION) Review comment: It is better to add the partition size of each partition in the comment. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org