This is an automated email from the ASF dual-hosted git repository.
github-bot pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new 258e18cf88 Use SortMergeJoinExec name consistently in physical plan
outputs (#19246)
258e18cf88 is described below
commit 258e18cf8895facc11903c922b800af38461ace1
Author: xavlee <[email protected]>
AuthorDate: Mon Dec 22 23:08:52 2025 -0500
Use SortMergeJoinExec name consistently in physical plan outputs (#19246)
## Which issue does this PR close?
- closes #19238
## Rationale for this change
`SortMergeJoinExec` is currently displayed inconsistently across
physical plan formats, see
[join.slt](https://github.com/apache/datafusion/blob/20870c18a418ec081d44ecf8a90a30a95aa53138/datafusion/sqllogictest/test_files/joins.slt#L2727)
vs.
[explain_tree.slt](https://github.com/apache/datafusion/blob/20870c18a418ec081d44ecf8a90a30a95aa53138/datafusion/sqllogictest/test_files/explain_tree.slt#L1203).
These examples show that the tree-fmt plan uses SortMergeJoinExec, while
the indent-fmt plan uses SortMergeJoin.
Standardizing the operator name improves clarity and aligns with the
naming conventions of other execution operators.
## What changes are included in this PR?
Updates the `DisplayAs` implementation for `SortMergeJoinExec` to output
`"SortMergeJoinExec: ..."`.
Updates SQL Logic Test expected outputs in `joins.slt` to reflect the
unified naming.
No functional behavior changes; this is a display/consistency fix.
## Are these changes tested?
Yes. This change is encapsulated in existing SQL Logic Tests. I updated
those expected outputs to match the new standardized naming.
All tests pass with the updated format.
## Are there any user-facing changes?
Yes—users inspecting physical plans will now consistently see
`SortMergeJoinExec` instead of `SortMergeJoin`.
---
.../physical_optimizer/enforce_distribution.rs | 48 +++++++++++-----------
.../tests/physical_optimizer/enforce_sorting.rs | 28 ++++++-------
.../tests/physical_optimizer/sanity_checker.rs | 6 +--
.../src/joins/sort_merge_join/exec.rs | 3 +-
datafusion/sqllogictest/test_files/joins.slt | 12 +++---
.../sqllogictest/test_files/sort_merge_join.slt | 2 +-
6 files changed, 50 insertions(+), 49 deletions(-)
diff --git a/datafusion/core/tests/physical_optimizer/enforce_distribution.rs
b/datafusion/core/tests/physical_optimizer/enforce_distribution.rs
index 4815f4a8a8..7cedaf86cb 100644
--- a/datafusion/core/tests/physical_optimizer/enforce_distribution.rs
+++ b/datafusion/core/tests/physical_optimizer/enforce_distribution.rs
@@ -1424,8 +1424,8 @@ fn multi_smj_joins() -> Result<()> {
// Should include 6 RepartitionExecs (3 hash, 3
round-robin), 3 SortExecs
JoinType::Inner | JoinType::Left | JoinType::LeftSemi |
JoinType::LeftAnti => {
assert_plan!(plan_distrib, @r"
- SortMergeJoin: join_type=..., on=[(a@0, c@2)]
- SortMergeJoin: join_type=..., on=[(a@0, b1@1)]
+ SortMergeJoinExec: join_type=..., on=[(a@0, c@2)]
+ SortMergeJoinExec: join_type=..., on=[(a@0, b1@1)]
SortExec: expr=[a@0 ASC],
preserve_partitioning=[true]
RepartitionExec: partitioning=Hash([a@0], 10),
input_partitions=1
DataSourceExec: file_groups={1 group: [[x]]},
projection=[a, b, c, d, e], file_type=parquet
@@ -1439,20 +1439,20 @@ fn multi_smj_joins() -> Result<()> {
");
}
// Should include 7 RepartitionExecs (4 hash, 3
round-robin), 4 SortExecs
- // Since ordering of the left child is not preserved after
SortMergeJoin
+ // Since ordering of the left child is not preserved after
SortMergeJoinExec
// when mode is Right, RightSemi, RightAnti, Full
- // - We need to add one additional SortExec after
SortMergeJoin in contrast the test cases
+ // - We need to add one additional SortExec after
SortMergeJoinExec in contrast the test cases
// when mode is Inner, Left, LeftSemi, LeftAnti
// Similarly, since partitioning of the left side is not
preserved
// when mode is Right, RightSemi, RightAnti, Full
- // - We need to add one additional Hash Repartition after
SortMergeJoin in contrast the test
+ // - We need to add one additional Hash Repartition after
SortMergeJoinExec in contrast the test
// cases when mode is Inner, Left, LeftSemi, LeftAnti
_ => {
assert_plan!(plan_distrib, @r"
- SortMergeJoin: join_type=..., on=[(a@0, c@2)]
+ SortMergeJoinExec: join_type=..., on=[(a@0, c@2)]
SortExec: expr=[a@0 ASC],
preserve_partitioning=[true]
RepartitionExec: partitioning=Hash([a@0], 10),
input_partitions=10
- SortMergeJoin: join_type=..., on=[(a@0, b1@1)]
+ SortMergeJoinExec: join_type=..., on=[(a@0,
b1@1)]
SortExec: expr=[a@0 ASC],
preserve_partitioning=[true]
RepartitionExec: partitioning=Hash([a@0],
10), input_partitions=1
DataSourceExec: file_groups={1 group:
[[x]]}, projection=[a, b, c, d, e], file_type=parquet
@@ -1474,8 +1474,8 @@ fn multi_smj_joins() -> Result<()> {
JoinType::Inner | JoinType::Left | JoinType::LeftSemi |
JoinType::LeftAnti => {
// TODO(wiedld): show different test result if enforce
distribution first.
assert_plan!(plan_sort, @r"
- SortMergeJoin: join_type=..., on=[(a@0, c@2)]
- SortMergeJoin: join_type=..., on=[(a@0, b1@1)]
+ SortMergeJoinExec: join_type=..., on=[(a@0, c@2)]
+ SortMergeJoinExec: join_type=..., on=[(a@0, b1@1)]
RepartitionExec: partitioning=Hash([a@0], 10),
input_partitions=1, maintains_sort_order=true
SortExec: expr=[a@0 ASC],
preserve_partitioning=[false]
DataSourceExec: file_groups={1 group: [[x]]},
projection=[a, b, c, d, e], file_type=parquet
@@ -1489,22 +1489,22 @@ fn multi_smj_joins() -> Result<()> {
");
}
// Should include 8 RepartitionExecs (4 hash, 8
round-robin), 4 SortExecs
- // Since ordering of the left child is not preserved after
SortMergeJoin
+ // Since ordering of the left child is not preserved after
SortMergeJoinExec
// when mode is Right, RightSemi, RightAnti, Full
- // - We need to add one additional SortExec after
SortMergeJoin in contrast the test cases
+ // - We need to add one additional SortExec after
SortMergeJoinExec in contrast the test cases
// when mode is Inner, Left, LeftSemi, LeftAnti
// Similarly, since partitioning of the left side is not
preserved
// when mode is Right, RightSemi, RightAnti, Full
// - We need to add one additional Hash Repartition and
Roundrobin repartition after
- // SortMergeJoin in contrast the test cases when mode is
Inner, Left, LeftSemi, LeftAnti
+ // SortMergeJoinExec in contrast the test cases when
mode is Inner, Left, LeftSemi, LeftAnti
_ => {
// TODO(wiedld): show different test result if enforce
distribution first.
assert_plan!(plan_sort, @r"
- SortMergeJoin: join_type=..., on=[(a@0, c@2)]
+ SortMergeJoinExec: join_type=..., on=[(a@0, c@2)]
RepartitionExec: partitioning=Hash([a@0], 10),
input_partitions=1, maintains_sort_order=true
SortExec: expr=[a@0 ASC],
preserve_partitioning=[false]
CoalescePartitionsExec
- SortMergeJoin: join_type=..., on=[(a@0, b1@1)]
+ SortMergeJoinExec: join_type=..., on=[(a@0,
b1@1)]
RepartitionExec: partitioning=Hash([a@0],
10), input_partitions=1, maintains_sort_order=true
SortExec: expr=[a@0 ASC],
preserve_partitioning=[false]
DataSourceExec: file_groups={1 group:
[[x]]}, projection=[a, b, c, d, e], file_type=parquet
@@ -1536,8 +1536,8 @@ fn multi_smj_joins() -> Result<()> {
JoinType::Inner | JoinType::Right => {
// TODO(wiedld): show different test result if
enforce sorting first.
assert_plan!(plan_distrib, @r"
- SortMergeJoin: join_type=..., on=[(b1@6, c@2)]
- SortMergeJoin: join_type=..., on=[(a@0,
b1@1)]
+ SortMergeJoinExec: join_type=..., on=[(b1@6,
c@2)]
+ SortMergeJoinExec: join_type=..., on=[(a@0,
b1@1)]
SortExec: expr=[a@0 ASC],
preserve_partitioning=[true]
RepartitionExec:
partitioning=Hash([a@0], 10), input_partitions=1
DataSourceExec: file_groups={1 group:
[[x]]}, projection=[a, b, c, d, e], file_type=parquet
@@ -1554,10 +1554,10 @@ fn multi_smj_joins() -> Result<()> {
JoinType::Left | JoinType::Full => {
// TODO(wiedld): show different test result if
enforce sorting first.
assert_plan!(plan_distrib, @r"
- SortMergeJoin: join_type=..., on=[(b1@6, c@2)]
+ SortMergeJoinExec: join_type=..., on=[(b1@6,
c@2)]
SortExec: expr=[b1@6 ASC],
preserve_partitioning=[true]
RepartitionExec: partitioning=Hash([b1@6],
10), input_partitions=10
- SortMergeJoin: join_type=..., on=[(a@0,
b1@1)]
+ SortMergeJoinExec: join_type=...,
on=[(a@0, b1@1)]
SortExec: expr=[a@0 ASC],
preserve_partitioning=[true]
RepartitionExec:
partitioning=Hash([a@0], 10), input_partitions=1
DataSourceExec: file_groups={1
group: [[x]]}, projection=[a, b, c, d, e], file_type=parquet
@@ -1581,8 +1581,8 @@ fn multi_smj_joins() -> Result<()> {
JoinType::Inner | JoinType::Right => {
// TODO(wiedld): show different test result if
enforce distribution first.
assert_plan!(plan_sort, @r"
- SortMergeJoin: join_type=..., on=[(b1@6, c@2)]
- SortMergeJoin: join_type=..., on=[(a@0,
b1@1)]
+ SortMergeJoinExec: join_type=..., on=[(b1@6,
c@2)]
+ SortMergeJoinExec: join_type=..., on=[(a@0,
b1@1)]
RepartitionExec: partitioning=Hash([a@0],
10), input_partitions=1, maintains_sort_order=true
SortExec: expr=[a@0 ASC],
preserve_partitioning=[false]
DataSourceExec: file_groups={1 group:
[[x]]}, projection=[a, b, c, d, e], file_type=parquet
@@ -1599,11 +1599,11 @@ fn multi_smj_joins() -> Result<()> {
JoinType::Left | JoinType::Full => {
// TODO(wiedld): show different test result if
enforce distribution first.
assert_plan!(plan_sort, @r"
- SortMergeJoin: join_type=..., on=[(b1@6, c@2)]
+ SortMergeJoinExec: join_type=..., on=[(b1@6,
c@2)]
RepartitionExec: partitioning=Hash([b1@6],
10), input_partitions=1, maintains_sort_order=true
SortExec: expr=[b1@6 ASC],
preserve_partitioning=[false]
CoalescePartitionsExec
- SortMergeJoin: join_type=...,
on=[(a@0, b1@1)]
+ SortMergeJoinExec: join_type=...,
on=[(a@0, b1@1)]
RepartitionExec:
partitioning=Hash([a@0], 10), input_partitions=1, maintains_sort_order=true
SortExec: expr=[a@0 ASC],
preserve_partitioning=[false]
DataSourceExec: file_groups={1
group: [[x]]}, projection=[a, b, c, d, e], file_type=parquet
@@ -1682,7 +1682,7 @@ fn smj_join_key_ordering() -> Result<()> {
// Only two RepartitionExecs added
let plan_distrib = test_config.to_plan(join.clone(),
&DISTRIB_DISTRIB_SORT);
assert_plan!(plan_distrib, @r"
- SortMergeJoin: join_type=Inner, on=[(b3@1, b2@1), (a3@0, a2@0)]
+ SortMergeJoinExec: join_type=Inner, on=[(b3@1, b2@1), (a3@0, a2@0)]
SortExec: expr=[b3@1 ASC, a3@0 ASC], preserve_partitioning=[true]
ProjectionExec: expr=[a1@0 as a3, b1@1 as b3]
ProjectionExec: expr=[a1@1 as a1, b1@0 as b1]
@@ -1703,7 +1703,7 @@ fn smj_join_key_ordering() -> Result<()> {
// Test: result IS DIFFERENT, if EnforceSorting is run first:
let plan_sort = test_config.to_plan(join, &SORT_DISTRIB_DISTRIB);
assert_plan!(plan_sort, @r"
- SortMergeJoin: join_type=Inner, on=[(b3@1, b2@1), (a3@0, a2@0)]
+ SortMergeJoinExec: join_type=Inner, on=[(b3@1, b2@1), (a3@0, a2@0)]
RepartitionExec: partitioning=Hash([b3@1, a3@0], 10),
input_partitions=1, maintains_sort_order=true
SortExec: expr=[b3@1 ASC, a3@0 ASC], preserve_partitioning=[false]
CoalescePartitionsExec
diff --git a/datafusion/core/tests/physical_optimizer/enforce_sorting.rs
b/datafusion/core/tests/physical_optimizer/enforce_sorting.rs
index ca99b17be0..47e3adb455 100644
--- a/datafusion/core/tests/physical_optimizer/enforce_sorting.rs
+++ b/datafusion/core/tests/physical_optimizer/enforce_sorting.rs
@@ -1340,12 +1340,12 @@ async fn test_sort_merge_join_order_by_left() ->
Result<()> {
assert_snapshot!(test.run(), @r"
Input Plan:
SortPreservingMergeExec: [nullable_col@0 ASC,
non_nullable_col@1 ASC]
- SortMergeJoin: join_type=..., on=[(nullable_col@0, col_a@0)]
+ SortMergeJoinExec: join_type=..., on=[(nullable_col@0,
col_a@0)]
DataSourceExec: file_groups={1 group: [[x]]},
projection=[nullable_col, non_nullable_col], file_type=parquet
DataSourceExec: file_groups={1 group: [[x]]},
projection=[col_a, col_b], file_type=parquet
Optimized Plan:
- SortMergeJoin: join_type=..., on=[(nullable_col@0, col_a@0)]
+ SortMergeJoinExec: join_type=..., on=[(nullable_col@0,
col_a@0)]
SortExec: expr=[nullable_col@0 ASC, non_nullable_col@1 ASC],
preserve_partitioning=[false]
DataSourceExec: file_groups={1 group: [[x]]},
projection=[nullable_col, non_nullable_col], file_type=parquet
SortExec: expr=[col_a@0 ASC], preserve_partitioning=[false]
@@ -1357,13 +1357,13 @@ async fn test_sort_merge_join_order_by_left() ->
Result<()> {
assert_snapshot!(test.run(), @r"
Input Plan:
SortPreservingMergeExec: [nullable_col@0 ASC,
non_nullable_col@1 ASC]
- SortMergeJoin: join_type=..., on=[(nullable_col@0, col_a@0)]
+ SortMergeJoinExec: join_type=..., on=[(nullable_col@0,
col_a@0)]
DataSourceExec: file_groups={1 group: [[x]]},
projection=[nullable_col, non_nullable_col], file_type=parquet
DataSourceExec: file_groups={1 group: [[x]]},
projection=[col_a, col_b], file_type=parquet
Optimized Plan:
SortExec: expr=[nullable_col@0 ASC, non_nullable_col@1 ASC],
preserve_partitioning=[false]
- SortMergeJoin: join_type=..., on=[(nullable_col@0, col_a@0)]
+ SortMergeJoinExec: join_type=..., on=[(nullable_col@0,
col_a@0)]
SortExec: expr=[nullable_col@0 ASC],
preserve_partitioning=[false]
DataSourceExec: file_groups={1 group: [[x]]},
projection=[nullable_col, non_nullable_col], file_type=parquet
SortExec: expr=[col_a@0 ASC], preserve_partitioning=[false]
@@ -1430,12 +1430,12 @@ async fn test_sort_merge_join_order_by_right() ->
Result<()> {
assert_snapshot!(test.run(), @r"
Input Plan:
SortPreservingMergeExec: [col_a@2 ASC, col_b@3 ASC]
- SortMergeJoin: join_type=..., on=[(nullable_col@0, col_a@0)]
+ SortMergeJoinExec: join_type=..., on=[(nullable_col@0,
col_a@0)]
DataSourceExec: file_groups={1 group: [[x]]},
projection=[nullable_col, non_nullable_col], file_type=parquet
DataSourceExec: file_groups={1 group: [[x]]},
projection=[col_a, col_b], file_type=parquet
Optimized Plan:
- SortMergeJoin: join_type=..., on=[(nullable_col@0, col_a@0)]
+ SortMergeJoinExec: join_type=..., on=[(nullable_col@0,
col_a@0)]
SortExec: expr=[nullable_col@0 ASC],
preserve_partitioning=[false]
DataSourceExec: file_groups={1 group: [[x]]},
projection=[nullable_col, non_nullable_col], file_type=parquet
SortExec: expr=[col_a@0 ASC, col_b@1 ASC],
preserve_partitioning=[false]
@@ -1447,12 +1447,12 @@ async fn test_sort_merge_join_order_by_right() ->
Result<()> {
assert_snapshot!(test.run(), @r"
Input Plan:
SortPreservingMergeExec: [col_a@0 ASC, col_b@1 ASC]
- SortMergeJoin: join_type=..., on=[(nullable_col@0, col_a@0)]
+ SortMergeJoinExec: join_type=..., on=[(nullable_col@0,
col_a@0)]
DataSourceExec: file_groups={1 group: [[x]]},
projection=[nullable_col, non_nullable_col], file_type=parquet
DataSourceExec: file_groups={1 group: [[x]]},
projection=[col_a, col_b], file_type=parquet
Optimized Plan:
- SortMergeJoin: join_type=..., on=[(nullable_col@0, col_a@0)]
+ SortMergeJoinExec: join_type=..., on=[(nullable_col@0,
col_a@0)]
SortExec: expr=[nullable_col@0 ASC],
preserve_partitioning=[false]
DataSourceExec: file_groups={1 group: [[x]]},
projection=[nullable_col, non_nullable_col], file_type=parquet
SortExec: expr=[col_a@0 ASC, col_b@1 ASC],
preserve_partitioning=[false]
@@ -1464,13 +1464,13 @@ async fn test_sort_merge_join_order_by_right() ->
Result<()> {
assert_snapshot!(test.run(), @r"
Input Plan:
SortPreservingMergeExec: [col_a@2 ASC, col_b@3 ASC]
- SortMergeJoin: join_type=..., on=[(nullable_col@0, col_a@0)]
+ SortMergeJoinExec: join_type=..., on=[(nullable_col@0,
col_a@0)]
DataSourceExec: file_groups={1 group: [[x]]},
projection=[nullable_col, non_nullable_col], file_type=parquet
DataSourceExec: file_groups={1 group: [[x]]},
projection=[col_a, col_b], file_type=parquet
Optimized Plan:
SortExec: expr=[col_a@2 ASC, col_b@3 ASC],
preserve_partitioning=[false]
- SortMergeJoin: join_type=..., on=[(nullable_col@0, col_a@0)]
+ SortMergeJoinExec: join_type=..., on=[(nullable_col@0,
col_a@0)]
SortExec: expr=[nullable_col@0 ASC],
preserve_partitioning=[false]
DataSourceExec: file_groups={1 group: [[x]]},
projection=[nullable_col, non_nullable_col], file_type=parquet
SortExec: expr=[col_a@0 ASC], preserve_partitioning=[false]
@@ -1513,13 +1513,13 @@ async fn test_sort_merge_join_complex_order_by() ->
Result<()> {
assert_snapshot!(test.run(), @r"
Input Plan:
SortPreservingMergeExec: [col_b@3 ASC, col_a@2 ASC]
- SortMergeJoin: join_type=Inner, on=[(nullable_col@0, col_a@0)]
+ SortMergeJoinExec: join_type=Inner, on=[(nullable_col@0, col_a@0)]
DataSourceExec: file_groups={1 group: [[x]]},
projection=[nullable_col, non_nullable_col], file_type=parquet
DataSourceExec: file_groups={1 group: [[x]]}, projection=[col_a,
col_b], file_type=parquet
Optimized Plan:
SortExec: expr=[col_b@3 ASC, nullable_col@0 ASC],
preserve_partitioning=[false]
- SortMergeJoin: join_type=Inner, on=[(nullable_col@0, col_a@0)]
+ SortMergeJoinExec: join_type=Inner, on=[(nullable_col@0, col_a@0)]
SortExec: expr=[nullable_col@0 ASC], preserve_partitioning=[false]
DataSourceExec: file_groups={1 group: [[x]]},
projection=[nullable_col, non_nullable_col], file_type=parquet
SortExec: expr=[col_a@0 ASC], preserve_partitioning=[false]
@@ -1540,12 +1540,12 @@ async fn test_sort_merge_join_complex_order_by() ->
Result<()> {
assert_snapshot!(test.run(), @r"
Input Plan:
SortPreservingMergeExec: [nullable_col@0 ASC, col_b@3 ASC, col_a@2 ASC]
- SortMergeJoin: join_type=Inner, on=[(nullable_col@0, col_a@0)]
+ SortMergeJoinExec: join_type=Inner, on=[(nullable_col@0, col_a@0)]
DataSourceExec: file_groups={1 group: [[x]]},
projection=[nullable_col, non_nullable_col], file_type=parquet
DataSourceExec: file_groups={1 group: [[x]]}, projection=[col_a,
col_b], file_type=parquet
Optimized Plan:
- SortMergeJoin: join_type=Inner, on=[(nullable_col@0, col_a@0)]
+ SortMergeJoinExec: join_type=Inner, on=[(nullable_col@0, col_a@0)]
SortExec: expr=[nullable_col@0 ASC], preserve_partitioning=[false]
DataSourceExec: file_groups={1 group: [[x]]},
projection=[nullable_col, non_nullable_col], file_type=parquet
SortExec: expr=[col_a@0 ASC, col_b@1 ASC], preserve_partitioning=[false]
diff --git a/datafusion/core/tests/physical_optimizer/sanity_checker.rs
b/datafusion/core/tests/physical_optimizer/sanity_checker.rs
index 5aa3925e27..217570846d 100644
--- a/datafusion/core/tests/physical_optimizer/sanity_checker.rs
+++ b/datafusion/core/tests/physical_optimizer/sanity_checker.rs
@@ -555,7 +555,7 @@ async fn test_sort_merge_join_satisfied() -> Result<()> {
assert_snapshot!(
actual,
@r"
- SortMergeJoin: join_type=Inner, on=[(c9@0, a@0)]
+ SortMergeJoinExec: join_type=Inner, on=[(c9@0, a@0)]
RepartitionExec: partitioning=Hash([c9@0], 10), input_partitions=1,
maintains_sort_order=true
SortExec: expr=[c9@0 ASC], preserve_partitioning=[false]
DataSourceExec: partitions=1, partition_sizes=[0]
@@ -605,7 +605,7 @@ async fn test_sort_merge_join_order_missing() -> Result<()>
{
assert_snapshot!(
actual,
@r"
- SortMergeJoin: join_type=Inner, on=[(c9@0, a@0)]
+ SortMergeJoinExec: join_type=Inner, on=[(c9@0, a@0)]
RepartitionExec: partitioning=Hash([c9@0], 10), input_partitions=1,
maintains_sort_order=true
SortExec: expr=[c9@0 ASC], preserve_partitioning=[false]
DataSourceExec: partitions=1, partition_sizes=[0]
@@ -653,7 +653,7 @@ async fn test_sort_merge_join_dist_missing() -> Result<()> {
assert_snapshot!(
actual,
@r"
- SortMergeJoin: join_type=Inner, on=[(c9@0, a@0)]
+ SortMergeJoinExec: join_type=Inner, on=[(c9@0, a@0)]
RepartitionExec: partitioning=Hash([c9@0], 10), input_partitions=1,
maintains_sort_order=true
SortExec: expr=[c9@0 ASC], preserve_partitioning=[false]
DataSourceExec: partitions=1, partition_sizes=[0]
diff --git a/datafusion/physical-plan/src/joins/sort_merge_join/exec.rs
b/datafusion/physical-plan/src/joins/sort_merge_join/exec.rs
index 5362259d22..5560c29d54 100644
--- a/datafusion/physical-plan/src/joins/sort_merge_join/exec.rs
+++ b/datafusion/physical-plan/src/joins/sort_merge_join/exec.rs
@@ -360,7 +360,8 @@ impl DisplayAs for SortMergeJoinExec {
};
write!(
f,
- "SortMergeJoin: join_type={:?}, on=[{}]{}{}",
+ "{}: join_type={:?}, on=[{}]{}{}",
+ Self::static_name(),
self.join_type,
on,
self.filter.as_ref().map_or_else(
diff --git a/datafusion/sqllogictest/test_files/joins.slt
b/datafusion/sqllogictest/test_files/joins.slt
index 3259ef9cd8..38037ede21 100644
--- a/datafusion/sqllogictest/test_files/joins.slt
+++ b/datafusion/sqllogictest/test_files/joins.slt
@@ -2721,7 +2721,7 @@ logical_plan
04)--SubqueryAlias: t2
05)----TableScan: hashjoin_datatype_table_t2 projection=[c1, c2, c3, c4]
physical_plan
-01)SortMergeJoin: join_type=Inner, on=[(c1@0, c1@0)]
+01)SortMergeJoinExec: join_type=Inner, on=[(c1@0, c1@0)]
02)--SortExec: expr=[c1@0 ASC], preserve_partitioning=[true]
03)----RepartitionExec: partitioning=Hash([c1@0], 2), input_partitions=1
04)------DataSourceExec: partitions=1, partition_sizes=[1]
@@ -2748,7 +2748,7 @@ logical_plan
05)----TableScan: hashjoin_datatype_table_t2 projection=[c1, c2, c3, c4]
physical_plan
01)ProjectionExec: expr=[c1@0 as c1, c2@1 as c2, c3@2 as c3, c4@3 as c4, c1@5
as c1, c2@6 as c2, c3@7 as c3, c4@8 as c4]
-02)--SortMergeJoin: join_type=Right, on=[(CAST(t1.c3 AS Decimal128(10, 2))@4,
c3@2)]
+02)--SortMergeJoinExec: join_type=Right, on=[(CAST(t1.c3 AS Decimal128(10,
2))@4, c3@2)]
03)----SortExec: expr=[CAST(t1.c3 AS Decimal128(10, 2))@4 ASC],
preserve_partitioning=[true]
04)------RepartitionExec: partitioning=Hash([CAST(t1.c3 AS Decimal128(10,
2))@4], 2), input_partitions=2
05)--------ProjectionExec: expr=[c1@0 as c1, c2@1 as c2, c3@2 as c3, c4@3 as
c4, CAST(c3@2 AS Decimal128(10, 2)) as CAST(t1.c3 AS Decimal128(10, 2))]
@@ -3157,7 +3157,7 @@ logical_plan
08)------TableScan: annotated_data projection=[a0, a, b, c, d]
physical_plan
01)SortPreservingMergeExec: [rn1@5 ASC NULLS LAST]
-02)--SortMergeJoin: join_type=Inner, on=[(a@1, a@1)]
+02)--SortMergeJoinExec: join_type=Inner, on=[(a@1, a@1)]
03)----RepartitionExec: partitioning=Hash([a@1], 2), input_partitions=1,
maintains_sort_order=true
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]
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]
@@ -3187,7 +3187,7 @@ logical_plan
08)----------TableScan: annotated_data projection=[a0, a, b, c, d]
physical_plan
01)SortPreservingMergeExec: [rn1@10 ASC NULLS LAST]
-02)--SortMergeJoin: join_type=Right, on=[(a@1, a@1)]
+02)--SortMergeJoinExec: join_type=Right, on=[(a@1, a@1)]
03)----RepartitionExec: partitioning=Hash([a@1], 2), input_partitions=1,
maintains_sort_order=true
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
05)----RepartitionExec: partitioning=Hash([a@1], 2), input_partitions=1,
maintains_sort_order=true
@@ -3198,7 +3198,7 @@ physical_plan
statement ok
set datafusion.optimizer.prefer_existing_sort = false;
-# SortMergeJoin should add ordering equivalences of
+# SortMergeJoinExec should add ordering equivalences of
# right table as lexicographical append to the global ordering
# below query shouldn't add any SortExec for order by clause.
# since its requirement is already satisfied at the output of SortMergeJoinExec
@@ -3224,7 +3224,7 @@ logical_plan
10)----------TableScan: annotated_data projection=[a0, a, b, c, d]
physical_plan
01)SortPreservingMergeExec: [a@1 ASC, b@2 ASC NULLS LAST, c@3 ASC NULLS LAST,
rn1@11 ASC NULLS LAST]
-02)--SortMergeJoin: join_type=Inner, on=[(a@1, a@1)]
+02)--SortMergeJoinExec: join_type=Inner, on=[(a@1, a@1)]
03)----RepartitionExec: partitioning=Hash([a@1], 2), input_partitions=1,
maintains_sort_order=true
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]
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]
diff --git a/datafusion/sqllogictest/test_files/sort_merge_join.slt
b/datafusion/sqllogictest/test_files/sort_merge_join.slt
index 5f9276bdb7..d2fa37ef76 100644
--- a/datafusion/sqllogictest/test_files/sort_merge_join.slt
+++ b/datafusion/sqllogictest/test_files/sort_merge_join.slt
@@ -37,7 +37,7 @@ logical_plan
02)--TableScan: t1 projection=[a, b]
03)--TableScan: t2 projection=[a, b]
physical_plan
-01)SortMergeJoin: join_type=Inner, on=[(a@0, a@0)], filter=CAST(b@1 AS Int64)
* 50 <= CAST(b@0 AS Int64)
+01)SortMergeJoinExec: join_type=Inner, on=[(a@0, a@0)], filter=CAST(b@1 AS
Int64) * 50 <= CAST(b@0 AS Int64)
02)--SortExec: expr=[a@0 ASC], preserve_partitioning=[false]
03)----DataSourceExec: partitions=1, partition_sizes=[1]
04)--SortExec: expr=[a@0 ASC], preserve_partitioning=[false]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
