comphead commented on code in PR #26261:
URL: https://github.com/apache/datafusion/pull/26261#discussion_r4566541459
##########
datafusion/sqllogictest/test_files/sort_merge_join_sort_options.slt:
##########
@@ -0,0 +1,158 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: See the NOTICE file distributed with this work for
additional information regarding copyright ownership
+# SPDX-FileContributor: Licensed to the Apache Software Foundation (ASF) under
one or more contributor license agreements
+
+##########
+## The sort options of a SortMergeJoin follow the ordering both inputs declare
on the join keys,
+## so no SortExec is needed on top of inputs that are already sorted that way.
+##########
+
+statement ok
+set datafusion.optimizer.prefer_hash_join = false;
+
+statement ok
+set datafusion.execution.target_partitions = 2;
+
+statement ok
+COPY (SELECT * FROM (VALUES (3, 'c', 30), (2, 'b', 20), (1, 'a', 10)) AS t(k1,
k2, v) ORDER BY k1 DESC NULLS LAST, k2 ASC NULLS FIRST)
+TO 'test_files/scratch/sort_merge_join_sort_options/left.parquet';
+
+statement ok
+COPY (SELECT * FROM (VALUES (3, 'c', 300), (2, 'b', 200), (4, 'd', 400)) AS
t(k1, k2, w) ORDER BY k1 DESC NULLS LAST, k2 ASC NULLS FIRST)
Review Comment:
The fixtures have no NULL keys and every query is an inner join, so the new
`NULLS LAST` and `DESC` paths are only checked on matched non-null rows. The
SMJ unit tests use non-default options only as `DESC NULLS LAST` and never for
an outer join, and the join fuzz uses the default options. Could we add a NULL
key row to both files and one `FULL JOIN` query? I ran a local comparison of
SMJ against hash join for all join types, with `ASC NULLS LAST`, `DESC NULLS
FIRST` and `DESC NULLS LAST` inputs and NULL keys, and found no mismatch. So
this is about coverage, not a bug.
##########
datafusion/core/src/physical_planner.rs:
##########
@@ -1902,14 +1902,18 @@ impl DefaultPhysicalPlanner {
// Use ordinary sort-merge join when hash join is not
preferred.
// Null-aware joins not handled above fall back to
CollectLeft
// HashJoin because their semantics are not supported by
sort-merge join.
- let join_on_len = join_on.len();
+ // The join keys are an unordered set of equalities, so
they may be
+ // reshuffled (together with their sort options) to match
the input
+ // orderings and avoid sorting.
Review Comment:
`sort_merge_join_keys` already documents this, so the call-site comment can
go.
```suggestion
```
##########
datafusion/sqllogictest/test_files/sort_merge_join_sort_options.slt:
##########
@@ -0,0 +1,158 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: See the NOTICE file distributed with this work for
additional information regarding copyright ownership
+# SPDX-FileContributor: Licensed to the Apache Software Foundation (ASF) under
one or more contributor license agreements
+
+##########
+## The sort options of a SortMergeJoin follow the ordering both inputs declare
on the join keys,
+## so no SortExec is needed on top of inputs that are already sorted that way.
+##########
+
+statement ok
+set datafusion.optimizer.prefer_hash_join = false;
+
+statement ok
+set datafusion.execution.target_partitions = 2;
+
+statement ok
+COPY (SELECT * FROM (VALUES (3, 'c', 30), (2, 'b', 20), (1, 'a', 10)) AS t(k1,
k2, v) ORDER BY k1 DESC NULLS LAST, k2 ASC NULLS FIRST)
+TO 'test_files/scratch/sort_merge_join_sort_options/left.parquet';
+
+statement ok
+COPY (SELECT * FROM (VALUES (3, 'c', 300), (2, 'b', 200), (4, 'd', 400)) AS
t(k1, k2, w) ORDER BY k1 DESC NULLS LAST, k2 ASC NULLS FIRST)
+TO 'test_files/scratch/sort_merge_join_sort_options/right.parquet';
+
+# both inputs sorted `k1 DESC NULLS LAST, k2 ASC NULLS FIRST`: the join adopts
these per key options
+statement ok
+CREATE EXTERNAL TABLE l (k1 INT, k2 VARCHAR, v INT) STORED AS PARQUET
+LOCATION 'test_files/scratch/sort_merge_join_sort_options/left.parquet';
+
+statement ok
+CREATE EXTERNAL TABLE r (k1 INT, k2 VARCHAR, w INT) STORED AS PARQUET
+LOCATION 'test_files/scratch/sort_merge_join_sort_options/right.parquet';
+
+query TT
+EXPLAIN SELECT l.k1, l.k2, v, w FROM l JOIN r ON l.k1 = r.k1 AND l.k2 = r.k2
+----
+logical_plan
+01)Projection: l.k1, l.k2, l.v, r.w
+02)--Inner Join: l.k1 = r.k1, l.k2 = r.k2
+03)----TableScan: l projection=[k1, k2, v]
+04)----TableScan: r projection=[k1, k2, w]
+physical_plan
+01)SortMergeJoinExec: join_type=Inner, on=[(k1@0, k1@0), (k2@1, k2@1)],
projection=[k1@0, k2@1, v@2, w@5]
+02)--DataSourceExec: file_groups={1 group:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_merge_join_sort_options/left.parquet]]},
projection=[k1, k2, v], output_ordering=[k1@0 DESC NULLS LAST, k2@1 ASC],
file_type=parquet
+03)--DataSourceExec: file_groups={1 group:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_merge_join_sort_options/right.parquet]]},
projection=[k1, k2, w], output_ordering=[k1@0 DESC NULLS LAST, k2@1 ASC],
file_type=parquet
+
+query ITII rowsort
+SELECT l.k1, l.k2, v, w FROM l JOIN r ON l.k1 = r.k1 AND l.k2 = r.k2
+----
+2 b 20 200
+3 c 30 300
+
+# the right table declares a different ordering than the file holds, so only
the left ordering is adopted
+# and the right input is sorted
+statement ok
+CREATE EXTERNAL TABLE r_asc (k1 INT, k2 VARCHAR, w INT) STORED AS PARQUET
Review Comment:
`r_asc` declares an ordering that the file does not have, and the test works
only because the right side is sorted again. A right file written without
`ORDER BY` (so no `sorting_columns` to infer an ordering from) would cover the
one-sided case without the false declaration, and would let us check the result
too.
##########
datafusion/core/src/physical_planner.rs:
##########
@@ -3728,6 +3732,117 @@ enum PlanAsyncExpr {
Async(AsyncMapper, PlannedExprResult),
}
+/// Chooses the join key order and per-key [`SortOptions`] of a
[`SortMergeJoinExec`], so that as
+/// few inputs as possible need a `SortExec`.
+///
+/// The join keys are an unordered set of equalities, so they can be permuted
freely. Candidates
+/// are derived from each known output ordering of each input: its entries
that (after normalizing
+/// through the equivalence group) are join keys become the leading keys, in
that order and with
+/// those options, and the remaining keys follow with the default options.
Each candidate is
+/// scored by how many inputs `ordering_satisfy` confirms are already sorted
on it, so constants,
+/// equal columns and any key order are handled. The default (original key
order, default options)
+/// wins ties, so plans that already match are unchanged.
+///
+/// Cost: nothing is done if neither input has an ordering; otherwise one
candidate per known
+/// ordering, each verified with two `ordering_satisfy` calls.
+fn sort_merge_join_keys(
Review Comment:
Could this reuse `EquivalenceProperties::find_longest_permutation`? Windows
and aggregates already use it for the same problem. It returns the longest
permutation of a key list that an input is sorted on, with the options taken
from the ordering, and it handles equivalent columns and constants. That
removes the per-ordering loop, the manual `normalize_expr`, the `used` mask and
the two `left_side`/`driven_by_left` selections, and it leaves at most two
candidates instead of one per ordering:
```rust
/// Chooses the join key order and per-key [`SortOptions`] of a
[`SortMergeJoinExec`].
///
/// The keys are an unordered set of equalities, so each input proposes the
longest
/// permutation of its keys that it is already sorted on. A proposal
replaces the default
/// (original order, `ASC NULLS FIRST`) only if strictly more inputs are
sorted on it.
fn sort_merge_join_keys(
left: &Arc<dyn ExecutionPlan>,
right: &Arc<dyn ExecutionPlan>,
on: &join_utils::JoinOn,
) -> Result<(join_utils::JoinOn, Vec<SortOptions>)> {
let (left_keys, right_keys): (Vec<_>, Vec<_>) =
on.iter().cloned().unzip();
let inputs = [
(left.equivalence_properties(), left_keys),
(right.equivalence_properties(), right_keys),
];
// Number of inputs already sorted on the keys at `order` with `options`.
let sorted_inputs = |order: &[usize], options: &[SortOptions]| {
inputs.iter().try_fold(0, |sorted, (eq, keys)| {
let sort_exprs = order.iter().zip(options).map(|(&i, &options)| {
PhysicalSortExpr::new(Arc::clone(&keys[i]), options)
});
Ok::<_, DataFusionError>(
sorted + usize::from(eq.ordering_satisfy(sort_exprs)?),
)
})
};
let mut best_order: Vec<usize> = (0..on.len()).collect();
let mut best_options = vec![SortOptions::default(); on.len()];
let mut best_sorted = sorted_inputs(&best_order, &best_options)?;
for (eq, keys) in &inputs {
if best_sorted == inputs.len() || eq.oeq_class().is_empty() {
continue;
}
let (sort_exprs, mut order) = eq.find_longest_permutation(keys)?;
if order.is_empty() {
continue;
}
let mut options: Vec<_> = sort_exprs.iter().map(|e|
e.options).collect();
// Keys this input is not sorted on follow in their original order.
let unsorted: Vec<_> = (0..on.len()).filter(|i|
!order.contains(i)).collect();
options.resize(on.len(), SortOptions::default());
order.extend(unsorted);
let sorted = sorted_inputs(&order, &options)?;
if sorted > best_sorted {
(best_order, best_options, best_sorted) = (order, options,
sorted);
}
}
let on = best_order.into_iter().map(|i| on[i].clone()).collect();
Ok((on, best_options))
}
```
I tried this on top of the branch. Both slt files pass unchanged except the
constant `k2` case, which now plans `on=[(k2@1, k2@1), (k1@0, k1@0)]`, still
without a `SortExec`.
##########
datafusion/core/src/physical_planner.rs:
##########
@@ -3728,6 +3732,117 @@ enum PlanAsyncExpr {
Async(AsyncMapper, PlannedExprResult),
}
+/// Chooses the join key order and per-key [`SortOptions`] of a
[`SortMergeJoinExec`], so that as
+/// few inputs as possible need a `SortExec`.
+///
+/// The join keys are an unordered set of equalities, so they can be permuted
freely. Candidates
+/// are derived from each known output ordering of each input: its entries
that (after normalizing
+/// through the equivalence group) are join keys become the leading keys, in
that order and with
+/// those options, and the remaining keys follow with the default options.
Each candidate is
+/// scored by how many inputs `ordering_satisfy` confirms are already sorted
on it, so constants,
+/// equal columns and any key order are handled. The default (original key
order, default options)
+/// wins ties, so plans that already match are unchanged.
+///
+/// Cost: nothing is done if neither input has an ordering; otherwise one
candidate per known
+/// ordering, each verified with two `ordering_satisfy` calls.
+fn sort_merge_join_keys(
+ left: &Arc<dyn ExecutionPlan>,
+ right: &Arc<dyn ExecutionPlan>,
+ on: &join_utils::JoinOn,
+) -> Result<(join_utils::JoinOn, Vec<SortOptions>)> {
+ let left_eq = left.equivalence_properties();
+ let right_eq = right.equivalence_properties();
+
+ // The baseline candidate is the join keys in the order they were
provided, all with the
+ // default sort options. Other candidates replace it only if they strictly
reduce the number
+ // of inputs that still need sorting.
+ let mut best = (on.clone(), vec![SortOptions::default(); on.len()]);
+ if left_eq.oeq_class().is_empty() && right_eq.oeq_class().is_empty() {
+ return Ok(best);
+ }
+
+ // Scores a candidate (join keys and their sort options) by the number of
inputs that are
+ // already sorted on it and therefore need no `SortExec`: 0 (both inputs
need a sort),
+ // 1 (one input is already sorted) or 2 (neither input needs a sort).
+ let score = |candidate: &(join_utils::JoinOn, Vec<SortOptions>)| ->
Result<usize> {
+ let satisfied = |eq: &datafusion_physical_expr::EquivalenceProperties,
+ left_side: bool| {
+ eq.ordering_satisfy(candidate.0.iter().zip(&candidate.1).map(
+ |((l, r), options)| {
+ PhysicalSortExpr::new(
+ Arc::clone(if left_side { l } else { r }),
+ *options,
+ )
+ },
+ ))
+ };
+ Ok(usize::from(satisfied(left_eq, true)?)
+ + usize::from(satisfied(right_eq, false)?))
+ };
+ let mut best_score = score(&best)?;
+
+ // Derive candidates from the orderings of both inputs, left first and
then right. The left
+ // orderings may not lead to any improvement (e.g. the left input is
unsorted or sorted on
+ // other columns), but the right input's orderings can still give a better
score, so a
+ // candidate that only the right input is already sorted on is still
found. Stop early once
+ // a candidate scores 2, as no input needs sorting anymore.
+ for (eq, driven_by_left) in [(left_eq, true), (right_eq, false)] {
+ if best_score == 2 {
+ break;
+ }
+ // The join key of the driving side, normalized through its
equivalence group, so that
+ // equal columns match the entries of the ordering below.
+ let keys: Vec<_> = on
+ .iter()
+ .map(|(l, r)| {
+ eq.eq_group().normalize_expr(Arc::clone(if driven_by_left {
+ l
+ } else {
+ r
+ }))
+ })
+ .collect();
+ for ordering in eq.oeq_class().iter() {
+ // `used[j]` is true once join key `j` was found in this lex
ordering and moved to the
+ // front of the candidate. Each key is used at most once, so an
expression that
+ // appears twice in the ordering only matches one key.
+ let mut used = vec![false; on.len()];
+ let mut candidate =
+ (Vec::with_capacity(on.len()), Vec::with_capacity(on.len()));
+ // Walk the ordering in order: every sort expression that is a
(not yet used) join
+ // key adds that key to the candidate, with the sort options of
the ordering.
+ // Sort expressions that are not join keys are skipped;
`ordering_satisfy` in
+ // `score` decides whether the skipped ones (e.g. constants) are
harmless.
+ for sort_expr in ordering.iter() {
+ let expr =
eq.eq_group().normalize_expr(Arc::clone(&sort_expr.expr));
+ if let Some(j) = (0..on.len()).find(|&j| !used[j] &&
keys[j].eq(&expr)) {
+ used[j] = true;
+ candidate.0.push(on[j].clone());
+ candidate.1.push(sort_expr.options);
+ }
+ }
+ if candidate.0.is_empty() {
+ continue;
+ }
+ // The join keys that were not found in the ordering follow, in
their provided order
+ // and with the default sort options.
+ for (key, _) in on.iter().zip(&used).filter(|(_, used)| !**used) {
+ candidate.0.push(key.clone());
+ candidate.1.push(SortOptions::default());
+ }
+ let candidate_score = score(&candidate)?;
+ if candidate_score > best_score {
Review Comment:
When the inputs are sorted on only some of the keys, every candidate scores
0, so the default wins and the sorted prefix is dropped. For example, `l.k =
r.k AND l.v % 2 = r.w % 2` over inputs sorted `k DESC NULLS LAST` plans
`SortExec: expr=[k@0 ASC, l.v % Int64(2)@2 ASC]` on both sides. For bounded
inputs that costs about the same. For unbounded inputs,
`replace_with_partial_sort` can only use a `PartialSortExec` if the prefix is
kept. Would breaking ties by the length of the sorted prefix be worth it here,
or as a follow-up?
##########
datafusion/sqllogictest/test_files/sort_merge_join_sort_options.slt:
##########
@@ -0,0 +1,158 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: See the NOTICE file distributed with this work for
additional information regarding copyright ownership
+# SPDX-FileContributor: Licensed to the Apache Software Foundation (ASF) under
one or more contributor license agreements
+
+##########
+## The sort options of a SortMergeJoin follow the ordering both inputs declare
on the join keys,
+## so no SortExec is needed on top of inputs that are already sorted that way.
+##########
+
+statement ok
+set datafusion.optimizer.prefer_hash_join = false;
+
+statement ok
+set datafusion.execution.target_partitions = 2;
+
+statement ok
+COPY (SELECT * FROM (VALUES (3, 'c', 30), (2, 'b', 20), (1, 'a', 10)) AS t(k1,
k2, v) ORDER BY k1 DESC NULLS LAST, k2 ASC NULLS FIRST)
+TO 'test_files/scratch/sort_merge_join_sort_options/left.parquet';
+
+statement ok
+COPY (SELECT * FROM (VALUES (3, 'c', 300), (2, 'b', 200), (4, 'd', 400)) AS
t(k1, k2, w) ORDER BY k1 DESC NULLS LAST, k2 ASC NULLS FIRST)
+TO 'test_files/scratch/sort_merge_join_sort_options/right.parquet';
+
+# both inputs sorted `k1 DESC NULLS LAST, k2 ASC NULLS FIRST`: the join adopts
these per key options
+statement ok
+CREATE EXTERNAL TABLE l (k1 INT, k2 VARCHAR, v INT) STORED AS PARQUET
+LOCATION 'test_files/scratch/sort_merge_join_sort_options/left.parquet';
+
+statement ok
+CREATE EXTERNAL TABLE r (k1 INT, k2 VARCHAR, w INT) STORED AS PARQUET
+LOCATION 'test_files/scratch/sort_merge_join_sort_options/right.parquet';
+
+query TT
+EXPLAIN SELECT l.k1, l.k2, v, w FROM l JOIN r ON l.k1 = r.k1 AND l.k2 = r.k2
+----
+logical_plan
+01)Projection: l.k1, l.k2, l.v, r.w
+02)--Inner Join: l.k1 = r.k1, l.k2 = r.k2
+03)----TableScan: l projection=[k1, k2, v]
+04)----TableScan: r projection=[k1, k2, w]
+physical_plan
+01)SortMergeJoinExec: join_type=Inner, on=[(k1@0, k1@0), (k2@1, k2@1)],
projection=[k1@0, k2@1, v@2, w@5]
+02)--DataSourceExec: file_groups={1 group:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_merge_join_sort_options/left.parquet]]},
projection=[k1, k2, v], output_ordering=[k1@0 DESC NULLS LAST, k2@1 ASC],
file_type=parquet
+03)--DataSourceExec: file_groups={1 group:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_merge_join_sort_options/right.parquet]]},
projection=[k1, k2, w], output_ordering=[k1@0 DESC NULLS LAST, k2@1 ASC],
file_type=parquet
+
+query ITII rowsort
+SELECT l.k1, l.k2, v, w FROM l JOIN r ON l.k1 = r.k1 AND l.k2 = r.k2
+----
+2 b 20 200
+3 c 30 300
+
+# the right table declares a different ordering than the file holds, so only
the left ordering is adopted
+# and the right input is sorted
+statement ok
+CREATE EXTERNAL TABLE r_asc (k1 INT, k2 VARCHAR, w INT) STORED AS PARQUET
+WITH ORDER (k1 ASC NULLS LAST, k2 ASC NULLS FIRST)
+LOCATION 'test_files/scratch/sort_merge_join_sort_options/right.parquet';
+
+query TT
+EXPLAIN SELECT l.k1, l.k2, v, w FROM l JOIN r_asc r ON l.k1 = r.k1 AND l.k2 =
r.k2
+----
+logical_plan
+01)Projection: l.k1, l.k2, l.v, r.w
+02)--Inner Join: l.k1 = r.k1, l.k2 = r.k2
+03)----TableScan: l projection=[k1, k2, v]
+04)----SubqueryAlias: r
+05)------TableScan: r_asc projection=[k1, k2, w]
+physical_plan
+01)SortMergeJoinExec: join_type=Inner, on=[(k1@0, k1@0), (k2@1, k2@1)],
projection=[k1@0, k2@1, v@2, w@5]
+02)--DataSourceExec: file_groups={1 group:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_merge_join_sort_options/left.parquet]]},
projection=[k1, k2, v], output_ordering=[k1@0 DESC NULLS LAST, k2@1 ASC],
file_type=parquet
+03)--SortExec: expr=[k1@0 DESC NULLS LAST, k2@1 ASC],
preserve_partitioning=[false]
+04)----DataSourceExec: file_groups={1 group:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_merge_join_sort_options/right.parquet]]},
projection=[k1, k2, w], file_type=parquet, sort_order_for_reorder=[k1@0 DESC
NULLS LAST, k2@1 ASC], reverse_row_groups=true
+
+# the join keys are listed in a different order than the input ordering: the
keys are reordered
+# to follow the ordering, so no SortExec is needed
+query TT
+EXPLAIN SELECT l.k1, l.k2, v, w FROM l JOIN r ON l.k2 = r.k2 AND l.k1 = r.k1
+----
+logical_plan
+01)Projection: l.k1, l.k2, l.v, r.w
+02)--Inner Join: l.k2 = r.k2, l.k1 = r.k1
+03)----TableScan: l projection=[k1, k2, v]
+04)----TableScan: r projection=[k1, k2, w]
+physical_plan
+01)SortMergeJoinExec: join_type=Inner, on=[(k1@0, k1@0), (k2@1, k2@1)],
projection=[k1@0, k2@1, v@2, w@5]
+02)--DataSourceExec: file_groups={1 group:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_merge_join_sort_options/left.parquet]]},
projection=[k1, k2, v], output_ordering=[k1@0 DESC NULLS LAST, k2@1 ASC],
file_type=parquet
+03)--DataSourceExec: file_groups={1 group:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_merge_join_sort_options/right.parquet]]},
projection=[k1, k2, w], output_ordering=[k1@0 DESC NULLS LAST, k2@1 ASC],
file_type=parquet
+
+query ITII rowsort
+SELECT l.k1, l.k2, v, w FROM l JOIN r ON l.k2 = r.k2 AND l.k1 = r.k1
+----
+2 b 20 200
+3 c 30 300
+
+# only the leading ordering column is a join key: its options are adopted and
no SortExec is needed
Review Comment:
This takes the same path as the first query, with one key instead of two.
Could it be replaced by the NULL key / `FULL JOIN` case?
##########
datafusion/sqllogictest/test_files/sort_merge_join_sort_options.slt:
##########
@@ -0,0 +1,158 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: See the NOTICE file distributed with this work for
additional information regarding copyright ownership
+# SPDX-FileContributor: Licensed to the Apache Software Foundation (ASF) under
one or more contributor license agreements
+
+##########
+## The sort options of a SortMergeJoin follow the ordering both inputs declare
on the join keys,
+## so no SortExec is needed on top of inputs that are already sorted that way.
+##########
+
+statement ok
+set datafusion.optimizer.prefer_hash_join = false;
+
+statement ok
+set datafusion.execution.target_partitions = 2;
+
+statement ok
+COPY (SELECT * FROM (VALUES (3, 'c', 30), (2, 'b', 20), (1, 'a', 10)) AS t(k1,
k2, v) ORDER BY k1 DESC NULLS LAST, k2 ASC NULLS FIRST)
+TO 'test_files/scratch/sort_merge_join_sort_options/left.parquet';
+
+statement ok
+COPY (SELECT * FROM (VALUES (3, 'c', 300), (2, 'b', 200), (4, 'd', 400)) AS
t(k1, k2, w) ORDER BY k1 DESC NULLS LAST, k2 ASC NULLS FIRST)
+TO 'test_files/scratch/sort_merge_join_sort_options/right.parquet';
+
+# both inputs sorted `k1 DESC NULLS LAST, k2 ASC NULLS FIRST`: the join adopts
these per key options
+statement ok
+CREATE EXTERNAL TABLE l (k1 INT, k2 VARCHAR, v INT) STORED AS PARQUET
+LOCATION 'test_files/scratch/sort_merge_join_sort_options/left.parquet';
+
+statement ok
+CREATE EXTERNAL TABLE r (k1 INT, k2 VARCHAR, w INT) STORED AS PARQUET
+LOCATION 'test_files/scratch/sort_merge_join_sort_options/right.parquet';
+
+query TT
+EXPLAIN SELECT l.k1, l.k2, v, w FROM l JOIN r ON l.k1 = r.k1 AND l.k2 = r.k2
+----
+logical_plan
+01)Projection: l.k1, l.k2, l.v, r.w
+02)--Inner Join: l.k1 = r.k1, l.k2 = r.k2
+03)----TableScan: l projection=[k1, k2, v]
+04)----TableScan: r projection=[k1, k2, w]
+physical_plan
+01)SortMergeJoinExec: join_type=Inner, on=[(k1@0, k1@0), (k2@1, k2@1)],
projection=[k1@0, k2@1, v@2, w@5]
+02)--DataSourceExec: file_groups={1 group:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_merge_join_sort_options/left.parquet]]},
projection=[k1, k2, v], output_ordering=[k1@0 DESC NULLS LAST, k2@1 ASC],
file_type=parquet
+03)--DataSourceExec: file_groups={1 group:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_merge_join_sort_options/right.parquet]]},
projection=[k1, k2, w], output_ordering=[k1@0 DESC NULLS LAST, k2@1 ASC],
file_type=parquet
+
+query ITII rowsort
+SELECT l.k1, l.k2, v, w FROM l JOIN r ON l.k1 = r.k1 AND l.k2 = r.k2
+----
+2 b 20 200
+3 c 30 300
+
+# the right table declares a different ordering than the file holds, so only
the left ordering is adopted
+# and the right input is sorted
+statement ok
+CREATE EXTERNAL TABLE r_asc (k1 INT, k2 VARCHAR, w INT) STORED AS PARQUET
+WITH ORDER (k1 ASC NULLS LAST, k2 ASC NULLS FIRST)
+LOCATION 'test_files/scratch/sort_merge_join_sort_options/right.parquet';
+
+query TT
+EXPLAIN SELECT l.k1, l.k2, v, w FROM l JOIN r_asc r ON l.k1 = r.k1 AND l.k2 =
r.k2
+----
+logical_plan
+01)Projection: l.k1, l.k2, l.v, r.w
+02)--Inner Join: l.k1 = r.k1, l.k2 = r.k2
+03)----TableScan: l projection=[k1, k2, v]
+04)----SubqueryAlias: r
+05)------TableScan: r_asc projection=[k1, k2, w]
+physical_plan
+01)SortMergeJoinExec: join_type=Inner, on=[(k1@0, k1@0), (k2@1, k2@1)],
projection=[k1@0, k2@1, v@2, w@5]
+02)--DataSourceExec: file_groups={1 group:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_merge_join_sort_options/left.parquet]]},
projection=[k1, k2, v], output_ordering=[k1@0 DESC NULLS LAST, k2@1 ASC],
file_type=parquet
+03)--SortExec: expr=[k1@0 DESC NULLS LAST, k2@1 ASC],
preserve_partitioning=[false]
+04)----DataSourceExec: file_groups={1 group:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_merge_join_sort_options/right.parquet]]},
projection=[k1, k2, w], file_type=parquet, sort_order_for_reorder=[k1@0 DESC
NULLS LAST, k2@1 ASC], reverse_row_groups=true
+
+# the join keys are listed in a different order than the input ordering: the
keys are reordered
+# to follow the ordering, so no SortExec is needed
+query TT
+EXPLAIN SELECT l.k1, l.k2, v, w FROM l JOIN r ON l.k2 = r.k2 AND l.k1 = r.k1
+----
+logical_plan
+01)Projection: l.k1, l.k2, l.v, r.w
+02)--Inner Join: l.k2 = r.k2, l.k1 = r.k1
+03)----TableScan: l projection=[k1, k2, v]
+04)----TableScan: r projection=[k1, k2, w]
+physical_plan
+01)SortMergeJoinExec: join_type=Inner, on=[(k1@0, k1@0), (k2@1, k2@1)],
projection=[k1@0, k2@1, v@2, w@5]
+02)--DataSourceExec: file_groups={1 group:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_merge_join_sort_options/left.parquet]]},
projection=[k1, k2, v], output_ordering=[k1@0 DESC NULLS LAST, k2@1 ASC],
file_type=parquet
+03)--DataSourceExec: file_groups={1 group:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_merge_join_sort_options/right.parquet]]},
projection=[k1, k2, w], output_ordering=[k1@0 DESC NULLS LAST, k2@1 ASC],
file_type=parquet
+
+query ITII rowsort
+SELECT l.k1, l.k2, v, w FROM l JOIN r ON l.k2 = r.k2 AND l.k1 = r.k1
+----
+2 b 20 200
+3 c 30 300
+
+# only the leading ordering column is a join key: its options are adopted and
no SortExec is needed
+query TT
+EXPLAIN SELECT l.k1, v, w FROM l JOIN r ON l.k1 = r.k1
+----
+logical_plan
+01)Projection: l.k1, l.v, r.w
+02)--Inner Join: l.k1 = r.k1
+03)----TableScan: l projection=[k1, v]
+04)----TableScan: r projection=[k1, w]
+physical_plan
+01)SortMergeJoinExec: join_type=Inner, on=[(k1@0, k1@0)], projection=[k1@0,
v@1, w@3]
+02)--DataSourceExec: file_groups={1 group:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_merge_join_sort_options/left.parquet]]},
projection=[k1, v], output_ordering=[k1@0 DESC NULLS LAST], file_type=parquet
+03)--DataSourceExec: file_groups={1 group:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_merge_join_sort_options/right.parquet]]},
projection=[k1, w], output_ordering=[k1@0 DESC NULLS LAST], file_type=parquet
+
+query III rowsort
+SELECT l.k1, v, w FROM l JOIN r ON l.k1 = r.k1
+----
+2 20 200
+3 30 300
+
+# k2 is constant after the filter, so it does not need to be in the ordering:
the join adopts the
+# options of the k1 ordering and no SortExec is needed (a single partition
avoids a repartition)
+statement ok
+set datafusion.execution.target_partitions = 1;
+
+query TT
+EXPLAIN SELECT l.k1, l.k2, v, w FROM l JOIN r ON l.k2 = r.k2 AND l.k1 = r.k1
WHERE l.k2 = 'b'
+----
+logical_plan
+01)Projection: l.k1, l.k2, l.v, r.w
+02)--Inner Join: l.k2 = r.k2, l.k1 = r.k1
+03)----Filter: l.k2 = Utf8View("b")
+04)------TableScan: l projection=[k1, k2, v], partial_filters=[l.k2 =
Utf8View("b")]
+05)----Filter: r.k2 = Utf8View("b")
+06)------TableScan: r projection=[k1, k2, w], partial_filters=[r.k2 =
Utf8View("b")]
+physical_plan
+01)SortMergeJoinExec: join_type=Inner, on=[(k1@0, k1@0), (k2@1, k2@1)],
projection=[k1@0, k2@1, v@2, w@5]
+02)--FilterExec: k2@1 = b
+03)----DataSourceExec: file_groups={1 group:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_merge_join_sort_options/left.parquet]]},
projection=[k1, k2, v], output_ordering=[k1@0 DESC NULLS LAST, k2@1 ASC],
file_type=parquet, predicate=k2@1 = b, pruning_predicate=k2_null_count@2 !=
row_count@3 AND k2_min@0 <= b AND b <= k2_max@1, required_guarantees=[k2 in (b)]
+04)--FilterExec: k2@1 = b
+05)----DataSourceExec: file_groups={1 group:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_merge_join_sort_options/right.parquet]]},
projection=[k1, k2, w], output_ordering=[k1@0 DESC NULLS LAST, k2@1 ASC],
file_type=parquet, predicate=k2@1 = b, pruning_predicate=k2_null_count@2 !=
row_count@3 AND k2_min@0 <= b AND b <= k2_max@1, required_guarantees=[k2 in (b)]
+
+query ITII rowsort
+SELECT l.k1, l.k2, v, w FROM l JOIN r ON l.k2 = r.k2 AND l.k1 = r.k1 WHERE
l.k2 = 'b'
+----
+2 b 20 200
+
+statement ok
+set datafusion.execution.target_partitions = 2;
Review Comment:
This setting is overwritten before any other query runs.
```suggestion
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
