diff --git a/datafusion/physical-expr/src/equivalence/class.rs b/datafusion/physical-expr/src/equivalence/class.rs index b68d20fccf76..ff34190bb1c1 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 { 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::>>()?, ), ); @@ -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. + /// + #[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 = 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 df8a3a8aad42..03e5c740a92e 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, schema: SchemaRef, ) -> Result { - // 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 0d48421a833b..3514c30e3cce 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;