gene-bordegaray commented on code in PR #23487:
URL: https://github.com/apache/datafusion/pull/23487#discussion_r3622576193
##########
datafusion/core/tests/physical_optimizer/enforce_distribution.rs:
##########
@@ -933,6 +933,79 @@ fn range_window_rehashes_incompatible_range_partitioning()
-> Result<()> {
Ok(())
}
+#[test]
+fn range_left_mark_hash_join_reuses_range_partitioning() -> Result<()> {
+ let left = parquet_exec_with_output_partitioning(range_partitioning(
+ "a",
+ [10, 20, 30],
+ SortOptions::default(),
+ )?);
+ let right = parquet_exec_with_output_partitioning(range_partitioning(
+ "a",
+ [10, 20, 30],
+ SortOptions::default(),
+ )?);
+ let join_on = vec![(
+ Arc::new(Column::new_with_schema("a", &left.schema())?) as _,
+ Arc::new(Column::new_with_schema("a", &right.schema())?) as _,
+ )];
+ let join = hash_join_exec(left, right, &join_on, &JoinType::LeftMark);
+
+ let plan = TestConfig::default()
+ .with_query_execution_partitions(4)
+ .to_plan(join, &DISTRIB_DISTRIB_SORT);
+
+ assert_plan!(
+ plan,
+ @r"
+ HashJoinExec: mode=Partitioned, join_type=LeftMark, on=[(a@0, a@0)]
+ DataSourceExec: file_groups={4 groups: [[p0], [p1], [p2], [p3]]},
projection=[a, b, c, d, e], output_partitioning=Range([a@0 ASC], [(10), (20),
(30)], 4), file_type=parquet
+ DataSourceExec: file_groups={4 groups: [[p0], [p1], [p2], [p3]]},
projection=[a, b, c, d, e], output_partitioning=Range([a@0 ASC], [(10), (20),
(30)], 4), file_type=parquet
+ "
+ );
+
+ Ok(())
+}
+
+#[test]
+fn range_left_anti_hash_join_rehashes_incompatible_sort_options() ->
Result<()> {
+ let left = parquet_exec_with_output_partitioning(range_partitioning(
+ "a",
+ [10, 20, 30],
+ SortOptions::default(),
+ )?);
+ let right = parquet_exec_with_output_partitioning(range_partitioning(
Review Comment:
we test the sort options with a right test in #23484 . I expect the covergae
to largerly spill over to left joins here. Coul.d we replace this with a nulls
options test for that coverage?
##########
datafusion/sqllogictest/src/test_context/range_partitioning.rs:
##########
@@ -130,6 +130,34 @@ pub(super) fn register_range_partitioned_table(ctx:
&SessionContext) {
],
Some(shifted_output_partitioning),
);
+
+ // Same rows as `range_partitioned` but split into only three range
+ // partitions on `range_key`. Used to exercise the co-partition check when
+ // two Range inputs disagree on partition count.
+ let narrow_output_partitioning = Partitioning::Range(
Review Comment:
just a personal note: this should disappear after merge with updated main
happens
##########
datafusion/sqllogictest/test_files/range_partitioning.slt:
##########
@@ -395,27 +395,123 @@ ORDER BY l.non_range_key, l.value, r.value;
2 350 350
##########
-# TEST 12: Non-Inner Range Join Repartitions
-# Only inner partitioned hash joins opt in to Range satisfying KeyPartitioned
-# requirements. Non-inner joins keep using Hash repartitioning.
+# TEST 12: Left-Side Range Hash Joins
Review Comment:
lets keep the comment scoped to just what test 12 i showing which is that
compatbiel ranges satisfy the join requirement
##########
datafusion/sqllogictest/test_files/range_partitioning.slt:
##########
@@ -395,27 +395,123 @@ ORDER BY l.non_range_key, l.value, r.value;
2 350 350
##########
-# TEST 12: Non-Inner Range Join Repartitions
-# Only inner partitioned hash joins opt in to Range satisfying KeyPartitioned
-# requirements. Non-inner joins keep using Hash repartitioning.
+# TEST 12: Left-Side Range Hash Joins
+# Left-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. Incompatible split points, partition counts,
+# or join key expressions still use Hash repartitioning to repair the inputs.
##########
query TT
EXPLAIN SELECT l.range_key, l.value, r.value
FROM range_partitioned l
-LEFT JOIN range_partitioned r ON l.range_key = r.range_key;
+LEFT JOIN (SELECT range_key, value FROM range_partitioned WHERE value <= 150) r
+ON l.range_key = r.range_key;
+----
+physical_plan
+01)HashJoinExec: mode=Partitioned, join_type=Left, on=[(range_key@0,
range_key@0)], projection=[range_key@0, value@1, value@3]
+02)--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
+03)--FilterExec: value@1 <= 150
Review Comment:
I like 👍
##########
datafusion/sqllogictest/test_files/range_partitioning.slt:
##########
@@ -395,27 +395,123 @@ ORDER BY l.non_range_key, l.value, r.value;
2 350 350
##########
-# TEST 12: Non-Inner Range Join Repartitions
-# Only inner partitioned hash joins opt in to Range satisfying KeyPartitioned
-# requirements. Non-inner joins keep using Hash repartitioning.
+# TEST 12: Left-Side Range Hash Joins
+# Left-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. Incompatible split points, partition counts,
+# or join key expressions still use Hash repartitioning to repair the inputs.
##########
query TT
EXPLAIN SELECT l.range_key, l.value, r.value
FROM range_partitioned l
-LEFT JOIN range_partitioned r ON l.range_key = r.range_key;
+LEFT JOIN (SELECT range_key, value FROM range_partitioned WHERE value <= 150) r
+ON l.range_key = r.range_key;
+----
+physical_plan
+01)HashJoinExec: mode=Partitioned, join_type=Left, on=[(range_key@0,
range_key@0)], projection=[range_key@0, value@1, value@3]
+02)--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
+03)--FilterExec: value@1 <= 150
+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
+
+query III
+SELECT l.range_key, l.value, r.value
+FROM range_partitioned l
+LEFT JOIN (SELECT range_key, value FROM range_partitioned WHERE value <= 150) 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 NULL
+25 250 NULL
+30 300 NULL
+35 350 NULL
+
+query TT
+EXPLAIN SELECT l.range_key, l.value
+FROM range_partitioned l
+LEFT SEMI JOIN (SELECT range_key FROM range_partitioned WHERE value <= 150) r
+ON l.range_key = r.range_key;
+----
+physical_plan
+01)HashJoinExec: mode=Partitioned, join_type=LeftSemi, on=[(range_key@0,
range_key@0)]
+02)--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
+03)--FilterExec: value@1 <= 150, projection=[range_key@0]
+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
+
+query II
+SELECT l.range_key, l.value
+FROM range_partitioned l
+LEFT SEMI JOIN (SELECT range_key FROM range_partitioned WHERE value <= 150) r
+ON l.range_key = r.range_key
+ORDER BY l.range_key;
+----
+1 10
+5 50
+10 100
+15 150
+
+query TT
+EXPLAIN SELECT l.range_key, l.value
+FROM range_partitioned l
+LEFT ANTI JOIN (SELECT range_key FROM range_partitioned WHERE value <= 150) r
+ON l.range_key = r.range_key;
+----
+physical_plan
+01)HashJoinExec: mode=Partitioned, join_type=LeftAnti, on=[(range_key@0,
range_key@0)]
+02)--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
+03)--FilterExec: value@1 <= 150, projection=[range_key@0]
+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
+
+query II
+SELECT l.range_key, l.value
+FROM range_partitioned l
+LEFT ANTI JOIN (SELECT range_key FROM range_partitioned WHERE value <= 150) r
+ON l.range_key = r.range_key
+ORDER BY l.range_key;
+----
+20 200
+25 250
+30 300
+35 350
+
+# Range([range_key]) is only a subset of the composite join key, so the
Review Comment:
I would prefer these to be separated by a new
```text
##########
# TEST ...
##########
```
block. I think separating at the naive satisfaction cases, then separate at
subset, then at diff partition counts, then the subquery
--
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]
