This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-23583-a8d1af6b8a82ded56eb547ad698b224c1f8bf7be in repository https://gitbox.apache.org/repos/asf/datafusion.git
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]
