Repository navigation
[SPARK-40429][SQL] Only set KeyGroupedPartitioning when the referenced column is in the output - #37886
Closed
huaxingao wants to merge 3 commits into
Closed
[SPARK-40429][SQL] Only set KeyGroupedPartitioning when the referenced column is in the output#37886huaxingao wants to merge 3 commits into
huaxingao wants to merge 3 commits into
Conversation
…d column is in the output
Contributor
Author
cloud-fan
reviewed
Sep 15, 2022
| None | ||
| } else { | ||
| val ref = AttributeSet.fromAttributeSets(partitioning.get.map(_.references)) | ||
| if (ref.subsetOf(AttributeSet(d.output))) { |
Contributor
There was a problem hiding this comment.
Suggested change
| if (ref.subsetOf(AttributeSet(d.output))) { | |
| if (ref.subsetOf(d.outputSet)) { |
cloud-fan
reviewed
Sep 15, 2022
| if (partitioning.isEmpty) { | ||
| None | ||
| } else { | ||
| val ref = AttributeSet.fromAttributeSets(partitioning.get.map(_.references)) |
Contributor
There was a problem hiding this comment.
how about partitioning.get.forall(p => p.references.subsetOf(...))
Contributor
Author
There was a problem hiding this comment.
Sounds good! Changed.
dongjoon-hyun
approved these changes
Sep 15, 2022
Member
There was a problem hiding this comment.
+1, LGTM. Thank you, @huaxingao and @cloud-fan .
Merged to master.
Member
|
Could you make a backporting PR, @huaxingao ? |
Contributor
Author
|
Thanks @cloud-fan @dongjoon-hyun |
dongjoon-hyun
pushed a commit
that referenced
this pull request
Sep 15, 2022
…renced column is in the output ### What changes were proposed in this pull request? back porting [PR](#37886) to 3.3. Only set `KeyGroupedPartitioning` when the referenced column is in the output ### Why are the changes needed? bug fixing ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? New test Closes #37901 from huaxingao/3.3. Authored-by: huaxingao <huaxin_gao@apple.com> Signed-off-by: Dongjoon Hyun <dongjoon@apache.org>
szehon-ho
added a commit
to szehon-ho/spark
that referenced
this pull request
Aug 24, 2023
…keys ### What changes were proposed in this pull request? - Add new conf spark.sql.sources.v2.bucketing.allowJoinKeysSubsetOfPartitionKeys.enabled - Change key compatibility checks in EnsureRequirements. Remove checks where all partition keys must be in join keys to allow isKeyCompatible = true in this case. - Change BatchScanExec/DataSourceV2Relation to group splits by join keys (previously grouped only by partition values) - Implement partiallyClustered skew-handling. - Group only the replicate side (now by join key as well) - add an additional sort in the end of partitions based on join key, as when we group the non-replicate side, partition ordering becomes out of order. ### Why are the changes needed? - Support Storage Partition Join in cases where the join condition does not contain all the partition keys, but just some of them ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? -Added tests in KeyGroupedPartitioningSuite -Found two problems, will address in separate PR: - apache#37886 made another change so that we have to select all join keys, otherwise DSV2 scan does not report KeyGroupedPartitioning and SPJ does not get triggered. Need to see how to relax this. - https://issues.apache.org/jira/browse/SPARK-44641 was found when testing this change. This pr refactors some of those code to add group-by-join-key, but doesnt change the underlying logic, so issue continues to exist. Hopefully this will also get fixed in another way.
szehon-ho
added a commit
to szehon-ho/spark
that referenced
this pull request
Aug 24, 2023
…keys ### What changes were proposed in this pull request? - Add new conf spark.sql.sources.v2.bucketing.allowJoinKeysSubsetOfPartitionKeys.enabled - Change key compatibility checks in EnsureRequirements. Remove checks where all partition keys must be in join keys to allow isKeyCompatible = true in this case. - Change BatchScanExec/DataSourceV2Relation to group splits by join keys (previously grouped only by partition values) - Implement partiallyClustered skew-handling. - Group only the replicate side (now by join key as well) - add an additional sort in the end of partitions based on join key, as when we group the non-replicate side, partition ordering becomes out of order. ### Why are the changes needed? - Support Storage Partition Join in cases where the join condition does not contain all the partition keys, but just some of them ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? -Added tests in KeyGroupedPartitioningSuite -Found two problems, will address in separate PR: - apache#37886 made another change so that we have to select all join keys, otherwise DSV2 scan does not report KeyGroupedPartitioning and SPJ does not get triggered. Need to see how to relax this. - https://issues.apache.org/jira/browse/SPARK-44641 was found when testing this change. This pr refactors some of those code to add group-by-join-key, but doesnt change the underlying logic, so issue continues to exist. Hopefully this will also get fixed in another way.
szehon-ho
added a commit
to szehon-ho/spark
that referenced
this pull request
Sep 8, 2023
…keys
### What changes were proposed in this pull request?
- Add new conf spark.sql.sources.v2.bucketing.allowJoinKeysSubsetOfPartitionKeys.enabled
- Change key compatibility checks in EnsureRequirements. Remove checks where all partition keys must be in join keys to allow isKeyCompatible = true in this case (if this flag is enabled)
- "Project" partitions by join keys in KeyGroupedPartitioning/KeyGroupedShuffleSpec
- Add join key grouping to the partition grouping in BatchScanExec
### Why are the changes needed?
- Support Storage Partition Join in cases where the join condition does not contain all the partition keys, but just some of them
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
-Added tests in KeyGroupedPartitioningSuite
-Because of apache#37886 we have to select all join keys to trigger SPJ in this case, otherwise DSV2 scan does not report KeyGroupedPartitioning and SPJ does not get triggered. Need to see how to relax this in separate PR.
dongjoon-hyun
pushed a commit
that referenced
this pull request
Sep 11, 2023
…keys ### What changes were proposed in this pull request? - Add new conf spark.sql.sources.v2.bucketing.allowJoinKeysSubsetOfPartitionKeys.enabled - Change key compatibility checks in EnsureRequirements. Remove checks where all partition keys must be in join keys to allow isKeyCompatible = true in this case (if this flag is enabled) - Change BatchScanExec/DataSourceV2Relation to group splits by join keys if they differ from partition keys (previously grouped only by partition values). Do same for all auxiliary data structure, like commonPartValues. - Implement partiallyClustered skew-handling. - Group only the replicate side (now by join key as well), replicate by the total size of other-side partitions that share the join key. - add an additional sort for partitions based on join key, as when we group the replicate side, partition ordering becomes out of order from the non-replicate side. ### Why are the changes needed? - Support Storage Partition Join in cases where the join condition does not contain all the partition keys, but just some of them ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? -Added tests in KeyGroupedPartitioningSuite -Found two existing problems, will address in separate PR: - Because of #37886 we have to select all join keys to trigger SPJ in this case, otherwise DSV2 scan does not report KeyGroupedPartitioning and SPJ does not get triggered. Need to see how to relax this. - https://issues.apache.org/jira/browse/SPARK-44641 was found when testing this change. This pr refactors some of those code to add group-by-join-key, but doesnt change the underlying logic, so issue continues to exist. Hopefully this will also get fixed in another way. Closes #42306 from szehon-ho/spj_attempt_master. Authored-by: Szehon Ho <szehon.apache@gmail.com> Signed-off-by: Dongjoon Hyun <dhyun@apple.com>
snmvaughan
pushed a commit
to snmvaughan/spark
that referenced
this pull request
Mar 26, 2024
…keys - Add new conf spark.sql.sources.v2.bucketing.allowJoinKeysSubsetOfPartitionKeys.enabled - Change key compatibility checks in EnsureRequirements. Remove checks where all partition keys must be in join keys to allow isKeyCompatible = true in this case (if this flag is enabled) - Change BatchScanExec/DataSourceV2Relation to group splits by join keys if they differ from partition keys (previously grouped only by partition values). Do same for all auxiliary data structure, like commonPartValues. - Implement partiallyClustered skew-handling. - Group only the replicate side (now by join key as well), replicate by the total size of other-side partitions that share the join key. - add an additional sort for partitions based on join key, as when we group the replicate side, partition ordering becomes out of order from the non-replicate side. - Support Storage Partition Join in cases where the join condition does not contain all the partition keys, but just some of them No -Added tests in KeyGroupedPartitioningSuite -Found two existing problems, will address in separate PR: - Because of apache#37886 we have to select all join keys to trigger SPJ in this case, otherwise DSV2 scan does not report KeyGroupedPartitioning and SPJ does not get triggered. Need to see how to relax this. - https://issues.apache.org/jira/browse/SPARK-44641 was found when testing this change. This pr refactors some of those code to add group-by-join-key, but doesnt change the underlying logic, so issue continues to exist. Hopefully this will also get fixed in another way. Closes apache#42306 from szehon-ho/spj_attempt_master. Authored-by: Szehon Ho <szehon.apache@gmail.com> Signed-off-by: Dongjoon Hyun <dhyun@apple.com>
ulysses-you
added a commit
that referenced
this pull request
Sep 8, 2026
…are pruned from the scan output ### What changes were proposed in this pull request? Keep the reported key-grouped partitioning when some of its keys are column-pruned out of the scan output, instead of dropping the whole partitioning, so a storage-partitioned join still applies. - `V2ScanPartitioningAndOrdering` keeps the reported keys as long as at least one of them is still in the scan output. The full key list is kept, not the surviving subset, to stay positionally aligned with the raw partition-key rows. - `DataSourceV2ScanExecBase` splits the two views of that report. `reportedKeyedPartitioning` is the source's own, at full key width, and is what reads and orders the raw `HasPartitionKey` rows (`BatchScanExec.filteredPartitions`). `outputPartitioning` projects the pruned key positions away, so the partitioning Spark plans against only references output columns. `replanWithRuntimeFilters` takes `Option[KeyedPartitioning]`, so the projected view cannot reach the raw key rows. - `DataSourceV2ScanRelation.doCanonicalize` and `BatchScanExec` (`doCanonicalize` / `equals` / `hashCode`, through a shared `prunedKeyGroupedPartitioning`) ignore partitioning keys that reference pruned columns, and the canonicalized ordering keeps only its leading run of sort orders over output columns. Otherwise a dangling attribute's exprId keeps otherwise-equivalent scans unequal and defeats subplan merging and exchange reuse. Whether the surviving projection is usable is decided one layer down, by `KeyedPartitioning`: pruning a key can map keys that were distinct onto the same key, and such a projection is marked `isCollapsed`, so grouping it stays gated on `spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled` in `mayGroupToSatisfy`. A projection that collapses nothing leaves the keys unique and needs no opt-in. No config doc changes: the config keeps the roles it already has. ### Why are the changes needed? An extra partition key that is column-pruned out of the scan output should not disable a storage-partitioned join. Dropping the whole report also lost it for the case where the surviving keys are still unique, which is sound with no opt-in at all. Background: SPARK-40429 (#37886) introduced the drop-to-`None` guard to avoid a "missing inputs" plan-validation error when a partitioning key referenced a pruned column. That failure mode no longer applies: SPARK-40259 made `DataSourceV2ScanRelation.references` return empty (partitioning, ordering and pushed filters are scan metadata, not references to resolve), so a dangling partitioning key no longer trips plan validation. This change relies on that. SPARK-40259 is on `master` only, so this change is master-only too and is not backportable to `branch-4.x` as it stands. ### Does this PR introduce _any_ user-facing change? Yes. When a partition key is pruned out of the scan output, Spark now reports the projected partitioning instead of no partitioning, so a query that previously shuffled can plan a storage-partitioned join: - with `spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled` off (default), when the surviving keys are still unique; - with it on, also when pruning collapses distinct keys onto the same key. Results are unchanged either way. ### How was this patch tested? New tests in `KeyGroupedPartitioningSuite`: - a join whose key is a subset of the partition keys, with the extra key pruned, triggers SPJ with `allowKeysSubsetOfPartitionKeys` on and shuffles with it off; - a pruned key that collapses nothing keeps SPJ with the config off; - a pruned *leading* key keeps the raw key rows aligned: one test on row loss, one on the `ClassCastException` from reading a leading `String` key at the projected `Integer` type; - runtime filtering with a pruned leading key re-plans the scan's partitions against the full-width key rows. This one lives in `KeyGroupedPartitioningRuntimeFilterTests`, so it runs against both runtime-filtering fixtures; only the V2 one prunes a column, since the Catalyst fixture's scan keeps the full table schema; - all partition keys pruned reports no partitioning, even with the config on; - a self-join with a pruned key stays canonicalizable, and a pruned key does not defeat plan reuse. Pointing `filteredPartitions` at the projected partitioning fails exactly the three tests that pin it (row loss, `ClassCastException`, runtime filtering) and nothing else. `MergeSubplansSuite`: identical scans whose reported ordering is on a pruned column are deduplicated, not fused. Verified this test fails without the ordering half of the canonicalization change. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (qwen3.8-max, Claude Opus 5) Closes #58522 from ulysses-you/worktree-spj-columnprune. Authored-by: Xiduo You <ulyssesyou18@gmail.com> Signed-off-by: Xiduo You <ulyssesyou@apache.org>
ulysses-you
added a commit
that referenced
this pull request
Sep 8, 2026
…are pruned from the scan output ### What changes were proposed in this pull request? Keep the reported key-grouped partitioning when some of its keys are column-pruned out of the scan output, instead of dropping the whole partitioning, so a storage-partitioned join still applies. - `V2ScanPartitioningAndOrdering` keeps the reported keys as long as at least one of them is still in the scan output. The full key list is kept, not the surviving subset, to stay positionally aligned with the raw partition-key rows. - `DataSourceV2ScanExecBase` splits the two views of that report. `reportedKeyedPartitioning` is the source's own, at full key width, and is what reads and orders the raw `HasPartitionKey` rows (`BatchScanExec.filteredPartitions`). `outputPartitioning` projects the pruned key positions away, so the partitioning Spark plans against only references output columns. `replanWithRuntimeFilters` takes `Option[KeyedPartitioning]`, so the projected view cannot reach the raw key rows. - `DataSourceV2ScanRelation.doCanonicalize` and `BatchScanExec` (`doCanonicalize` / `equals` / `hashCode`, through a shared `prunedKeyGroupedPartitioning`) ignore partitioning keys that reference pruned columns, and the canonicalized ordering keeps only its leading run of sort orders over output columns. Otherwise a dangling attribute's exprId keeps otherwise-equivalent scans unequal and defeats subplan merging and exchange reuse. Whether the surviving projection is usable is decided one layer down, by `KeyedPartitioning`: pruning a key can map keys that were distinct onto the same key, and such a projection is marked `isCollapsed`, so grouping it stays gated on `spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled` in `mayGroupToSatisfy`. A projection that collapses nothing leaves the keys unique and needs no opt-in. No config doc changes: the config keeps the roles it already has. ### Why are the changes needed? An extra partition key that is column-pruned out of the scan output should not disable a storage-partitioned join. Dropping the whole report also lost it for the case where the surviving keys are still unique, which is sound with no opt-in at all. Background: SPARK-40429 (#37886) introduced the drop-to-`None` guard to avoid a "missing inputs" plan-validation error when a partitioning key referenced a pruned column. That failure mode no longer applies: SPARK-40259 made `DataSourceV2ScanRelation.references` return empty (partitioning, ordering and pushed filters are scan metadata, not references to resolve), so a dangling partitioning key no longer trips plan validation. This change relies on that. SPARK-40259 is on `master` only, so this change is master-only too and is not backportable to `branch-4.x` as it stands. ### Does this PR introduce _any_ user-facing change? Yes. When a partition key is pruned out of the scan output, Spark now reports the projected partitioning instead of no partitioning, so a query that previously shuffled can plan a storage-partitioned join: - with `spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled` off (default), when the surviving keys are still unique; - with it on, also when pruning collapses distinct keys onto the same key. Results are unchanged either way. ### How was this patch tested? New tests in `KeyGroupedPartitioningSuite`: - a join whose key is a subset of the partition keys, with the extra key pruned, triggers SPJ with `allowKeysSubsetOfPartitionKeys` on and shuffles with it off; - a pruned key that collapses nothing keeps SPJ with the config off; - a pruned *leading* key keeps the raw key rows aligned: one test on row loss, one on the `ClassCastException` from reading a leading `String` key at the projected `Integer` type; - runtime filtering with a pruned leading key re-plans the scan's partitions against the full-width key rows. This one lives in `KeyGroupedPartitioningRuntimeFilterTests`, so it runs against both runtime-filtering fixtures; only the V2 one prunes a column, since the Catalyst fixture's scan keeps the full table schema; - all partition keys pruned reports no partitioning, even with the config on; - a self-join with a pruned key stays canonicalizable, and a pruned key does not defeat plan reuse. Pointing `filteredPartitions` at the projected partitioning fails exactly the three tests that pin it (row loss, `ClassCastException`, runtime filtering) and nothing else. `MergeSubplansSuite`: identical scans whose reported ordering is on a pruned column are deduplicated, not fused. Verified this test fails without the ordering half of the canonicalization change. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (qwen3.8-max, Claude Opus 5) Closes #58522 from ulysses-you/worktree-spj-columnprune. Authored-by: Xiduo You <ulyssesyou18@gmail.com> Signed-off-by: Xiduo You <ulyssesyou@apache.org> (cherry picked from commit cfbec6d) Signed-off-by: Xiduo You <ulyssesyou@apache.org>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this pull request?
Only set
KeyGroupedPartitioningwhen the referenced column is in the outputWhy are the changes needed?
bug fixing
Does this PR introduce any user-facing change?
no
How was this patch tested?
new test