You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "peter-toth (via GitHub)" <gi...@apache.org> on 2026/01/19 18:53:47 UTC
[PR] [SPARK-55092][SQL] `KeyGroupedPartitioning` don't group partitions when not needed [spark]
peter-toth opened a new pull request, #53859:
URL: https://github.com/apache/spark/pull/53859
### What changes were proposed in this pull request?
Currently `KeyGroupedPartitioning` always groups partitions regardless if grouping is actually needed or not. This beahaviour decreases parallelism and can lead to slower performance.
This PR disables grouping when a shuffle is inserted above a scan with `KeyGroupedPartitioning` output partitioning, but it is not needed for the intermediate nodes either.
TODO: grouping might not be needed even if there is no shuffle...
### Why are the changes needed?
Improve performance.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
New UT added.
### Was this patch authored or co-authored using generative AI tooling?
No.
--
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
Re: [PR] [SPARK-55092][SQL] Disable partition grouping in `KeyGroupedPartitioning` when not needed [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #53859:
URL: https://github.com/apache/spark/pull/53859#issuecomment-3813265464
@chirag-s-db if you also want to take a look?
--
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
Re: [PR] [SPARK-55092][SQL] Disable partition grouping in `KeyGroupedPartitioning` when not needed [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #53859:
URL: https://github.com/apache/spark/pull/53859#issuecomment-3774698844
cc @szehon-ho , @sunchao, @viirya, @dongjoon-hyun
--
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
Re: [PR] [SPARK-55092][SQL] Disable partition grouping in `KeyGroupedPartitioning` when not needed [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #53859:
URL: https://github.com/apache/spark/pull/53859#discussion_r2710505965
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -688,21 +691,40 @@ case class EnsureRequirements(
replicatePartitions = replicatePartitions
)
)
+ newScan.copyTagsFrom(scan)
Review Comment:
Ya, this sounds like a new `copyTagFrom`-related bug to me, too. In this case, could you make a new JIRA issue (and its corresponding PR) for traceability , @peter-toth ?
- https://github.com/apache/spark/pull/50862
- https://github.com/apache/spark/pull/34172
- https://github.com/apache/spark/pull/31260
- https://github.com/apache/spark/pull/29593
--
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
Re: [PR] [SPARK-55092][SQL] Disable partition grouping in `KeyGroupedPartitioning` when not needed [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #53859:
URL: https://github.com/apache/spark/pull/53859#issuecomment-3783295843
> BTW, this PR is not rebased to the master after merging #53884 . Did I understand correctly?
You are right. I've just rebased 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
Re: [PR] [SPARK-55092][SQL] Disable partition grouping in `KeyGroupedPartitioning` when not needed [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #53859:
URL: https://github.com/apache/spark/pull/53859#discussion_r2715756442
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -688,21 +691,40 @@ case class EnsureRequirements(
replicatePartitions = replicatePartitions
)
)
+ newScan.copyTagsFrom(scan)
Review Comment:
https://github.com/apache/spark/pull/53884 was merged.
--
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
Re: [PR] [SPARK-55092][SQL] `KeyGroupedPartitioning` don't group partitions when not needed [spark]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #53859:
URL: https://github.com/apache/spark/pull/53859#issuecomment-3769712582
## JIRA Issue Information
=== Improvement SPARK-55092 ===
Summary: KeyGroupedPartitionig shouldn't group partitions when not needed
Assignee: None
Status: Open
Affected: ["4.2.0"]
---
*This comment was automatically generated by GitHub Actions*
--
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
Re: [PR] [SPARK-55092][SQL] Disable partition grouping in `KeyGroupedPartitioning` when not needed [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #53859:
URL: https://github.com/apache/spark/pull/53859#issuecomment-3802448094
hi , this seems useful, i will try to review this week, but if it looks ok to @sunchao @viirya , go ahead
--
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
Re: [PR] [SPARK-55092][SQL] Disable partition grouping in `KeyGroupedPartitioning` when not needed [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #53859:
URL: https://github.com/apache/spark/pull/53859#issuecomment-3822791189
Please note that not only checkpointed RDDs, but cached RDDs would also need an extra shuffle.
Actually, I wonder if partition grouping by key is at the right place in `BatchScanExec` or it could be a separate operator that does the grouping. The operator should reside between a consumer that requires `ClusteredDistribution` and a producer that provides "partitions with keys" partitioning. `EnsureRequirements` could insert such operators if needed similarly to how it inserts exchanges now. As it is would be a new operator it could be inserted on the top of `BatchScanExec` and `LogicalRDD` (checkpointed plan) and `InMemoryTableScanExec` (cached plan) 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
Re: [PR] [SPARK-55092][SQL] Disable partition grouping in `KeyGroupedPartitioning` when not needed [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #53859:
URL: https://github.com/apache/spark/pull/53859#issuecomment-3775515226
Thank you for pinging me, @peter-toth .
--
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
Re: [PR] [SPARK-55092][SQL] Disable partition grouping in `KeyGroupedPartitioning` when not needed [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #53859:
URL: https://github.com/apache/spark/pull/53859#issuecomment-3783297905
Thank you!
--
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
Re: [PR] [SPARK-55092][SQL] Disable partition grouping in `KeyGroupedPartitioning` when not needed [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #53859:
URL: https://github.com/apache/spark/pull/53859#discussion_r2710505965
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -688,21 +691,40 @@ case class EnsureRequirements(
replicatePartitions = replicatePartitions
)
)
+ newScan.copyTagsFrom(scan)
Review Comment:
Ya, this sounds like a new `copyTagFrom`-related bug to me, too. In this case, could you make a new JIRA issue (and its corresponding PR) for traceability , @peter-toth ?
- https://github.com/apache/spark/pull/50862
- https://github.com/apache/spark/pull/34172
- https://github.com/apache/spark/pull/31260
--
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
Re: [PR] [SPARK-55092][SQL] Disable partition grouping in `KeyGroupedPartitioning` when not needed [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #53859:
URL: https://github.com/apache/spark/pull/53859#discussion_r2711564985
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -688,21 +691,40 @@ case class EnsureRequirements(
replicatePartitions = replicatePartitions
)
)
+ newScan.copyTagsFrom(scan)
Review Comment:
Extracted the fix to this separate ticket/PR: https://github.com/apache/spark/pull/53884
--
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
Re: [PR] [SPARK-55092][SQL] Disable partition grouping in `KeyGroupedPartitioning` when not needed [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #53859:
URL: https://github.com/apache/spark/pull/53859#issuecomment-3783306323
@szehon-ho , @sunchao , @viirya , do you have any concerns or 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
Re: [PR] [SPARK-55092][SQL] Disable partition grouping in `KeyGroupedPartitioning` when not needed [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #53859:
URL: https://github.com/apache/spark/pull/53859#discussion_r2740844979
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala:
##########
@@ -622,6 +623,11 @@ object QueryExecution {
sparkSession: SparkSession,
adaptiveExecutionRule: Option[InsertAdaptiveSparkPlan] = None,
subquery: Boolean): Seq[Rule[SparkPlan]] = {
+ val requiredDistribution = if (subquery) {
Review Comment:
Fixed in https://github.com/apache/spark/pull/53859/commits/b04bb614cd5b5d56b40855c509ef770949b0eaa1.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -224,8 +227,11 @@ case class EnsureRequirements(
child match {
case ShuffleExchangeExec(_, c, so, ps) =>
- ShuffleExchangeExec(newPartitioning, c, so, ps)
- case _ => ShuffleExchangeExec(newPartitioning, child)
+ val newChild = disableKeyGroupingIfNotNeeded(c)
+ ShuffleExchangeExec(newPartitioning, newChild, so, ps)
+ case _ =>
+ val newChild = disableKeyGroupingIfNotNeeded(child)
Review Comment:
Done in https://github.com/apache/spark/pull/53859/commits/b04bb614cd5b5d56b40855c509ef770949b0eaa1.
--
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
Re: [PR] [SPARK-55092][SQL] Disable partition grouping in `KeyGroupedPartitioning` when not needed [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #53859:
URL: https://github.com/apache/spark/pull/53859#issuecomment-3778443598
> Although we need to handle `copyFromTag` independently, the proposal itself sounds reasonable to me. Do you think you can share some supporting performance numbers based on the existing benchmark or from your production environment?
Numbers depend heavily on the usecase. In our case a customer would like to use SPJ, between table `A` and `B`. Both tables are storage partitoned, but `B` is storage partitioned by some columns that don't match the join condition. In this case "one side shuffle" can help if `spark.sql.sources.v2.bucketing.shuffle.enabled` is enabled and only `B` will be shuffled, but the unecessary grouping of partitions still happens in case of `B`. And while storage partitioning of `B` helps in other queries, in this particular usecase it significantly decreases partitioning and slows down the stage before the shuffle.
The optimization in this PR is similar to what `spark.sql.sources.v2.bucketing.partiallyClusteredDistribution.enabled` does to keep one side partially clustered, but it works with "one side 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
Re: [PR] [SPARK-55092][SQL] Disable partition grouping in `KeyGroupedPartitioning` when not needed [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #53859:
URL: https://github.com/apache/spark/pull/53859#discussion_r2710507265
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -688,21 +691,40 @@ case class EnsureRequirements(
replicatePartitions = replicatePartitions
)
)
+ newScan.copyTagsFrom(scan)
+ newScan
case node =>
node.mapChildren(child => populateCommonPartitionInfo(
child, values, joinKeyPositions, reducers, applyPartialClustering, replicatePartitions))
}
+ private def disableKeyGroupingIfNotNeeded(child: SparkPlan) = {
+ if (canApplyPartialClusteredDistribution(child)) {
+ populateNoGroupingPartitionInfo(child)
+ } else {
+ child
+ }
+ }
+
+ private def populateNoGroupingPartitionInfo(plan: SparkPlan): SparkPlan = plan match {
+ case scan: BatchScanExec =>
+ val newScan = scan.copy(spjParams = scan.spjParams.copy(noGrouping = true))
+ newScan.copyTagsFrom(scan)
+ newScan
+ case node => node.mapChildren(child => populateNoGroupingPartitionInfo(child))
+ }
private def populateJoinKeyPositions(
plan: SparkPlan,
joinKeyPositions: Option[Seq[Int]]): SparkPlan = plan match {
case scan: BatchScanExec =>
- scan.copy(
+ val newScan = scan.copy(
spjParams = scan.spjParams.copy(
joinKeyPositions = joinKeyPositions
)
)
+ newScan.copyTagsFrom(scan)
Review Comment:
ditto. We need a new JIRA for the missing `copyTagsFrom` issue.
--
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
Re: [PR] [SPARK-55092][SQL] Disable partition grouping in `KeyGroupedPartitioning` when not needed [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #53859:
URL: https://github.com/apache/spark/pull/53859#issuecomment-3819334983
> One advantage of this approach is that it allows us to avoid grouping for the (presumably not uncommon) case of a simple scan from a partitioned table, and it should still be safe for checkpointed scans (as the checkpointed scans would have a `KeyGroupedPartitioning` w/ `disableGrouping=true`, which would not satisfy most required distributions). This approach should also decrease the complexity of the EnsureRequirements changes (since we wouldn't have to catch all the cases in which a KeyGroupedPartitioning scan doesn't contribute to the output partitioning of the plan).
My concern with this approach is that we can introduce an extra shuffle above the checkpointed (ungrouped) data.
--
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
Re: [PR] [SPARK-55092][SQL] Disable partition grouping in `KeyGroupedPartitioning` when not needed [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #53859:
URL: https://github.com/apache/spark/pull/53859#discussion_r2740844979
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala:
##########
@@ -622,6 +623,11 @@ object QueryExecution {
sparkSession: SparkSession,
adaptiveExecutionRule: Option[InsertAdaptiveSparkPlan] = None,
subquery: Boolean): Seq[Rule[SparkPlan]] = {
+ val requiredDistribution = if (subquery) {
Review Comment:
Fixed in https://github.com/apache/spark/pull/53859/commits/b04bb614cd5b5d56b40855c509ef770949b0eaa1 and added comments in https://github.com/apache/spark/pull/53859/commits/c28fc3fdd7ddca3e7431fd749b3cb4878ca45c32.
--
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
Re: [PR] [SPARK-55092][SQL] Disable partition grouping in `KeyGroupedPartitioning` when not needed [spark]
Posted by "chirag-s-db (via GitHub)" <gi...@apache.org>.
chirag-s-db commented on PR #53859:
URL: https://github.com/apache/spark/pull/53859#issuecomment-3818751056
Could we also support the case where the KeyGroupedPartitioning is the output of the plan with the following approach?
1. By default, `disableGrouping` is true in both the scan and in a field in the `KeyGroupedPartitioning`.
2. With `disableGrouping=true`, a `KeyGroupedPartitioning` can only satisfy the requirements of an UnspecifiedDistribution (or any distribution if there is only a single partition). However, with `disableGrouping=true`, we don't allow a `KeyGroupedPartitioning` (w/ > 1 partition) to satisfy the requirements of a Clustered or Ordered distribution.
3. In `EnsureRequirements`, we add a new case [here](https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala#L70) that checks if the `KeyGroupedPartitioning` _could_ satisfy the requirement if it were to be grouped (essentially, if the current implementation of `satisfies` for `KeyGroupedPartitioning` that take into consideration whether grouping is enabled or disabled). If so, then we push down `disableGrouping=false` to _both_ the scan and the `KeyGroupedPartitioning`, after which point we know that the partitioning has been used to satisfy some requirements, so we must do grouping. If we can't push down to the scan (for example, if the plan reporting `KeyGroupedPartitioning` is checkpointed), then we just add a shuffle as normal.
One advantage of this approach is that it allows us to avoid grouping for the (presumably not uncommon) case of a simple scan from a partitioned table, and it should still be safe for checkpointed scans (as the checkpointed scans would have a `KeyGroupedPartitioning` w/ `disableGrouping=true`, which would not satisfy most required distributions). This approach should also decrease the complexity of the EnsureRequirements changes (since we wouldn't have to catch all the cases in which a KeyGroupedPartitioning scan doesn't contribute to the output partitioning of the plan).
FYI @szehon-ho
--
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
Re: [PR] [SPARK-55092][SQL] Disable partition grouping in `KeyGroupedPartitioning` when not needed [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #53859:
URL: https://github.com/apache/spark/pull/53859#issuecomment-3820786152
I really like @chirag-s-db 's idea, it is quite clean. Else we have to guard everywhere we make a Shuffle to disable it. But i also see the point about missing the opportunity to avoid shuffle for a checkpointed KeyGrouped RDD. Although i guess its not a very common case. So in short , no strong opinion either way. @sunchao @viirya wondering any opinion?
--
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
Re: [PR] [SPARK-55092][SQL] Disable partition grouping in `KeyGroupedPartitioning` when not needed [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #53859:
URL: https://github.com/apache/spark/pull/53859#discussion_r2738362715
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala:
##########
@@ -622,6 +623,11 @@ object QueryExecution {
sparkSession: SparkSession,
adaptiveExecutionRule: Option[InsertAdaptiveSparkPlan] = None,
subquery: Boolean): Seq[Rule[SparkPlan]] = {
+ val requiredDistribution = if (subquery) {
Review Comment:
Can we add more detailed comments here? It looks confusing without any context when looking code 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
Re: [PR] [SPARK-55092][SQL] Disable partition grouping in `KeyGroupedPartitioning` when not needed [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #53859:
URL: https://github.com/apache/spark/pull/53859#discussion_r2740851220
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -695,6 +701,21 @@ case class EnsureRequirements(
child, values, joinKeyPositions, reducers, applyPartialClustering, replicatePartitions))
}
+ private def disableKeyGroupingIfNotNeeded(child: SparkPlan) = {
Review Comment:
I added comments in https://github.com/apache/spark/commit/b04bb614cd5b5d56b40855c509ef770949b0eaa1.
--
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
Re: [PR] [SPARK-55092][SQL] Disable partition grouping in `KeyGroupedPartitioning` when not needed [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #53859:
URL: https://github.com/apache/spark/pull/53859#discussion_r2738250760
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -695,6 +701,21 @@ case class EnsureRequirements(
child, values, joinKeyPositions, reducers, applyPartialClustering, replicatePartitions))
}
+ private def disableKeyGroupingIfNotNeeded(child: SparkPlan) = {
Review Comment:
More detailed comments on this method would be good, e.g., the conditions under which grouping can be safely disabled, etc.
--
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
Re: [PR] [SPARK-55092][SQL] Disable partition grouping in `KeyGroupedPartitioning` when not needed [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #53859:
URL: https://github.com/apache/spark/pull/53859#discussion_r2707992946
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -688,21 +691,40 @@ case class EnsureRequirements(
replicatePartitions = replicatePartitions
)
)
+ newScan.copyTagsFrom(scan)
Review Comment:
The changes here in `populateCommonPartitionInfo()` and `populateJoinKeyPositions()` are not needed for this PR, but it turned out that copying tags is necessary for the new `populateNoGroupingPartitionInfo()`, so it is better to copy tags here as well to fix possible hidden bugs.
--
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
Re: [PR] [SPARK-55092][SQL] Disable partition grouping in `KeyGroupedPartitioning` when not needed [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #53859:
URL: https://github.com/apache/spark/pull/53859#issuecomment-3905115978
Just a quick update, that I have opeped a draft PR: https://github.com/apache/spark/pull/54330 to implement the above idea and extract the partition grouping logic from `BatchScanExec` to a new operator. It is still draft, but feedback is always welcome.
--
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
Re: [PR] [SPARK-55092][SQL] Disable partition grouping in `KeyGroupedPartitioning` when not needed [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #53859:
URL: https://github.com/apache/spark/pull/53859#discussion_r2738237433
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -695,6 +701,21 @@ case class EnsureRequirements(
child, values, joinKeyPositions, reducers, applyPartialClustering, replicatePartitions))
}
+ private def disableKeyGroupingIfNotNeeded(child: SparkPlan) = {
+ if (canApplyPartialClusteredDistribution(child)) {
+ populateNoGroupingPartitionInfo(child)
+ } else {
+ child
+ }
+ }
+
+ private def populateNoGroupingPartitionInfo(plan: SparkPlan): SparkPlan = plan match {
Review Comment:
This looks like can be done with transform api?
--
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
Re: [PR] [SPARK-55092][SQL] Disable partition grouping in `KeyGroupedPartitioning` when not needed [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #53859:
URL: https://github.com/apache/spark/pull/53859#discussion_r2740872012
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala:
##########
@@ -622,6 +623,11 @@ object QueryExecution {
sparkSession: SparkSession,
adaptiveExecutionRule: Option[InsertAdaptiveSparkPlan] = None,
subquery: Boolean): Seq[Rule[SparkPlan]] = {
+ val requiredDistribution = if (subquery) {
Review Comment:
I changed this and now passing in `subquery` and added comments in https://github.com/apache/spark/pull/53859/commits/c28fc3fdd7ddca3e7431fd749b3cb4878ca45c32.
--
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
Re: [PR] [SPARK-55092][SQL] Disable partition grouping in `KeyGroupedPartitioning` when not needed [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth closed pull request #53859: [SPARK-55092][SQL] Disable partition grouping in `KeyGroupedPartitioning` when not needed
URL: https://github.com/apache/spark/pull/53859
--
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
Re: [PR] [SPARK-55092][SQL] Disable partition grouping in `KeyGroupedPartitioning` when not needed [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #53859:
URL: https://github.com/apache/spark/pull/53859#issuecomment-4032119938
I'm closing this because https://github.com/apache/spark/pull/54330 was merged.
--
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
Re: [PR] [SPARK-55092][SQL] Disable partition grouping in `KeyGroupedPartitioning` when not needed [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #53859:
URL: https://github.com/apache/spark/pull/53859#discussion_r2740849454
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -695,6 +701,21 @@ case class EnsureRequirements(
child, values, joinKeyPositions, reducers, applyPartialClustering, replicatePartitions))
}
+ private def disableKeyGroupingIfNotNeeded(child: SparkPlan) = {
+ if (canApplyPartialClusteredDistribution(child)) {
+ populateNoGroupingPartitionInfo(child)
+ } else {
+ child
+ }
+ }
+
+ private def populateNoGroupingPartitionInfo(plan: SparkPlan): SparkPlan = plan match {
Review Comment:
I modified all 3 `populate...()`s in https://github.com/apache/spark/commit/b04bb614cd5b5d56b40855c509ef770949b0eaa1.
--
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
Re: [PR] [SPARK-55092][SQL] Disable partition grouping in `KeyGroupedPartitioning` when not needed [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #53859:
URL: https://github.com/apache/spark/pull/53859#discussion_r2740846141
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/StoragePartitionJoinParams.scala:
##########
@@ -29,13 +29,15 @@ case class StoragePartitionJoinParams(
commonPartitionValues: Option[Seq[(InternalRow, Int)]] = None,
reducers: Option[Seq[Option[Reducer[_, _]]]] = None,
applyPartialClustering: Boolean = false,
- replicatePartitions: Boolean = false) {
+ replicatePartitions: Boolean = false,
+ noGrouping: Boolean = false) {
Review Comment:
Changed in https://github.com/apache/spark/pull/53859/commits/b04bb614cd5b5d56b40855c509ef770949b0eaa1.
--
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
Re: [PR] [SPARK-55092][SQL] Disable partition grouping in `KeyGroupedPartitioning` when not needed [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #53859:
URL: https://github.com/apache/spark/pull/53859#discussion_r2738431098
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala:
##########
@@ -622,6 +623,11 @@ object QueryExecution {
sparkSession: SparkSession,
adaptiveExecutionRule: Option[InsertAdaptiveSparkPlan] = None,
subquery: Boolean): Seq[Rule[SparkPlan]] = {
+ val requiredDistribution = if (subquery) {
Review Comment:
Yeah, let me change this tomorrow and pass in `subquery` directly into `EnsureRequirements`, that way this will be much cleaner.
--
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
Re: [PR] [SPARK-55092][SQL] Disable partition grouping in `KeyGroupedPartitioning` when not needed [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #53859:
URL: https://github.com/apache/spark/pull/53859#discussion_r2738064271
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -224,8 +227,11 @@ case class EnsureRequirements(
child match {
case ShuffleExchangeExec(_, c, so, ps) =>
- ShuffleExchangeExec(newPartitioning, c, so, ps)
- case _ => ShuffleExchangeExec(newPartitioning, child)
+ val newChild = disableKeyGroupingIfNotNeeded(c)
+ ShuffleExchangeExec(newPartitioning, newChild, so, ps)
+ case _ =>
+ val newChild = disableKeyGroupingIfNotNeeded(child)
Review Comment:
could we make a method createShuffleExchangeExec(..., disableGrouping: Boolean) to reduce duplication?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala:
##########
@@ -622,6 +623,11 @@ object QueryExecution {
sparkSession: SparkSession,
adaptiveExecutionRule: Option[InsertAdaptiveSparkPlan] = None,
subquery: Boolean): Seq[Rule[SparkPlan]] = {
+ val requiredDistribution = if (subquery) {
Review Comment:
not sure i get this, if its not a subquery we pass in any requiredDistribution?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/StoragePartitionJoinParams.scala:
##########
@@ -29,13 +29,15 @@ case class StoragePartitionJoinParams(
commonPartitionValues: Option[Seq[(InternalRow, Int)]] = None,
reducers: Option[Seq[Option[Reducer[_, _]]]] = None,
applyPartialClustering: Boolean = false,
- replicatePartitions: Boolean = false) {
+ replicatePartitions: Boolean = false,
+ noGrouping: Boolean = false) {
Review Comment:
nit: disableGrouping? to be consistent with the EnsureRequirements terminology
--
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
Re: [PR] [SPARK-55092][SQL] Disable partition grouping in `KeyGroupedPartitioning` when not needed [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #53859:
URL: https://github.com/apache/spark/pull/53859#discussion_r2738424594
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -695,6 +701,21 @@ case class EnsureRequirements(
child, values, joinKeyPositions, reducers, applyPartialClustering, replicatePartitions))
}
+ private def disableKeyGroupingIfNotNeeded(child: SparkPlan) = {
+ if (canApplyPartialClusteredDistribution(child)) {
+ populateNoGroupingPartitionInfo(child)
+ } else {
+ child
+ }
+ }
+
+ private def populateNoGroupingPartitionInfo(plan: SparkPlan): SparkPlan = plan match {
Review Comment:
Yes, I can change this to use `transform()` APIs.
Wanted to make it similar to the other 2 `populate...()` methods. Shall I change those 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
Re: [PR] [SPARK-55092][SQL] Disable partition grouping in `KeyGroupedPartitioning` when not needed [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #53859:
URL: https://github.com/apache/spark/pull/53859#issuecomment-3830302253
yea that sounds like it would be cleaner, and cover the checkpoint/ cache case. im not sure the detail about the disableGrouping by default will work, but curious to see, thanks @peter-toth
--
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