You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "gene-bordegaray (via GitHub)" <gi...@apache.org> on 2026/07/11 13:51:28 UTC
[PR] Enforce co-partitioning for sort merge and symmetric hash joins [datafusion]
gene-bordegaray opened a new pull request, #23480:
URL: https://github.com/apache/datafusion/pull/23480
## Which issue does this PR close?
- Closes #23451
## Rationale for this change
Partitioned sort merge and symmetric hash joins process matching child partition indexes together. Independent `KeyPartitioned` requirements do not express that cross-child invariant.
## What changes are included in this PR?
- Declare co-partitioned input requirements for `SortMergeJoinExec` and partitioned `SymmetricHashJoinExec`.
- Add sanity-check coverage for compatible and incompatible hash-partitioned children.
- Add range-partitioning SLT coverage showing both joins remain conservatively repartitioned until their dedicated Range opt-in work lands.
## Are these changes tested?
- `cargo fmt --all --check`
- `cargo clippy --all-targets --all-features -- -D warnings`
- `cargo test -p datafusion --test core_integration co_partitioned_children`
- `cargo test -p datafusion-sqllogictest --test sqllogictests -- range_partitioning`
## Are there any user-facing changes?
None.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Enforce co-partitioning for sort merge and symmetric hash joins [datafusion]
Posted by "gabotechs (via GitHub)" <gi...@apache.org>.
gabotechs merged PR #23480:
URL: https://github.com/apache/datafusion/pull/23480
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Enforce co-partitioning for sort merge and symmetric hash joins [datafusion]
Posted by "gene-bordegaray (via GitHub)" <gi...@apache.org>.
gene-bordegaray commented on code in PR #23480:
URL: https://github.com/apache/datafusion/pull/23480#discussion_r3579773116
##########
datafusion/physical-plan/src/joins/symmetric_hash_join.rs:
##########
@@ -415,23 +416,26 @@ impl ExecutionPlan for SymmetricHashJoinExec {
self.input_distribution_requirements().into_per_child()
}
- fn input_distribution_requirements(&self) -> crate::InputDistributionRequirements {
- crate::InputDistributionRequirements::new(match self.mode {
+ fn input_distribution_requirements(&self) -> InputDistributionRequirements {
+ match self.mode {
StreamJoinPartitionMode::Partitioned => {
let (left_expr, right_expr) = self
.on
.iter()
.map(|(l, r)| (Arc::clone(l) as _, Arc::clone(r) as _))
.unzip();
- vec![
+ InputDistributionRequirements::co_partitioned(vec![
Distribution::KeyPartitioned(left_expr),
Distribution::KeyPartitioned(right_expr),
- ]
+ ])
Review Comment:
Ya that was prbably too small of changes per PR, I addressed this adding the support.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Enforce co-partitioning for sort merge and symmetric hash joins [datafusion]
Posted by "gabotechs (via GitHub)" <gi...@apache.org>.
gabotechs commented on code in PR #23480:
URL: https://github.com/apache/datafusion/pull/23480#discussion_r3579484956
##########
datafusion/sqllogictest/test_files/range_partitioning.slt:
##########
@@ -634,6 +634,80 @@ ORDER BY l.range_key;
30 600
35 700
+##########
+# TEST 18: Sort Merge Join Avoids Repartition for Compatible Range Inputs
+# Compatible Range inputs satisfy SortMergeJoinExec's co-partitioned
+# KeyPartitioned requirements.
+##########
+
+statement ok
+set datafusion.optimizer.prefer_hash_join = false;
+
+query TT
+EXPLAIN SELECT l.range_key, l.value, r.value
+FROM range_partitioned l
+JOIN range_partitioned r ON l.range_key = r.range_key;
+----
+physical_plan
+01)ProjectionExec: expr=[range_key@0 as range_key, value@1 as value, value@3 as value]
+02)--SortMergeJoinExec: join_type=Inner, on=[(range_key@0, range_key@0)]
+03)----SortExec: expr=[range_key@0 ASC], preserve_partitioning=[true]
+04)------DataSourceExec: file_groups=<slt:ignore>, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false
+05)----SortExec: expr=[range_key@0 ASC], preserve_partitioning=[true]
+06)------DataSourceExec: file_groups=<slt:ignore>, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false
+
+query III
+SELECT l.range_key, l.value, r.value
+FROM range_partitioned l
+JOIN range_partitioned r ON l.range_key = r.range_key
+ORDER BY l.range_key;
+----
+1 10 10
+5 50 50
+10 100 100
+15 150 150
+20 200 200
+25 250 250
+30 300 300
+35 350 350
+
+statement ok
+reset datafusion.optimizer.prefer_hash_join;
+
+##########
+# TEST 19: Symmetric Hash Join Avoids Repartition for Compatible Range Inputs
+# Compatible Range streams satisfy SymmetricHashJoinExec's co-partitioned
+# KeyPartitioned requirements.
+##########
+
+statement ok
+set datafusion.optimizer.prefer_hash_join = true;
+
+query TT
+EXPLAIN SELECT l.range_key, l.value, r.value
+FROM unbounded_range_like l
+FULL JOIN unbounded_range_like r ON l.range_key = r.range_key;
+----
+physical_plan
+01)ProjectionExec: expr=[range_key@0 as range_key, value@1 as value, value@3 as value]
+02)--SymmetricHashJoinExec: mode=Partitioned, join_type=Full, on=[(range_key@0, range_key@0)]
+03)----StreamingTableExec: partition_sizes=4, projection=[range_key, value], infinite_source=true, output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4)
+04)----StreamingTableExec: partition_sizes=4, projection=[range_key, value], infinite_source=true, output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4)
Review Comment:
Nice! how about adding one or two negative tests 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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Enforce co-partitioning for sort merge and symmetric hash joins [datafusion]
Posted by "gene-bordegaray (via GitHub)" <gi...@apache.org>.
gene-bordegaray commented on code in PR #23480:
URL: https://github.com/apache/datafusion/pull/23480#discussion_r3578074238
##########
datafusion/sqllogictest/test_files/range_partitioning.slt:
##########
@@ -634,6 +634,80 @@ ORDER BY l.range_key;
30 600
35 700
+##########
+# TEST 18: Sort Merge Join Repartitions Compatible Range Inputs
+# SortMergeJoinExec does not opt into Range satisfying KeyPartitioned. These
+# compatible Range inputs receive Hash repartitions, a future
+# Range satisfaction implementation should remove them.
+##########
+
+statement ok
+set datafusion.optimizer.prefer_hash_join = false;
+
+query TT
+EXPLAIN SELECT l.range_key, l.value, r.value
+FROM range_partitioned l
+JOIN range_partitioned r ON l.range_key = r.range_key;
+----
+physical_plan
+01)ProjectionExec: expr=[range_key@0 as range_key, value@1 as value, value@3 as value]
+02)--SortMergeJoinExec: join_type=Inner, on=[(range_key@0, range_key@0)]
+03)----SortExec: expr=[range_key@0 ASC], preserve_partitioning=[true]
+04)------RepartitionExec: partitioning=Hash([range_key@0], 4), input_partitions=4
+05)--------DataSourceExec: file_groups=<slt:ignore>, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false
+06)----SortExec: expr=[range_key@0 ASC], preserve_partitioning=[true]
+07)------RepartitionExec: partitioning=Hash([range_key@0], 4), input_partitions=4
+08)--------DataSourceExec: file_groups=<slt:ignore>, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false
Review Comment:
it is because I did not opt into the range satisfaction since it isn't general yet with: `allow_range_satisfaction_for_key_partitioning()`
I didn't do this just because I wanted to do the other operators first but with others picking those up I guess there is no reason fro me not to 👍
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Enforce co-partitioning for sort merge and symmetric hash joins [datafusion]
Posted by "stuhood (via GitHub)" <gi...@apache.org>.
stuhood commented on code in PR #23480:
URL: https://github.com/apache/datafusion/pull/23480#discussion_r3575240899
##########
datafusion/physical-plan/src/joins/symmetric_hash_join.rs:
##########
@@ -415,23 +416,26 @@ impl ExecutionPlan for SymmetricHashJoinExec {
self.input_distribution_requirements().into_per_child()
}
- fn input_distribution_requirements(&self) -> crate::InputDistributionRequirements {
- crate::InputDistributionRequirements::new(match self.mode {
+ fn input_distribution_requirements(&self) -> InputDistributionRequirements {
+ match self.mode {
StreamJoinPartitionMode::Partitioned => {
let (left_expr, right_expr) = self
.on
.iter()
.map(|(l, r)| (Arc::clone(l) as _, Arc::clone(r) as _))
.unzip();
- vec![
+ InputDistributionRequirements::co_partitioned(vec![
Distribution::KeyPartitioned(left_expr),
Distribution::KeyPartitioned(right_expr),
- ]
+ ])
Review Comment:
> Add range-partitioning SLT coverage showing both joins repartitioned until their dedicated Range opt-in work lands.
Is it worth TODOs next to both of the uses of `InputDistributionRequirements::co_partitioned` linking to the relevant issues which will do it? Alternatively, would it be reasonably to inline that enabling? Otherwise it feels like this PR is nearly a noop.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Enforce co-partitioning for sort merge and symmetric hash joins [datafusion]
Posted by "gene-bordegaray (via GitHub)" <gi...@apache.org>.
gene-bordegaray commented on code in PR #23480:
URL: https://github.com/apache/datafusion/pull/23480#discussion_r3579778025
##########
datafusion/physical-plan/src/streaming.rs:
##########
@@ -115,6 +116,25 @@ impl StreamingTableExec {
})
}
+ /// Declares the output partitioning of this stream.
+ ///
+ /// `output_partitioning` must describe this plan's current output and have
+ /// the same number of partitions as the stream.
+ pub fn with_output_partitioning(
Review Comment:
surprised this wasn't avaialble before, wondering if there was a reason but could not think of why 🤔
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Enforce co-partitioning for sort merge and symmetric hash joins [datafusion]
Posted by "gabotechs (via GitHub)" <gi...@apache.org>.
gabotechs commented on PR #23480:
URL: https://github.com/apache/datafusion/pull/23480#issuecomment-4980596788
Nice, merging this one in 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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Enforce co-partitioning for sort merge and symmetric hash joins [datafusion]
Posted by "gene-bordegaray (via GitHub)" <gi...@apache.org>.
gene-bordegaray commented on code in PR #23480:
URL: https://github.com/apache/datafusion/pull/23480#discussion_r3564304623
##########
datafusion/core/tests/physical_optimizer/sanity_checker.rs:
##########
@@ -417,6 +419,26 @@ fn range_partitioned_exec(
.map(|exec| Arc::new(exec) as Arc<dyn ExecutionPlan>)
}
+fn hash_partitioned_exec(
Review Comment:
using hash to prove that co-partitioning dows work fro an accepted partitioning type
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Enforce co-partitioning for sort merge and symmetric hash joins [datafusion]
Posted by "gabotechs (via GitHub)" <gi...@apache.org>.
gabotechs commented on code in PR #23480:
URL: https://github.com/apache/datafusion/pull/23480#discussion_r3577737302
##########
datafusion/sqllogictest/test_files/range_partitioning.slt:
##########
@@ -634,6 +634,80 @@ ORDER BY l.range_key;
30 600
35 700
+##########
+# TEST 18: Sort Merge Join Repartitions Compatible Range Inputs
+# SortMergeJoinExec does not opt into Range satisfying KeyPartitioned. These
+# compatible Range inputs receive Hash repartitions, a future
+# Range satisfaction implementation should remove them.
+##########
+
+statement ok
+set datafusion.optimizer.prefer_hash_join = false;
+
+query TT
+EXPLAIN SELECT l.range_key, l.value, r.value
+FROM range_partitioned l
+JOIN range_partitioned r ON l.range_key = r.range_key;
+----
+physical_plan
+01)ProjectionExec: expr=[range_key@0 as range_key, value@1 as value, value@3 as value]
+02)--SortMergeJoinExec: join_type=Inner, on=[(range_key@0, range_key@0)]
+03)----SortExec: expr=[range_key@0 ASC], preserve_partitioning=[true]
+04)------RepartitionExec: partitioning=Hash([range_key@0], 4), input_partitions=4
+05)--------DataSourceExec: file_groups=<slt:ignore>, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false
+06)----SortExec: expr=[range_key@0 ASC], preserve_partitioning=[true]
+07)------RepartitionExec: partitioning=Hash([range_key@0], 4), input_partitions=4
+08)--------DataSourceExec: file_groups=<slt:ignore>, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false
Review Comment:
Can you remind me how is it that, even if `SortMergeJoinExec` now requires both children to be co-partitioned, and the join is in the `range_key`, we are still seeing `RepartitionExec`s injected here? I bet this is tracked in an issue. Can we link the issue in the test description?
##########
datafusion/physical-plan/src/joins/symmetric_hash_join.rs:
##########
@@ -415,23 +416,26 @@ impl ExecutionPlan for SymmetricHashJoinExec {
self.input_distribution_requirements().into_per_child()
}
- fn input_distribution_requirements(&self) -> crate::InputDistributionRequirements {
- crate::InputDistributionRequirements::new(match self.mode {
+ fn input_distribution_requirements(&self) -> InputDistributionRequirements {
+ match self.mode {
StreamJoinPartitionMode::Partitioned => {
let (left_expr, right_expr) = self
.on
.iter()
.map(|(l, r)| (Arc::clone(l) as _, Arc::clone(r) as _))
.unzip();
- vec![
+ InputDistributionRequirements::co_partitioned(vec![
Distribution::KeyPartitioned(left_expr),
Distribution::KeyPartitioned(right_expr),
- ]
+ ])
Review Comment:
That's also my feeling after reviewing this PR, it seems to be doing close to nothing.
Rather than changing 2 LOC in `symteric_hash_join.rs` and `sort_merge_join/mod.rs`, and adding some tests demonstrating it still does not work, would it be too hard to actually make it work in this same PR?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Enforce co-partitioning for sort merge and symmetric hash joins [datafusion]
Posted by "gene-bordegaray (via GitHub)" <gi...@apache.org>.
gene-bordegaray commented on code in PR #23480:
URL: https://github.com/apache/datafusion/pull/23480#discussion_r3587085324
##########
datafusion/sqllogictest/test_files/range_partitioning.slt:
##########
@@ -634,6 +634,80 @@ ORDER BY l.range_key;
30 600
35 700
+##########
+# TEST 18: Sort Merge Join Avoids Repartition for Compatible Range Inputs
+# Compatible Range inputs satisfy SortMergeJoinExec's co-partitioned
+# KeyPartitioned requirements.
+##########
+
+statement ok
+set datafusion.optimizer.prefer_hash_join = false;
+
+query TT
+EXPLAIN SELECT l.range_key, l.value, r.value
+FROM range_partitioned l
+JOIN range_partitioned r ON l.range_key = r.range_key;
+----
+physical_plan
+01)ProjectionExec: expr=[range_key@0 as range_key, value@1 as value, value@3 as value]
+02)--SortMergeJoinExec: join_type=Inner, on=[(range_key@0, range_key@0)]
+03)----SortExec: expr=[range_key@0 ASC], preserve_partitioning=[true]
+04)------DataSourceExec: file_groups=<slt:ignore>, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false
+05)----SortExec: expr=[range_key@0 ASC], preserve_partitioning=[true]
+06)------DataSourceExec: file_groups=<slt:ignore>, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false
+
+query III
+SELECT l.range_key, l.value, r.value
+FROM range_partitioned l
+JOIN range_partitioned r ON l.range_key = r.range_key
+ORDER BY l.range_key;
+----
+1 10 10
+5 50 50
+10 100 100
+15 150 150
+20 200 200
+25 250 250
+30 300 300
+35 350 350
+
+statement ok
+reset datafusion.optimizer.prefer_hash_join;
+
+##########
+# TEST 19: Symmetric Hash Join Avoids Repartition for Compatible Range Inputs
+# Compatible Range streams satisfy SymmetricHashJoinExec's co-partitioned
+# KeyPartitioned requirements.
+##########
+
+statement ok
+set datafusion.optimizer.prefer_hash_join = true;
+
+query TT
+EXPLAIN SELECT l.range_key, l.value, r.value
+FROM unbounded_range_like l
+FULL JOIN unbounded_range_like r ON l.range_key = r.range_key;
+----
+physical_plan
+01)ProjectionExec: expr=[range_key@0 as range_key, value@1 as value, value@3 as value]
+02)--SymmetricHashJoinExec: mode=Partitioned, join_type=Full, on=[(range_key@0, range_key@0)]
+03)----StreamingTableExec: partition_sizes=4, projection=[range_key, value], infinite_source=true, output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4)
+04)----StreamingTableExec: partition_sizes=4, projection=[range_key, value], infinite_source=true, output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4)
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Enforce co-partitioning for sort merge and symmetric hash joins [datafusion]
Posted by "gene-bordegaray (via GitHub)" <gi...@apache.org>.
gene-bordegaray commented on PR #23480:
URL: https://github.com/apache/datafusion/pull/23480#issuecomment-4946385072
@stuhood @mithuncy @gabotechs
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org