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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
