fix: do not treat concat as preserving lexicographical ordering

datafusion/expr-common/src/sort_properties.rs

@@ -140,8 +140,7 @@ pub struct ExprProperties {
     /// the expression. Used to compute reliable bounds.
     pub range: Interval,
     /// Indicates whether the expression preserves lexicographical ordering
-    /// of its inputs. For example, string concatenation preserves ordering,
-    /// while addition does not.
+    /// of its inputs.
     pub preserves_lex_ordering: bool,
 }
 

datafusion/expr/src/udf.rs

@@ -979,8 +979,6 @@ pub trait ScalarUDFImpl: Debug + DynEq + DynHash + Send + Sync + Any {
 
     /// Returns true if the function preserves lexicographical ordering based on
     /// the input ordering.
-    ///
-    /// For example, `concat(a || b)` preserves lexicographical ordering, but `abs(a)` does not.
     fn preserves_lex_ordering(&self, _inputs: &[ExprProperties]) -> Result<bool> {
         Ok(false)
     }

datafusion/functions/src/string/concat.rs

@@ -30,7 +30,6 @@ use datafusion_common::{
 };
 use datafusion_expr::expr::ScalarFunction;
 use datafusion_expr::simplify::{ExprSimplifyResult, SimplifyContext};
-use datafusion_expr::sort_properties::ExprProperties;
 use datafusion_expr::{ColumnarValue, Documentation, Expr, Volatility, lit};
 use datafusion_expr::{ScalarFunctionArgs, ScalarUDFImpl, Signature};
 use datafusion_macros::user_doc;
@@ -253,10 +252,6 @@ impl ScalarUDFImpl for ConcatFunc {
     fn documentation(&self) -> Option<&Documentation> {
         self.doc()
     }
-
-    fn preserves_lex_ordering(&self, _inputs: &[ExprProperties]) -> Result<bool> {
-        Ok(true)
-    }
 }
 
 pub(crate) fn deduce_return_type(arg_types: &[DataType]) -> DataType {

datafusion/physical-expr/src/equivalence/properties/dependency.rs

@@ -1011,7 +1011,7 @@ mod tests {
     }
 
     #[test]
-    fn test_ordering_equivalence_with_lex_monotonic_concat() -> Result<()> {
+    fn test_ordering_equivalence_with_non_lex_monotonic_concat() -> Result<()> {
         let schema = Arc::new(Schema::new(vec![
             Field::new("a", DataType::Utf8, false),
             Field::new("b", DataType::Utf8, false),
@@ -1033,28 +1033,23 @@ mod tests {
         // Assume existing ordering is [c ASC, a ASC, b ASC]
         let mut eq_properties = EquivalenceProperties::new(Arc::clone(&schema));
 
-        eq_properties.add_ordering([
+        let initial_ordering: LexOrdering = [
             PhysicalSortExpr::new_default(Arc::clone(&col_c)).asc(),
             PhysicalSortExpr::new_default(Arc::clone(&col_a)).asc(),
             PhysicalSortExpr::new_default(Arc::clone(&col_b)).asc(),
-        ]);
+        ]
+        .into();
+
+        eq_properties.add_ordering(initial_ordering.clone());
 
         // Add equality condition c = concat(a, b)
         eq_properties.add_equal_conditions(Arc::clone(&col_c), a_concat_b)?;
 
         let orderings = eq_properties.oeq_class();
 
-        let expected_ordering1 = [PhysicalSortExpr::new_default(col_c).asc()].into();
-        let expected_ordering2 = [
-            PhysicalSortExpr::new_default(col_a).asc(),
-            PhysicalSortExpr::new_default(col_b).asc(),
-        ]
-        .into();
-
-        // The ordering should be [c ASC] and [a ASC, b ASC]
-        assert_eq!(orderings.len(), 2);
-        assert!(orderings.contains(&expected_ordering1));
-        assert!(orderings.contains(&expected_ordering2));
+        // The ordering should remain unchanged since concat is not lex-monotonic
+        assert_eq!(orderings.len(), 1);
+        assert!(orderings.contains(&initial_ordering));
 
         Ok(())
     }
@@ -1101,55 +1096,6 @@ mod tests {
         Ok(())
     }
 
-    #[test]
-    fn test_ordering_equivalence_with_concat_equality() -> Result<()> {
-        let schema = Arc::new(Schema::new(vec![
-            Field::new("a", DataType::Utf8, false),
-            Field::new("b", DataType::Utf8, false),
-            Field::new("c", DataType::Utf8, false),
-        ]));
-
-        let col_a = col("a", &schema)?;
-        let col_b = col("b", &schema)?;
-        let col_c = col("c", &schema)?;
-
-        let a_concat_b = Arc::new(ScalarFunctionExpr::new(
-            "concat",
-            concat(),
-            vec![Arc::clone(&col_a), Arc::clone(&col_b)],
-            Field::new("f", DataType::Utf8, true).into(),
-            Arc::new(ConfigOptions::default()),
-        )) as _;
-
-        // Assume existing ordering is [concat(a, b) ASC, a ASC, b ASC]
-        let mut eq_properties = EquivalenceProperties::new(Arc::clone(&schema));
-
-        eq_properties.add_ordering([
-            PhysicalSortExpr::new_default(Arc::clone(&a_concat_b)).asc(),
-            PhysicalSortExpr::new_default(Arc::clone(&col_a)).asc(),
-            PhysicalSortExpr::new_default(Arc::clone(&col_b)).asc(),
-        ]);
-
-        // Add equality condition c = concat(a, b)
-        eq_properties.add_equal_conditions(col_c, Arc::clone(&a_concat_b))?;
-
-        let orderings = eq_properties.oeq_class();
-
-        let expected_ordering1 = [PhysicalSortExpr::new_default(a_concat_b).asc()].into();
-        let expected_ordering2 = [
-            PhysicalSortExpr::new_default(col_a).asc(),
-            PhysicalSortExpr::new_default(col_b).asc(),
-        ]
-        .into();
-
-        // The ordering should be [c ASC] and [a ASC, b ASC]
-        assert_eq!(orderings.len(), 2);
-        assert!(orderings.contains(&expected_ordering1));
-        assert!(orderings.contains(&expected_ordering2));
-
-        Ok(())
-    }
-
     #[test]
     fn test_requirements_compatible() -> Result<()> {
         let schema = Arc::new(Schema::new(vec![

datafusion/sqllogictest/test_files/monotonic_projection_test.slt

@@ -168,3 +168,87 @@ physical_plan
 03)----ProjectionExec: expr=[CAST(a@0 + b@1 AS Int64) as sum_expr, a@0 as a, b@1 as b]
 04)------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1, maintains_sort_order=true
 05)--------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a, b], output_ordering=[a@0 ASC NULLS LAST, b@1 ASC NULLS LAST], file_type=csv, has_header=true
+
+# concat(a, b) is not lexicographically ordered just because a is ordered:
+# "a" < "a0", but "a1" > "a01". The projected result still needs a sort.
+query I
+COPY (
+  SELECT * FROM (VALUES ('a', '1'), ('a0', '1')) AS t(a, b) ORDER BY a
+) TO 'test_files/scratch/monotonic_projection_test/concat_ordered.parquet';
+----
+2
+
+statement ok
+CREATE EXTERNAL TABLE concat_ordered (a VARCHAR, b VARCHAR)
+STORED AS PARQUET
+WITH ORDER (a)
+WITH ORDER (b)
+LOCATION 'test_files/scratch/monotonic_projection_test/concat_ordered.parquet';
+
+query TT
+EXPLAIN
+SELECT concat(a, b) AS c
+FROM concat_ordered
+ORDER BY c;
+----
+logical_plan
+01)Sort: c ASC NULLS LAST
+02)--Projection: concat(concat_ordered.a, concat_ordered.b) AS c
+03)----TableScan: concat_ordered projection=[a, b]
+physical_plan
+01)SortExec: expr=[c@0 ASC NULLS LAST], preserve_partitioning=[false]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/monotonic_projection_test/concat_ordered.parquet]]}, projection=[concat(a@0, b@1) as c], file_type=parquet
+
+query T
+SELECT concat(a, b) AS c
+FROM concat_ordered
+ORDER BY c;
+----
+a01
+a1
+
+# An ordering on (c, a, b) does not imply an ordering on (a, b), even when
+# FilterExec establishes c = concat(a, b). EnsureRequirements must retain the
+# sort required by ORDER BY a, b.
+query I
+COPY (
+  SELECT concat(a, b) AS c, a, b
+  FROM (VALUES ('a0', '1'), ('a', '1')) AS t(a, b)
+  ORDER BY c, a, b
+) TO 'test_files/scratch/monotonic_projection_test/concat_equality_ordered.parquet';
+----
+2
+
+statement ok
+CREATE EXTERNAL TABLE concat_equality_ordered (c VARCHAR, a VARCHAR, b VARCHAR)
+STORED AS PARQUET
+WITH ORDER (c, a, b)
+LOCATION 'test_files/scratch/monotonic_projection_test/concat_equality_ordered.parquet';
+
+query TT
+EXPLAIN
+SELECT a, b
+FROM concat_equality_ordered
+WHERE c = concat(a, b)
+ORDER BY a, b;
+----
+logical_plan
+01)Sort: concat_equality_ordered.a ASC NULLS LAST, concat_equality_ordered.b ASC NULLS LAST
+02)--Projection: concat_equality_ordered.a, concat_equality_ordered.b
+03)----Filter: concat_equality_ordered.c = concat(concat_equality_ordered.a, concat_equality_ordered.b)
+04)------TableScan: concat_equality_ordered projection=[c, a, b], partial_filters=[concat_equality_ordered.c = concat(concat_equality_ordered.a, concat_equality_ordered.b)]
+physical_plan
+01)SortPreservingMergeExec: [a@0 ASC NULLS LAST, b@1 ASC NULLS LAST]
+02)--SortExec: expr=[a@0 ASC NULLS LAST, b@1 ASC NULLS LAST], preserve_partitioning=[true]
+03)----FilterExec: c@0 = concat(a@1, b@2), projection=[a@1, b@2]
+04)------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1, maintains_sort_order=true
+05)--------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/monotonic_projection_test/concat_equality_ordered.parquet]]}, projection=[c, a, b], output_ordering=[c@0 ASC NULLS LAST, a@1 ASC NULLS LAST, b@2 ASC NULLS LAST], file_type=parquet, predicate=c@0 = concat(a@1, b@2)
+
+query TT
+SELECT a, b
+FROM concat_equality_ordered
+WHERE c = concat(a, b)
+ORDER BY a, b;
+----
+a 1
+a0 1