You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2022/08/22 10:04:52 UTC
[GitHub] [spark] ulysses-you opened a new pull request, #37612: [SPARK-39915][SQL] Ensure the output partitioning is user-specified in AQE
ulysses-you opened a new pull request, #37612:
URL: https://github.com/apache/spark/pull/37612
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'core/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
Add `userSpecifiedDistribution` validate in `AdaptiveSparkPlanExec`, so we can check and revert the optimized SparkPlan which does not satisfy the user-specified.
The definition of `userSpecifiedDistribution` is more stricter than the existed `requiredDistribution`. It will through all unary node to find the repartition shuffle.
To approach all user-specified distribution, this pr adds a new `Distribution`: `RandomDistribution` which is only for checking the output partition number. e.g. ensure the output partition number of RoundRobin based repartition.
Enhance the `ValidateSparkPlan` to support check the output partitioning.
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
AQE can not completely respect the user-specified repartition. The main reasons are:
1. the AQE optimzier will convert empty to local relation which does not reserve the partitioning info
2. the machine of AQE `requiredDistribution` only restore the repartition which only support through some unary nodes, and it is not safe to support through all unary nodes to restore repartition.
For example:
The partition number of `spark.range(0).repartition(5).rdd.getNumPartitions` should be 5.
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as the documentation fix.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
yes, ensure the user-specified distribution.
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
If benchmark tests were added, please run the benchmarks in GitHub Actions for the consistent environment, and the instructions could accord to: https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
-->
add tests
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on pull request #37612: [SPARK-39915][SQL] Ensure the output partitioning is user-specified in AQE
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on PR #37612:
URL: https://github.com/apache/spark/pull/37612#issuecomment-1245375874
@ulysses-you can you update the PR description? I think this is ready to merge.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ulysses-you commented on a diff in pull request #37612: [SPARK-39915][SQL] Ensure the output partitioning is user-specified in AQE
Posted by GitBox <gi...@apache.org>.
ulysses-you commented on code in PR #37612:
URL: https://github.com/apache/spark/pull/37612#discussion_r954422378
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/ValidateSparkPlan.scala:
##########
@@ -39,30 +45,41 @@ object ValidateSparkPlan extends Rule[SparkPlan] {
* - BroadcastQueryStage only appears as the immediate child and the build side of a broadcast
* hash join or broadcast nested loop join.
*/
- private def validate(plan: SparkPlan): Unit = plan match {
+ private def validateBroadcastQueryStage(plan: SparkPlan): Unit = plan match {
case b: BroadcastHashJoinExec =>
val (buildPlan, probePlan) = b.buildSide match {
case BuildLeft => (b.left, b.right)
case BuildRight => (b.right, b.left)
}
if (!buildPlan.isInstanceOf[BroadcastQueryStageExec]) {
- validate(buildPlan)
+ validateBroadcastQueryStage(buildPlan)
}
- validate(probePlan)
+ validateBroadcastQueryStage(probePlan)
case b: BroadcastNestedLoopJoinExec =>
val (buildPlan, probePlan) = b.buildSide match {
case BuildLeft => (b.left, b.right)
case BuildRight => (b.right, b.left)
}
if (!buildPlan.isInstanceOf[BroadcastQueryStageExec]) {
- validate(buildPlan)
+ validateBroadcastQueryStage(buildPlan)
}
- validate(probePlan)
+ validateBroadcastQueryStage(probePlan)
case q: BroadcastQueryStageExec => errorOnInvalidBroadcastQueryStage(q)
- case _ => plan.children.foreach(validate)
+ case _ => plan.children.foreach(validateBroadcastQueryStage)
}
private def errorOnInvalidBroadcastQueryStage(plan: SparkPlan): Unit = {
throw InvalidAQEPlanException("Invalid broadcast query stage", plan)
}
+
+ /**
+ * Validate that the output partitioning of plan satisfies the `userSpecifiedDistribution`.
+ */
+ private def validateOutputPartitioning(plan: SparkPlan): Unit = {
+ if (userSpecifiedDistribution.isDefined &&
+ !plan.outputPartitioning.satisfies(userSpecifiedDistribution.get)) {
+ throw InvalidAQEPlanException(s"Output partitioning does not satisfy the specified, " +
Review Comment:
It's a little overkill to validate parititoning again after EnsureRequirement. I removed this part of code.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ulysses-you commented on a diff in pull request #37612: [SPARK-39915][SQL] Ensure the output partitioning is user-specified in AQE
Posted by GitBox <gi...@apache.org>.
ulysses-you commented on code in PR #37612:
URL: https://github.com/apache/spark/pull/37612#discussion_r953555245
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AQEUtils.scala:
##########
@@ -28,16 +28,28 @@ object AQEUtils {
def getRequiredDistribution(p: SparkPlan): Option[Distribution] = p match {
// User-specified repartition is only effective when it's the root node, or under
// Project/Filter/LocalSort/CollectMetrics.
- // Note: we only care about `HashPartitioning` as `EnsureRequirements` can only optimize out
- // user-specified repartition with `HashPartitioning`.
Review Comment:
I only removed the `with HashPartitioning` since we are going to check all partitionings which come from repartition.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a diff in pull request #37612: [SPARK-39915][SQL] Ensure the output partitioning is user-specified in AQE
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #37612:
URL: https://github.com/apache/spark/pull/37612#discussion_r954458141
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AQEUtils.scala:
##########
@@ -28,16 +28,31 @@ object AQEUtils {
def getRequiredDistribution(p: SparkPlan): Option[Distribution] = p match {
// User-specified repartition is only effective when it's the root node, or under
// Project/Filter/LocalSort/CollectMetrics.
- // Note: we only care about `HashPartitioning` as `EnsureRequirements` can only optimize out
- // user-specified repartition with `HashPartitioning`.
- case ShuffleExchangeExec(h: HashPartitioning, _, shuffleOrigin)
+ // Note, here are two cases of how user-specified repartition can be optimized out:
+ // 1. `EnsureRequirements` can only optimize out user-specified repartition with
+ // `HashPartitioning`.
+ // 2. `AQEOptimizer` can optimize out user-specified repartition with all `Partitioning`,
+ // e.g. convert empty to local relation.
Review Comment:
Or, we can still optimize out `Repartition`, but we should assign a `Partitioning` to `LocalRelation`. Ideally a empty data can be any partitioning.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a diff in pull request #37612: [SPARK-39915][SQL] Ensure the output partitioning is user-specified in AQE
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #37612:
URL: https://github.com/apache/spark/pull/37612#discussion_r953531474
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AQEUtils.scala:
##########
@@ -28,16 +28,28 @@ object AQEUtils {
def getRequiredDistribution(p: SparkPlan): Option[Distribution] = p match {
// User-specified repartition is only effective when it's the root node, or under
// Project/Filter/LocalSort/CollectMetrics.
- // Note: we only care about `HashPartitioning` as `EnsureRequirements` can only optimize out
- // user-specified repartition with `HashPartitioning`.
Review Comment:
```
as `EnsureRequirements` can only optimize out user-specified repartition with `HashPartitioning`
```
This is not true any more?
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a diff in pull request #37612: [SPARK-39915][SQL] Ensure the output partitioning is user-specified in AQE
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #37612:
URL: https://github.com/apache/spark/pull/37612#discussion_r954456484
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/InsertAdaptiveSparkPlan.scala:
##########
@@ -51,6 +51,11 @@ case class InsertAdaptiveSparkPlan(
case c: DataWritingCommandExec
if !c.cmd.isInstanceOf[V1WriteCommand] || !conf.plannedWriteEnabled =>
c.copy(child = apply(c.child))
+ // `DeserializeToObjectExec` is used by Spark internal e.g. `Dataset.rdd`.
+ // It should always be a root node, and does not allow a shuffle on top of it.
+ // This is conflict with AQE framework since we may add shuffle back during re-optimize
Review Comment:
```suggestion
// This conflicts with AQE framework since we may add shuffle back during re-optimize
```
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ulysses-you commented on a diff in pull request #37612: [SPARK-39915][SQL] Ensure the output partitioning is user-specified in AQE
Posted by GitBox <gi...@apache.org>.
ulysses-you commented on code in PR #37612:
URL: https://github.com/apache/spark/pull/37612#discussion_r954473375
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AQEUtils.scala:
##########
@@ -28,16 +28,31 @@ object AQEUtils {
def getRequiredDistribution(p: SparkPlan): Option[Distribution] = p match {
// User-specified repartition is only effective when it's the root node, or under
// Project/Filter/LocalSort/CollectMetrics.
- // Note: we only care about `HashPartitioning` as `EnsureRequirements` can only optimize out
- // user-specified repartition with `HashPartitioning`.
- case ShuffleExchangeExec(h: HashPartitioning, _, shuffleOrigin)
+ // Note, here are two cases of how user-specified repartition can be optimized out:
+ // 1. `EnsureRequirements` can only optimize out user-specified repartition with
+ // `HashPartitioning`.
+ // 2. `AQEOptimizer` can optimize out user-specified repartition with all `Partitioning`,
+ // e.g. convert empty to local relation.
Review Comment:
> shall we fix PropagateEmptyRelationBase instead? I don't think we can optimize out Repartition which breaks user expectations. The change here only covers AQE and I think this is a problem for non AQE as well.
if we do not optimize the repartition at the top of empty relation, we also can not optimize other plan at the top of repartition. It may become a regression. We should only preserve the final repatition which can affect the final output partition as the `requiredDistribution` did.
> Or, we can still optimize out Repartition, but we should assign a Partitioning to LocalRelation. Ideally a empty data can be any partitioning.
I actually thought about it, but it's not easy. If we want to assign a `Partitioning` to `LocalRelation`, we also need to consider how to propagate the `Partitioning` when we through other plan otherwise it does less meaning. Then we need care the output partitioning for all logical plan. This should not be expected since the partitioning is a physical concept.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ulysses-you commented on a diff in pull request #37612: [SPARK-39915][SQL] Ensure the output partitioning is user-specified in AQE
Posted by GitBox <gi...@apache.org>.
ulysses-you commented on code in PR #37612:
URL: https://github.com/apache/spark/pull/37612#discussion_r954454246
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AQEUtils.scala:
##########
@@ -28,16 +28,28 @@ object AQEUtils {
def getRequiredDistribution(p: SparkPlan): Option[Distribution] = p match {
// User-specified repartition is only effective when it's the root node, or under
// Project/Filter/LocalSort/CollectMetrics.
- // Note: we only care about `HashPartitioning` as `EnsureRequirements` can only optimize out
- // user-specified repartition with `HashPartitioning`.
Review Comment:
That is right, but it can not describe the following code equivalently.. I changed this comments to:
```
// Note, here are two cases of how user-specified repartition can be optimized out:
// 1. `EnsureRequirements` can only optimize out user-specified repartition with
// `HashPartitioning`.
// 2. `AQEOptimizer` can optimize out user-specified repartition with all `Partitioning`,
// e.g. convert empty to local relation.
```
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ulysses-you commented on a diff in pull request #37612: [SPARK-39915][SQL] Ensure the output partitioning is user-specified in AQE
Posted by GitBox <gi...@apache.org>.
ulysses-you commented on code in PR #37612:
URL: https://github.com/apache/spark/pull/37612#discussion_r960155139
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/LogicalQueryStage.scala:
##########
@@ -40,7 +40,15 @@ case class LogicalQueryStage(
override def output: Seq[Attribute] = logicalPlan.output
override val isStreaming: Boolean = logicalPlan.isStreaming
override val outputOrdering: Seq[SortOrder] = physicalPlan.outputOrdering
- override protected val nodePatterns: Seq[TreePattern] = Seq(LOGICAL_QUERY_STAGE)
+ override protected val nodePatterns: Seq[TreePattern] = {
+ // Repartition is a special node that it represents a shuffle exchange,
+ // then in AQE the repartition will be always wrapped into `LogicalQueryStage`
+ val repartitionPattern = logicalPlan match {
+ case _: RepartitionOperation => Some(REPARTITION_OPERATION)
+ case _ => None
+ }
+ Seq(LOGICAL_QUERY_STAGE) ++ repartitionPattern
Review Comment:
this is for AQE side to tag `TreePattern`. cc @cloud-fan
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ulysses-you commented on a diff in pull request #37612: [SPARK-39915][SQL] Ensure the output partitioning is user-specified in AQE
Posted by GitBox <gi...@apache.org>.
ulysses-you commented on code in PR #37612:
URL: https://github.com/apache/spark/pull/37612#discussion_r960162375
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala:
##########
@@ -2643,6 +2643,49 @@ class AdaptiveQueryExecSuite
assert(findTopLevelBroadcastNestedLoopJoin(adaptivePlan).size == 1)
}
}
+
+ test("SPARK-39915: Dataset.repartition(N) may not create N partitions") {
+ withSQLConf(SQLConf.SHUFFLE_PARTITIONS.key -> "6") {
+ // partitioning: HashPartitioning
+ // shuffleOrigin: REPARTITION_BY_NUM
+ assert(spark.range(0).repartition(5, $"id").rdd.getNumPartitions == 5)
+ // shuffleOrigin: REPARTITION_BY_COL
+ // The minimum partition number after AQE coalesce is 1
+ assert(spark.range(0).repartition($"id").rdd.getNumPartitions == 1)
+ // through project
+ assert(spark.range(0).selectExpr("id % 3 as c1", "id % 7 as c2")
+ .repartition(5, $"c1").select($"c2").rdd.getNumPartitions == 5)
+
+ // partitioning: RangePartitioning
+ // shuffleOrigin: REPARTITION_BY_NUM
+ // The minimum partition number of RangePartitioner is 1
+ assert(spark.range(0).repartitionByRange(5, $"id").rdd.getNumPartitions == 1)
Review Comment:
it's a magic from `RangePartitioner`, not sure it is a bug. `RangePartitioner` do sample to decide the range partition bounds, and the bounds will be empty if the data is empty. Then the output partition number is 1.
https://github.com/apache/spark/blob/a027db1dc0379d823cbd638181bb25095f4e6577/core/src/main/scala/org/apache/spark/Partitioner.scala#L220
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a diff in pull request #37612: [SPARK-39915][SQL] Ensure the output partitioning is user-specified in AQE
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #37612:
URL: https://github.com/apache/spark/pull/37612#discussion_r954456371
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/InsertAdaptiveSparkPlan.scala:
##########
@@ -51,6 +51,11 @@ case class InsertAdaptiveSparkPlan(
case c: DataWritingCommandExec
if !c.cmd.isInstanceOf[V1WriteCommand] || !conf.plannedWriteEnabled =>
c.copy(child = apply(c.child))
+ // `DeserializeToObjectExec` is used by Spark internal e.g. `Dataset.rdd`.
Review Comment:
```suggestion
// `DeserializeToObjectExec` is used by Spark internally e.g. `Dataset.rdd`.
```
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a diff in pull request #37612: [SPARK-39915][SQL] Ensure the output partitioning is user-specified in AQE
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #37612:
URL: https://github.com/apache/spark/pull/37612#discussion_r955758719
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PropagateEmptyRelation.scala:
##########
@@ -44,6 +45,9 @@ import org.apache.spark.sql.catalyst.trees.TreePattern.{LOCAL_RELATION, TRUE_OR_
* - Generate(Explode) with all empty children. Others like Hive UDTF may return results.
*/
abstract class PropagateEmptyRelationBase extends Rule[LogicalPlan] with CastSupport {
+ // This tag is used to mark a repartition as a root repartition which is user-specified
Review Comment:
can we make a new PR for this change? It's not only related to AQE.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ulysses-you commented on pull request #37612: [SPARK-39915][SQL] Ensure the output partitioning is user-specified in AQE
Posted by GitBox <gi...@apache.org>.
ulysses-you commented on PR #37612:
URL: https://github.com/apache/spark/pull/37612#issuecomment-1226672854
It should be clear now. This pr only did two things:
1. Only apply AQE for the children of DeserializeToObjectExec
2. Check all partitioning for requiredDistribution so we can have a chance to add it back
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a diff in pull request #37612: [SPARK-39915][SQL] Ensure the output partitioning is user-specified in AQE
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #37612:
URL: https://github.com/apache/spark/pull/37612#discussion_r953535869
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/InsertAdaptiveSparkPlan.scala:
##########
@@ -51,6 +51,7 @@ case class InsertAdaptiveSparkPlan(
case c: DataWritingCommandExec
if !c.cmd.isInstanceOf[V1WriteCommand] || !conf.plannedWriteEnabled =>
c.copy(child = apply(c.child))
+ case d: DeserializeToObjectExec => d.withNewChildren(d.children.map(apply))
Review Comment:
let's add comments to explain it
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ulysses-you commented on a diff in pull request #37612: [SPARK-39915][SQL] Ensure the output partitioning is user-specified in AQE
Posted by GitBox <gi...@apache.org>.
ulysses-you commented on code in PR #37612:
URL: https://github.com/apache/spark/pull/37612#discussion_r955557324
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AQEUtils.scala:
##########
@@ -28,16 +28,31 @@ object AQEUtils {
def getRequiredDistribution(p: SparkPlan): Option[Distribution] = p match {
// User-specified repartition is only effective when it's the root node, or under
// Project/Filter/LocalSort/CollectMetrics.
- // Note: we only care about `HashPartitioning` as `EnsureRequirements` can only optimize out
- // user-specified repartition with `HashPartitioning`.
- case ShuffleExchangeExec(h: HashPartitioning, _, shuffleOrigin)
+ // Note, here are two cases of how user-specified repartition can be optimized out:
+ // 1. `EnsureRequirements` can only optimize out user-specified repartition with
+ // `HashPartitioning`.
+ // 2. `AQEOptimizer` can optimize out user-specified repartition with all `Partitioning`,
+ // e.g. convert empty to local relation.
Review Comment:
skip optimize root user-specified repartition by https://github.com/apache/spark/pull/37612/commits/546628906f0329f523fc26ead4913f3032ec9851
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a diff in pull request #37612: [SPARK-39915][SQL] Ensure the output partitioning is user-specified in AQE
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #37612:
URL: https://github.com/apache/spark/pull/37612#discussion_r954457377
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AQEUtils.scala:
##########
@@ -28,16 +28,31 @@ object AQEUtils {
def getRequiredDistribution(p: SparkPlan): Option[Distribution] = p match {
// User-specified repartition is only effective when it's the root node, or under
// Project/Filter/LocalSort/CollectMetrics.
- // Note: we only care about `HashPartitioning` as `EnsureRequirements` can only optimize out
- // user-specified repartition with `HashPartitioning`.
- case ShuffleExchangeExec(h: HashPartitioning, _, shuffleOrigin)
+ // Note, here are two cases of how user-specified repartition can be optimized out:
+ // 1. `EnsureRequirements` can only optimize out user-specified repartition with
+ // `HashPartitioning`.
+ // 2. `AQEOptimizer` can optimize out user-specified repartition with all `Partitioning`,
+ // e.g. convert empty to local relation.
Review Comment:
shall we fix `PropagateEmptyRelationBase` instead? I don't think we can optimize out `Repartition` which breaks user expectations. The change here only covers AQE and I think this is a problem for non AQE as well.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a diff in pull request #37612: [SPARK-39915][SQL] Ensure the output partitioning is user-specified in AQE
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #37612:
URL: https://github.com/apache/spark/pull/37612#discussion_r954444592
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AQEUtils.scala:
##########
@@ -28,16 +28,28 @@ object AQEUtils {
def getRequiredDistribution(p: SparkPlan): Option[Distribution] = p match {
// User-specified repartition is only effective when it's the root node, or under
// Project/Filter/LocalSort/CollectMetrics.
- // Note: we only care about `HashPartitioning` as `EnsureRequirements` can only optimize out
- // user-specified repartition with `HashPartitioning`.
Review Comment:
But if `EnsureRequirements` can't remove user-specified repartition that is not `HashPartitioning`, we don't need to change 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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a diff in pull request #37612: [SPARK-39915][SQL] Ensure the output partitioning is user-specified in AQE
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #37612:
URL: https://github.com/apache/spark/pull/37612#discussion_r960158919
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala:
##########
@@ -2643,6 +2643,49 @@ class AdaptiveQueryExecSuite
assert(findTopLevelBroadcastNestedLoopJoin(adaptivePlan).size == 1)
}
}
+
+ test("SPARK-39915: Dataset.repartition(N) may not create N partitions") {
+ withSQLConf(SQLConf.SHUFFLE_PARTITIONS.key -> "6") {
+ // partitioning: HashPartitioning
+ // shuffleOrigin: REPARTITION_BY_NUM
+ assert(spark.range(0).repartition(5, $"id").rdd.getNumPartitions == 5)
+ // shuffleOrigin: REPARTITION_BY_COL
+ // The minimum partition number after AQE coalesce is 1
+ assert(spark.range(0).repartition($"id").rdd.getNumPartitions == 1)
+ // through project
+ assert(spark.range(0).selectExpr("id % 3 as c1", "id % 7 as c2")
+ .repartition(5, $"c1").select($"c2").rdd.getNumPartitions == 5)
+
+ // partitioning: RangePartitioning
+ // shuffleOrigin: REPARTITION_BY_NUM
+ // The minimum partition number of RangePartitioner is 1
+ assert(spark.range(0).repartitionByRange(5, $"id").rdd.getNumPartitions == 1)
Review Comment:
is it another bug? the num partitions is not respected.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ulysses-you commented on a diff in pull request #37612: [SPARK-39915][SQL] Ensure the output partitioning is user-specified in AQE
Posted by GitBox <gi...@apache.org>.
ulysses-you commented on code in PR #37612:
URL: https://github.com/apache/spark/pull/37612#discussion_r959116737
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala:
##########
@@ -2643,6 +2643,50 @@ class AdaptiveQueryExecSuite
assert(findTopLevelBroadcastNestedLoopJoin(adaptivePlan).size == 1)
}
}
+
+ test("SPARK-39915: Dataset.repartition(N) may not create N partitions") {
+ withSQLConf(SQLConf.SHUFFLE_PARTITIONS.key -> "6") {
+ // partitioning: HashPartitioning
+ // shuffleOrigin: REPARTITION_BY_NUM
+ assert(spark.range(0).repartition(5).rdd.getNumPartitions == 5)
+ assert(spark.range(0).repartition(5, $"id").rdd.getNumPartitions == 5)
+ // shuffleOrigin: REPARTITION_BY_COL
+ // The minimum partition number after AQE coalesce is 1
+ assert(spark.range(0).repartition($"id").rdd.getNumPartitions == 1)
+ // not retained output partitioning
Review Comment:
the test is outdate, fixed them in 048e781bcaf5f78d28666c5af16745c6a93cc08c
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a diff in pull request #37612: [SPARK-39915][SQL] Ensure the output partitioning is user-specified in AQE
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #37612:
URL: https://github.com/apache/spark/pull/37612#discussion_r958612230
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala:
##########
@@ -2643,6 +2643,50 @@ class AdaptiveQueryExecSuite
assert(findTopLevelBroadcastNestedLoopJoin(adaptivePlan).size == 1)
}
}
+
+ test("SPARK-39915: Dataset.repartition(N) may not create N partitions") {
+ withSQLConf(SQLConf.SHUFFLE_PARTITIONS.key -> "6") {
+ // partitioning: HashPartitioning
+ // shuffleOrigin: REPARTITION_BY_NUM
+ assert(spark.range(0).repartition(5).rdd.getNumPartitions == 5)
+ assert(spark.range(0).repartition(5, $"id").rdd.getNumPartitions == 5)
+ // shuffleOrigin: REPARTITION_BY_COL
+ // The minimum partition number after AQE coalesce is 1
+ assert(spark.range(0).repartition($"id").rdd.getNumPartitions == 1)
+ // not retained output partitioning
Review Comment:
why is this not retained? repartition under Project should be respected.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a diff in pull request #37612: [SPARK-39915][SQL] Ensure the output partitioning is user-specified in AQE
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #37612:
URL: https://github.com/apache/spark/pull/37612#discussion_r960157740
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PropagateEmptyRelation.scala:
##########
@@ -130,7 +130,8 @@ abstract class PropagateEmptyRelationBase extends Rule[LogicalPlan] with CastSup
p
}
- // the only case can be matched here is that LogicalQueryStage is empty
+ // In AQE, the `Repartition` will be replaced with `LogicalQueryStage`,
Review Comment:
do we still need this comment change?
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a diff in pull request #37612: [SPARK-39915][SQL] Ensure the output partitioning is user-specified in AQE
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #37612:
URL: https://github.com/apache/spark/pull/37612#discussion_r958613405
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala:
##########
@@ -2643,6 +2643,50 @@ class AdaptiveQueryExecSuite
assert(findTopLevelBroadcastNestedLoopJoin(adaptivePlan).size == 1)
}
}
+
+ test("SPARK-39915: Dataset.repartition(N) may not create N partitions") {
+ withSQLConf(SQLConf.SHUFFLE_PARTITIONS.key -> "6") {
+ // partitioning: HashPartitioning
Review Comment:
`HashPartitioning`? `repartition(5)` is RoundRobinPartitioning
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ulysses-you commented on a diff in pull request #37612: [SPARK-39915][SQL] Ensure the output partitioning is user-specified in AQE
Posted by GitBox <gi...@apache.org>.
ulysses-you commented on code in PR #37612:
URL: https://github.com/apache/spark/pull/37612#discussion_r969216245
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AdaptiveSparkPlanExec.scala:
##########
@@ -122,7 +122,8 @@ case class AdaptiveSparkPlanExec(
ReplaceHashWithSortAgg,
RemoveRedundantSorts,
DisableUnnecessaryBucketedScan,
- OptimizeSkewedJoin(ensureRequirements)
+ OptimizeSkewedJoin(ensureRequirements),
+ AdjustShuffleExchangePosition
Review Comment:
addressed
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ulysses-you commented on pull request #37612: [SPARK-39915][SQL] Ensure the output partitioning is user-specified in AQE
Posted by GitBox <gi...@apache.org>.
ulysses-you commented on PR #37612:
URL: https://github.com/apache/spark/pull/37612#issuecomment-1245393222
thank you @cloud-fan, updated
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ulysses-you commented on a diff in pull request #37612: [SPARK-39915][SQL] Ensure the output partitioning is user-specified in AQE
Posted by GitBox <gi...@apache.org>.
ulysses-you commented on code in PR #37612:
URL: https://github.com/apache/spark/pull/37612#discussion_r955851760
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PropagateEmptyRelation.scala:
##########
@@ -44,6 +45,9 @@ import org.apache.spark.sql.catalyst.trees.TreePattern.{LOCAL_RELATION, TRUE_OR_
* - Generate(Explode) with all empty children. Others like Hive UDTF may return results.
*/
abstract class PropagateEmptyRelationBase extends Rule[LogicalPlan] with CastSupport {
+ // This tag is used to mark a repartition as a root repartition which is user-specified
Review Comment:
ok, will send a new for non-AQE part
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on pull request #37612: [SPARK-39915][SQL] Ensure the output partitioning is user-specified in AQE
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on PR #37612:
URL: https://github.com/apache/spark/pull/37612#issuecomment-1246091278
thanks, merging to master! (the last commit only add comments)
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan closed pull request #37612: [SPARK-39915][SQL] Ensure the output partitioning is user-specified in AQE
Posted by GitBox <gi...@apache.org>.
cloud-fan closed pull request #37612: [SPARK-39915][SQL] Ensure the output partitioning is user-specified in AQE
URL: https://github.com/apache/spark/pull/37612
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ulysses-you commented on a diff in pull request #37612: [SPARK-39915][SQL] Ensure the output partitioning is user-specified in AQE
Posted by GitBox <gi...@apache.org>.
ulysses-you commented on code in PR #37612:
URL: https://github.com/apache/spark/pull/37612#discussion_r969357388
##########
sql/core/src/test/scala/org/apache/spark/sql/DataFrameSuite.scala:
##########
@@ -483,7 +483,7 @@ class DataFrameSuite extends QueryTest
testData.select("key").coalesce(1).select("key"),
testData.select("key").collect().toSeq)
- assert(spark.emptyDataFrame.coalesce(1).rdd.partitions.size === 0)
+ assert(spark.emptyDataFrame.coalesce(1).rdd.partitions.size === 1)
Review Comment:
one more failed test
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ulysses-you commented on pull request #37612: [SPARK-39915][SQL] Ensure the output partitioning is user-specified in AQE
Posted by GitBox <gi...@apache.org>.
ulysses-you commented on PR #37612:
URL: https://github.com/apache/spark/pull/37612#issuecomment-1223589809
cc @wangyum @zsxwing @cloud-fan @maryannxue if you have time to review
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a diff in pull request #37612: [SPARK-39915][SQL] Ensure the output partitioning is user-specified in AQE
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #37612:
URL: https://github.com/apache/spark/pull/37612#discussion_r953537151
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/ValidateSparkPlan.scala:
##########
@@ -39,30 +45,41 @@ object ValidateSparkPlan extends Rule[SparkPlan] {
* - BroadcastQueryStage only appears as the immediate child and the build side of a broadcast
* hash join or broadcast nested loop join.
*/
- private def validate(plan: SparkPlan): Unit = plan match {
+ private def validateBroadcastQueryStage(plan: SparkPlan): Unit = plan match {
case b: BroadcastHashJoinExec =>
val (buildPlan, probePlan) = b.buildSide match {
case BuildLeft => (b.left, b.right)
case BuildRight => (b.right, b.left)
}
if (!buildPlan.isInstanceOf[BroadcastQueryStageExec]) {
- validate(buildPlan)
+ validateBroadcastQueryStage(buildPlan)
}
- validate(probePlan)
+ validateBroadcastQueryStage(probePlan)
case b: BroadcastNestedLoopJoinExec =>
val (buildPlan, probePlan) = b.buildSide match {
case BuildLeft => (b.left, b.right)
case BuildRight => (b.right, b.left)
}
if (!buildPlan.isInstanceOf[BroadcastQueryStageExec]) {
- validate(buildPlan)
+ validateBroadcastQueryStage(buildPlan)
}
- validate(probePlan)
+ validateBroadcastQueryStage(probePlan)
case q: BroadcastQueryStageExec => errorOnInvalidBroadcastQueryStage(q)
- case _ => plan.children.foreach(validate)
+ case _ => plan.children.foreach(validateBroadcastQueryStage)
}
private def errorOnInvalidBroadcastQueryStage(plan: SparkPlan): Unit = {
throw InvalidAQEPlanException("Invalid broadcast query stage", plan)
}
+
+ /**
+ * Validate that the output partitioning of plan satisfies the `userSpecifiedDistribution`.
+ */
+ private def validateOutputPartitioning(plan: SparkPlan): Unit = {
+ if (userSpecifiedDistribution.isDefined &&
+ !plan.outputPartitioning.satisfies(userSpecifiedDistribution.get)) {
+ throw InvalidAQEPlanException(s"Output partitioning does not satisfy the specified, " +
Review Comment:
do we have a test case to cover this case, or it's just a safe guard for now?
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a diff in pull request #37612: [SPARK-39915][SQL] Ensure the output partitioning is user-specified in AQE
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #37612:
URL: https://github.com/apache/spark/pull/37612#discussion_r955761128
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PropagateEmptyRelation.scala:
##########
@@ -162,14 +177,53 @@ abstract class PropagateEmptyRelationBase extends Rule[LogicalPlan] with CastSup
case _ => p
}
}
+
+ protected val repartitionTreePattern: TreePattern
+ protected def userSpecifiedRepartition(p: LogicalPlan): Boolean
+
+ /**
+ * Add a [[ROOT_REPARTITION]] tag for the root user-specified repartition so this rule can
+ * skip optimize it.
+ */
+ protected def addTagForRootRepartition(plan: LogicalPlan): LogicalPlan = {
+ var isRootRepartition = true
+ plan.transformDownWithPruning(_.containsPattern(repartitionTreePattern)) {
+ case repartition if repartition.getTagValue(ROOT_REPARTITION).isEmpty &&
+ isRootRepartition && userSpecifiedRepartition(repartition) =>
+ repartition.setTagValue(ROOT_REPARTITION, ())
+ repartition
+ case project: Project => project
+ case filter: Filter => filter
+ case other if isRootRepartition =>
+ isRootRepartition = false
+ other
+ case other => other
+ }
+ }
}
/**
* This rule runs in the normal optimizer
*/
object PropagateEmptyRelation extends PropagateEmptyRelationBase {
- override def apply(plan: LogicalPlan): LogicalPlan = plan.transformUpWithPruning(
- _.containsAnyPattern(LOCAL_RELATION, TRUE_OR_FALSE_LITERAL), ruleId) {
- commonApplyFunc
+ override protected val repartitionTreePattern: TreePattern = REPARTITION_OPERATION
+
+ protected def userSpecifiedRepartition(p: LogicalPlan): Boolean = p match {
+ case _: Repartition => true
+ case r: RepartitionByExpression
+ if r.optNumPartitions.isDefined || r.partitionExpressions.nonEmpty => true
+ case _ => false
+ }
+
+ override def apply(plan: LogicalPlan): LogicalPlan = {
Review Comment:
is it possible to move the `apply` method to the base class, so that we can share more code?
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a diff in pull request #37612: [SPARK-39915][SQL] Ensure the output partitioning is user-specified in AQE
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #37612:
URL: https://github.com/apache/spark/pull/37612#discussion_r955759026
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PropagateEmptyRelation.scala:
##########
@@ -127,7 +131,13 @@ abstract class PropagateEmptyRelationBase extends Rule[LogicalPlan] with CastSup
}
// the only case can be matched here is that LogicalQueryStage is empty
- case p: LeafNode if !p.isInstanceOf[LocalRelation] && isEmpty(p) => empty(p)
+ case p: LeafNode if !p.isInstanceOf[LocalRelation] && isEmpty(p) =>
+ if (p.getTagValue(ROOT_REPARTITION).isEmpty) {
Review Comment:
how can a leaf node be root repartition?
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ulysses-you commented on a diff in pull request #37612: [SPARK-39915][SQL] Ensure the output partitioning is user-specified in AQE
Posted by GitBox <gi...@apache.org>.
ulysses-you commented on code in PR #37612:
URL: https://github.com/apache/spark/pull/37612#discussion_r955770780
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PropagateEmptyRelation.scala:
##########
@@ -127,7 +131,13 @@ abstract class PropagateEmptyRelationBase extends Rule[LogicalPlan] with CastSup
}
// the only case can be matched here is that LogicalQueryStage is empty
- case p: LeafNode if !p.isInstanceOf[LocalRelation] && isEmpty(p) => empty(p)
+ case p: LeafNode if !p.isInstanceOf[LocalRelation] && isEmpty(p) =>
+ if (p.getTagValue(ROOT_REPARTITION).isEmpty) {
Review Comment:
it is `LogicalQueryStage` in AQE since the reparition is planned to shuffle
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a diff in pull request #37612: [SPARK-39915][SQL] Ensure the output partitioning is user-specified in AQE
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #37612:
URL: https://github.com/apache/spark/pull/37612#discussion_r954560331
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AQEUtils.scala:
##########
@@ -28,16 +28,31 @@ object AQEUtils {
def getRequiredDistribution(p: SparkPlan): Option[Distribution] = p match {
// User-specified repartition is only effective when it's the root node, or under
// Project/Filter/LocalSort/CollectMetrics.
- // Note: we only care about `HashPartitioning` as `EnsureRequirements` can only optimize out
- // user-specified repartition with `HashPartitioning`.
- case ShuffleExchangeExec(h: HashPartitioning, _, shuffleOrigin)
+ // Note, here are two cases of how user-specified repartition can be optimized out:
+ // 1. `EnsureRequirements` can only optimize out user-specified repartition with
+ // `HashPartitioning`.
+ // 2. `AQEOptimizer` can optimize out user-specified repartition with all `Partitioning`,
+ // e.g. convert empty to local relation.
Review Comment:
OK let me make my proposal clear: let's not optimize out repartition if it's the root node, or below Project/Filter, in any cases. What do you think?
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ulysses-you commented on a diff in pull request #37612: [SPARK-39915][SQL] Ensure the output partitioning is user-specified in AQE
Posted by GitBox <gi...@apache.org>.
ulysses-you commented on code in PR #37612:
URL: https://github.com/apache/spark/pull/37612#discussion_r953558684
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/ValidateSparkPlan.scala:
##########
@@ -39,30 +45,41 @@ object ValidateSparkPlan extends Rule[SparkPlan] {
* - BroadcastQueryStage only appears as the immediate child and the build side of a broadcast
* hash join or broadcast nested loop join.
*/
- private def validate(plan: SparkPlan): Unit = plan match {
+ private def validateBroadcastQueryStage(plan: SparkPlan): Unit = plan match {
case b: BroadcastHashJoinExec =>
val (buildPlan, probePlan) = b.buildSide match {
case BuildLeft => (b.left, b.right)
case BuildRight => (b.right, b.left)
}
if (!buildPlan.isInstanceOf[BroadcastQueryStageExec]) {
- validate(buildPlan)
+ validateBroadcastQueryStage(buildPlan)
}
- validate(probePlan)
+ validateBroadcastQueryStage(probePlan)
case b: BroadcastNestedLoopJoinExec =>
val (buildPlan, probePlan) = b.buildSide match {
case BuildLeft => (b.left, b.right)
case BuildRight => (b.right, b.left)
}
if (!buildPlan.isInstanceOf[BroadcastQueryStageExec]) {
- validate(buildPlan)
+ validateBroadcastQueryStage(buildPlan)
}
- validate(probePlan)
+ validateBroadcastQueryStage(probePlan)
case q: BroadcastQueryStageExec => errorOnInvalidBroadcastQueryStage(q)
- case _ => plan.children.foreach(validate)
+ case _ => plan.children.foreach(validateBroadcastQueryStage)
}
private def errorOnInvalidBroadcastQueryStage(plan: SparkPlan): Unit = {
throw InvalidAQEPlanException("Invalid broadcast query stage", plan)
}
+
+ /**
+ * Validate that the output partitioning of plan satisfies the `userSpecifiedDistribution`.
+ */
+ private def validateOutputPartitioning(plan: SparkPlan): Unit = {
+ if (userSpecifiedDistribution.isDefined &&
+ !plan.outputPartitioning.satisfies(userSpecifiedDistribution.get)) {
+ throw InvalidAQEPlanException(s"Output partitioning does not satisfy the specified, " +
Review Comment:
I think it's just a safe guard, the error msg is logging by `logOnLevel` which is `debug` by default at `reOptimize`
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a diff in pull request #37612: [SPARK-39915][SQL] Ensure the output partitioning is user-specified in AQE
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #37612:
URL: https://github.com/apache/spark/pull/37612#discussion_r969205676
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AdaptiveSparkPlanExec.scala:
##########
@@ -122,7 +122,8 @@ case class AdaptiveSparkPlanExec(
ReplaceHashWithSortAgg,
RemoveRedundantSorts,
DisableUnnecessaryBucketedScan,
- OptimizeSkewedJoin(ensureRequirements)
+ OptimizeSkewedJoin(ensureRequirements),
+ AdjustShuffleExchangePosition
Review Comment:
shall we put it right after ensureRequirements?
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org