From cd34d9be6553cb4790cb4a1883eebb5aa808f2c9 Mon Sep 17 00:00:00 2001 From: buraksenn Date: Wed, 22 Jul 2026 17:07:47 +0300 Subject: [PATCH 1/2] dont preserve lex order in concat --- .../physical_optimizer/enforce_sorting.rs | 134 ++++++++++++++++++ datafusion/expr-common/src/sort_properties.rs | 3 +- datafusion/expr/src/udf.rs | 2 - datafusion/functions/src/string/concat.rs | 5 - .../src/equivalence/properties/dependency.rs | 89 ++++++++---- 5 files changed, 196 insertions(+), 37 deletions(-) diff --git a/datafusion/core/tests/physical_optimizer/enforce_sorting.rs b/datafusion/core/tests/physical_optimizer/enforce_sorting.rs index e9ad978b2e0cb..3102c9a64b756 100644 --- a/datafusion/core/tests/physical_optimizer/enforce_sorting.rs +++ b/datafusion/core/tests/physical_optimizer/enforce_sorting.rs @@ -3255,3 +3255,137 @@ async fn test_does_not_push_fetch_sort_through_projection_over_union() -> Result Ok(()) } + +#[tokio::test] +async fn test_concat_lex_sort_in_project_with_optimization() -> Result<()> { + test_concat_lex_ordering_with_project(true).await +} + +#[tokio::test] +async fn test_concat_lex_sort_in_project_without_optimization() -> Result<()> { + test_concat_lex_ordering_with_project(false).await +} + +/// The input plan is: +/// ```text +/// SortExec: expr=[c@0 ASC], preserve_partitioning=[false] +/// ProjectionExec: expr=[concat(a@0, b@1) as c] +/// DataSourceExec: partitions=1, partition_sizes=[1], output_ordering=a@0 ASC +/// ``` +/// +/// The output must be sorted whether or not sort enforcement is optimized. +async fn test_concat_lex_ordering_with_project(optimize: bool) -> Result<()> { + // Sorted by a ASC ("a" < "a0") and by b ASC ("1" == "1"), but + // concat(a, b) is NOT sorted: "a1" > "a01". + let batch = record_batch!(("a", Utf8, ["a", "a0"]), ("b", Utf8, ["1", "1"]))?; + let schema = batch.schema(); + let source = Arc::new(DataSourceExec::new(Arc::new( + datafusion::datasource::memory::MemorySourceConfig::try_new( + &[vec![batch]], + schema.clone(), + None, + )? + .try_with_sort_information(vec![ + LexOrdering::new([sort_expr("a", &schema)]).unwrap(), + LexOrdering::new([sort_expr("b", &schema)]).unwrap(), + ])?, + ))) as Arc; + + let concat = Arc::new(datafusion_physical_expr::ScalarFunctionExpr::try_new( + datafusion::functions::string::concat(), + vec![col("a", &schema)?, col("b", &schema)?], + &schema, + Arc::new(ConfigOptions::default()), + )?) as Arc; + let projection = projection_exec(vec![(concat, "c".to_string())], source)?; + let plan = sort_exec( + LexOrdering::new([sort_expr("c", &projection.schema())]).unwrap(), + projection, + ); + + assert_plan_output_eq(plan, optimize, vec!["a01", "a1"]).await +} + +#[tokio::test] +async fn test_concat_lex_sort_in_filter_with_optimization() -> Result<()> { + test_concat_lex_ordering_with_filter(true).await +} + +#[tokio::test] +async fn test_concat_lex_sort_in_filter_without_optimization() -> Result<()> { + test_concat_lex_ordering_with_filter(false).await +} + +/// The input plan is: +/// ```text +/// SortExec: expr=[a@0 ASC, b@1 ASC], preserve_partitioning=[false] +/// FilterExec: c@2 = concat(a@0, b@1) +/// SortExec: expr=[c@2 ASC, a@0 ASC, b@1 ASC], preserve_partitioning=[false] +/// DataSourceExec: partitions=1, partition_sizes=[1] +/// ``` +/// +/// The output must be sorted whether or not sort enforcement is optimized. +async fn test_concat_lex_ordering_with_filter(optimize: bool) -> Result<()> { + // Rows are sorted by (c, a, b) and satisfy c = concat(a, b), but are + // NOT sorted by (a, b): ("a0", "1") > ("a", "1"). + let batch = record_batch!( + ("a", Utf8, ["a0", "a"]), + ("b", Utf8, ["1", "1"]), + ("c", Utf8, ["a01", "a1"]) + )?; + let schema = batch.schema(); + let source = datafusion::datasource::memory::MemorySourceConfig::try_new_exec( + &[vec![batch]], + schema.clone(), + None, + )?; + let sort_cab = sort_exec( + LexOrdering::new([ + sort_expr("c", &schema), + sort_expr("a", &schema), + sort_expr("b", &schema), + ]) + .unwrap(), + source, + ); + + let concat = Arc::new(datafusion_physical_expr::ScalarFunctionExpr::try_new( + datafusion::functions::string::concat(), + vec![col("a", &schema)?, col("b", &schema)?], + &schema, + Arc::new(ConfigOptions::default()), + )?) as Arc; + let predicate = + Arc::new(BinaryExpr::new(col("c", &schema)?, Operator::Eq, concat)) as _; + let filter = filter_exec(predicate, sort_cab); + let plan = sort_exec( + LexOrdering::new([sort_expr("a", &schema), sort_expr("b", &schema)]).unwrap(), + filter, + ); + + assert_plan_output_eq(plan, optimize, vec!["a", "a0"]).await +} + +async fn assert_plan_output_eq( + plan: Arc, + optimize: bool, + expected_result: Vec<&str>, +) -> Result<()> { + let plan = if optimize { + EnsureRequirements::new().optimize(plan, &ConfigOptions::default())? + } else { + plan + }; + let plan_str = displayable(plan.as_ref()).indent(true).to_string(); + + let results = + datafusion_physical_plan::collect(plan, Arc::new(TaskContext::default())).await?; + let all = arrow::compute::concat_batches(&results[0].schema(), &results)?; + let expected = create_array!(Utf8, expected_result) as ArrayRef; + assert_eq!( + all.column(0), + &expected, + "output must satisfy the requested ordering (optimize={optimize}); plan:\n{plan_str}" + ); + Ok(()) +} diff --git a/datafusion/expr-common/src/sort_properties.rs b/datafusion/expr-common/src/sort_properties.rs index 5d17a34a96fbc..04da574882d30 100644 --- a/datafusion/expr-common/src/sort_properties.rs +++ b/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, } diff --git a/datafusion/expr/src/udf.rs b/datafusion/expr/src/udf.rs index e206ce8b29108..4c51ff46f7365 100644 --- a/datafusion/expr/src/udf.rs +++ b/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 { Ok(false) } diff --git a/datafusion/functions/src/string/concat.rs b/datafusion/functions/src/string/concat.rs index af51f66faa97c..1c1f6d640798a 100644 --- a/datafusion/functions/src/string/concat.rs +++ b/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 { - Ok(true) - } } pub(crate) fn deduce_return_type(arg_types: &[DataType]) -> DataType { diff --git a/datafusion/physical-expr/src/equivalence/properties/dependency.rs b/datafusion/physical-expr/src/equivalence/properties/dependency.rs index 2ebc71559fcf4..171dd6817298e 100644 --- a/datafusion/physical-expr/src/equivalence/properties/dependency.rs +++ b/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(()) } @@ -1102,7 +1097,7 @@ mod tests { } #[test] - fn test_ordering_equivalence_with_concat_equality() -> Result<()> { + fn test_ordering_equivalence_with_non_lex_monotonic_concat_equality() -> Result<()> { let schema = Arc::new(Schema::new(vec![ Field::new("a", DataType::Utf8, false), Field::new("b", DataType::Utf8, false), @@ -1124,28 +1119,23 @@ mod tests { // 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([ + let initial_ordering: LexOrdering = [ 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(), - ]); + ] + .into(); + + eq_properties.add_ordering(initial_ordering.clone()); // 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)); + // The ordering should remain unchanged since concat is not lex-monotonic + assert_eq!(orderings.len(), 1); + assert!(orderings.contains(&initial_ordering)); Ok(()) } @@ -1617,4 +1607,47 @@ mod tests { Ok(()) } + + #[test] + fn test_concat_projection_does_not_preserve_lex_ordering() -> Result<()> { + let input_schema = Arc::new(Schema::new(vec![ + Field::new("a", DataType::Utf8, false), + Field::new("b", DataType::Utf8, false), + ])); + let col_a = col("a", &input_schema)?; + let col_b = col("b", &input_schema)?; + + let mut input_properties = EquivalenceProperties::new(Arc::clone(&input_schema)); + input_properties.add_ordering([ + PhysicalSortExpr::new_default(Arc::clone(&col_a)), + PhysicalSortExpr::new_default(Arc::clone(&col_b)), + ]); + + let a_concat_b = Arc::new(ScalarFunctionExpr::new( + "concat", + concat(), + vec![col_a, Arc::clone(&col_b)], + Field::new("ab", DataType::Utf8, true).into(), + Arc::new(ConfigOptions::default()), + )) as Arc; + let projection_mapping = ProjectionMapping::try_new( + vec![(a_concat_b, "ab".to_string()), (col_b, "b".to_string())], + &input_schema, + )?; + let output_schema = output_schema(&projection_mapping, &input_schema)?; + let projected = + input_properties.project(&projection_mapping, Arc::clone(&output_schema)); + let unsafe_projected_ordering: LexOrdering = [ + PhysicalSortExpr::new_default(col("ab", &output_schema)?), + PhysicalSortExpr::new_default(col("b", &output_schema)?), + ] + .into(); + + assert!( + !projected.ordering_satisfy(unsafe_projected_ordering)?, + "concat(a, b) must not replace a in the projected [a, b] ordering" + ); + + Ok(()) + } } From 6cf2d537bd1102442139ec04ac13cf7b88216bba Mon Sep 17 00:00:00 2001 From: buraksenn Date: Wed, 22 Jul 2026 18:24:38 +0000 Subject: [PATCH 2/2] delete existing tests and add slt ones --- .../physical_optimizer/enforce_sorting.rs | 134 ------------------ .../src/equivalence/properties/dependency.rs | 87 ------------ .../test_files/monotonic_projection_test.slt | 84 +++++++++++ 3 files changed, 84 insertions(+), 221 deletions(-) diff --git a/datafusion/core/tests/physical_optimizer/enforce_sorting.rs b/datafusion/core/tests/physical_optimizer/enforce_sorting.rs index 3102c9a64b756..e9ad978b2e0cb 100644 --- a/datafusion/core/tests/physical_optimizer/enforce_sorting.rs +++ b/datafusion/core/tests/physical_optimizer/enforce_sorting.rs @@ -3255,137 +3255,3 @@ async fn test_does_not_push_fetch_sort_through_projection_over_union() -> Result Ok(()) } - -#[tokio::test] -async fn test_concat_lex_sort_in_project_with_optimization() -> Result<()> { - test_concat_lex_ordering_with_project(true).await -} - -#[tokio::test] -async fn test_concat_lex_sort_in_project_without_optimization() -> Result<()> { - test_concat_lex_ordering_with_project(false).await -} - -/// The input plan is: -/// ```text -/// SortExec: expr=[c@0 ASC], preserve_partitioning=[false] -/// ProjectionExec: expr=[concat(a@0, b@1) as c] -/// DataSourceExec: partitions=1, partition_sizes=[1], output_ordering=a@0 ASC -/// ``` -/// -/// The output must be sorted whether or not sort enforcement is optimized. -async fn test_concat_lex_ordering_with_project(optimize: bool) -> Result<()> { - // Sorted by a ASC ("a" < "a0") and by b ASC ("1" == "1"), but - // concat(a, b) is NOT sorted: "a1" > "a01". - let batch = record_batch!(("a", Utf8, ["a", "a0"]), ("b", Utf8, ["1", "1"]))?; - let schema = batch.schema(); - let source = Arc::new(DataSourceExec::new(Arc::new( - datafusion::datasource::memory::MemorySourceConfig::try_new( - &[vec![batch]], - schema.clone(), - None, - )? - .try_with_sort_information(vec![ - LexOrdering::new([sort_expr("a", &schema)]).unwrap(), - LexOrdering::new([sort_expr("b", &schema)]).unwrap(), - ])?, - ))) as Arc; - - let concat = Arc::new(datafusion_physical_expr::ScalarFunctionExpr::try_new( - datafusion::functions::string::concat(), - vec![col("a", &schema)?, col("b", &schema)?], - &schema, - Arc::new(ConfigOptions::default()), - )?) as Arc; - let projection = projection_exec(vec![(concat, "c".to_string())], source)?; - let plan = sort_exec( - LexOrdering::new([sort_expr("c", &projection.schema())]).unwrap(), - projection, - ); - - assert_plan_output_eq(plan, optimize, vec!["a01", "a1"]).await -} - -#[tokio::test] -async fn test_concat_lex_sort_in_filter_with_optimization() -> Result<()> { - test_concat_lex_ordering_with_filter(true).await -} - -#[tokio::test] -async fn test_concat_lex_sort_in_filter_without_optimization() -> Result<()> { - test_concat_lex_ordering_with_filter(false).await -} - -/// The input plan is: -/// ```text -/// SortExec: expr=[a@0 ASC, b@1 ASC], preserve_partitioning=[false] -/// FilterExec: c@2 = concat(a@0, b@1) -/// SortExec: expr=[c@2 ASC, a@0 ASC, b@1 ASC], preserve_partitioning=[false] -/// DataSourceExec: partitions=1, partition_sizes=[1] -/// ``` -/// -/// The output must be sorted whether or not sort enforcement is optimized. -async fn test_concat_lex_ordering_with_filter(optimize: bool) -> Result<()> { - // Rows are sorted by (c, a, b) and satisfy c = concat(a, b), but are - // NOT sorted by (a, b): ("a0", "1") > ("a", "1"). - let batch = record_batch!( - ("a", Utf8, ["a0", "a"]), - ("b", Utf8, ["1", "1"]), - ("c", Utf8, ["a01", "a1"]) - )?; - let schema = batch.schema(); - let source = datafusion::datasource::memory::MemorySourceConfig::try_new_exec( - &[vec![batch]], - schema.clone(), - None, - )?; - let sort_cab = sort_exec( - LexOrdering::new([ - sort_expr("c", &schema), - sort_expr("a", &schema), - sort_expr("b", &schema), - ]) - .unwrap(), - source, - ); - - let concat = Arc::new(datafusion_physical_expr::ScalarFunctionExpr::try_new( - datafusion::functions::string::concat(), - vec![col("a", &schema)?, col("b", &schema)?], - &schema, - Arc::new(ConfigOptions::default()), - )?) as Arc; - let predicate = - Arc::new(BinaryExpr::new(col("c", &schema)?, Operator::Eq, concat)) as _; - let filter = filter_exec(predicate, sort_cab); - let plan = sort_exec( - LexOrdering::new([sort_expr("a", &schema), sort_expr("b", &schema)]).unwrap(), - filter, - ); - - assert_plan_output_eq(plan, optimize, vec!["a", "a0"]).await -} - -async fn assert_plan_output_eq( - plan: Arc, - optimize: bool, - expected_result: Vec<&str>, -) -> Result<()> { - let plan = if optimize { - EnsureRequirements::new().optimize(plan, &ConfigOptions::default())? - } else { - plan - }; - let plan_str = displayable(plan.as_ref()).indent(true).to_string(); - - let results = - datafusion_physical_plan::collect(plan, Arc::new(TaskContext::default())).await?; - let all = arrow::compute::concat_batches(&results[0].schema(), &results)?; - let expected = create_array!(Utf8, expected_result) as ArrayRef; - assert_eq!( - all.column(0), - &expected, - "output must satisfy the requested ordering (optimize={optimize}); plan:\n{plan_str}" - ); - Ok(()) -} diff --git a/datafusion/physical-expr/src/equivalence/properties/dependency.rs b/datafusion/physical-expr/src/equivalence/properties/dependency.rs index 171dd6817298e..d2a8c2f654cf0 100644 --- a/datafusion/physical-expr/src/equivalence/properties/dependency.rs +++ b/datafusion/physical-expr/src/equivalence/properties/dependency.rs @@ -1096,50 +1096,6 @@ mod tests { Ok(()) } - #[test] - fn test_ordering_equivalence_with_non_lex_monotonic_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)); - - let initial_ordering: LexOrdering = [ - 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(), - ] - .into(); - - eq_properties.add_ordering(initial_ordering.clone()); - - // 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(); - - // The ordering should remain unchanged since concat is not lex-monotonic - assert_eq!(orderings.len(), 1); - assert!(orderings.contains(&initial_ordering)); - - Ok(()) - } - #[test] fn test_requirements_compatible() -> Result<()> { let schema = Arc::new(Schema::new(vec![ @@ -1607,47 +1563,4 @@ mod tests { Ok(()) } - - #[test] - fn test_concat_projection_does_not_preserve_lex_ordering() -> Result<()> { - let input_schema = Arc::new(Schema::new(vec![ - Field::new("a", DataType::Utf8, false), - Field::new("b", DataType::Utf8, false), - ])); - let col_a = col("a", &input_schema)?; - let col_b = col("b", &input_schema)?; - - let mut input_properties = EquivalenceProperties::new(Arc::clone(&input_schema)); - input_properties.add_ordering([ - PhysicalSortExpr::new_default(Arc::clone(&col_a)), - PhysicalSortExpr::new_default(Arc::clone(&col_b)), - ]); - - let a_concat_b = Arc::new(ScalarFunctionExpr::new( - "concat", - concat(), - vec![col_a, Arc::clone(&col_b)], - Field::new("ab", DataType::Utf8, true).into(), - Arc::new(ConfigOptions::default()), - )) as Arc; - let projection_mapping = ProjectionMapping::try_new( - vec![(a_concat_b, "ab".to_string()), (col_b, "b".to_string())], - &input_schema, - )?; - let output_schema = output_schema(&projection_mapping, &input_schema)?; - let projected = - input_properties.project(&projection_mapping, Arc::clone(&output_schema)); - let unsafe_projected_ordering: LexOrdering = [ - PhysicalSortExpr::new_default(col("ab", &output_schema)?), - PhysicalSortExpr::new_default(col("b", &output_schema)?), - ] - .into(); - - assert!( - !projected.ordering_satisfy(unsafe_projected_ordering)?, - "concat(a, b) must not replace a in the projected [a, b] ordering" - ); - - Ok(()) - } } diff --git a/datafusion/sqllogictest/test_files/monotonic_projection_test.slt b/datafusion/sqllogictest/test_files/monotonic_projection_test.slt index 7feefc169fcab..0045e51715980 100644 --- a/datafusion/sqllogictest/test_files/monotonic_projection_test.slt +++ b/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