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/10/23 14:04:15 UTC
(datafusion) 01/01: fix: keep probe-side constants in join equivalence properties (#26459)
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-26459-c76ed5396ad3f4a4dc9d9ceeddc419a146b871f9
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 4518a33072e6750ed2e8a9780e2d331a363984d9
Author: Jay Zhan <ja...@gmail.com>
AuthorDate: Fri Oct 23 14:03:50 2026 +0000
fix: keep probe-side constants in join equivalence properties (#26459)
## Which issue does this PR close?
- Closes #26283.
## Rationale for this change
A query that orders by a column that is constant on the probe side of a
join fails to plan:
```text
SanityCheckPlan
caused by
Error during planning: Plan: ["SortPreservingMergeExec: [k@0 DESC]", " HashJoinExec: ..."] does not satisfy order requirements: [k@0 DESC]. Child-0 order: []
```
`EquivalenceClass::try_with_offset` builds a new class from the
expressions and does not copy `constant`. `EquivalenceGroup::join` uses
it for every class of the right side, so the join output forgets that
`k` is constant. `EnsureRequirements` removes the sort because `k` is
constant below the join, and `SanityCheckPlan` then rejects the plan
because the join output no longer says so.
## What changes are included in this PR?
- `try_with_offset` keeps the class's constant value.
- `EquivalenceGroup::join` clears the right side's constants for `Left`
and `Full` joins. Unmatched rows pad that side with NULLs, so a constant
there no longer holds. Without this, copying the constant would drop a
sort that is needed.
The first version of this PR added a sort in `EnsureRequirements`
wherever the requirement could not be proven. Following review, it now
fixes the join instead. The plan for #26283 then has no sort at all,
instead of a sort on a constant column.
The same problem exists the other way round: the left side's constants
are kept for `Right` and `Full` joins, where that side is padded with
NULLs too. For example, this returns `5` before the NULLs:
```sql
CREATE TABLE t(a INT) AS VALUES (1), (2), (3);
CREATE TABLE u(a INT) AS VALUES (9), (10), (1);
SELECT s.k, u.a FROM (SELECT a, 5 AS k FROM t) s RIGHT JOIN u ON s.a = u.a
ORDER BY s.k DESC NULLS FIRST;
```
This PR does not change the left side, because #26042 already rebuilds
the classes of the side that is padded with NULLs, which drops these
constants as well.
## What is the testing strategy for this PR?
- `class.rs`: a unit test that a right-side constant stays constant
after `Inner` and `Right` joins, and not after `Left` and `Full` joins.
- `order.slt`: both queries from #26283, with results and `EXPLAIN`.
Without the change, both fail to plan.
- `order.slt`: a `LEFT JOIN` where the probe-side constant is padded
with NULLs. If the constant is not cleared, the sort on it is dropped
and the rows come back in the wrong order.
## Are there any user-facing changes?
Queries like the one in #26283 plan and run instead of failing. There
are no API changes.
---
datafusion/physical-expr/src/equivalence/class.rs | 43 ++++++-
datafusion/sqllogictest/test_files/order.slt | 133 ++++++++++++++++++++++
2 files changed, 174 insertions(+), 2 deletions(-)
diff --git a/datafusion/physical-expr/src/equivalence/class.rs b/datafusion/physical-expr/src/equivalence/class.rs
index b68d20fccf..ff34190bb1 100644
--- a/datafusion/physical-expr/src/equivalence/class.rs
+++ b/datafusion/physical-expr/src/equivalence/class.rs
@@ -245,7 +245,8 @@ impl EquivalenceClass {
}
/// Adds the given offset to all columns in the expressions inside this
- /// class. This is used when schemas are appended, e.g. in joins.
+ /// class, keeping its constant value. This is used when schemas are
+ /// appended, e.g. in joins.
pub fn try_with_offset(&self, offset: isize) -> Result<Self> {
let mut cls = Self::default();
for expr_result in self
@@ -256,6 +257,7 @@ impl EquivalenceClass {
{
cls.push(expr_result?);
}
+ cls.constant.clone_from(&self.constant);
Ok(cls)
}
}
@@ -823,7 +825,16 @@ impl EquivalenceGroup {
self.iter().cloned().chain(
right_equivalences
.iter()
- .map(|cls| cls.try_with_offset(left_size as _))
+ .map(|cls| {
+ let mut cls = cls.try_with_offset(left_size as _)?;
+ // Rows without a match on the right side are
+ // padded with NULLs, so its constants no
+ // longer hold.
+ if matches!(join_type, JoinType::Left | JoinType::Full) {
+ cls.constant = None;
+ }
+ Ok(cls)
+ })
.collect::<Result<Vec<_>>>()?,
),
);
@@ -1351,6 +1362,34 @@ mod tests {
Ok(())
}
+ /// A constant on the right side of a join stays constant in the join
+ /// output, unless unmatched rows pad the right side with NULLs.
+ /// <https://github.com/apache/datafusion/issues/26283>
+ #[test]
+ fn test_join_keeps_right_constants() -> Result<()> {
+ let schema = abc_schema();
+ let mut right = EquivalenceGroup::default();
+ right.add_constant(ConstExpr::from(col("a", &schema)?));
+ // `a` of the right side is column 4 of the join output
+ let right_a: Arc<dyn PhysicalExpr> = Arc::new(Column::new("a", 4));
+
+ for (join_type, expected) in [
+ (JoinType::Inner, true),
+ (JoinType::Right, true),
+ (JoinType::Left, false),
+ (JoinType::Full, false),
+ ] {
+ let joined = EquivalenceGroup::default().join(&right, &join_type, 4, &[])?;
+ assert_eq!(
+ joined.is_expr_constant(&right_a).is_some(),
+ expected,
+ "{join_type}"
+ );
+ }
+
+ Ok(())
+ }
+
#[test]
fn test_has_same_classes_ignores_the_map() -> Result<()> {
// `map` indexes into `classes`, so equal classes must imply equal
diff --git a/datafusion/sqllogictest/test_files/order.slt b/datafusion/sqllogictest/test_files/order.slt
index 0d48421a83..3514c30e3c 100644
--- a/datafusion/sqllogictest/test_files/order.slt
+++ b/datafusion/sqllogictest/test_files/order.slt
@@ -2333,3 +2333,136 @@ DROP TABLE spm_orderings;
statement ok
set datafusion.execution.target_partitions = 4;
+
+# `k` is constant on the probe side of the join, so the join output reports
+# it as constant and no sort on `k` is needed.
+# https://github.com/apache/datafusion/issues/26283
+statement ok
+CREATE TABLE probe_constant_t(a INT, b INT) AS VALUES (1, 1), (2, 2), (3, 3), (4, 4);
+
+statement ok
+CREATE TABLE probe_constant_u(a INT) AS VALUES (1);
+
+statement ok
+CREATE TABLE probe_constant_v(x INT) AS VALUES (1), (2);
+
+query II
+SELECT s.k, s.a
+FROM (
+ SELECT probe_constant_t.a, -2 AS k
+ FROM probe_constant_t JOIN probe_constant_v ON probe_constant_t.b <> probe_constant_v.x
+) s
+JOIN probe_constant_u ON s.a = probe_constant_u.a
+ORDER BY s.k DESC;
+----
+-2 1
+
+query TT
+EXPLAIN SELECT s.k, s.a
+FROM (
+ SELECT probe_constant_t.a, -2 AS k
+ FROM probe_constant_t JOIN probe_constant_v ON probe_constant_t.b <> probe_constant_v.x
+) s
+JOIN probe_constant_u ON s.a = probe_constant_u.a
+ORDER BY s.k DESC;
+----
+logical_plan
+01)Sort: s.k DESC NULLS FIRST
+02)--Projection: s.k, s.a
+03)----Inner Join: s.a = probe_constant_u.a
+04)------SubqueryAlias: s
+05)--------Projection: probe_constant_t.a, Int64(-2) AS k
+06)----------Inner Join: Filter: probe_constant_t.b != probe_constant_v.x
+07)------------TableScan: probe_constant_t projection=[a, b]
+08)------------TableScan: probe_constant_v projection=[x]
+09)------TableScan: probe_constant_u projection=[a]
+physical_plan
+01)CoalescePartitionsExec
+02)--HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(a@0, a@0)], projection=[k@2, a@1]
+03)----DataSourceExec: partitions=1, partition_sizes=[1]
+04)----RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+05)------ProjectionExec: expr=[a@0 as a, -2 as k]
+06)--------NestedLoopJoinExec: join_type=Inner, filter=b@0 != x@1, projection=[a@1]
+07)----------DataSourceExec: partitions=1, partition_sizes=[1]
+08)----------DataSourceExec: partitions=1, partition_sizes=[1]
+
+statement ok
+SET datafusion.execution.target_partitions = 1;
+
+query ITI
+SELECT s.a, s.k, row_number() OVER (PARTITION BY s.a ORDER BY s.k) AS r
+FROM probe_constant_u JOIN (SELECT a, 'x' AS k FROM probe_constant_t) AS s
+ ON s.a = probe_constant_u.a;
+----
+1 x 1
+
+query TT
+EXPLAIN SELECT s.a, s.k, row_number() OVER (PARTITION BY s.a ORDER BY s.k) AS r
+FROM probe_constant_u JOIN (SELECT a, 'x' AS k FROM probe_constant_t) AS s
+ ON s.a = probe_constant_u.a;
+----
+logical_plan
+01)Projection: s.a, s.k, row_number() PARTITION BY [s.a] ORDER BY [s.k ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW AS r
+02)--WindowAggr: windowExpr=[[row_number() PARTITION BY [s.a] ORDER BY [s.k ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]]
+03)----Projection: s.a, s.k
+04)------Inner Join: probe_constant_u.a = s.a
+05)--------TableScan: probe_constant_u projection=[a]
+06)--------SubqueryAlias: s
+07)----------Projection: probe_constant_t.a, Utf8("x") AS k
+08)------------TableScan: probe_constant_t projection=[a]
+physical_plan
+01)ProjectionExec: expr=[a@0 as a, k@1 as k, row_number() PARTITION BY [s.a] ORDER BY [s.k ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW@2 as r]
+02)--BoundedWindowAggExec: wdw=[row_number() PARTITION BY [s.a] ORDER BY [s.k ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Field { "row_number() PARTITION BY [s.a] ORDER BY [s.k ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW": UInt64 }, frame: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW], mode=[Sorted]
+03)----HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(a@0, a@0)], projection=[a@1, k@2]
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+05)------SortExec: expr=[a@0 ASC NULLS LAST], preserve_partitioning=[false]
+06)--------FilterExec: DynamicFilter [ empty ]
+07)----------ProjectionExec: expr=[a@0 as a, x as k]
+08)------------DataSourceExec: partitions=1, partition_sizes=[1]
+
+# Unmatched rows of a left join pad the probe side with NULLs, so `k` is not
+# constant in the join output and the sort on `k` is kept.
+query TT
+EXPLAIN SELECT u.a, s.k
+FROM (VALUES (1), (10)) AS u(a) LEFT JOIN (SELECT a, 'x' AS k FROM probe_constant_t) AS s
+ ON s.a = u.a
+ORDER BY s.k DESC NULLS FIRST, u.a;
+----
+logical_plan
+01)Sort: s.k DESC NULLS FIRST, u.a ASC NULLS LAST
+02)--Projection: u.a, s.k
+03)----Left Join: u.a = CAST(s.a AS Int64)
+04)------SubqueryAlias: u
+05)--------Projection: column1 AS a
+06)----------Values: (Int64(1)), (Int64(10))
+07)------SubqueryAlias: s
+08)--------Projection: probe_constant_t.a, Utf8("x") AS k
+09)----------TableScan: probe_constant_t projection=[a]
+physical_plan
+01)SortExec: expr=[k@1 DESC, a@0 ASC NULLS LAST], preserve_partitioning=[false]
+02)--HashJoinExec: mode=CollectLeft, join_type=Left, on=[(a@0, CAST(s.a AS Int64)@2)], projection=[a@0, k@2]
+03)----ProjectionExec: expr=[column1@0 as a]
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+05)----ProjectionExec: expr=[a@0 as a, x as k, CAST(a@0 AS Int64) as CAST(s.a AS Int64)]
+06)------DataSourceExec: partitions=1, partition_sizes=[1]
+
+query IT
+SELECT u.a, s.k
+FROM (VALUES (1), (10)) AS u(a) LEFT JOIN (SELECT a, 'x' AS k FROM probe_constant_t) AS s
+ ON s.a = u.a
+ORDER BY s.k DESC NULLS FIRST, u.a;
+----
+10 NULL
+1 x
+
+statement ok
+SET datafusion.execution.target_partitions = 4;
+
+statement ok
+DROP TABLE probe_constant_t;
+
+statement ok
+DROP TABLE probe_constant_u;
+
+statement ok
+DROP TABLE probe_constant_v;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org