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 ed22a4aa19 feat: support co-partitioned range right-side equi hash
joins (#23484)
ed22a4aa19 is described below
commit ed22a4aa19693fccfd47a5beded42d4be1ac3dee
Author: Gustavo Schneiter <[email protected]>
AuthorDate: Tue Jul 21 04:06:08 2026 -0300
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.
---
.../physical_optimizer/enforce_distribution.rs | 73 +++++
.../tests/physical_optimizer/sanity_checker.rs | 26 ++
.../physical-plan/src/joins/hash_join/exec.rs | 10 +-
.../src/test_context/range_partitioning.rs | 28 ++
.../sqllogictest/test_files/range_partitioning.slt | 338 +++++++++++++++++++--
5 files changed, 449 insertions(+), 26 deletions(-)
diff --git a/datafusion/core/tests/physical_optimizer/enforce_distribution.rs
b/datafusion/core/tests/physical_optimizer/enforce_distribution.rs
index 27079011ea..5d405c50cb 100644
--- a/datafusion/core/tests/physical_optimizer/enforce_distribution.rs
+++ b/datafusion/core/tests/physical_optimizer/enforce_distribution.rs
@@ -870,6 +870,79 @@ fn
range_inner_hash_join_rehashes_incompatible_range_partitioning() -> Result<()
Ok(())
}
+#[test]
+fn range_right_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::RightMark);
+
+ let plan = TestConfig::default()
+ .with_query_execution_partitions(4)
+ .to_plan(join, &DISTRIB_DISTRIB_SORT);
+
+ assert_plan!(
+ plan,
+ @r"
+ HashJoinExec: mode=Partitioned, join_type=RightMark, 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_right_semi_hash_join_rehashes_incompatible_sort_options() ->
Result<()> {
+ let left = parquet_exec_with_output_partitioning(range_partitioning(
+ "a",
+ [20],
+ SortOptions::default(),
+ )?);
+ let right = parquet_exec_with_output_partitioning(range_partitioning(
+ "a",
+ [20],
+ SortOptions {
+ descending: true,
+ nulls_first: true,
+ },
+ )?);
+ 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::RightSemi);
+
+ let plan = TestConfig::default()
+ .with_query_execution_partitions(4)
+ .to_plan(join, &DISTRIB_DISTRIB_SORT);
+
+ assert_plan!(
+ plan,
+ @r"
+ HashJoinExec: mode=Partitioned, join_type=RightSemi, on=[(a@0, a@0)]
+ RepartitionExec: partitioning=Hash([a@0], 4), input_partitions=2
+ DataSourceExec: file_groups={2 groups: [[p0], [p1]]}, projection=[a,
b, c, d, e], output_partitioning=Range([a@0 ASC], [(20)], 2), file_type=parquet
+ RepartitionExec: partitioning=Hash([a@0], 4), input_partitions=2
+ DataSourceExec: file_groups={2 groups: [[p0], [p1]]}, projection=[a,
b, c, d, e], output_partitioning=Range([a@0 DESC], [(20)], 2), file_type=parquet
+ "
+ );
+
+ Ok(())
+}
+
#[test]
fn range_window_reuses_range_partitioning() -> Result<()> {
let input = parquet_exec_with_output_partitioning(range_partitioning(
diff --git a/datafusion/core/tests/physical_optimizer/sanity_checker.rs
b/datafusion/core/tests/physical_optimizer/sanity_checker.rs
index 3c426e2b09..184125dcbe 100644
--- a/datafusion/core/tests/physical_optimizer/sanity_checker.rs
+++ b/datafusion/core/tests/physical_optimizer/sanity_checker.rs
@@ -445,6 +445,32 @@ fn
test_partitioned_hash_join_requires_co_partitioned_children() -> Result<()> {
Ok(())
}
+#[test]
+fn test_partitioned_right_hash_join_requires_co_partitioned_children() ->
Result<()> {
+ let schema = create_test_schema2();
+ let join_on = vec![(col("a", &schema)?, col("a", &schema)?)];
+
+ let compatible_join = hash_join_exec(
+ range_partitioned_exec(&schema, "a", [10])?,
+ range_partitioned_exec(&schema, "a", [10])?,
+ join_on.clone(),
+ None,
+ &JoinType::Right,
+ )?;
+ assert_sanity_check(&compatible_join, true);
+
+ let incompatible_join = hash_join_exec(
+ range_partitioned_exec(&schema, "a", [10])?,
+ range_partitioned_exec(&schema, "a", [20])?,
+ join_on,
+ None,
+ &JoinType::Right,
+ )?;
+ assert_sanity_check(&incompatible_join, false);
+
+ Ok(())
+}
+
#[test]
fn test_sort_merge_join_requires_co_partitioned_children() -> Result<()> {
let schema = create_test_schema2();
diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs
b/datafusion/physical-plan/src/joins/hash_join/exec.rs
index 89fc4b5a81..64eb85ffee 100644
--- a/datafusion/physical-plan/src/joins/hash_join/exec.rs
+++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs
@@ -1294,7 +1294,15 @@ impl ExecutionPlan for HashJoinExec {
};
if self.mode == PartitionMode::Partitioned
- && matches!(self.join_type, JoinType::Inner | JoinType::Full)
+ && matches!(
+ self.join_type,
+ JoinType::Inner
+ | JoinType::Full
+ | JoinType::Right
+ | JoinType::RightSemi
+ | JoinType::RightAnti
+ | JoinType::RightMark
+ )
{
requirements.allow_range_satisfaction_for_key_partitioning()
} else {
diff --git a/datafusion/sqllogictest/src/test_context/range_partitioning.rs
b/datafusion/sqllogictest/src/test_context/range_partitioning.rs
index d60fdcecf4..4141e00014 100644
--- a/datafusion/sqllogictest/src/test_context/range_partitioning.rs
+++ b/datafusion/sqllogictest/src/test_context/range_partitioning.rs
@@ -131,6 +131,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(
+ 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))]),
+ ],
+ )
+ .expect("range partitioning should be valid"),
+ );
+
+ register_csv_listing_table(
+ ctx,
+ "range_partitioned_narrow",
+ Path::new(env!("CARGO_MANIFEST_DIR"))
+
.join("test_files/scratch_range_partitioning/range_partitioned_narrow"),
+ Arc::clone(&schema),
+ [
+ "1,1,10\n5,2,50\n",
+ "10,1,100\n15,2,150\n",
+ "20,1,200\n25,2,250\n30,1,300\n35,2,350\n",
+ ],
+ Some(narrow_output_partitioning),
+ );
+
let sparse_output_partitioning = Partitioning::Range(
RangePartitioning::try_new(
vec![col("range_key").sort(true, true)],
diff --git a/datafusion/sqllogictest/test_files/range_partitioning.slt
b/datafusion/sqllogictest/test_files/range_partitioning.slt
index 52bb695dbe..3a84e2c6d0 100644
--- a/datafusion/sqllogictest/test_files/range_partitioning.slt
+++ b/datafusion/sqllogictest/test_files/range_partitioning.slt
@@ -395,9 +395,10 @@ ORDER BY l.non_range_key, l.value, r.value;
2 350 350
##########
-# TEST 12: Unsupported
-# Only Inner and Full partitioned hash joins opt in to Range satisfying
-# KeyPartitioned requirements. Other join types keep using Hash repartitioning.
+# TEST 12: Left Range Join Repartitions
+# Only Inner, Full, and right-side (Right/RightSemi/RightAnti/RightMark)
+# partitioned hash joins opt in to Range satisfying KeyPartitioned
+# requirements. Other join types, such as Left, keep using Hash repartitioning.
##########
query TT
@@ -629,7 +630,294 @@ 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
+# Compatible Range inputs satisfy the join's partitioning requirements, so no
+# Hash repartitioning is inserted. The left filter keeps its Range partitioning
+# and the unmatched right rows above 150 are preserved.
+##########
+
+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=<slt:ignore>, 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=<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.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=<slt:ignore>, 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=<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 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=<slt:ignore>, 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=<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 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=<slt:ignore>, 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=<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.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: Composite-Key Right Join Repartitions
+# Range([range_key]) does not satisfy a partitioned join on
+# (range_key, non_range_key), so both sides repartition on the full key.
+##########
+
+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=<slt:ignore>, 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=<slt:ignore>, 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=<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=3
+05)----DataSourceExec: file_groups=<slt:ignore>, 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=<slt:ignore>, 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=<slt:ignore>, 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
+# Mark joins preserve matched, unmatched, and NULL-key marker behavior over
+# range-partitioned inputs.
+##########
+
+query TT
+EXPLAIN SELECT r.range_key, r.value
+FROM range_partitioned r
+WHERE r.non_range_key = 2 OR r.range_key IN (
+ SELECT range_key FROM range_partitioned WHERE value <= 150);
+----
+physical_plan
+01)FilterExec: non_range_key@1 = 2 OR mark@3, projection=[range_key@0, value@2]
+02)--HashJoinExec: mode=Partitioned, join_type=LeftMark, on=[(range_key@0,
range_key@0)]
+03)----RepartitionExec: partitioning=Hash([range_key@0], 4), input_partitions=4
+04)------DataSourceExec: file_groups=<slt:ignore>, 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([range_key@0], 4), input_partitions=4
+06)------FilterExec: value@1 <= 150, projection=[range_key@0]
+07)--------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
+
+# Matched rows have mark=true and are returned; unmatched rows have
+# mark=false and are only returned when non_range_key = 2.
+query II
+SELECT r.range_key, r.value
+FROM range_partitioned r
+WHERE r.non_range_key = 2 OR r.range_key IN (
+ SELECT range_key FROM range_partitioned WHERE value <= 150)
+ORDER BY r.range_key;
+----
+1 10
+5 50
+10 100
+15 150
+25 250
+35 350
+
+# NULL join keys on the build side never match: rows whose keys only "match"
+# the NULL entries keep a non-true marker and are filtered out unless the
+# non_range_key = 2 disjunct covers them.
+query II
+SELECT r.range_key, r.value
+FROM range_partitioned r
+WHERE r.non_range_key = 2 OR r.range_key IN (
+ SELECT CASE WHEN value <= 150 THEN range_key ELSE NULL END FROM
range_partitioned)
+ORDER BY r.range_key;
+----
+1 10
+5 50
+10 100
+15 150
+25 250
+35 350
+
+##########
+# TEST 26: Sort Merge Join Avoids Repartition for Compatible Range Inputs
# Compatible Range inputs satisfy SortMergeJoinExec's co-partitioned
# KeyPartitioned requirements.
##########
@@ -666,7 +954,7 @@ ORDER BY l.range_key;
35 350 350
##########
-# TEST 19: Sort Merge Join Repartitions Incompatible Range Inputs
+# TEST 27: Sort Merge Join Repartitions Incompatible Range Inputs
# Different Range split points do not satisfy SortMergeJoinExec's
# co-partitioned KeyPartitioned requirements.
##########
@@ -705,7 +993,7 @@ statement ok
reset datafusion.optimizer.prefer_hash_join;
##########
-# TEST 20: Symmetric Hash Join Avoids Repartition for Compatible Range Inputs
+# TEST 28: Symmetric Hash Join Avoids Repartition for Compatible Range Inputs
# Compatible Range streams satisfy SymmetricHashJoinExec's co-partitioned
# KeyPartitioned requirements.
##########
@@ -739,7 +1027,7 @@ FULL JOIN unbounded_range_like r ON l.range_key =
r.range_key;
5 50 50
##########
-# TEST 21: Symmetric Hash Join Repartitions Incompatible Range Inputs
+# TEST 29: Symmetric Hash Join Repartitions Incompatible Range Inputs
# Different Range split points do not satisfy SymmetricHashJoinExec's
# co-partitioned KeyPartitioned requirements.
##########
@@ -772,7 +1060,7 @@ FULL JOIN unbounded_range_like_shifted r ON l.range_key =
r.range_key;
5 50 50
##########
-# TEST 22: Full Outer Join on Range Partition Column
+# TEST 30: 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.
##########
@@ -803,7 +1091,7 @@ ORDER BY l.range_key;
35 350 350
##########
-# TEST 23: Full Outer Join Incompatible Range Repartitions
+# TEST 31: 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.
##########
@@ -836,7 +1124,7 @@ ORDER BY l.range_key;
35 350 350
##########
-# TEST 24: Full Outer Join Produces Matched and Unmatched Rows
+# TEST 32: 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
@@ -881,7 +1169,7 @@ statement ok
reset datafusion.optimizer.preserve_file_partitions;
##########
-# TEST 25: Union of Range Partitioned Inputs
+# TEST 33: 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.
##########
@@ -930,7 +1218,7 @@ set datafusion.optimizer.preserve_file_partitions = 0;
##########
-# TEST 26: Window on Range Partition Column
+# TEST 34: Window on Range Partition Column
# Range([range_key]) colocates equal range_key values, so
# PARTITION BY range_key is satisfied without a hash repartition.
##########
@@ -958,7 +1246,7 @@ SELECT range_key, SUM(value) OVER (PARTITION BY range_key
ORDER BY value) FROM r
##########
-# TEST 27: Unbounded-Frame Window on Range Partition Column
+# TEST 35: 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.
@@ -987,7 +1275,7 @@ SELECT range_key, SUM(value) OVER (PARTITION BY range_key
ORDER BY value ROWS BE
##########
-# TEST 28: Window on Non-Range Column Rehashes
+# TEST 36: 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.
##########
@@ -1016,7 +1304,7 @@ SELECT non_range_key, value, SUM(value) OVER (PARTITION
BY non_range_key ORDER B
##########
-# TEST 29: Unbounded-Frame Window on Non-Range Column Rehashes
+# TEST 37: 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.
@@ -1046,7 +1334,7 @@ SELECT non_range_key, value, SUM(value) OVER (PARTITION
BY non_range_key ORDER B
##########
-# TEST 30: Window Subset Satisfaction on Range Partition Column
+# TEST 38: 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.
@@ -1078,7 +1366,7 @@ SELECT range_key, SUM(value) OVER (PARTITION BY
range_key, non_range_key ORDER B
##########
-# TEST 31: Window Subset Rehashes Below Subset Threshold
+# TEST 39: 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.
@@ -1117,7 +1405,7 @@ reset datafusion.optimizer.preserve_file_partitions;
##########
-# TEST 32: Window Without Partition Keys Uses a Single Partition
+# TEST 40: Window Without Partition Keys Uses a Single Partition
# A window with no PARTITION BY requires a single partition; range
# partitioning is not applicable.
##########
@@ -1147,7 +1435,7 @@ SELECT range_key, SUM(value) OVER (ORDER BY value) FROM
range_partitioned ORDER
##########
-# TEST 33: PartitionedTopK on Range Partition Column
+# TEST 41: PartitionedTopK on Range Partition Column
# Exact Range([range_key]) satisfies the TopK partition key and avoids
repartitioning.
##########
@@ -1190,7 +1478,7 @@ ORDER BY range_key;
##########
-# TEST 34: PartitionedTopK on Non-Range Column
+# TEST 42: PartitionedTopK on Non-Range Column
# Partitioning on a non-range key cannot reuse Range([range_key]) and
# requires hash repartitioning.
##########
@@ -1226,7 +1514,7 @@ ORDER BY non_range_key;
##########
-# TEST 35: PartitionedTopK Reuses Range Subset Partitioning
+# TEST 43: PartitionedTopK Reuses Range Subset Partitioning
# With subset threshold met and preserve-file disabled, Range([range_key])
# satisfies partitioning by (range_key, non_range_key).
##########
@@ -1267,7 +1555,7 @@ ORDER BY range_key, non_range_key;
##########
-# TEST 36: Range Subset PartitionedTopK Rehashes Below Subset Threshold
+# TEST 44: 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.
@@ -1305,7 +1593,7 @@ statement ok
reset datafusion.optimizer.enable_window_topn;
##########
-# TEST 34: Subset of Inputs Compatible Does Not Trigger InterleaveExec
+# TEST 45: Subset of Inputs Compatible Does Not Trigger InterleaveExec
# In a three-way union, two inputs share the same Range split points [10,20,30]
# while the third has a partially-overlapping but different set [15,20,30].
# can_interleave requires ALL inputs to match, so UnionExec is kept.
@@ -1361,7 +1649,7 @@ ORDER BY range_key, value;
35 350
##########
-# TEST 35: Incompatible Range Split Points Falls Back to UnionExec
+# TEST 46: Incompatible Range Split Points Falls Back to UnionExec
# Two range-partitioned inputs with different split points cannot be
interleaved,
# so the optimizer keeps UnionExec instead of converting to InterleaveExec.
##########
@@ -1403,7 +1691,7 @@ ORDER BY range_key, value;
35 350
##########
-# TEST 36: InterleaveExec Propagates Range Partitioning to Aggregate
+# TEST 47: InterleaveExec Propagates Range Partitioning to Aggregate
# InterleaveExec outputs the same Range partitioning as its compatible inputs,
# allowing a downstream aggregate on range_key to run SinglePartitioned without
# a Hash repartition.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
