Skip to content

Commit 1f5ff8c

Browse files
lyne7-sclyne7-sc
committed
fix: preserve only the maintained input ordering in joins
1 parent f96892a commit 1f5ff8c

3 files changed

Lines changed: 116 additions & 125 deletions

File tree

‎datafusion/physical-expr/src/equivalence/properties/joins.rs‎

Lines changed: 12 additions & 63 deletions
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,7 @@ pub fn join_equivalence_properties(
2828
join_type: &JoinType,
2929
join_schema: SchemaRef,
3030
maintains_input_order: &[bool],
31-
probe_side: Option<JoinSide>,
31+
_probe_side: Option<JoinSide>,
3232
on: &[(PhysicalExprRef, PhysicalExprRef)],
3333
) -> Result<EquivalenceProperties> {
3434
let left_size = left.schema.fields.len();
@@ -49,56 +49,17 @@ pub fn join_equivalence_properties(
4949
..
5050
} = right;
5151
match maintains_input_order {
52-
[true, false] => {
53-
// In this special case, right side ordering can be prefixed with
54-
// the left side ordering.
55-
if matches!(join_type, JoinType::Inner | JoinType::Left)
56-
&& probe_side == Some(JoinSide::Left)
57-
{
58-
updated_right_ordering_equivalence_class(
59-
&mut right_oeq_class,
60-
join_type,
61-
left_size,
62-
)?;
63-
64-
// Right side ordering equivalence properties should be prepended
65-
// with those of the left side while constructing output ordering
66-
// equivalence properties since stream side is the left side.
67-
//
68-
// For example, if the right side ordering equivalences contain
69-
// `b ASC`, and the left side ordering equivalences contain `a ASC`,
70-
// then we should add `a ASC, b ASC` to the ordering equivalences
71-
// of the join output.
72-
let out_oeq_class = left_oeq_class.join_suffix(&right_oeq_class);
73-
result.add_orderings(out_oeq_class);
74-
} else {
75-
result.add_orderings(left_oeq_class);
76-
}
77-
}
52+
// Only the maintained side's ordering is guaranteed. With duplicate
53+
// probe keys, the other side's matches repeat for each probe row, so
54+
// its ordering cannot be appended as a suffix.
55+
[true, false] => result.add_orderings(left_oeq_class),
7856
[false, true] => {
7957
updated_right_ordering_equivalence_class(
8058
&mut right_oeq_class,
8159
join_type,
8260
left_size,
8361
)?;
84-
// In this special case, left side ordering can be prefixed with
85-
// the right side ordering.
86-
if matches!(join_type, JoinType::Inner | JoinType::Right)
87-
&& probe_side == Some(JoinSide::Right)
88-
{
89-
// Left side ordering equivalence properties should be prepended
90-
// with those of the right side while constructing output ordering
91-
// equivalence properties since stream side is the right side.
92-
//
93-
// For example, if the left side ordering equivalences contain
94-
// `a ASC`, and the right side ordering equivalences contain `b ASC`,
95-
// then we should add `b ASC, a ASC` to the ordering equivalences
96-
// of the join output.
97-
let out_oeq_class = right_oeq_class.join_suffix(&left_oeq_class);
98-
result.add_orderings(out_oeq_class);
99-
} else {
100-
result.add_orderings(right_oeq_class);
101-
}
62+
result.add_orderings(right_oeq_class);
10263
}
10364
[false, false] => {}
10465
[true, true] => unreachable!("Cannot maintain ordering of both sides"),
@@ -136,7 +97,6 @@ mod tests {
13697
use crate::equivalence::convert_to_orderings;
13798
use crate::equivalence::tests::create_test_schema;
13899
use crate::expressions::col;
139-
use crate::physical_expr::add_offset_to_expr;
140100

141101
use arrow::compute::SortOptions;
142102
use arrow::datatypes::{DataType, Field, Fields, Schema};
@@ -147,9 +107,6 @@ mod tests {
147107
let col_a = &col("a", &schema)?;
148108
let col_b = &col("b", &schema)?;
149109
let col_c = &col("c", &schema)?;
150-
let offset = schema.fields.len() as _;
151-
let col_a2 = &add_offset_to_expr(Arc::clone(col_a), offset)?;
152-
let col_b2 = &add_offset_to_expr(Arc::clone(col_b), offset)?;
153110
let option_asc = SortOptions {
154111
descending: false,
155112
nulls_first: false,
@@ -162,13 +119,8 @@ mod tests {
162119
vec![vec![(col_a, option_asc)], vec![(col_b, option_asc)]],
163120
// [a ASC], [b ASC]
164121
vec![vec![(col_a, option_asc)], vec![(col_b, option_asc)]],
165-
// expected [a ASC, a2 ASC], [a ASC, b2 ASC], [b ASC, a2 ASC], [b ASC, b2 ASC]
166-
vec![
167-
vec![(col_a, option_asc), (col_a2, option_asc)],
168-
vec![(col_a, option_asc), (col_b2, option_asc)],
169-
vec![(col_b, option_asc), (col_a2, option_asc)],
170-
vec![(col_b, option_asc), (col_b2, option_asc)],
171-
],
122+
// Only the left input's orderings are preserved.
123+
vec![vec![(col_a, option_asc)], vec![(col_b, option_asc)]],
172124
),
173125
// ------- TEST CASE 2 --------
174126
// [a ASC], [b ASC]
@@ -181,14 +133,11 @@ mod tests {
181133
],
182134
// [a ASC], [b ASC]
183135
vec![vec![(col_a, option_asc)], vec![(col_b, option_asc)]],
184-
// expected [a ASC, a2 ASC], [a ASC, b2 ASC], [b ASC, a2 ASC], [b ASC, b2 ASC], [c ASC, a2 ASC], [c ASC, b2 ASC]
136+
// Only the left input's orderings are preserved.
185137
vec![
186-
vec![(col_a, option_asc), (col_a2, option_asc)],
187-
vec![(col_a, option_asc), (col_b2, option_asc)],
188-
vec![(col_b, option_asc), (col_a2, option_asc)],
189-
vec![(col_b, option_asc), (col_b2, option_asc)],
190-
vec![(col_c, option_asc), (col_a2, option_asc)],
191-
vec![(col_c, option_asc), (col_b2, option_asc)],
138+
vec![(col_a, option_asc)],
139+
vec![(col_b, option_asc)],
140+
vec![(col_c, option_asc)],
192141
],
193142
),
194143
];

‎datafusion/physical-plan/src/joins/utils.rs‎

Lines changed: 5 additions & 46 deletions
Original file line numberDiff line numberDiff line change
@@ -170,45 +170,13 @@ pub fn calculate_join_output_ordering(
170170
join_type: JoinType,
171171
left_columns_len: usize,
172172
maintains_input_order: &[bool],
173-
probe_side: Option<JoinSide>,
173+
_probe_side: Option<JoinSide>,
174174
) -> Result<Option<LexOrdering>> {
175175
match maintains_input_order {
176-
[true, false] => {
177-
// Special case, we can prefix ordering of right side with the ordering of left side.
178-
if join_type == JoinType::Inner
179-
&& probe_side == Some(JoinSide::Left)
180-
&& let Some(right_ordering) = right_ordering.cloned()
181-
{
182-
let right_offset = add_offset_to_physical_sort_exprs(
183-
right_ordering,
184-
left_columns_len as _,
185-
)?;
186-
return if let Some(left_ordering) = left_ordering {
187-
let mut result = left_ordering.clone();
188-
result.extend(right_offset);
189-
Ok(Some(result))
190-
} else {
191-
Ok(LexOrdering::new(right_offset))
192-
};
193-
}
194-
Ok(left_ordering.cloned())
195-
}
176+
// Matches from the other side can repeat for duplicate probe keys,
177+
// so only the maintained side's ordering is guaranteed.
178+
[true, false] => Ok(left_ordering.cloned()),
196179
[false, true] => {
197-
// Special case, we can prefix ordering of left side with the ordering of right side.
198-
if join_type == JoinType::Inner && probe_side == Some(JoinSide::Right) {
199-
return if let Some(right_ordering) = right_ordering.cloned() {
200-
let mut right_offset = add_offset_to_physical_sort_exprs(
201-
right_ordering,
202-
left_columns_len as _,
203-
)?;
204-
if let Some(left_ordering) = left_ordering {
205-
right_offset.extend(left_ordering.clone());
206-
}
207-
Ok(LexOrdering::new(right_offset))
208-
} else {
209-
Ok(left_ordering.cloned())
210-
};
211-
}
212180
let Some(right_ordering) = right_ordering else {
213181
return Ok(None);
214182
};
@@ -4483,19 +4451,10 @@ mod tests {
44834451
let probe_sides = [Some(JoinSide::Left), Some(JoinSide::Right)];
44844452

44854453
let expected = [
4486-
LexOrdering::new(vec![
4487-
PhysicalSortExpr::new_default(Arc::new(Column::new("a", 0))),
4488-
PhysicalSortExpr::new_default(Arc::new(Column::new("c", 2))),
4489-
PhysicalSortExpr::new_default(Arc::new(Column::new("d", 3))),
4490-
PhysicalSortExpr::new_default(Arc::new(Column::new("z", 7))),
4491-
PhysicalSortExpr::new_default(Arc::new(Column::new("y", 6))),
4492-
]),
4454+
left_ordering.clone(),
44934455
LexOrdering::new(vec![
44944456
PhysicalSortExpr::new_default(Arc::new(Column::new("z", 7))),
44954457
PhysicalSortExpr::new_default(Arc::new(Column::new("y", 6))),
4496-
PhysicalSortExpr::new_default(Arc::new(Column::new("a", 0))),
4497-
PhysicalSortExpr::new_default(Arc::new(Column::new("c", 2))),
4498-
PhysicalSortExpr::new_default(Arc::new(Column::new("d", 3))),
44994458
]),
45004459
];
45014460

‎datafusion/sqllogictest/test_files/joins.slt‎

Lines changed: 99 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -3376,10 +3376,8 @@ physical_plan
33763376
statement ok
33773377
set datafusion.optimizer.prefer_existing_sort = false;
33783378

3379-
# SortMergeJoinExec should add ordering equivalences of
3380-
# right table as lexicographical append to the global ordering
3381-
# below query shouldn't add any SortExec for order by clause.
3382-
# since its requirement is already satisfied at the output of SortMergeJoinExec
3379+
# The right ordering cannot be appended to the left ordering: duplicate
3380+
# left keys restart the right matches. Keep a SortExec above SortMergeJoinExec.
33833381
query TT
33843382
EXPLAIN SELECT *
33853383
FROM (SELECT *, ROW_NUMBER() OVER() as rn1
@@ -3402,15 +3400,16 @@ logical_plan
34023400
10)----------TableScan: annotated_data projection=[a0, a, b, c, d]
34033401
physical_plan
34043402
01)SortPreservingMergeExec: [a@1 ASC, b@2 ASC NULLS LAST, c@3 ASC NULLS LAST, rn1@11 ASC NULLS LAST]
3405-
02)--SortMergeJoinExec: join_type=Inner, on=[(a@1, a@1)]
3406-
03)----RepartitionExec: partitioning=Hash([a@1], 2), input_partitions=1, maintains_sort_order=true
3407-
04)------ProjectionExec: expr=[a0@0 as a0, a@1 as a, b@2 as b, c@3 as c, d@4 as d, row_number() ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING@5 as rn1]
3408-
05)--------BoundedWindowAggExec: wdw=[row_number() ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING: Field { "row_number() ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING": UInt64 }, frame: ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING], mode=[Sorted]
3409-
06)----------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a0, a, b, c, d], output_ordering=[a@1 ASC, b@2 ASC NULLS LAST, c@3 ASC NULLS LAST], file_type=csv, has_header=true
3410-
07)----RepartitionExec: partitioning=Hash([a@1], 2), input_partitions=1, maintains_sort_order=true
3411-
08)------ProjectionExec: expr=[a0@0 as a0, a@1 as a, b@2 as b, c@3 as c, d@4 as d, row_number() ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING@5 as rn1]
3412-
09)--------BoundedWindowAggExec: wdw=[row_number() ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING: Field { "row_number() ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING": UInt64 }, frame: ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING], mode=[Sorted]
3413-
10)----------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a0, a, b, c, d], output_ordering=[a@1 ASC, b@2 ASC NULLS LAST, c@3 ASC NULLS LAST], file_type=csv, has_header=true
3403+
02)--SortExec: expr=[a@1 ASC, b@2 ASC NULLS LAST, c@3 ASC NULLS LAST, rn1@11 ASC NULLS LAST], preserve_partitioning=[true]
3404+
03)----SortMergeJoinExec: join_type=Inner, on=[(a@1, a@1)]
3405+
04)------RepartitionExec: partitioning=Hash([a@1], 2), input_partitions=1, maintains_sort_order=true
3406+
05)--------ProjectionExec: expr=[a0@0 as a0, a@1 as a, b@2 as b, c@3 as c, d@4 as d, row_number() ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING@5 as rn1]
3407+
06)----------BoundedWindowAggExec: wdw=[row_number() ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING: Field { "row_number() ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING": UInt64 }, frame: ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING], mode=[Sorted]
3408+
07)------------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a0, a, b, c, d], output_ordering=[a@1 ASC, b@2 ASC NULLS LAST, c@3 ASC NULLS LAST], file_type=csv, has_header=true
3409+
08)------RepartitionExec: partitioning=Hash([a@1], 2), input_partitions=1, maintains_sort_order=true
3410+
09)--------ProjectionExec: expr=[a0@0 as a0, a@1 as a, b@2 as b, c@3 as c, d@4 as d, row_number() ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING@5 as rn1]
3411+
10)----------BoundedWindowAggExec: wdw=[row_number() ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING: Field { "row_number() ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING": UInt64 }, frame: ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING], mode=[Sorted]
3412+
11)------------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a0, a, b, c, d], output_ordering=[a@1 ASC, b@2 ASC NULLS LAST, c@3 ASC NULLS LAST], file_type=csv, has_header=true
34143413

34153414
statement ok
34163415
set datafusion.optimizer.prefer_hash_join = true;
@@ -3488,9 +3487,10 @@ logical_plan
34883487
05)----SubqueryAlias: r_table
34893488
06)------TableScan: annotated_data projection=[a0, a, b, c, d]
34903489
physical_plan
3491-
01)HashJoinExec: mode=CollectLeft, join_type=Right, on=[(b@2, b@2)]
3492-
02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a0, a, b, c, d], output_ordering=[a@1 ASC, b@2 ASC NULLS LAST, c@3 ASC NULLS LAST], file_type=csv, has_header=true
3493-
03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a0, a, b, c, d], output_ordering=[a@1 ASC, b@2 ASC NULLS LAST, c@3 ASC NULLS LAST], file_type=csv, has_header=true
3490+
01)SortExec: expr=[a@6 ASC, b@7 ASC NULLS LAST, c@8 ASC NULLS LAST, a@1 ASC], preserve_partitioning=[false]
3491+
02)--HashJoinExec: mode=CollectLeft, join_type=Right, on=[(b@2, b@2)]
3492+
03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a0, a, b, c, d], output_ordering=[a@1 ASC, b@2 ASC NULLS LAST, c@3 ASC NULLS LAST], file_type=csv, has_header=true
3493+
04)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a0, a, b, c, d], output_ordering=[a@1 ASC, b@2 ASC NULLS LAST, c@3 ASC NULLS LAST], file_type=csv, has_header=true
34943494

34953495
query TT
34963496
EXPLAIN SELECT l.a, LAST_VALUE(r.b ORDER BY r.a ASC NULLS FIRST) as last_col1
@@ -6300,3 +6300,86 @@ reset datafusion.optimizer.repartition_file_scans;
63006300

63016301
statement ok
63026302
set datafusion.execution.target_partitions = 4;
6303+
6304+
# Duplicate probe keys restart the build-side ordering for each matching row.
6305+
statement ok
6306+
set datafusion.execution.target_partitions = 2;
6307+
6308+
statement ok
6309+
set datafusion.optimizer.prefer_hash_join = false;
6310+
6311+
statement ok
6312+
CREATE TABLE duplicate_join_left AS VALUES (1, 10), (1, 20), (2, 30);
6313+
6314+
statement ok
6315+
CREATE TABLE duplicate_join_right AS VALUES (1, 100), (1, 200), (2, 300);
6316+
6317+
# Sorting the build side alone must not satisfy the output ordering.
6318+
query II
6319+
SELECT l.column1, r.column2
6320+
FROM duplicate_join_left l JOIN duplicate_join_right r ON l.column1 = r.column1
6321+
ORDER BY l.column1, r.column2;
6322+
----
6323+
1 100
6324+
1 100
6325+
1 200
6326+
1 200
6327+
2 300
6328+
6329+
query II
6330+
SELECT l.column1, r.column2
6331+
FROM duplicate_join_left l JOIN duplicate_join_right r ON l.column1 = r.column1
6332+
ORDER BY l.column1, r.column2 LIMIT 2;
6333+
----
6334+
1 100
6335+
1 100
6336+
6337+
query IIII rowsort
6338+
SELECT r.column2,
6339+
row_number() OVER (ORDER BY l.column1, r.column2),
6340+
rank() OVER (ORDER BY l.column1, r.column2),
6341+
dense_rank() OVER (ORDER BY l.column1, r.column2)
6342+
FROM duplicate_join_left l JOIN duplicate_join_right r ON l.column1 = r.column1;
6343+
----
6344+
100 1 1 1
6345+
100 2 1 1
6346+
200 3 3 2
6347+
200 4 3 2
6348+
300 5 5 3
6349+
6350+
# Ordered sources exercise HashJoin's output properties. Filtering the build
6351+
# input also leaves an unmatched probe row, covering null extension.
6352+
statement ok
6353+
set datafusion.execution.target_partitions = 1;
6354+
6355+
statement ok
6356+
set datafusion.optimizer.prefer_hash_join = true;
6357+
6358+
statement ok
6359+
COPY (SELECT * FROM duplicate_join_left ORDER BY column1, column2)
6360+
TO 'test_files/scratch/joins/duplicate_ordering.csv'
6361+
STORED AS CSV OPTIONS ('format.has_header' 'true');
6362+
6363+
statement ok
6364+
CREATE EXTERNAL TABLE duplicate_ordering (column1 BIGINT, column2 BIGINT)
6365+
STORED AS CSV WITH ORDER (column1 ASC, column2 ASC)
6366+
LOCATION 'test_files/scratch/joins/duplicate_ordering.csv'
6367+
OPTIONS ('format.has_header' 'true');
6368+
6369+
query III
6370+
SELECT r.column1, l.column1, l.column2
6371+
FROM (SELECT * FROM duplicate_ordering WHERE column1 = 1) l
6372+
RIGHT JOIN duplicate_ordering r ON l.column1 = r.column1
6373+
ORDER BY r.column1, l.column1, l.column2;
6374+
----
6375+
1 1 10
6376+
1 1 10
6377+
1 1 20
6378+
1 1 20
6379+
2 NULL NULL
6380+
6381+
statement ok
6382+
set datafusion.execution.target_partitions = 4;
6383+
6384+
statement ok
6385+
reset datafusion.optimizer.prefer_hash_join;

0 commit comments

Comments
 (0)