gene-bordegaray commented on code in PR #23484:
URL: https://github.com/apache/datafusion/pull/23484#discussion_r3606077462
##########
datafusion/sqllogictest/test_files/range_partitioning.slt:
##########
@@ -629,7 +630,304 @@ ORDER BY l.range_key;
35 700
##########
-# TEST 18: Sort Merge Join Avoids Repartition for Compatible Range Inputs
+# TEST 18: Right Join on Range Partition Column
+# Right-side partitioned hash joins also opt in to Range satisfying
Review Comment:
lets keep descriptions generic so they remain valid after the PR merges
##########
datafusion/sqllogictest/test_files/range_partitioning.slt:
##########
@@ -629,7 +630,304 @@ ORDER BY l.range_key;
35 700
##########
-# TEST 18: Sort Merge Join Avoids Repartition for Compatible Range Inputs
+# TEST 18: Right Join on Range Partition Column
+# Right-side partitioned hash joins also opt in to Range satisfying
+# KeyPartitioned requirements. Compatible Range layouts satisfy both the
+# per-child key requirements and the cross-child layout requirement, so no
+# Hash repartitioning is inserted. The filter on the left input keeps its
+# Range partitioning and leaves the right rows above 150 unmatched.
+##########
+
+query TT
+EXPLAIN SELECT l.value, r.range_key, r.value
+FROM (SELECT range_key, value FROM range_partitioned WHERE value <= 150) l
+RIGHT JOIN range_partitioned r ON l.range_key = r.range_key;
+----
+physical_plan
+01)HashJoinExec: mode=Partitioned, join_type=Right, on=[(range_key@0,
range_key@0)], projection=[value@1, range_key@2, value@3]
+02)--FilterExec: value@1 <= 150
+03)----DataSourceExec: file_groups={4 groups:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]},
projection=[range_key, value], output_partitioning=Range([range_key@0 ASC],
[(10), (20), (30)], 4), file_type=csv, has_header=false
+04)--DataSourceExec: file_groups={4 groups:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]},
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.value, r.range_key, r.value
+FROM (SELECT range_key, value FROM range_partitioned WHERE value <= 150) l
+RIGHT JOIN range_partitioned r ON l.range_key = r.range_key
+ORDER BY r.range_key;
+----
+10 1 10
+50 5 50
+100 10 100
+150 15 150
+NULL 20 200
+NULL 25 250
+NULL 30 300
+NULL 35 350
+
+##########
+# TEST 19: Right Semi Join on Range Partition Column
+# Compatible Range inputs avoid Hash repartitioning for RightSemi joins.
+# Only right rows with a match on the filtered left side are returned.
+##########
+
+query TT
+EXPLAIN SELECT r.range_key, r.value
+FROM (SELECT range_key FROM range_partitioned WHERE value <= 150) l
+RIGHT SEMI JOIN range_partitioned r ON l.range_key = r.range_key;
+----
+physical_plan
+01)HashJoinExec: mode=Partitioned, join_type=RightSemi, on=[(range_key@0,
range_key@0)]
+02)--FilterExec: value@1 <= 150, projection=[range_key@0]
+03)----DataSourceExec: file_groups={4 groups:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]},
projection=[range_key, value], output_partitioning=Range([range_key@0 ASC],
[(10), (20), (30)], 4), file_type=csv, has_header=false
+04)--DataSourceExec: file_groups={4 groups:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]},
projection=[range_key, value], output_partitioning=Range([range_key@0 ASC],
[(10), (20), (30)], 4), file_type=csv, has_header=false
+
+query II
+SELECT r.range_key, r.value
+FROM (SELECT range_key FROM range_partitioned WHERE value <= 150) l
+RIGHT SEMI JOIN range_partitioned r ON l.range_key = r.range_key
+ORDER BY r.range_key;
+----
+1 10
+5 50
+10 100
+15 150
+
+##########
+# TEST 20: Right Anti Join on Range Partition Column
+# Compatible Range inputs avoid Hash repartitioning for RightAnti joins.
+# Only right rows without a match on the filtered left side are returned.
+##########
+
+query TT
+EXPLAIN SELECT r.range_key, r.value
+FROM (SELECT range_key FROM range_partitioned WHERE value <= 150) l
+RIGHT ANTI JOIN range_partitioned r ON l.range_key = r.range_key;
+----
+physical_plan
+01)HashJoinExec: mode=Partitioned, join_type=RightAnti, on=[(range_key@0,
range_key@0)]
+02)--FilterExec: value@1 <= 150, projection=[range_key@0]
+03)----DataSourceExec: file_groups={4 groups:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]},
projection=[range_key, value], output_partitioning=Range([range_key@0 ASC],
[(10), (20), (30)], 4), file_type=csv, has_header=false
+04)--DataSourceExec: file_groups={4 groups:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]},
projection=[range_key, value], output_partitioning=Range([range_key@0 ASC],
[(10), (20), (30)], 4), file_type=csv, has_header=false
+
+query II
+SELECT r.range_key, r.value
+FROM (SELECT range_key FROM range_partitioned WHERE value <= 150) l
+RIGHT ANTI JOIN range_partitioned r ON l.range_key = r.range_key
+ORDER BY r.range_key;
+----
+20 200
+25 250
+30 300
+35 350
+
+##########
+# TEST 21: Incompatible Range Right Join Repartitions
+# The split points of the two inputs differ, so the co-partitioned layout
+# requirement cannot be satisfied and Hash repartitioning repairs both sides
+# of the right join. Results stay correct on the repartitioned path.
+##########
+
+query TT
+EXPLAIN SELECT l.value, r.range_key, r.value
+FROM (SELECT range_key, value FROM range_partitioned WHERE value <= 150) l
+RIGHT JOIN range_partitioned_shifted r ON l.range_key = r.range_key;
+----
+physical_plan
+01)HashJoinExec: mode=Partitioned, join_type=Right, on=[(range_key@0,
range_key@0)], projection=[value@1, range_key@2, value@3]
+02)--RepartitionExec: partitioning=Hash([range_key@0], 4), input_partitions=4
+03)----FilterExec: value@1 <= 150
+04)------DataSourceExec: file_groups={4 groups:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]},
projection=[range_key, value], output_partitioning=Range([range_key@0 ASC],
[(10), (20), (30)], 4), file_type=csv, has_header=false
+05)--RepartitionExec: partitioning=Hash([range_key@0], 4), input_partitions=4
+06)----DataSourceExec: file_groups={4 groups:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned_shifted/part-0.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned_shifted/part-1.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned_shifted/part-2.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned_shifted/part-3.csv]]},
projection=[range_key, value], output_partitioning=Range([range_key@0 ASC],
[(15), (20), (30)], 4), file_type=csv, has_header=false
+
+query III
+SELECT l.value, r.range_key, r.value
+FROM (SELECT range_key, value FROM range_partitioned WHERE value <= 150) l
+RIGHT JOIN range_partitioned_shifted r ON l.range_key = r.range_key
+ORDER BY r.range_key;
+----
+10 1 10
+50 5 50
+100 10 100
+150 15 150
+NULL 20 200
+NULL 25 250
+NULL 30 300
+NULL 35 350
+
+##########
+# TEST 22: Right Join on Composite Key With Range Subset Rehashes
+# The join key (range_key, non_range_key) is a strict superset of the exposed
+# Range([range_key]) partitioning column. Unlike aggregates (TEST 3), subset
+# satisfaction is disabled for partitioned joins -- allowing Range([range_key])
+# to satisfy a wider key requirement could let matching rows land in different
+# partitions -- so planning still Hash repartitions both sides on the full
+# join key, even though subset_repartition_threshold is met.
+##########
+
+statement ok
+set datafusion.optimizer.subset_repartition_threshold = 4;
+
+query TT
+EXPLAIN SELECT l.range_key, l.non_range_key, l.value, r.value
+FROM range_partitioned l
+RIGHT JOIN range_partitioned r ON l.range_key = r.range_key AND
l.non_range_key = r.non_range_key;
+----
+physical_plan
+01)HashJoinExec: mode=Partitioned, join_type=Right, on=[(range_key@0,
range_key@0), (non_range_key@1, non_range_key@1)], projection=[range_key@0,
non_range_key@1, value@2, value@5]
+02)--RepartitionExec: partitioning=Hash([range_key@0, non_range_key@1], 4),
input_partitions=4
+03)----DataSourceExec: file_groups={4 groups:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]},
projection=[range_key, non_range_key, value],
output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4),
file_type=csv, has_header=false
+04)--RepartitionExec: partitioning=Hash([range_key@0, non_range_key@1], 4),
input_partitions=4
+05)----DataSourceExec: file_groups={4 groups:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]},
projection=[range_key, non_range_key, value],
output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4),
file_type=csv, has_header=false
+
+query IIII
+SELECT l.range_key, l.non_range_key, l.value, r.value
+FROM range_partitioned l
+RIGHT JOIN range_partitioned r ON l.range_key = r.range_key AND
l.non_range_key = r.non_range_key
+ORDER BY l.range_key;
+----
+1 1 10 10
+5 2 50 50
+10 1 100 100
+15 2 150 150
+20 1 200 200
+25 2 250 250
+30 1 300 300
+35 2 350 350
+
+statement ok
+reset datafusion.optimizer.subset_repartition_threshold;
+
+##########
+# TEST 23: Right Join with Mismatched Range Partition Counts Repartitions
+# Both inputs are range partitioned on range_key, but declare a different
number
+# of partitions (four vs three). The per-child key requirements can be
satisfied
+# by Range, but the co-partitioned layout requirement cannot, so Hash
+# repartitioning repairs both sides of the right join.
+##########
+
+query TT
+EXPLAIN SELECT l.value, r.range_key, r.value
+FROM range_partitioned l
+RIGHT JOIN range_partitioned_narrow r ON l.range_key = r.range_key;
+----
+physical_plan
+01)HashJoinExec: mode=Partitioned, join_type=Right, on=[(range_key@0,
range_key@0)], projection=[value@1, range_key@2, value@3]
+02)--RepartitionExec: partitioning=Hash([range_key@0], 4), input_partitions=4
+03)----DataSourceExec: file_groups={4 groups:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]},
projection=[range_key, value], output_partitioning=Range([range_key@0 ASC],
[(10), (20), (30)], 4), file_type=csv, has_header=false
+04)--RepartitionExec: partitioning=Hash([range_key@0], 4), input_partitions=3
+05)----DataSourceExec: file_groups={3 groups:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned_narrow/part-0.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned_narrow/part-1.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned_narrow/part-2.csv]]},
projection=[range_key, value], output_partitioning=Range([range_key@0 ASC],
[(10), (20)], 3), file_type=csv, has_header=false
+
+query III
+SELECT l.value, r.range_key, r.value
+FROM range_partitioned l
+RIGHT JOIN range_partitioned_narrow r ON l.range_key = r.range_key
+ORDER BY r.range_key;
+----
+10 1 10
+50 5 50
+100 10 100
+150 15 150
+200 20 200
+250 25 250
+300 30 300
+350 35 350
+
+##########
+# TEST 24: Right Join on Non-Range Key Repartitions
+# Both inputs expose Range([range_key]), but the join key is non_range_key.
+# Range([range_key]) does not satisfy KeyPartitioned([non_range_key]), so
+# planning inserts Hash repartitioning on the actual join key for the right
join.
+##########
+
+query TT
+EXPLAIN SELECT l.value, r.range_key, r.value
+FROM (SELECT non_range_key, value FROM range_partitioned WHERE range_key < 10)
l
+RIGHT JOIN range_partitioned r ON l.non_range_key = r.non_range_key;
+----
+physical_plan
+01)HashJoinExec: mode=Partitioned, join_type=Right, on=[(non_range_key@0,
non_range_key@1)], projection=[value@1, range_key@2, value@4]
+02)--RepartitionExec: partitioning=Hash([non_range_key@0], 4),
input_partitions=4
+03)----FilterExec: range_key@0 < 10, projection=[non_range_key@1, value@2]
+04)------DataSourceExec: file_groups={4 groups:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]},
projection=[range_key, non_range_key, value],
output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4),
file_type=csv, has_header=false
+05)--RepartitionExec: partitioning=Hash([non_range_key@1], 4),
input_partitions=4
+06)----DataSourceExec: file_groups={4 groups:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]},
projection=[range_key, non_range_key, value],
output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4),
file_type=csv, has_header=false
+
+query III
+SELECT l.value, r.range_key, r.value
+FROM (SELECT non_range_key, value FROM range_partitioned WHERE range_key < 10)
l
+RIGHT JOIN range_partitioned r ON l.non_range_key = r.non_range_key
+ORDER BY r.range_key;
+----
+10 1 10
+50 5 50
+10 10 100
+50 15 150
+10 20 200
+50 25 250
+10 30 300
+50 35 350
+
+##########
+# TEST 25: Mark Join Marker Semantics over Range Partitioned Inputs
Review Comment:
lets just keep descriptions specific to what this test is showing. Not
mentioning other things in teh codebase
##########
datafusion/sqllogictest/test_files/range_partitioning.slt:
##########
@@ -629,7 +630,304 @@ ORDER BY l.range_key;
35 700
##########
-# TEST 18: Sort Merge Join Avoids Repartition for Compatible Range Inputs
+# TEST 18: Right Join on Range Partition Column
+# Right-side partitioned hash joins also opt in to Range satisfying
+# KeyPartitioned requirements. Compatible Range layouts satisfy both the
+# per-child key requirements and the cross-child layout requirement, so no
+# Hash repartitioning is inserted. The filter on the left input keeps its
+# Range partitioning and leaves the right rows above 150 unmatched.
+##########
+
+query TT
+EXPLAIN SELECT l.value, r.range_key, r.value
+FROM (SELECT range_key, value FROM range_partitioned WHERE value <= 150) l
+RIGHT JOIN range_partitioned r ON l.range_key = r.range_key;
+----
+physical_plan
+01)HashJoinExec: mode=Partitioned, join_type=Right, on=[(range_key@0,
range_key@0)], projection=[value@1, range_key@2, value@3]
+02)--FilterExec: value@1 <= 150
+03)----DataSourceExec: file_groups={4 groups:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]},
projection=[range_key, value], output_partitioning=Range([range_key@0 ASC],
[(10), (20), (30)], 4), file_type=csv, has_header=false
+04)--DataSourceExec: file_groups={4 groups:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]},
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.value, r.range_key, r.value
+FROM (SELECT range_key, value FROM range_partitioned WHERE value <= 150) l
+RIGHT JOIN range_partitioned r ON l.range_key = r.range_key
+ORDER BY r.range_key;
+----
+10 1 10
+50 5 50
+100 10 100
+150 15 150
+NULL 20 200
+NULL 25 250
+NULL 30 300
+NULL 35 350
+
+##########
+# TEST 19: Right Semi Join on Range Partition Column
+# Compatible Range inputs avoid Hash repartitioning for RightSemi joins.
+# Only right rows with a match on the filtered left side are returned.
+##########
+
+query TT
+EXPLAIN SELECT r.range_key, r.value
+FROM (SELECT range_key FROM range_partitioned WHERE value <= 150) l
+RIGHT SEMI JOIN range_partitioned r ON l.range_key = r.range_key;
+----
+physical_plan
+01)HashJoinExec: mode=Partitioned, join_type=RightSemi, on=[(range_key@0,
range_key@0)]
+02)--FilterExec: value@1 <= 150, projection=[range_key@0]
+03)----DataSourceExec: file_groups={4 groups:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]},
projection=[range_key, value], output_partitioning=Range([range_key@0 ASC],
[(10), (20), (30)], 4), file_type=csv, has_header=false
+04)--DataSourceExec: file_groups={4 groups:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]},
projection=[range_key, value], output_partitioning=Range([range_key@0 ASC],
[(10), (20), (30)], 4), file_type=csv, has_header=false
+
+query II
+SELECT r.range_key, r.value
+FROM (SELECT range_key FROM range_partitioned WHERE value <= 150) l
+RIGHT SEMI JOIN range_partitioned r ON l.range_key = r.range_key
+ORDER BY r.range_key;
+----
+1 10
+5 50
+10 100
+15 150
+
+##########
+# TEST 20: Right Anti Join on Range Partition Column
+# Compatible Range inputs avoid Hash repartitioning for RightAnti joins.
+# Only right rows without a match on the filtered left side are returned.
+##########
+
+query TT
+EXPLAIN SELECT r.range_key, r.value
+FROM (SELECT range_key FROM range_partitioned WHERE value <= 150) l
+RIGHT ANTI JOIN range_partitioned r ON l.range_key = r.range_key;
+----
+physical_plan
+01)HashJoinExec: mode=Partitioned, join_type=RightAnti, on=[(range_key@0,
range_key@0)]
+02)--FilterExec: value@1 <= 150, projection=[range_key@0]
+03)----DataSourceExec: file_groups={4 groups:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]},
projection=[range_key, value], output_partitioning=Range([range_key@0 ASC],
[(10), (20), (30)], 4), file_type=csv, has_header=false
Review Comment:
can we ignore files 👍
##########
datafusion/sqllogictest/test_files/range_partitioning.slt:
##########
@@ -629,7 +630,304 @@ ORDER BY l.range_key;
35 700
##########
-# TEST 18: Sort Merge Join Avoids Repartition for Compatible Range Inputs
+# TEST 18: Right Join on Range Partition Column
+# Right-side partitioned hash joins also opt in to Range satisfying
+# KeyPartitioned requirements. Compatible Range layouts satisfy both the
+# per-child key requirements and the cross-child layout requirement, so no
+# Hash repartitioning is inserted. The filter on the left input keeps its
+# Range partitioning and leaves the right rows above 150 unmatched.
+##########
+
+query TT
+EXPLAIN SELECT l.value, r.range_key, r.value
+FROM (SELECT range_key, value FROM range_partitioned WHERE value <= 150) l
+RIGHT JOIN range_partitioned r ON l.range_key = r.range_key;
+----
+physical_plan
+01)HashJoinExec: mode=Partitioned, join_type=Right, on=[(range_key@0,
range_key@0)], projection=[value@1, range_key@2, value@3]
+02)--FilterExec: value@1 <= 150
+03)----DataSourceExec: file_groups={4 groups:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]},
projection=[range_key, value], output_partitioning=Range([range_key@0 ASC],
[(10), (20), (30)], 4), file_type=csv, has_header=false
+04)--DataSourceExec: file_groups={4 groups:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]},
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.value, r.range_key, r.value
+FROM (SELECT range_key, value FROM range_partitioned WHERE value <= 150) l
+RIGHT JOIN range_partitioned r ON l.range_key = r.range_key
+ORDER BY r.range_key;
+----
+10 1 10
+50 5 50
+100 10 100
+150 15 150
+NULL 20 200
+NULL 25 250
+NULL 30 300
+NULL 35 350
+
+##########
+# TEST 19: Right Semi Join on Range Partition Column
+# Compatible Range inputs avoid Hash repartitioning for RightSemi joins.
+# Only right rows with a match on the filtered left side are returned.
+##########
+
+query TT
+EXPLAIN SELECT r.range_key, r.value
+FROM (SELECT range_key FROM range_partitioned WHERE value <= 150) l
+RIGHT SEMI JOIN range_partitioned r ON l.range_key = r.range_key;
+----
+physical_plan
+01)HashJoinExec: mode=Partitioned, join_type=RightSemi, on=[(range_key@0,
range_key@0)]
+02)--FilterExec: value@1 <= 150, projection=[range_key@0]
+03)----DataSourceExec: file_groups={4 groups:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]},
projection=[range_key, value], output_partitioning=Range([range_key@0 ASC],
[(10), (20), (30)], 4), file_type=csv, has_header=false
+04)--DataSourceExec: file_groups={4 groups:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]},
projection=[range_key, value], output_partitioning=Range([range_key@0 ASC],
[(10), (20), (30)], 4), file_type=csv, has_header=false
+
+query II
+SELECT r.range_key, r.value
+FROM (SELECT range_key FROM range_partitioned WHERE value <= 150) l
+RIGHT SEMI JOIN range_partitioned r ON l.range_key = r.range_key
+ORDER BY r.range_key;
+----
+1 10
+5 50
+10 100
+15 150
+
+##########
+# TEST 20: Right Anti Join on Range Partition Column
+# Compatible Range inputs avoid Hash repartitioning for RightAnti joins.
+# Only right rows without a match on the filtered left side are returned.
+##########
+
+query TT
+EXPLAIN SELECT r.range_key, r.value
+FROM (SELECT range_key FROM range_partitioned WHERE value <= 150) l
+RIGHT ANTI JOIN range_partitioned r ON l.range_key = r.range_key;
+----
+physical_plan
+01)HashJoinExec: mode=Partitioned, join_type=RightAnti, on=[(range_key@0,
range_key@0)]
+02)--FilterExec: value@1 <= 150, projection=[range_key@0]
+03)----DataSourceExec: file_groups={4 groups:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]},
projection=[range_key, value], output_partitioning=Range([range_key@0 ASC],
[(10), (20), (30)], 4), file_type=csv, has_header=false
+04)--DataSourceExec: file_groups={4 groups:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]},
projection=[range_key, value], output_partitioning=Range([range_key@0 ASC],
[(10), (20), (30)], 4), file_type=csv, has_header=false
+
+query II
+SELECT r.range_key, r.value
+FROM (SELECT range_key FROM range_partitioned WHERE value <= 150) l
+RIGHT ANTI JOIN range_partitioned r ON l.range_key = r.range_key
+ORDER BY r.range_key;
+----
+20 200
+25 250
+30 300
+35 350
+
+##########
+# TEST 21: Incompatible Range Right Join Repartitions
+# The split points of the two inputs differ, so the co-partitioned layout
+# requirement cannot be satisfied and Hash repartitioning repairs both sides
+# of the right join. Results stay correct on the repartitioned path.
+##########
+
+query TT
+EXPLAIN SELECT l.value, r.range_key, r.value
+FROM (SELECT range_key, value FROM range_partitioned WHERE value <= 150) l
+RIGHT JOIN range_partitioned_shifted r ON l.range_key = r.range_key;
+----
+physical_plan
+01)HashJoinExec: mode=Partitioned, join_type=Right, on=[(range_key@0,
range_key@0)], projection=[value@1, range_key@2, value@3]
+02)--RepartitionExec: partitioning=Hash([range_key@0], 4), input_partitions=4
+03)----FilterExec: value@1 <= 150
+04)------DataSourceExec: file_groups={4 groups:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]},
projection=[range_key, value], output_partitioning=Range([range_key@0 ASC],
[(10), (20), (30)], 4), file_type=csv, has_header=false
+05)--RepartitionExec: partitioning=Hash([range_key@0], 4), input_partitions=4
+06)----DataSourceExec: file_groups={4 groups:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned_shifted/part-0.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned_shifted/part-1.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned_shifted/part-2.csv],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned_shifted/part-3.csv]]},
projection=[range_key, value], output_partitioning=Range([range_key@0 ASC],
[(15), (20), (30)], 4), file_type=csv, has_header=false
+
+query III
+SELECT l.value, r.range_key, r.value
+FROM (SELECT range_key, value FROM range_partitioned WHERE value <= 150) l
+RIGHT JOIN range_partitioned_shifted r ON l.range_key = r.range_key
+ORDER BY r.range_key;
+----
+10 1 10
+50 5 50
+100 10 100
+150 15 150
+NULL 20 200
+NULL 25 250
+NULL 30 300
+NULL 35 350
+
+##########
+# TEST 22: Right Join on Composite Key With Range Subset Rehashes
+# The join key (range_key, non_range_key) is a strict superset of the exposed
Review Comment:
coudl we keep explanations concise 👍
--
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]
