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:56:34 UTC
(datafusion) branch gh-readonly-queue/main/pr-26459-c76ed5396ad3f4a4dc9d9ceeddc419a146b871f9 created (now bd219dc192)
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-26459-c76ed5396ad3f4a4dc9d9ceeddc419a146b871f9
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at bd219dc192 fix: keep probe-side constants in join equivalence properties (#26459)
This branch includes the following new commits:
new bd219dc192 fix: keep probe-side constants in join equivalence properties (#26459)
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: fix: keep probe-side constants in join equivalence properties (#26459)
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-26459-c76ed5396ad3f4a4dc9d9ceeddc419a146b871f9
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit bd219dc192b94edde9acd9f27323aa7d484a6add
Author: Jay Zhan <ja...@gmail.com>
AuthorDate: Fri Oct 23 14:56:12 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/physical-plan/src/joins/cross_join.rs | 6 +-
datafusion/sqllogictest/test_files/order.slt | 133 ++++++++++++++++++++++
3 files changed, 178 insertions(+), 4 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/physical-plan/src/joins/cross_join.rs b/datafusion/physical-plan/src/joins/cross_join.rs
index df8a3a8aad..03e5c740a9 100644
--- a/datafusion/physical-plan/src/joins/cross_join.rs
+++ b/datafusion/physical-plan/src/joins/cross_join.rs
@@ -149,13 +149,15 @@ impl CrossJoinExec {
right: &Arc<dyn ExecutionPlan>,
schema: SchemaRef,
) -> Result<PlanProperties> {
- // Calculate equivalence properties
+ // Calculate equivalence properties. A cross join is an inner join
+ // without keys: no side is padded with NULLs, so the constants of
+ // both sides still hold.
// TODO: Check equivalence properties of cross join, it may preserve
// ordering in some cases.
let eq_properties = join_equivalence_properties(
left.equivalence_properties().clone(),
right.equivalence_properties().clone(),
- &JoinType::Full,
+ &JoinType::Inner,
schema,
&[false, false],
None,
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