From 22e4e4a48a1f24b728458d0f7971cfeaa6a375f2 Mon Sep 17 00:00:00 2001 From: Justin O'Dwyer Date: Sun, 12 Jul 2026 00:35:15 +0200 Subject: [PATCH 1/5] Add tests showing repartitions. --- .../enforce_distribution.rs | 108 ++++++++++++++++++ 1 file changed, 108 insertions(+) diff --git a/datafusion/core/tests/physical_optimizer/enforce_distribution.rs b/datafusion/core/tests/physical_optimizer/enforce_distribution.rs index e01311e25be8b..0669808ee10fd 100644 --- a/datafusion/core/tests/physical_optimizer/enforce_distribution.rs +++ b/datafusion/core/tests/physical_optimizer/enforce_distribution.rs @@ -870,6 +870,114 @@ fn range_inner_hash_join_rehashes_incompatible_range_partitioning() -> Result<() Ok(()) } +#[test] +fn left_hash_join_uses_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::Left); + + let plan = TestConfig::default() + .with_query_execution_partitions(4) + .to_plan(join, &DISTRIB_DISTRIB_SORT); + + println!("{}", displayable(plan.as_ref()).indent(true)); + + Ok(()) +} + +#[test] +fn left_semi_hash_join_uses_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::LeftSemi); + + let plan = TestConfig::default() + .with_query_execution_partitions(4) + .to_plan(join, &DISTRIB_DISTRIB_SORT); + + println!("{}", displayable(plan.as_ref()).indent(true)); + + Ok(()) +} + +#[test] +fn left_anti_hash_join_uses_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::LeftAnti); + + let plan = TestConfig::default() + .with_query_execution_partitions(4) + .to_plan(join, &DISTRIB_DISTRIB_SORT); + + println!("{}", displayable(plan.as_ref()).indent(true)); + + Ok(()) +} + +#[test] +fn left_mark_hash_join_uses_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); + + println!("{}", displayable(plan.as_ref()).indent(true)); + + Ok(()) +} + #[test] fn multi_hash_joins() -> Result<()> { let left = parquet_exec(); From b5135118b7dcb95f980626ab5863aa10d63b7379 Mon Sep 17 00:00:00 2001 From: Justin O'Dwyer Date: Sun, 12 Jul 2026 10:58:27 +0200 Subject: [PATCH 2/5] Adding more tests. --- .../enforce_distribution.rs | 108 ++++++++++++++++++ .../test_files/range_partitioning.slt | 7 +- 2 files changed, 112 insertions(+), 3 deletions(-) diff --git a/datafusion/core/tests/physical_optimizer/enforce_distribution.rs b/datafusion/core/tests/physical_optimizer/enforce_distribution.rs index 0669808ee10fd..d20ff3f4267b7 100644 --- a/datafusion/core/tests/physical_optimizer/enforce_distribution.rs +++ b/datafusion/core/tests/physical_optimizer/enforce_distribution.rs @@ -978,6 +978,114 @@ fn left_mark_hash_join_uses_range_partitioning() -> Result<()> { Ok(()) } +#[test] +fn left_hash_join_range_rehash_incompatible_split_points() -> Result<()> { + let left = parquet_exec_with_output_partitioning(range_partitioning( + "a", + [10, 15, 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::Left); + + let plan = TestConfig::default() + .with_query_execution_partitions(4) + .to_plan(join, &DISTRIB_DISTRIB_SORT); + + println!("{}", displayable(plan.as_ref()).indent(true)); + + Ok(()) +} + +#[test] +fn left_hash_join_range_rehash_incompatible_partition_counts() -> 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], + 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::LeftSemi); + + let plan = TestConfig::default() + .with_query_execution_partitions(4) + .to_plan(join, &DISTRIB_DISTRIB_SORT); + + println!("{}", displayable(plan.as_ref()).indent(true)); + + Ok(()) +} + +#[test] +fn left_hash_join_range_rehash_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( + "a", + [30, 20, 10], + SortOptions::default().with_descending(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::LeftAnti); + + let plan = TestConfig::default() + .with_query_execution_partitions(4) + .to_plan(join, &DISTRIB_DISTRIB_SORT); + + println!("{}", displayable(plan.as_ref()).indent(true)); + + Ok(()) +} + +#[test] +fn left_hash_join_range_rehash_incompatible_key_expressions() -> 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("b", &left.schema())?) as _, + Arc::new(Column::new_with_schema("b", &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); + + println!("{}", displayable(plan.as_ref()).indent(true)); + + Ok(()) +} + #[test] fn multi_hash_joins() -> Result<()> { let left = parquet_exec(); diff --git a/datafusion/sqllogictest/test_files/range_partitioning.slt b/datafusion/sqllogictest/test_files/range_partitioning.slt index 5d004ca7fdfa5..9f597f28bf660 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: 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 Range Join Repartitions +# Only inner 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 From b1991a4e4138c14f9172f6a993f336e3c090e135 Mon Sep 17 00:00:00 2001 From: Justin O'Dwyer Date: Wed, 15 Jul 2026 18:09:15 +0200 Subject: [PATCH 3/5] co-partitioned range left-side hash join --- .../enforce_distribution.rs | 191 +++--------------- .../physical-plan/src/joins/hash_join/exec.rs | 11 +- .../src/test_context/range_partitioning.rs | 30 ++- .../test_files/range_partitioning.slt | 180 ++++++++++++++++- 4 files changed, 236 insertions(+), 176 deletions(-) diff --git a/datafusion/core/tests/physical_optimizer/enforce_distribution.rs b/datafusion/core/tests/physical_optimizer/enforce_distribution.rs index d03358996a972..5687c60048295 100644 --- a/datafusion/core/tests/physical_optimizer/enforce_distribution.rs +++ b/datafusion/core/tests/physical_optimizer/enforce_distribution.rs @@ -934,88 +934,7 @@ fn range_window_rehashes_incompatible_range_partitioning() -> Result<()> { } #[test] -fn left_hash_join_uses_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::Left); - - let plan = TestConfig::default() - .with_query_execution_partitions(4) - .to_plan(join, &DISTRIB_DISTRIB_SORT); - - println!("{}", displayable(plan.as_ref()).indent(true)); - - Ok(()) -} - -#[test] -fn left_semi_hash_join_uses_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::LeftSemi); - - let plan = TestConfig::default() - .with_query_execution_partitions(4) - .to_plan(join, &DISTRIB_DISTRIB_SORT); - - println!("{}", displayable(plan.as_ref()).indent(true)); - - Ok(()) -} - -#[test] -fn left_anti_hash_join_uses_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::LeftAnti); - - let plan = TestConfig::default() - .with_query_execution_partitions(4) - .to_plan(join, &DISTRIB_DISTRIB_SORT); - - println!("{}", displayable(plan.as_ref()).indent(true)); - - Ok(()) -} - -#[test] -fn left_mark_hash_join_uses_range_partitioning() -> Result<()> { +fn range_left_mark_hash_join_reuses_range_partitioning() -> Result<()> { let left = parquet_exec_with_output_partitioning(range_partitioning( "a", [10, 20, 30], @@ -1036,67 +955,20 @@ fn left_mark_hash_join_uses_range_partitioning() -> Result<()> { .with_query_execution_partitions(4) .to_plan(join, &DISTRIB_DISTRIB_SORT); - println!("{}", displayable(plan.as_ref()).indent(true)); - - Ok(()) -} - -#[test] -fn left_hash_join_range_rehash_incompatible_split_points() -> Result<()> { - let left = parquet_exec_with_output_partitioning(range_partitioning( - "a", - [10, 15, 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::Left); - - let plan = TestConfig::default() - .with_query_execution_partitions(4) - .to_plan(join, &DISTRIB_DISTRIB_SORT); - - println!("{}", displayable(plan.as_ref()).indent(true)); - - Ok(()) -} - -#[test] -fn left_hash_join_range_rehash_incompatible_partition_counts() -> 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], - 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::LeftSemi); - - let plan = TestConfig::default() - .with_query_execution_partitions(4) - .to_plan(join, &DISTRIB_DISTRIB_SORT); - - println!("{}", displayable(plan.as_ref()).indent(true)); + 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 left_hash_join_range_rehash_incompatible_sort_options() -> Result<()> { +fn range_left_anti_hash_join_rehashes_incompatible_sort_options() -> Result<()> { let left = parquet_exec_with_output_partitioning(range_partitioning( "a", [10, 20, 30], @@ -1105,7 +977,10 @@ fn left_hash_join_range_rehash_incompatible_sort_options() -> Result<()> { let right = parquet_exec_with_output_partitioning(range_partitioning( "a", [30, 20, 10], - SortOptions::default().with_descending(true), + SortOptions { + descending: true, + nulls_first: true, + }, )?); let join_on = vec![( Arc::new(Column::new_with_schema("a", &left.schema())?) as _, @@ -1117,34 +992,16 @@ fn left_hash_join_range_rehash_incompatible_sort_options() -> Result<()> { .with_query_execution_partitions(4) .to_plan(join, &DISTRIB_DISTRIB_SORT); - println!("{}", displayable(plan.as_ref()).indent(true)); - - Ok(()) -} - -#[test] -fn left_hash_join_range_rehash_incompatible_key_expressions() -> 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("b", &left.schema())?) as _, - Arc::new(Column::new_with_schema("b", &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); - - println!("{}", displayable(plan.as_ref()).indent(true)); + assert_plan!( + plan, + @r" + HashJoinExec: mode=Partitioned, join_type=LeftAnti, on=[(a@0, a@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([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 DESC], [(30), (20), (10)], 4), file_type=parquet + " + ); Ok(()) } diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index 6f60155ea34a4..4426a53bcc26c 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -1293,7 +1293,16 @@ 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::Left + | JoinType::LeftSemi + | JoinType::LeftAnti + | JoinType::LeftMark + ) + { 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 4c8545fecaa16..8c2142568cedf 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,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"), + 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), + ); } fn register_csv_listing_table( diff --git a/datafusion/sqllogictest/test_files/range_partitioning.slt b/datafusion/sqllogictest/test_files/range_partitioning.slt index 0bef0f0adaed1..ace752db686fd 100644 --- a/datafusion/sqllogictest/test_files/range_partitioning.slt +++ b/datafusion/sqllogictest/test_files/range_partitioning.slt @@ -395,28 +395,123 @@ ORDER BY l.non_range_key, l.value, r.value; 2 350 350 ########## -# TEST 12: Left Range Join Repartitions -# Only inner 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. +# 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=, 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=, 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=, 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=, 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=, 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=, 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 +# co-partitioned hash join requirement is repaired with Hash repartitioning. +query TT +EXPLAIN SELECT l.range_key, l.non_range_key, l.value, r.value +FROM range_partitioned l +LEFT JOIN (SELECT range_key, non_range_key, value FROM range_partitioned WHERE value <= 150) 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=Left, 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=, 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)----FilterExec: value@2 <= 150 +06)------DataSourceExec: file_groups=, projection=[range_key, non_range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false + +# Different split points do not satisfy the co-partitioned layout requirement. +query TT +EXPLAIN SELECT l.range_key, l.value, r.value +FROM range_partitioned l +LEFT JOIN range_partitioned_shifted 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)--RepartitionExec: partitioning=Hash([range_key@0], 4), input_partitions=4 03)----DataSourceExec: file_groups=, 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=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +05)----DataSourceExec: file_groups=, 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 -LEFT JOIN range_partitioned r ON l.range_key = r.range_key +LEFT JOIN range_partitioned_shifted r ON l.range_key = r.range_key ORDER BY l.range_key; ---- 1 10 10 @@ -428,6 +523,77 @@ ORDER BY l.range_key; 30 300 300 35 350 350 +# Different partition counts do not satisfy the co-partitioned layout +# requirement. +query TT +EXPLAIN SELECT l.range_key, l.value, r.value +FROM range_partitioned l +LEFT JOIN range_partitioned_narrow 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)--RepartitionExec: partitioning=Hash([range_key@0], 4), input_partitions=4 +03)----DataSourceExec: file_groups=, 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=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20)], 3), file_type=csv, has_header=false + +# Range([range_key]) does not satisfy a join keyed on non_range_key. +query TT +EXPLAIN SELECT l.range_key, l.non_range_key, l.value, r.value +FROM range_partitioned l +LEFT JOIN range_partitioned r ON l.non_range_key = r.non_range_key; +---- +physical_plan +01)HashJoinExec: mode=Partitioned, join_type=Left, on=[(non_range_key@1, non_range_key@0)], projection=[range_key@0, non_range_key@1, value@2, value@4] +02)--RepartitionExec: partitioning=Hash([non_range_key@1], 4), input_partitions=4 +03)----DataSourceExec: file_groups=, 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([non_range_key@0], 4), input_partitions=4 +05)----DataSourceExec: file_groups=, projection=[non_range_key, value], output_partitioning=UnknownPartitioning(4), file_type=csv, has_header=false + +# SQL IN subqueries decorrelate to LeftMark joins. These queries pin matched, +# unmatched, and NULL marker behavior over compatible Range inputs. +query TT +EXPLAIN SELECT l.range_key, l.value +FROM range_partitioned l +WHERE l.non_range_key = 2 OR l.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)----DataSourceExec: file_groups=, 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)----FilterExec: value@1 <= 150, projection=[range_key@0] +05)------DataSourceExec: file_groups=, 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 +WHERE l.non_range_key = 2 OR l.range_key IN ( + SELECT range_key FROM range_partitioned WHERE value <= 150) +ORDER BY l.range_key; +---- +1 10 +5 50 +10 100 +15 150 +25 250 +35 350 + +query II +SELECT l.range_key, l.value +FROM range_partitioned l +WHERE l.non_range_key = 2 OR l.range_key IN ( + SELECT CASE WHEN value <= 150 THEN range_key ELSE NULL END + FROM range_partitioned) +ORDER BY l.range_key; +---- +1 10 +5 50 +10 100 +15 150 +25 250 +35 350 + ########## # TEST 13: Compatible Range Join Repartitions to Increase Parallelism # Co-partitioning satisfaction does not prevent a repartition that increases From f3a38ce7a07b94bbc985d8539d7b0630b50370a8 Mon Sep 17 00:00:00 2001 From: Justin O'Dwyer Date: Tue, 21 Jul 2026 10:25:02 -0400 Subject: [PATCH 4/5] Updating from comments. --- .../enforce_distribution.rs | 10 +- .../test_files/range_partitioning.slt | 119 ++++++++++-------- 2 files changed, 71 insertions(+), 58 deletions(-) diff --git a/datafusion/core/tests/physical_optimizer/enforce_distribution.rs b/datafusion/core/tests/physical_optimizer/enforce_distribution.rs index df919c6db12ff..189650fe4afca 100644 --- a/datafusion/core/tests/physical_optimizer/enforce_distribution.rs +++ b/datafusion/core/tests/physical_optimizer/enforce_distribution.rs @@ -1125,7 +1125,7 @@ fn range_left_mark_hash_join_reuses_range_partitioning() -> Result<()> { } #[test] -fn range_left_anti_hash_join_rehashes_incompatible_sort_options() -> Result<()> { +fn range_left_anti_hash_join_rehashes_incompatible_null_options() -> Result<()> { let left = parquet_exec_with_output_partitioning(range_partitioning( "a", [10, 20, 30], @@ -1133,10 +1133,10 @@ fn range_left_anti_hash_join_rehashes_incompatible_sort_options() -> Result<()> )?); let right = parquet_exec_with_output_partitioning(range_partitioning( "a", - [30, 20, 10], + [10, 20, 30], SortOptions { - descending: true, - nulls_first: true, + descending: false, + nulls_first: false, }, )?); let join_on = vec![( @@ -1156,7 +1156,7 @@ fn range_left_anti_hash_join_rehashes_incompatible_sort_options() -> Result<()> 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([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 DESC], [(30), (20), (10)], 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 NULLS LAST], [(10), (20), (30)], 4), file_type=parquet " ); diff --git a/datafusion/sqllogictest/test_files/range_partitioning.slt b/datafusion/sqllogictest/test_files/range_partitioning.slt index 7244698a962dd..1e0a1582eac65 100644 --- a/datafusion/sqllogictest/test_files/range_partitioning.slt +++ b/datafusion/sqllogictest/test_files/range_partitioning.slt @@ -396,11 +396,8 @@ ORDER BY l.non_range_key, l.value, r.value; ########## # 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. +# Compatible Range layouts satisfy left-side partitioned hash join +# requirements without Hash repartitioning. ########## query TT @@ -479,6 +476,12 @@ ORDER BY l.range_key; 30 300 35 350 +########## +# TEST 13: Left-Side Range Hash Joins With Incomplete Range Keys +# Range partitioning covers only range_key, so joins requiring additional +# or different keys are repaired with Hash repartitioning. +########## + # Range([range_key]) is only a subset of the composite join key, so the # co-partitioned hash join requirement is repaired with Hash repartitioning. query TT @@ -495,6 +498,25 @@ physical_plan 05)----FilterExec: value@2 <= 150 06)------DataSourceExec: file_groups=, projection=[range_key, non_range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +# Range([range_key]) does not satisfy a join keyed on non_range_key. +query TT +EXPLAIN SELECT l.range_key, l.non_range_key, l.value, r.value +FROM range_partitioned l +LEFT JOIN range_partitioned r ON l.non_range_key = r.non_range_key; +---- +physical_plan +01)HashJoinExec: mode=Partitioned, join_type=Left, on=[(non_range_key@1, non_range_key@0)], projection=[range_key@0, non_range_key@1, value@2, value@4] +02)--RepartitionExec: partitioning=Hash([non_range_key@1], 4), input_partitions=4 +03)----DataSourceExec: file_groups=, 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([non_range_key@0], 4), input_partitions=4 +05)----DataSourceExec: file_groups=, projection=[non_range_key, value], output_partitioning=UnknownPartitioning(4), file_type=csv, has_header=false + +########## +# TEST 14: Left-Side Range Hash Joins With Incompatible Range Layouts +# Different split points or partition counts do not satisfy the +# co-partitioned layout requirement. +########## + # Different split points do not satisfy the co-partitioned layout requirement. query TT EXPLAIN SELECT l.range_key, l.value, r.value @@ -537,21 +559,12 @@ physical_plan 04)--RepartitionExec: partitioning=Hash([range_key@0], 4), input_partitions=3 05)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20)], 3), file_type=csv, has_header=false -# Range([range_key]) does not satisfy a join keyed on non_range_key. -query TT -EXPLAIN SELECT l.range_key, l.non_range_key, l.value, r.value -FROM range_partitioned l -LEFT JOIN range_partitioned r ON l.non_range_key = r.non_range_key; ----- -physical_plan -01)HashJoinExec: mode=Partitioned, join_type=Left, on=[(non_range_key@1, non_range_key@0)], projection=[range_key@0, non_range_key@1, value@2, value@4] -02)--RepartitionExec: partitioning=Hash([non_range_key@1], 4), input_partitions=4 -03)----DataSourceExec: file_groups=, 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([non_range_key@0], 4), input_partitions=4 -05)----DataSourceExec: file_groups=, projection=[non_range_key, value], output_partitioning=UnknownPartitioning(4), file_type=csv, has_header=false - +########## +# TEST 15: LeftMark Subqueries Over Range Hash Joins # SQL IN subqueries decorrelate to LeftMark joins. These queries pin matched, # unmatched, and NULL marker behavior over compatible Range inputs. +########## + query TT EXPLAIN SELECT l.range_key, l.value FROM range_partitioned l @@ -595,7 +608,7 @@ ORDER BY l.range_key; 35 350 ########## -# TEST 13: Compatible Range Join Repartitions to Increase Parallelism +# TEST 16: Compatible Range Join Repartitions to Increase Parallelism # Co-partitioning satisfaction does not prevent a repartition that increases # parallelism. With target_partitions larger than the Range partition count, # both sides are hash repartitioned. @@ -632,7 +645,7 @@ ORDER BY l.range_key; 35 350 350 ########## -# TEST 14: Preserve File Partitions Preserves Range Join Inputs +# TEST 17: Preserve File Partitions Preserves Range Join Inputs # preserve_file_partitions preserves compatible Range inputs for partitioned # joins even when target_partitions is higher than the input partition count. ########## @@ -672,7 +685,7 @@ statement ok set datafusion.optimizer.preserve_file_partitions = 0; ########## -# TEST 15: Nested Range Joins +# TEST 18: Nested Range Joins # Compatible Range partitioning is preserved through the lower join, allowing # the upper join to consume it without Hash repartitioning either input. ########## @@ -707,7 +720,7 @@ ORDER BY l.range_key; 35 350 350 350 ########## -# TEST 16: Range Aggregates Feed Range Join +# TEST 19: Range Aggregates Feed Range Join # Aggregates on range_key preserve reusable partitioning for the downstream # partitioned join. ########## @@ -762,7 +775,7 @@ ORDER BY l.range_key; 35 350 350 ########## -# TEST 17: Range Join Feeds Aggregate +# TEST 20: Range Join Feeds Aggregate # The join preserves compatible Range partitioning on range_key, allowing the # aggregate above it to avoid Hash repartitioning. ########## @@ -796,7 +809,7 @@ ORDER BY l.range_key; 35 700 ########## -# TEST 18: Right Join on Range Partition Column +# TEST 21: 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. @@ -829,7 +842,7 @@ NULL 30 300 NULL 35 350 ########## -# TEST 19: Right Semi Join on Range Partition Column +# TEST 22: 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. ########## @@ -857,7 +870,7 @@ ORDER BY r.range_key; 15 150 ########## -# TEST 20: Right Anti Join on Range Partition Column +# TEST 23: 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. ########## @@ -885,7 +898,7 @@ ORDER BY r.range_key; 35 350 ########## -# TEST 21: Incompatible Range Right Join Repartitions +# TEST 24: 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. @@ -920,7 +933,7 @@ NULL 30 300 NULL 35 350 ########## -# TEST 22: Composite-Key Right Join Repartitions +# TEST 25: 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. ########## @@ -959,7 +972,7 @@ statement ok reset datafusion.optimizer.subset_repartition_threshold; ########## -# TEST 23: Right Join with Mismatched Range Partition Counts Repartitions +# TEST 26: 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 @@ -994,7 +1007,7 @@ ORDER BY r.range_key; 350 35 350 ########## -# TEST 24: Right Join on Non-Range Key Repartitions +# TEST 27: 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. @@ -1029,7 +1042,7 @@ ORDER BY r.range_key; 50 35 350 ########## -# TEST 25: Mark Join Marker Semantics +# TEST 28: Mark Join Marker Semantics # Mark joins preserve matched, unmatched, and NULL-key marker behavior over # range-partitioned inputs. ########## @@ -1081,7 +1094,7 @@ ORDER BY r.range_key; 35 350 ########## -# TEST 26: Sort Merge Join Avoids Repartition for Compatible Range Inputs +# TEST 29: Sort Merge Join Avoids Repartition for Compatible Range Inputs # Compatible Range inputs satisfy SortMergeJoinExec's co-partitioned # KeyPartitioned requirements. ########## @@ -1118,7 +1131,7 @@ ORDER BY l.range_key; 35 350 350 ########## -# TEST 27: Sort Merge Join Repartitions Incompatible Range Inputs +# TEST 30: Sort Merge Join Repartitions Incompatible Range Inputs # Different Range split points do not satisfy SortMergeJoinExec's # co-partitioned KeyPartitioned requirements. ########## @@ -1157,7 +1170,7 @@ statement ok reset datafusion.optimizer.prefer_hash_join; ########## -# TEST 28: Symmetric Hash Join Avoids Repartition for Compatible Range Inputs +# TEST 31: Symmetric Hash Join Avoids Repartition for Compatible Range Inputs # Compatible Range streams satisfy SymmetricHashJoinExec's co-partitioned # KeyPartitioned requirements. ########## @@ -1191,7 +1204,7 @@ FULL JOIN unbounded_range_like r ON l.range_key = r.range_key; 5 50 50 ########## -# TEST 29: Symmetric Hash Join Repartitions Incompatible Range Inputs +# TEST 32: Symmetric Hash Join Repartitions Incompatible Range Inputs # Different Range split points do not satisfy SymmetricHashJoinExec's # co-partitioned KeyPartitioned requirements. ########## @@ -1224,7 +1237,7 @@ FULL JOIN unbounded_range_like_shifted r ON l.range_key = r.range_key; 5 50 50 ########## -# TEST 30: Full Outer Join on Range Partition Column +# TEST 33: 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. ########## @@ -1255,7 +1268,7 @@ ORDER BY l.range_key; 35 350 350 ########## -# TEST 31: Full Outer Join Incompatible Range Repartitions +# TEST 34: 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. ########## @@ -1288,7 +1301,7 @@ ORDER BY l.range_key; 35 350 350 ########## -# TEST 32: Full Outer Join Produces Matched and Unmatched Rows +# TEST 35: 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 @@ -1333,7 +1346,7 @@ statement ok reset datafusion.optimizer.preserve_file_partitions; ########## -# TEST 33: Union of Range Partitioned Inputs +# TEST 36: 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. ########## @@ -1382,7 +1395,7 @@ set datafusion.optimizer.preserve_file_partitions = 0; ########## -# TEST 34: Window on Range Partition Column +# TEST 37: Window on Range Partition Column # Range([range_key]) colocates equal range_key values, so # PARTITION BY range_key is satisfied without a hash repartition. ########## @@ -1410,7 +1423,7 @@ SELECT range_key, SUM(value) OVER (PARTITION BY range_key ORDER BY value) FROM r ########## -# TEST 35: Unbounded-Frame Window on Range Partition Column +# TEST 38: 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. @@ -1439,7 +1452,7 @@ SELECT range_key, SUM(value) OVER (PARTITION BY range_key ORDER BY value ROWS BE ########## -# TEST 36: Window on Non-Range Column Rehashes +# TEST 39: 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. ########## @@ -1468,7 +1481,7 @@ SELECT non_range_key, value, SUM(value) OVER (PARTITION BY non_range_key ORDER B ########## -# TEST 37: Unbounded-Frame Window on Non-Range Column Rehashes +# TEST 40: 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. @@ -1498,7 +1511,7 @@ SELECT non_range_key, value, SUM(value) OVER (PARTITION BY non_range_key ORDER B ########## -# TEST 38: Window Subset Satisfaction on Range Partition Column +# TEST 41: 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. @@ -1530,7 +1543,7 @@ SELECT range_key, SUM(value) OVER (PARTITION BY range_key, non_range_key ORDER B ########## -# TEST 39: Window Subset Rehashes Below Subset Threshold +# TEST 42: 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. @@ -1569,7 +1582,7 @@ reset datafusion.optimizer.preserve_file_partitions; ########## -# TEST 40: Window Without Partition Keys Uses a Single Partition +# TEST 43: Window Without Partition Keys Uses a Single Partition # A window with no PARTITION BY requires a single partition; range # partitioning is not applicable. ########## @@ -1599,7 +1612,7 @@ SELECT range_key, SUM(value) OVER (ORDER BY value) FROM range_partitioned ORDER ########## -# TEST 41: PartitionedTopK on Range Partition Column +# TEST 44: PartitionedTopK on Range Partition Column # Exact Range([range_key]) satisfies the TopK partition key and avoids repartitioning. ########## @@ -1642,7 +1655,7 @@ ORDER BY range_key; ########## -# TEST 42: PartitionedTopK on Non-Range Column +# TEST 45: PartitionedTopK on Non-Range Column # Partitioning on a non-range key cannot reuse Range([range_key]) and # requires hash repartitioning. ########## @@ -1678,7 +1691,7 @@ ORDER BY non_range_key; ########## -# TEST 43: PartitionedTopK Reuses Range Subset Partitioning +# TEST 46: PartitionedTopK Reuses Range Subset Partitioning # With subset threshold met and preserve-file disabled, Range([range_key]) # satisfies partitioning by (range_key, non_range_key). ########## @@ -1719,7 +1732,7 @@ ORDER BY range_key, non_range_key; ########## -# TEST 44: Range Subset PartitionedTopK Rehashes Below Subset Threshold +# TEST 47: 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. @@ -1757,7 +1770,7 @@ statement ok reset datafusion.optimizer.enable_window_topn; ########## -# TEST 45: Subset of Inputs Compatible Does Not Trigger InterleaveExec +# TEST 48: 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. @@ -1813,7 +1826,7 @@ ORDER BY range_key, value; 35 350 ########## -# TEST 46: Incompatible Range Split Points Falls Back to UnionExec +# TEST 49: 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. ########## @@ -1855,7 +1868,7 @@ ORDER BY range_key, value; 35 350 ########## -# TEST 47: InterleaveExec Propagates Range Partitioning to Aggregate +# TEST 50: 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. From fb9c59f7af2ec46a0827b5bc1d87c53984f0f3ec Mon Sep 17 00:00:00 2001 From: Justin O'Dwyer Date: Tue, 21 Jul 2026 11:02:17 -0400 Subject: [PATCH 5/5] Simplify. --- .../physical-plan/src/joins/hash_join/exec.rs | 18 +++--------------- 1 file changed, 3 insertions(+), 15 deletions(-) diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index f69136653ce82..e941bb0898fed 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -1293,21 +1293,9 @@ impl ExecutionPlan for HashJoinExec { ]), }; - if self.mode == PartitionMode::Partitioned - && matches!( - self.join_type, - JoinType::Inner - | JoinType::Full - | JoinType::Left - | JoinType::LeftSemi - | JoinType::LeftAnti - | JoinType::LeftMark - | JoinType::Right - | JoinType::RightSemi - | JoinType::RightAnti - | JoinType::RightMark - ) - { + if self.mode == PartitionMode::Partitioned { + // Compatible Range inputs co-locate equal join keys, which + // satisfies the co-partitioned requirement for hash joins. requirements.allow_range_satisfaction_for_key_partitioning() } else { requirements