You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/08/03 19:35:48 UTC
(datafusion) branch gh-readonly-queue/main/pr-23566-f9dde71ec315833ea0bc39bd63104e613109e915 created (now a93d065731)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-23566-f9dde71ec315833ea0bc39bd63104e613109e915
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at a93d065731 feat: eliminate LEFT/RIGHT JOINs with redundant sides (#23566)
This branch includes the following new commits:
new a93d065731 feat: eliminate LEFT/RIGHT JOINs with redundant sides (#23566)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion) 01/01: feat: eliminate LEFT/RIGHT JOINs with redundant sides (#23566)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-23566-f9dde71ec315833ea0bc39bd63104e613109e915
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit a93d06573182d6622331133506525b93bac4ebf9
Author: Simon Vandel Sillesen <si...@gmail.com>
AuthorDate: Mon Aug 3 21:35:24 2026 +0200
feat: eliminate LEFT/RIGHT JOINs with redundant sides (#23566)
## Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and
enhancements and this helps us generate change logs for our releases.
You can link an issue to this PR using the GitHub syntax. For example
`Closes #123` indicates that this PR will close issue #123.
-->
- Closes #.
## Rationale for this change
Join elimination is useful in e.g. generated queries or views, where
joined columns can end up not being used.
An explanation of when the optimization is valid, can be seen in the
docs update for `EliminateJoin`.
<!--
Why are you proposing this change? If this is already explained clearly
in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand
your changes and offer better suggestions for fixes.
-->
## What changes are included in this PR?
Builds upon the analysis that
https://github.com/apache/datafusion/pull/22652 introduced for the
`EliminateJoin` optimization pass. We use it to detect when the
right-side of a left-join can be removed. Same symmetrical rule for
right-join elimination.
<!--
There is no need to duplicate the description in the issue here but it
is sometimes worth providing a summary of the individual changes in this
PR.
-->
## Are these changes tested?
Yes, SLT additions that test the feature end-to-end.
<!--
We typically require tests for all PRs in order to:
1. Prevent the code from being accidentally broken by subsequent changes
2. Serve as another way to document the expected behavior of the code
If tests are not included in your PR, please explain why (for example,
are they covered by existing tests)?
-->
## Are there any user-facing changes?
Yes, faster queries!
<!--
If there are user-facing changes then we may require documentation to be
updated before approving the PR.
-->
<!--
If there are any breaking changes to public APIs, please add the `api
change` label.
-->
---------
Co-authored-by: Claude <no...@anthropic.com>
---
datafusion/optimizer/src/eliminate_join.rs | 126 +++--
.../test_files/functional_dependencies.slt | 10 +-
datafusion/sqllogictest/test_files/group_by.slt | 15 +-
datafusion/sqllogictest/test_files/joins.slt | 539 +++++++++++++++++++++
4 files changed, 642 insertions(+), 48 deletions(-)
diff --git a/datafusion/optimizer/src/eliminate_join.rs b/datafusion/optimizer/src/eliminate_join.rs
index cce17c07b5..56aa888706 100644
--- a/datafusion/optimizer/src/eliminate_join.rs
+++ b/datafusion/optimizer/src/eliminate_join.rs
@@ -15,8 +15,8 @@
// specific language governing permissions and limitations
// under the License.
-//! [`EliminateJoin`] rewrites inner joins to simpler forms to make them cheaper
-//! to evaluate. We implement two distinct rewrites:
+//! [`EliminateJoin`] rewrites joins to simpler forms to make them cheaper
+//! to evaluate. We implement three distinct rewrites:
//!
//! * An inner join can be rewritten to an empty relation if the join condition
//! is trivially false.
@@ -32,6 +32,18 @@
//! functional dependencies to prove that each L row matches at most one R
//! row (R is provably unique on the join keys).
//!
+//! * A left outer join `L ⟕ R` can be removed entirely, i.e. replaced by `L`,
+//! under the same two conditions. Unlike an inner join, a left join
+//! preserves every row of L whether or not it has a match in R, so when R's
+//! columns are unused and R cannot multiply L's rows the join has no
+//! observable effect at all. Such joins commonly appear in generated SQL
+//! and in queries over views that join in lookup tables the query does not
+//! read. A join filter does not prevent this rewrite: for a left join it
+//! only decides whether a left row is matched or null-padded, and either
+//! way the row is emitted. Symmetrically, a right outer join `L ⟖ R` can be
+//! replaced by `R` when L's columns are unused and L cannot multiply R's
+//! rows.
+//!
//! # Overview
//!
//! `rewrite_subtree` walks the plan top-down, threading two pieces of context
@@ -52,7 +64,8 @@
//! so a duplicate-sensitive node further above does not matter.
//!
//! At each join, `rewritten_join_type` combines this context with the side's
-//! functional dependencies to choose `Inner`, `LeftSemi`, or `RightSemi`. Most
+//! functional dependencies to choose `Inner`, `LeftSemi`, or `RightSemi`, or
+//! to eliminate the join entirely in favor of its preserved input. Most
//! node types just forward the context to their single child via
//! `rewrite_single_input`; nodes that alter column requirements or
//! duplicate-sensitivity (projection, aggregate, sort, ...) adjust it first.
@@ -142,8 +155,10 @@ impl LiveColumns {
}
}
-/// Rewrites an inner join to a semi join when one input only filters the other,
-/// and replaces an always-false inner join with an empty relation.
+/// Rewrites an inner join to a semi join when one input only filters the
+/// other, removes an outer join whose non-preserved side is unused and cannot
+/// multiply the preserved side's rows, and replaces an always-false inner join
+/// with an empty relation.
#[derive(Default, Debug)]
pub struct EliminateJoin;
@@ -394,8 +409,30 @@ fn rewrite_join(
let (visible_left, visible_right) = split_join_output_columns(&join, live);
- let rewritten_join_type =
- rewritten_join_type(&join, &visible_left, &visible_right, duplicate_insensitive);
+ let rewritten_join_type = match rewritten_join_type(
+ &join,
+ &visible_left,
+ &visible_right,
+ duplicate_insensitive,
+ ) {
+ JoinRewrite::ReplaceWithLeft => {
+ let left = rewrite_subtree(
+ Arc::unwrap_or_clone(join.left),
+ visible_left,
+ duplicate_insensitive,
+ )?;
+ return Ok(Transformed::yes(left.data));
+ }
+ JoinRewrite::ReplaceWithRight => {
+ let right = rewrite_subtree(
+ Arc::unwrap_or_clone(join.right),
+ visible_right,
+ duplicate_insensitive,
+ )?;
+ return Ok(Transformed::yes(right.data));
+ }
+ JoinRewrite::Join(join_type) => join_type,
+ };
let (mut left_live, mut right_live) = match rewritten_join_type {
JoinType::LeftSemi | JoinType::LeftAnti | JoinType::LeftMark => {
@@ -477,40 +514,69 @@ fn child_duplicate_insensitivity(
}
}
-/// Rewrites an inner join to a semi join when the removed side has no
-/// parent-visible columns and either the parent ignores duplicate output rows or
-/// the removed side is unique on the join keys.
+/// The rewrite chosen for a join by [`rewritten_join_type`].
+enum JoinRewrite {
+ /// Keep the join, with this (possibly rewritten) join type.
+ Join(JoinType),
+ /// The join has no observable effect; replace it with its left input.
+ ReplaceWithLeft,
+ /// The join has no observable effect; replace it with its right input.
+ ReplaceWithRight,
+}
+
+/// Chooses a cheaper form for a join: removes an outer join whose non-preserved
+/// side is redundant, or rewrites an inner join to a semi join when the
+/// removed side has no parent-visible columns and either the parent ignores
+/// duplicate output rows or the removed side is unique on the join keys.
fn rewritten_join_type(
join: &Join,
visible_left: &LiveColumns,
visible_right: &LiveColumns,
duplicate_insensitive: bool,
-) -> JoinType {
- if join.join_type != JoinType::Inner || join.on.is_empty() {
- return join.join_type;
+) -> JoinRewrite {
+ // A side is redundant when nothing above the join references its columns
+ // and it cannot multiply the other side's rows (the ancestors are
+ // duplicate-insensitive, or the side is unique on the join keys).
+ let can_remove_right = visible_right.is_empty()
+ && (duplicate_insensitive
+ || side_unique_on_join(
+ join.right.schema(),
+ join.on.iter().map(|(_, right)| right),
+ join.null_equality,
+ ));
+
+ // A LEFT JOIN preserves every left row, so with a redundant right side the
+ // join has no observable effect and can be replaced by its left input. A
+ // join filter cannot prevent this: it only decides whether a left row is
+ // matched or null-padded, and either way the row is emitted.
+ if join.join_type == JoinType::Left && can_remove_right {
+ return JoinRewrite::ReplaceWithLeft;
+ }
+ let can_remove_left = visible_left.is_empty()
+ && (duplicate_insensitive
+ || side_unique_on_join(
+ join.left.schema(),
+ join.on.iter().map(|(left, _)| left),
+ join.null_equality,
+ ));
+
+ // Symmetrical rule for RIGHT JOIN removal (same explanation as above for the left-join case)
+ if join.join_type == JoinType::Right && can_remove_left {
+ return JoinRewrite::ReplaceWithRight;
}
- let can_remove_right = duplicate_insensitive
- || side_unique_on_join(
- join.right.schema(),
- join.on.iter().map(|(_, right)| right),
- join.null_equality,
- );
- if visible_right.is_empty() && can_remove_right {
- return JoinType::LeftSemi;
+ if join.join_type != JoinType::Inner || join.on.is_empty() {
+ return JoinRewrite::Join(join.join_type);
}
- let can_remove_left = duplicate_insensitive
- || side_unique_on_join(
- join.left.schema(),
- join.on.iter().map(|(left, _)| left),
- join.null_equality,
- );
- if visible_left.is_empty() && can_remove_left {
- return JoinType::RightSemi;
+ if can_remove_right {
+ return JoinRewrite::Join(JoinType::LeftSemi);
+ }
+ if can_remove_left {
+ return JoinRewrite::Join(JoinType::RightSemi);
}
- JoinType::Inner
+ JoinRewrite::Join(JoinType::Inner)
}
fn add_join_condition_columns(
diff --git a/datafusion/sqllogictest/test_files/functional_dependencies.slt b/datafusion/sqllogictest/test_files/functional_dependencies.slt
index 92aedf66e6..c49004190d 100644
--- a/datafusion/sqllogictest/test_files/functional_dependencies.slt
+++ b/datafusion/sqllogictest/test_files/functional_dependencies.slt
@@ -102,12 +102,8 @@ EXPLAIN SELECT DISTINCT p.x FROM t_pk p LEFT JOIN t_orders o ON p.x = o.x;
----
logical_plan
01)Aggregate: groupBy=[[p.x]], aggr=[[]]
-02)--Projection: p.x
-03)----Left Join: p.x = o.x
-04)------SubqueryAlias: p
-05)--------TableScan: t_pk projection=[x]
-06)------SubqueryAlias: o
-07)--------TableScan: t_orders projection=[x]
+02)--SubqueryAlias: p
+03)----TableScan: t_pk projection=[x]
statement ok
drop table t_orders;
@@ -283,7 +279,7 @@ logical_plan
10)--------------TableScan: t_null projection=[x]
# 5.2 The ORDER BY variant: `g.x` is NULL for both rows, so the `g.cnt`
-# tie-breaker is what orders them.
+# tie-breaker is what orders them.
query II
SELECT g.x, g.cnt
FROM t_probe a
diff --git a/datafusion/sqllogictest/test_files/group_by.slt b/datafusion/sqllogictest/test_files/group_by.slt
index afde5b9331..637b7c1882 100644
--- a/datafusion/sqllogictest/test_files/group_by.slt
+++ b/datafusion/sqllogictest/test_files/group_by.slt
@@ -5674,20 +5674,13 @@ EXPLAIN SELECT DISTINCT u.id
----
logical_plan
01)Aggregate: groupBy=[[u.id]], aggr=[[]]
-02)--Projection: u.id
-03)----Left Join: u.id = o.user_id
-04)------SubqueryAlias: u
-05)--------TableScan: users_with_pk projection=[id]
-06)------SubqueryAlias: o
-07)--------TableScan: user_orders projection=[user_id]
+02)--SubqueryAlias: u
+03)----TableScan: users_with_pk projection=[id]
physical_plan
01)AggregateExec: mode=FinalPartitioned, gby=[id@0 as id], aggr=[]
-02)--RepartitionExec: partitioning=Hash([id@0], 4), input_partitions=4
+02)--RepartitionExec: partitioning=Hash([id@0], 4), input_partitions=1
03)----AggregateExec: mode=Partial, gby=[id@0 as id], aggr=[]
-04)------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
-05)--------HashJoinExec: mode=CollectLeft, join_type=Left, on=[(id@0, user_id@0)], projection=[id@0]
-06)----------DataSourceExec: partitions=1, partition_sizes=[1]
-07)----------DataSourceExec: partitions=1, partition_sizes=[1]
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
statement ok
drop table users_with_pk;
diff --git a/datafusion/sqllogictest/test_files/joins.slt b/datafusion/sqllogictest/test_files/joins.slt
index 55efcc3874..7a706836f4 100644
--- a/datafusion/sqllogictest/test_files/joins.slt
+++ b/datafusion/sqllogictest/test_files/joins.slt
@@ -5609,6 +5609,545 @@ set datafusion.execution.target_partitions = 4;
statement ok
reset datafusion.execution.batch_size;
+##########
+# Eliminate unused outer joins (`EliminateJoin` rule)
+#
+# An outer join whose non-preserved side is unreferenced above the join is
+# removed entirely when it cannot duplicate the preserved side's rows: either
+# the non-preserved side is unique on the join keys (e.g. PRIMARY KEY /
+# UNIQUE constraint or GROUP BY), or the join's ancestors are
+# duplicate-insensitive. Most cases below exercise the LEFT JOIN direction;
+# RIGHT JOIN is symmetric and covered at the end of the section.
+##########
+
+statement ok
+CREATE TABLE elim_users (id INT primary key, name VARCHAR) AS VALUES
+ (1, 'alice'),
+ (2, 'bob'),
+ (4, 'dave');
+
+statement ok
+CREATE TABLE elim_orders (order_id INT, user_id INT, amount INT) AS VALUES
+ (1, 1, 100),
+ (2, 1, 200),
+ (3, 3, 50);
+
+# The right side is unique on the join key (primary key) and unused above the
+# join: the LEFT JOIN is removed from the plan.
+query TT
+EXPLAIN SELECT order_id, amount FROM elim_orders LEFT JOIN elim_users ON user_id = id;
+----
+logical_plan TableScan: elim_orders projection=[order_id, amount]
+physical_plan DataSourceExec: partitions=1, partition_sizes=[1]
+
+# All orders are returned, including the one with no matching user.
+query II rowsort
+SELECT order_id, amount FROM elim_orders LEFT JOIN elim_users ON user_id = id;
+----
+1 100
+2 200
+3 50
+
+# A WHERE clause on left-side columns does not block the rewrite.
+query TT
+EXPLAIN SELECT order_id FROM elim_orders LEFT JOIN elim_users ON user_id = id WHERE amount > 100;
+----
+logical_plan
+01)Projection: elim_orders.order_id
+02)--Filter: elim_orders.amount > Int32(100)
+03)----TableScan: elim_orders projection=[order_id, amount]
+physical_plan
+01)FilterExec: amount@1 > 100, projection=[order_id@0]
+02)--DataSourceExec: partitions=1, partition_sizes=[1]
+
+query I rowsort
+SELECT order_id FROM elim_orders LEFT JOIN elim_users ON user_id = id WHERE amount > 100;
+----
+2
+
+# An extra join filter on right-side columns does not block the rewrite: for a
+# left join it only decides whether a left row is matched or null-padded, and
+# either way the row is emitted.
+query TT
+EXPLAIN SELECT order_id FROM elim_orders LEFT JOIN elim_users ON user_id = id AND name <> 'bob';
+----
+logical_plan TableScan: elim_orders projection=[order_id]
+physical_plan DataSourceExec: partitions=1, partition_sizes=[1]
+
+query I rowsort
+SELECT order_id FROM elim_orders LEFT JOIN elim_users ON user_id = id AND name <> 'bob';
+----
+1
+2
+3
+
+# count(*) is duplicate-sensitive, but the unique join key guarantees each
+# order appears exactly once, so the join is still removed.
+query TT
+EXPLAIN SELECT count(*) FROM elim_orders LEFT JOIN elim_users ON user_id = id;
+----
+logical_plan
+01)Projection: count(Int64(1)) AS count(*)
+02)--Aggregate: groupBy=[[]], aggr=[[count(Int64(1))]]
+03)----TableScan: elim_orders projection=[]
+physical_plan
+01)ProjectionExec: expr=[3 as count(*)]
+02)--PlaceholderRowExec
+
+query I
+SELECT count(*) FROM elim_orders LEFT JOIN elim_users ON user_id = id;
+----
+3
+
+# A DISTINCT (or GROUP BY) right side is unique on its keys even without
+# declared constraints, so the join is removed.
+query TT
+EXPLAIN SELECT o.order_id FROM elim_orders o LEFT JOIN (SELECT DISTINCT user_id FROM elim_orders) d ON o.user_id = d.user_id;
+----
+logical_plan
+01)SubqueryAlias: o
+02)--TableScan: elim_orders projection=[order_id]
+physical_plan DataSourceExec: partitions=1, partition_sizes=[1]
+
+query I rowsort
+SELECT o.order_id FROM elim_orders o LEFT JOIN (SELECT DISTINCT user_id FROM elim_orders) d ON o.user_id = d.user_id;
+----
+1
+2
+3
+
+# Negative case: the right side is referenced in the SELECT list, so the join
+# must stay.
+query TT
+EXPLAIN SELECT order_id, name FROM elim_orders LEFT JOIN elim_users ON user_id = id;
+----
+logical_plan
+01)Projection: elim_orders.order_id, elim_users.name
+02)--Left Join: elim_orders.user_id = elim_users.id
+03)----TableScan: elim_orders projection=[order_id, user_id]
+04)----TableScan: elim_users projection=[id, name]
+physical_plan
+01)HashJoinExec: mode=CollectLeft, join_type=Left, on=[(user_id@1, id@0)], projection=[order_id@0, name@3]
+02)--DataSourceExec: partitions=1, partition_sizes=[1]
+03)--DataSourceExec: partitions=1, partition_sizes=[1]
+
+# Negative case: the right side is not unique on the join key, so a left row
+# may match several right rows; the join must stay.
+query TT
+EXPLAIN SELECT elim_users.id FROM elim_users LEFT JOIN elim_orders ON elim_users.id = elim_orders.user_id;
+----
+logical_plan
+01)Projection: elim_users.id
+02)--Left Join: elim_users.id = elim_orders.user_id
+03)----TableScan: elim_users projection=[id]
+04)----TableScan: elim_orders projection=[user_id]
+physical_plan
+01)HashJoinExec: mode=CollectLeft, join_type=Left, on=[(id@0, user_id@0)], projection=[id@0]
+02)--DataSourceExec: partitions=1, partition_sizes=[1]
+03)--DataSourceExec: partitions=1, partition_sizes=[1]
+
+# ... and the duplicates it produces are observable: user 1 has two orders.
+query I rowsort
+SELECT elim_users.id FROM elim_users LEFT JOIN elim_orders ON elim_users.id = elim_orders.user_id;
+----
+1
+1
+2
+4
+
+# The same non-unique right side under a DISTINCT: the join's ancestors are
+# duplicate-insensitive, so the extra matches only affect row multiplicity
+# and the join is removed even without uniqueness on the join key.
+query TT
+EXPLAIN SELECT DISTINCT name FROM elim_users LEFT JOIN elim_orders ON elim_users.id = elim_orders.user_id;
+----
+logical_plan
+01)Aggregate: groupBy=[[elim_users.name]], aggr=[[]]
+02)--TableScan: elim_users projection=[name]
+physical_plan
+01)AggregateExec: mode=FinalPartitioned, gby=[name@0 as name], aggr=[]
+02)--RepartitionExec: partitioning=Hash([name@0], 4), input_partitions=1
+03)----AggregateExec: mode=Partial, gby=[name@0 as name], aggr=[]
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+
+query T rowsort
+SELECT DISTINCT name FROM elim_users LEFT JOIN elim_orders ON elim_users.id = elim_orders.user_id;
+----
+alice
+bob
+dave
+
+# Negative case: count(*) observes row multiplicity and the right side is not
+# unique on the join key, so the join must stay (user 1 has two orders).
+query TT
+EXPLAIN SELECT count(*) FROM elim_users LEFT JOIN elim_orders ON elim_users.id = elim_orders.user_id;
+----
+logical_plan
+01)Projection: count(Int64(1)) AS count(*)
+02)--Aggregate: groupBy=[[]], aggr=[[count(Int64(1))]]
+03)----Projection:
+04)------Left Join: elim_users.id = elim_orders.user_id
+05)--------TableScan: elim_users projection=[id]
+06)--------TableScan: elim_orders projection=[user_id]
+physical_plan
+01)ProjectionExec: expr=[count(Int64(1))@0 as count(*)]
+02)--AggregateExec: mode=Final, gby=[], aggr=[count(Int64(1))]
+03)----CoalescePartitionsExec
+04)------AggregateExec: mode=Partial, gby=[], aggr=[count(Int64(1))]
+05)--------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+06)----------HashJoinExec: mode=CollectLeft, join_type=Left, on=[(id@0, user_id@0)], projection=[]
+07)------------DataSourceExec: partitions=1, partition_sizes=[1]
+08)------------DataSourceExec: partitions=1, partition_sizes=[1]
+
+query I
+SELECT count(*) FROM elim_users LEFT JOIN elim_orders ON elim_users.id = elim_orders.user_id;
+----
+4
+
+# A left join with no equi-join keys at all (ON true) matches every left row
+# with every right row. Under a duplicate-insensitive ancestor (DISTINCT) the
+# multiplication is unobservable and the join is removed.
+query TT
+EXPLAIN SELECT DISTINCT order_id FROM elim_orders LEFT JOIN elim_users ON true;
+----
+logical_plan
+01)Aggregate: groupBy=[[elim_orders.order_id]], aggr=[[]]
+02)--TableScan: elim_orders projection=[order_id]
+physical_plan
+01)AggregateExec: mode=FinalPartitioned, gby=[order_id@0 as order_id], aggr=[]
+02)--RepartitionExec: partitioning=Hash([order_id@0], 4), input_partitions=1
+03)----AggregateExec: mode=Partial, gby=[order_id@0 as order_id], aggr=[]
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+
+query I rowsort
+SELECT DISTINCT order_id FROM elim_orders LEFT JOIN elim_users ON true;
+----
+1
+2
+3
+
+# Negative case: without the DISTINCT the multiplication is observable (each
+# order is repeated once per user), so the join must stay.
+query TT
+EXPLAIN SELECT order_id FROM elim_orders LEFT JOIN elim_users ON true;
+----
+logical_plan
+01)Left Join:
+02)--TableScan: elim_orders projection=[order_id]
+03)--TableScan: elim_users projection=[]
+physical_plan
+01)NestedLoopJoinExec: join_type=Right
+02)--DataSourceExec: partitions=1, partition_sizes=[1]
+03)--DataSourceExec: partitions=1, partition_sizes=[1]
+
+query I rowsort
+SELECT order_id FROM elim_orders LEFT JOIN elim_users ON true;
+----
+1
+1
+1
+2
+2
+2
+3
+3
+3
+
+# A LIMIT makes the row count observable, but the uniqueness path does not
+# depend on duplicate-insensitivity: the unique (PK) right side is unused, so
+# the join is removed even under a LIMIT.
+query TT
+EXPLAIN SELECT order_id FROM elim_orders LEFT JOIN elim_users ON user_id = id LIMIT 2;
+----
+logical_plan
+01)Limit: skip=0, fetch=2
+02)--TableScan: elim_orders projection=[order_id], fetch=2
+physical_plan DataSourceExec: partitions=1, partition_sizes=[1], fetch=2
+
+query I rowsort
+SELECT order_id FROM elim_orders LEFT JOIN elim_users ON user_id = id LIMIT 2;
+----
+1
+2
+
+# Negative case: a LIMIT between the join and a DISTINCT ancestor makes the
+# row count observable, so the DISTINCT's duplicate-insensitivity does not
+# reach the join; with a non-unique right side the join must stay.
+query TT
+EXPLAIN SELECT DISTINCT id FROM (SELECT elim_users.id FROM elim_users LEFT JOIN elim_orders ON elim_users.id = elim_orders.user_id LIMIT 5);
+----
+logical_plan
+01)Aggregate: groupBy=[[elim_users.id]], aggr=[[]]
+02)--Projection: elim_users.id
+03)----Limit: skip=0, fetch=5
+04)------Left Join: elim_users.id = elim_orders.user_id
+05)--------Limit: skip=0, fetch=5
+06)----------TableScan: elim_users projection=[id], fetch=5
+07)--------TableScan: elim_orders projection=[user_id]
+physical_plan
+01)AggregateExec: mode=FinalPartitioned, gby=[id@0 as id], aggr=[]
+02)--RepartitionExec: partitioning=Hash([id@0], 4), input_partitions=4
+03)----AggregateExec: mode=Partial, gby=[id@0 as id], aggr=[]
+04)------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+05)--------HashJoinExec: mode=CollectLeft, join_type=Left, on=[(id@0, user_id@0)], projection=[id@0], fetch=5
+06)----------DataSourceExec: partitions=1, partition_sizes=[1], fetch=5
+07)----------DataSourceExec: partitions=1, partition_sizes=[1]
+
+query I rowsort
+SELECT DISTINCT id FROM (SELECT elim_users.id FROM elim_users LEFT JOIN elim_orders ON elim_users.id = elim_orders.user_id LIMIT 5);
+----
+1
+2
+4
+
+# LEFT JOIN LATERAL decorrelates into a plain left join, with equality
+# predicates extracted as join keys: the unique (PK) lateral side is unused,
+# so the join is removed.
+query TT
+EXPLAIN SELECT order_id FROM elim_orders LEFT JOIN LATERAL (SELECT * FROM elim_users WHERE id = user_id) AS u ON true;
+----
+logical_plan TableScan: elim_orders projection=[order_id]
+physical_plan DataSourceExec: partitions=1, partition_sizes=[1]
+
+query I rowsort
+SELECT order_id FROM elim_orders LEFT JOIN LATERAL (SELECT * FROM elim_users WHERE id = user_id) AS u ON true;
+----
+1
+2
+3
+
+# A non-equality lateral predicate becomes a join filter, which does not
+# block removal under a duplicate-insensitive ancestor (DISTINCT).
+query TT
+EXPLAIN SELECT DISTINCT order_id FROM elim_orders LEFT JOIN LATERAL (SELECT * FROM elim_users WHERE id > user_id) AS u ON true;
+----
+logical_plan
+01)Aggregate: groupBy=[[elim_orders.order_id]], aggr=[[]]
+02)--TableScan: elim_orders projection=[order_id]
+physical_plan
+01)AggregateExec: mode=FinalPartitioned, gby=[order_id@0 as order_id], aggr=[]
+02)--RepartitionExec: partitioning=Hash([order_id@0], 4), input_partitions=1
+03)----AggregateExec: mode=Partial, gby=[order_id@0 as order_id], aggr=[]
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+
+query I rowsort
+SELECT DISTINCT order_id FROM elim_orders LEFT JOIN LATERAL (SELECT * FROM elim_users WHERE id > user_id) AS u ON true;
+----
+1
+2
+3
+
+# Negative case: without the DISTINCT a filter-only lateral can multiply left
+# rows observably (each order matches every user with a greater id), so the
+# join must stay.
+query TT
+EXPLAIN SELECT order_id FROM elim_orders LEFT JOIN LATERAL (SELECT * FROM elim_users WHERE id > user_id) AS u ON true;
+----
+logical_plan
+01)Projection: elim_orders.order_id
+02)--Left Join: Filter: u.id > elim_orders.user_id
+03)----TableScan: elim_orders projection=[order_id, user_id]
+04)----SubqueryAlias: u
+05)------TableScan: elim_users projection=[id]
+physical_plan
+01)NestedLoopJoinExec: join_type=Right, filter=id@1 > user_id@0, projection=[order_id@1]
+02)--DataSourceExec: partitions=1, partition_sizes=[1]
+03)--DataSourceExec: partitions=1, partition_sizes=[1]
+
+query I rowsort
+SELECT order_id FROM elim_orders LEFT JOIN LATERAL (SELECT * FROM elim_users WHERE id > user_id) AS u ON true;
+----
+1
+1
+2
+2
+3
+
+# RIGHT JOIN is symmetric: the join is removed when its *left* side is
+# unreferenced above the join and cannot duplicate right rows.
+
+# The left side is unique on the join key (primary key) and unused above the
+# join: the RIGHT JOIN is removed from the plan.
+query TT
+EXPLAIN SELECT order_id, amount FROM elim_users RIGHT JOIN elim_orders ON id = user_id;
+----
+logical_plan TableScan: elim_orders projection=[order_id, amount]
+physical_plan DataSourceExec: partitions=1, partition_sizes=[1]
+
+# All orders are returned, including the one with no matching user.
+query II rowsort
+SELECT order_id, amount FROM elim_users RIGHT JOIN elim_orders ON id = user_id;
+----
+1 100
+2 200
+3 50
+
+# An extra join filter on left-side columns does not block the rewrite: for a
+# right join it only decides whether a right row is matched or null-padded,
+# and either way the row is emitted.
+query TT
+EXPLAIN SELECT order_id FROM elim_users RIGHT JOIN elim_orders ON id = user_id AND name <> 'bob';
+----
+logical_plan TableScan: elim_orders projection=[order_id]
+physical_plan DataSourceExec: partitions=1, partition_sizes=[1]
+
+query I rowsort
+SELECT order_id FROM elim_users RIGHT JOIN elim_orders ON id = user_id AND name <> 'bob';
+----
+1
+2
+3
+
+# count(*) is duplicate-sensitive, but the unique join key guarantees each
+# order appears exactly once, so the join is still removed.
+query TT
+EXPLAIN SELECT count(*) FROM elim_users RIGHT JOIN elim_orders ON id = user_id;
+----
+logical_plan
+01)Projection: count(Int64(1)) AS count(*)
+02)--Aggregate: groupBy=[[]], aggr=[[count(Int64(1))]]
+03)----TableScan: elim_orders projection=[]
+physical_plan
+01)ProjectionExec: expr=[3 as count(*)]
+02)--PlaceholderRowExec
+
+query I
+SELECT count(*) FROM elim_users RIGHT JOIN elim_orders ON id = user_id;
+----
+3
+
+# The left side is not unique on the join key, but a DISTINCT ancestor makes
+# the extra matches unobservable, so the join is removed even without
+# uniqueness on the join key.
+query TT
+EXPLAIN SELECT DISTINCT name FROM elim_orders RIGHT JOIN elim_users ON elim_orders.user_id = elim_users.id;
+----
+logical_plan
+01)Aggregate: groupBy=[[elim_users.name]], aggr=[[]]
+02)--TableScan: elim_users projection=[name]
+physical_plan
+01)AggregateExec: mode=FinalPartitioned, gby=[name@0 as name], aggr=[]
+02)--RepartitionExec: partitioning=Hash([name@0], 4), input_partitions=1
+03)----AggregateExec: mode=Partial, gby=[name@0 as name], aggr=[]
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+
+query T rowsort
+SELECT DISTINCT name FROM elim_orders RIGHT JOIN elim_users ON elim_orders.user_id = elim_users.id;
+----
+alice
+bob
+dave
+
+# Negative case: the left side is referenced in the SELECT list, so the join
+# must stay.
+query TT
+EXPLAIN SELECT order_id, name FROM elim_users RIGHT JOIN elim_orders ON id = user_id;
+----
+logical_plan
+01)Projection: elim_orders.order_id, elim_users.name
+02)--Right Join: elim_users.id = elim_orders.user_id
+03)----TableScan: elim_users projection=[id, name]
+04)----TableScan: elim_orders projection=[order_id, user_id]
+physical_plan
+01)HashJoinExec: mode=CollectLeft, join_type=Left, on=[(user_id@1, id@0)], projection=[order_id@0, name@3]
+02)--DataSourceExec: partitions=1, partition_sizes=[1]
+03)--DataSourceExec: partitions=1, partition_sizes=[1]
+
+# Negative case: the left side is not unique on the join key, so a right row
+# may match several left rows; the join must stay.
+query TT
+EXPLAIN SELECT elim_users.id FROM elim_orders RIGHT JOIN elim_users ON elim_orders.user_id = elim_users.id;
+----
+logical_plan
+01)Projection: elim_users.id
+02)--Right Join: elim_orders.user_id = elim_users.id
+03)----TableScan: elim_orders projection=[user_id]
+04)----TableScan: elim_users projection=[id]
+physical_plan
+01)HashJoinExec: mode=CollectLeft, join_type=Right, on=[(user_id@0, id@0)], projection=[id@1]
+02)--DataSourceExec: partitions=1, partition_sizes=[1]
+03)--DataSourceExec: partitions=1, partition_sizes=[1]
+
+# ... and the duplicates it produces are observable: user 1 has two orders.
+query I rowsort
+SELECT elim_users.id FROM elim_orders RIGHT JOIN elim_users ON elim_orders.user_id = elim_users.id;
+----
+1
+1
+2
+4
+
+statement ok
+DROP TABLE elim_users;
+
+statement ok
+DROP TABLE elim_orders;
+
+# A UNIQUE constraint, unlike PRIMARY KEY, permits NULLs — and per SQL
+# semantics several NULLs may coexist in a UNIQUE column. Whether a nullable
+# UNIQUE key proves uniqueness on the join keys therefore depends on the
+# join's null semantics.
+statement ok
+CREATE TABLE elim_null_keys (id INT, k INT) AS VALUES
+ (1, 10),
+ (2, NULL),
+ (3, 30);
+
+statement ok
+CREATE TABLE elim_null_lookup (ukey INT UNIQUE, payload INT) AS VALUES
+ (10, 100),
+ (NULL, 200),
+ (NULL, 300);
+
+# Under the default null semantics (`=`), NULL keys match nothing, so the
+# nullable UNIQUE right side still yields at most one match per left row and
+# the join is removed.
+query TT
+EXPLAIN SELECT id FROM elim_null_keys LEFT JOIN elim_null_lookup ON k = ukey;
+----
+logical_plan TableScan: elim_null_keys projection=[id]
+physical_plan DataSourceExec: partitions=1, partition_sizes=[1]
+
+# The NULL-keyed left row matches nothing and is emitted exactly once.
+query I rowsort
+SELECT id FROM elim_null_keys LEFT JOIN elim_null_lookup ON k = ukey;
+----
+1
+2
+3
+
+# Negative case: IS NOT DISTINCT FROM compares NULLs as equal, so both NULL
+# rows in the UNIQUE column match a NULL left key; the right side is not
+# unique under these semantics and the join must stay.
+query TT
+EXPLAIN SELECT id FROM elim_null_keys LEFT JOIN elim_null_lookup ON k IS NOT DISTINCT FROM ukey;
+----
+logical_plan
+01)Projection: elim_null_keys.id
+02)--Left Join: elim_null_keys.k = elim_null_lookup.ukey
+03)----TableScan: elim_null_keys projection=[id, k]
+04)----TableScan: elim_null_lookup projection=[ukey]
+physical_plan
+01)HashJoinExec: mode=CollectLeft, join_type=Right, on=[(ukey@0, k@1)], projection=[id@1], NullsEqual: true
+02)--DataSourceExec: partitions=1, partition_sizes=[1]
+03)--DataSourceExec: partitions=1, partition_sizes=[1]
+
+# ... and the duplicates are observable: the NULL-keyed left row matches both
+# NULL lookup rows.
+query I rowsort
+SELECT id FROM elim_null_keys LEFT JOIN elim_null_lookup ON k IS NOT DISTINCT FROM ukey;
+----
+1
+2
+2
+3
+
+statement ok
+DROP TABLE elim_null_keys;
+
+statement ok
+DROP TABLE elim_null_lookup;
+
# Regression test: a `CollectLeft` `HashJoinExec` requires `SinglePartition` on its build
# (left) child, and the `CoalescePartitionsExec` that satisfies it must survive the
# sort-parallelization phase of `EnsureRequirements`. It used to be removed positionally
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org