This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new 0f5790af04 feat: allow Full joins to reuse range co-partitioning in
HashJoinExec (#23583)
0f5790af04 is described below
commit 0f5790af040fc18cd941f6c6d6f0faeb3df60a9a
Author: Matthew Patton <[email protected]>
AuthorDate: Mon Jul 20 03:17:21 2026 -0400
feat: allow Full joins to reuse range co-partitioning in HashJoinExec
(#23583)
## Which issue does this PR close?
- Closes #23454.
## Rationale for this change
#23184 let compatible range-partitioned inputs satisfy inner partitioned
hash joins without repartitioning. Full partitioned equi joins still
always went through the conservative hash-repartition path, even when
both inputs were already co-partitioned by range on the join key(s). The
per-partition unmatched-row tracking in `HashJoinExec` is already
partition-local under `PartitionMode::Partitioned` (not shared globally
like in `CollectLeft`), so Full-join semantics generalize cleanly to
range co-partitioning with no additional bookkeeping required.
## What changes are included in this PR?
- Extend `HashJoinExec::input_distribution_requirements()` to opt
`JoinType::Full` in to
`allow_range_satisfaction_for_key_partitioning()`, alongside the
existing `JoinType::Inner` case. The underlying `co_partitioned` /
`compatible_co_partitioning_layout` / `co_partitioning_satisfied` logic
in `distribution_requirements.rs` was already join-type-agnostic, so no
changes were needed there.
- Add planner unit tests in `enforce_distribution.rs` covering both the
compatible-layout case (no repartition inserted) and the
incompatible-split-points case (repartition still inserted) for
`JoinType::Full`.
- Add a `range_partitioned_sparse` sqllogictest fixture table with the
same partition layout as `range_partitioned` but only partially
overlapping keys, and add sqllogictest coverage in
`range_partitioning.slt` for: a compatible Full join avoiding
repartition, an incompatible Full join still repartitioning, and
matched/left-only/right-only unmatched rows produced correctly by a
co-partitioned Full join.
## Are these changes tested?
Yes:
- Two new Rust unit tests in
`datafusion/core/tests/physical_optimizer/enforce_distribution.rs`
assert on the physical plan shape (repartition inserted or not) for
compatible and incompatible range layouts.
- Three new sqllogictest cases in
`datafusion/sqllogictest/test_files/range_partitioning.slt` exercise the
feature end-to-end against real data, including matched rows, left-only
unmatched rows, and right-only unmatched rows for a Full outer join.
- Existing `enforce_distribution` tests, `range_partitioning.slt`, and
proto roundtrip tests all continue to pass.
## Are there any user-facing changes?
Yes: `EXPLAIN` output for `Full` joins over compatible range-partitioned
inputs will no longer show a `RepartitionExec`, and such queries will
avoid the associated hash-shuffle cost at execution time. No public API
changes.
---------
Co-authored-by: Matthew Patton
<[email protected]>
---
.../physical_optimizer/enforce_distribution.rs | 84 +++++++++++++
.../physical-plan/src/joins/hash_join/exec.rs | 4 +-
.../src/test_context/range_partitioning.rs | 29 ++++-
.../sqllogictest/test_files/range_partitioning.slt | 130 ++++++++++++++++++---
4 files changed, 230 insertions(+), 17 deletions(-)
diff --git a/datafusion/core/tests/physical_optimizer/enforce_distribution.rs
b/datafusion/core/tests/physical_optimizer/enforce_distribution.rs
index 3292ada0a8..27079011ea 100644
--- a/datafusion/core/tests/physical_optimizer/enforce_distribution.rs
+++ b/datafusion/core/tests/physical_optimizer/enforce_distribution.rs
@@ -933,6 +933,90 @@ fn range_window_rehashes_incompatible_range_partitioning()
-> Result<()> {
Ok(())
}
+#[test]
+fn range_full_hash_join_reuses_compatible_range_partitioning() -> Result<()> {
+ let left = parquet_exec_with_output_partitioning(range_partitioning(
+ "a",
+ [10, 20, 30],
+ SortOptions::default(),
+ )?);
+ let right = projection_exec_with_alias(
+ parquet_exec_with_output_partitioning(range_partitioning(
+ "a",
+ [10, 20, 30],
+ SortOptions::default(),
+ )?),
+ vec![
+ ("a".to_string(), "a1".to_string()),
+ ("b".to_string(), "b1".to_string()),
+ ],
+ );
+ let join_on = vec![(
+ Arc::new(Column::new_with_schema("a", &left.schema())?) as _,
+ Arc::new(Column::new_with_schema("a1", &right.schema())?) as _,
+ )];
+ let join = hash_join_exec(left, right, &join_on, &JoinType::Full);
+
+ let plan = TestConfig::default()
+ .with_query_execution_partitions(4)
+ .to_plan(join, &DISTRIB_DISTRIB_SORT);
+
+ assert_plan!(
+ plan,
+ @r"
+ HashJoinExec: mode=Partitioned, join_type=Full, on=[(a@0, a1@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
+ ProjectionExec: expr=[a@0 as a1, b@1 as b1]
+ 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_full_hash_join_rehashes_incompatible_range_partitioning() ->
Result<()> {
+ let left = parquet_exec_with_output_partitioning(range_partitioning(
+ "a",
+ [10, 20, 30],
+ SortOptions::default(),
+ )?);
+ let right = projection_exec_with_alias(
+ parquet_exec_with_output_partitioning(range_partitioning(
+ "a",
+ [10, 30, 40],
+ SortOptions::default(),
+ )?),
+ vec![
+ ("a".to_string(), "a1".to_string()),
+ ("b".to_string(), "b1".to_string()),
+ ],
+ );
+ let join_on = vec![(
+ Arc::new(Column::new_with_schema("a", &left.schema())?) as _,
+ Arc::new(Column::new_with_schema("a1", &right.schema())?) as _,
+ )];
+ let join = hash_join_exec(left, right, &join_on, &JoinType::Full);
+
+ let plan = TestConfig::default()
+ .with_query_execution_partitions(4)
+ .to_plan(join, &DISTRIB_DISTRIB_SORT);
+
+ assert_plan!(
+ plan,
+ @r"
+ HashJoinExec: mode=Partitioned, join_type=Full, on=[(a@0, a1@0)]
+ RepartitionExec: partitioning=Hash([a@0], 4), input_partitions=4
+ 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
+ RepartitionExec: partitioning=Hash([a1@0], 4), input_partitions=4
+ ProjectionExec: expr=[a@0 as a1, b@1 as b1]
+ DataSourceExec: file_groups={4 groups: [[p0], [p1], [p2], [p3]]},
projection=[a, b, c, d, e], output_partitioning=Range([a@0 ASC], [(10), (30),
(40)], 4), file_type=parquet
+ "
+ );
+
+ Ok(())
+}
+
#[test]
fn multi_hash_joins() -> Result<()> {
let left = parquet_exec();
diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs
b/datafusion/physical-plan/src/joins/hash_join/exec.rs
index c622663f03..89fc4b5a81 100644
--- a/datafusion/physical-plan/src/joins/hash_join/exec.rs
+++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs
@@ -1293,7 +1293,9 @@ impl ExecutionPlan for HashJoinExec {
]),
};
- if self.mode == PartitionMode::Partitioned && self.join_type ==
JoinType::Inner {
+ if self.mode == PartitionMode::Partitioned
+ && matches!(self.join_type, JoinType::Inner | JoinType::Full)
+ {
requirements.allow_range_satisfaction_for_key_partitioning()
} else {
requirements
diff --git a/datafusion/sqllogictest/src/test_context/range_partitioning.rs
b/datafusion/sqllogictest/src/test_context/range_partitioning.rs
index 4c8545feca..d60fdcecf4 100644
--- a/datafusion/sqllogictest/src/test_context/range_partitioning.rs
+++ b/datafusion/sqllogictest/src/test_context/range_partitioning.rs
@@ -121,7 +121,7 @@ pub(super) fn register_range_partitioned_table(ctx:
&SessionContext) {
"range_partitioned_shifted",
Path::new(env!("CARGO_MANIFEST_DIR"))
.join("test_files/scratch_range_partitioning/range_partitioned_shifted"),
- schema,
+ Arc::clone(&schema),
[
"1,1,10\n5,2,50\n10,1,100\n",
"15,2,150\n",
@@ -130,6 +130,33 @@ pub(super) fn register_range_partitioned_table(ctx:
&SessionContext) {
],
Some(shifted_output_partitioning),
);
+
+ let sparse_output_partitioning = Partitioning::Range(
+ RangePartitioning::try_new(
+ vec![col("range_key").sort(true, true)],
+ vec![
+ SplitPoint::new(vec![ScalarValue::Int32(Some(10))]),
+ SplitPoint::new(vec![ScalarValue::Int32(Some(20))]),
+ SplitPoint::new(vec![ScalarValue::Int32(Some(30))]),
+ ],
+ )
+ .expect("range partitioning should be valid"),
+ );
+
+ register_csv_listing_table(
+ ctx,
+ "range_partitioned_sparse",
+ Path::new(env!("CARGO_MANIFEST_DIR"))
+
.join("test_files/scratch_range_partitioning/range_partitioned_sparse"),
+ schema,
+ [
+ "5,2,50\n8,3,80\n",
+ "10,1,100\n",
+ "20,1,200\n",
+ "30,1,300\n40,4,400\n",
+ ],
+ Some(sparse_output_partitioning),
+ );
}
fn register_csv_listing_table(
diff --git a/datafusion/sqllogictest/test_files/range_partitioning.slt
b/datafusion/sqllogictest/test_files/range_partitioning.slt
index f91b296a2a..52bb695dbe 100644
--- a/datafusion/sqllogictest/test_files/range_partitioning.slt
+++ b/datafusion/sqllogictest/test_files/range_partitioning.slt
@@ -395,9 +395,9 @@ 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: Unsupported
+# Only Inner and Full partitioned hash joins opt in to Range satisfying
+# KeyPartitioned requirements. Other join types keep using Hash repartitioning.
##########
query TT
@@ -771,6 +771,106 @@ FULL JOIN unbounded_range_like_shifted r ON l.range_key =
r.range_key;
35 350 350
5 50 50
+##########
+# TEST 22: Full Outer Join on Range Partition Column
+# Full partitioned hash joins also opt in to Range satisfying KeyPartitioned
+# requirements, so compatible Range layouts avoid Hash repartitioning here too.
+##########
+
+query TT
+EXPLAIN SELECT l.range_key, l.value, r.value
+FROM range_partitioned l
+FULL JOIN range_partitioned r ON l.range_key = r.range_key;
+----
+physical_plan
+01)HashJoinExec: mode=Partitioned, join_type=Full, 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)--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
+FULL 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
+
+##########
+# TEST 23: Full Outer Join Incompatible Range Repartitions
+# Same as TEST 10, but for Full: differing split points between the two
+# Range-partitioned inputs still require Hash repartitioning to co-partition.
+##########
+
+query TT
+EXPLAIN SELECT l.range_key, l.value, r.value
+FROM range_partitioned l
+FULL JOIN range_partitioned_shifted r ON l.range_key = r.range_key;
+----
+physical_plan
+01)HashJoinExec: mode=Partitioned, join_type=Full, on=[(range_key@0,
range_key@0)], projection=[range_key@0, value@1, value@3]
+02)--RepartitionExec: partitioning=Hash([range_key@0], 4), input_partitions=4
+03)----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
+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], [(15), (20), (30)], 4),
file_type=csv, has_header=false
+
+query III
+SELECT l.range_key, l.value, r.value
+FROM range_partitioned l
+FULL JOIN range_partitioned_shifted 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
+
+##########
+# TEST 24: Full Outer Join Produces Matched and Unmatched Rows
+# `range_partitioned` and `range_partitioned_sparse` share the same Range
+# split points/partition count but only partially overlapping range_key
+# values, so this exercises matched rows, left-only unmatched rows (NULLs on
+# the right), and right-only unmatched rows (NULLs on the left) while still
+# avoiding Hash repartitioning.
+##########
+
+query TT
+EXPLAIN SELECT l.range_key, r.range_key, l.value, r.value
+FROM range_partitioned l
+FULL JOIN range_partitioned_sparse r ON l.range_key = r.range_key;
+----
+physical_plan
+01)HashJoinExec: mode=Partitioned, join_type=Full, on=[(range_key@0,
range_key@0)], projection=[range_key@0, range_key@2, 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)--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 IIII
+SELECT l.range_key, r.range_key, l.value, r.value
+FROM range_partitioned l
+FULL JOIN range_partitioned_sparse r ON l.range_key = r.range_key
+ORDER BY l.range_key, r.range_key;
+----
+1 NULL 10 NULL
+5 5 50 50
+10 10 100 100
+15 NULL 150 NULL
+20 20 200 200
+25 NULL 250 NULL
+30 30 300 300
+35 NULL 350 NULL
+NULL 8 NULL 80
+NULL 40 NULL 400
+
statement ok
reset datafusion.optimizer.prefer_hash_join;
@@ -781,7 +881,7 @@ statement ok
reset datafusion.optimizer.preserve_file_partitions;
##########
-# TEST 22: Union of Range Partitioned Inputs
+# TEST 25: Union of Range Partitioned Inputs
# Each input exposes the same Range partitioning on range_key, so the optimizer
# converts UnionExec to InterleaveExec to avoid redundant repartitioning.
##########
@@ -830,7 +930,7 @@ set datafusion.optimizer.preserve_file_partitions = 0;
##########
-# TEST 23: Window on Range Partition Column
+# TEST 26: Window on Range Partition Column
# Range([range_key]) colocates equal range_key values, so
# PARTITION BY range_key is satisfied without a hash repartition.
##########
@@ -858,7 +958,7 @@ SELECT range_key, SUM(value) OVER (PARTITION BY range_key
ORDER BY value) FROM r
##########
-# TEST 24: Unbounded-Frame Window on Range Partition Column
+# TEST 27: Unbounded-Frame Window on Range Partition Column
# The unbounded frame makes DataFusion use WindowAggExec instead of
# BoundedWindowAggExec, which likewise reuses Range partitioning without a
# hash repartition.
@@ -887,7 +987,7 @@ SELECT range_key, SUM(value) OVER (PARTITION BY range_key
ORDER BY value ROWS BE
##########
-# TEST 25: Window on Non-Range Column Rehashes
+# TEST 28: Window on Non-Range Column Rehashes
# Range([range_key]) does not colocate non_range_key values, so
# PARTITION BY non_range_key still requires a hash repartition.
##########
@@ -916,7 +1016,7 @@ SELECT non_range_key, value, SUM(value) OVER (PARTITION BY
non_range_key ORDER B
##########
-# TEST 26: Unbounded-Frame Window on Non-Range Column Rehashes
+# TEST 29: Unbounded-Frame Window on Non-Range Column Rehashes
# The unbounded frame makes DataFusion use WindowAggExec; Range([range_key])
# does not colocate non_range_key values, so PARTITION BY non_range_key
# still requires a hash repartition.
@@ -946,7 +1046,7 @@ SELECT non_range_key, value, SUM(value) OVER (PARTITION BY
non_range_key ORDER B
##########
-# TEST 27: Window Subset Satisfaction on Range Partition Column
+# TEST 30: Window Subset Satisfaction on Range Partition Column
# With the subset threshold met, Range([range_key]) satisfies
# PARTITION BY (range_key, non_range_key): equal composite keys share the
# same range_key, so they are already colocated.
@@ -978,7 +1078,7 @@ SELECT range_key, SUM(value) OVER (PARTITION BY range_key,
non_range_key ORDER B
##########
-# TEST 28: Window Subset Rehashes Below Subset Threshold
+# TEST 31: Window Subset Rehashes Below Subset Threshold
# Range([range_key]) is only a subset of PARTITION BY
# (range_key, non_range_key), so it should not satisfy the window key when
# subset satisfaction is disabled.
@@ -1017,7 +1117,7 @@ reset datafusion.optimizer.preserve_file_partitions;
##########
-# TEST 29: Window Without Partition Keys Uses a Single Partition
+# TEST 32: Window Without Partition Keys Uses a Single Partition
# A window with no PARTITION BY requires a single partition; range
# partitioning is not applicable.
##########
@@ -1047,7 +1147,7 @@ SELECT range_key, SUM(value) OVER (ORDER BY value) FROM
range_partitioned ORDER
##########
-# TEST 30: PartitionedTopK on Range Partition Column
+# TEST 33: PartitionedTopK on Range Partition Column
# Exact Range([range_key]) satisfies the TopK partition key and avoids
repartitioning.
##########
@@ -1090,7 +1190,7 @@ ORDER BY range_key;
##########
-# TEST 31: PartitionedTopK on Non-Range Column
+# TEST 34: PartitionedTopK on Non-Range Column
# Partitioning on a non-range key cannot reuse Range([range_key]) and
# requires hash repartitioning.
##########
@@ -1126,7 +1226,7 @@ ORDER BY non_range_key;
##########
-# TEST 32: PartitionedTopK Reuses Range Subset Partitioning
+# TEST 35: PartitionedTopK Reuses Range Subset Partitioning
# With subset threshold met and preserve-file disabled, Range([range_key])
# satisfies partitioning by (range_key, non_range_key).
##########
@@ -1167,7 +1267,7 @@ ORDER BY range_key, non_range_key;
##########
-# TEST 33: Range Subset PartitionedTopK Rehashes Below Subset Threshold
+# TEST 36: Range Subset PartitionedTopK Rehashes Below Subset Threshold
# Range([range_key]) is only a subset of PARTITION BY (range_key,
non_range_key),
# so it should not satisfy the TopK partition key when subset satisfaction is
# disabled.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
