diff --git a/datafusion/core/tests/physical_optimizer/enforce_distribution.rs b/datafusion/core/tests/physical_optimizer/enforce_distribution.rs index e01311e25be8b..d913cd5016d86 100644 --- a/datafusion/core/tests/physical_optimizer/enforce_distribution.rs +++ b/datafusion/core/tests/physical_optimizer/enforce_distribution.rs @@ -870,6 +870,87 @@ fn range_inner_hash_join_rehashes_incompatible_range_partitioning() -> Result<() Ok(()) } +// Kept as a unit test: `RightMark` join plans cannot be produced from SQL (they +// only arise when a statistical swap flips a `LeftMark` join around), so this +// reuse plan shape has no `range_partitioning.slt` equivalent. The end-to-end +// mark-join coverage lives in that file's "Mark Join Marker Semantics" case. +#[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(()) +} + +// Kept as a unit test: a descending Range input cannot be registered through the +// sqllogictest fixtures (they only declare ascending Range layouts), so the +// sort-direction axis of the co-partition check has no `range_partitioning.slt` +// equivalent. Verifies that opposite sort options force Hash repartitioning. +#[test] +fn range_right_semi_hash_join_rehashes_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 { + 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=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(()) +} + #[test] fn multi_hash_joins() -> Result<()> { let left = parquet_exec(); diff --git a/datafusion/core/tests/physical_optimizer/sanity_checker.rs b/datafusion/core/tests/physical_optimizer/sanity_checker.rs index e759156282306..6edfa5a98a5a0 100644 --- a/datafusion/core/tests/physical_optimizer/sanity_checker.rs +++ b/datafusion/core/tests/physical_optimizer/sanity_checker.rs @@ -443,6 +443,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(()) +} + #[tokio::test] /// Tests that plan is valid when the sort requirements are satisfied. async fn test_bounded_window_agg_sort_requirement() -> Result<()> { diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index 56e7132dc4df7..734e885c34aeb 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::Right + | JoinType::RightSemi + | JoinType::RightAnti + | JoinType::RightMark + ) + { 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 aa741dded77be..258f3e6bbde5f 100644 --- a/datafusion/sqllogictest/src/test_context/range_partitioning.rs +++ b/datafusion/sqllogictest/src/test_context/range_partitioning.rs @@ -84,7 +84,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", @@ -93,6 +93,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 5d004ca7fdfa5..3dd7c362a3f9f 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 @@ -634,6 +635,260 @@ ORDER BY l.range_key; 30 600 35 700 +########## +# TEST 18: Right Join on Range Partition Column +# Right-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. The filter on the left input keeps its +# Range partitioning and leaves the right rows above 150 unmatched. +########## + +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={4 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]}, 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={4 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]}, 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={4 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]}, 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={4 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]}, 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={4 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]}, 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={4 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]}, 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={4 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]}, 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={4 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned_shifted/part-0.csv], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned_shifted/part-1.csv], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned_shifted/part-2.csv], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned_shifted/part-3.csv]]}, 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: Mark Join Marker Semantics over Range Partitioned Inputs +# SQL IN subqueries decorrelate to LeftMark joins, which stay on the Hash +# repartition path (RightMark only arises when a statistical swap turns a +# LeftMark join around; RightMark plan coverage lives in the +# enforce_distribution tests). These queries pin the marker semantics -- +# matched, unmatched, and NULL join keys -- for mark joins 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={4 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]}, 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={4 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]}, 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 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={4 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]}, 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={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned_narrow/part-0.csv], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned_narrow/part-1.csv], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned_narrow/part-2.csv]]}, 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={4 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]}, 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={4 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-0.csv], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-1.csv], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-2.csv], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_partitioned/part-3.csv]]}, 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 + statement ok reset datafusion.optimizer.prefer_hash_join; @@ -644,7 +899,7 @@ statement ok reset datafusion.optimizer.preserve_file_partitions; ########## -# TEST 18: Union of Range Partitioned Inputs +# TEST 25: Union of Range Partitioned Inputs # Each input exposes Range partitioning on range_key. These changes do not add a # cross-child Range relationship for UNION ALL. ##########