rluvaton opened a new issue, #23793:
URL: https://github.com/apache/datafusion/issues/23793
### Describe the bug
@EmilyMatt brought to my attention that `concat` can't be preserving lex
ordering with the following example:
`concat(my_col, "1")` would not preserve ordering in case my_col has values:
"a" and "a0" since before it was sorted and afterward it was not
### To Reproduce
```rust
#[tokio::test]
async fn test_concat_lex_sort_with_optimization() -> Result<()> {
test_concat_lex_ordering(true).await
}
#[tokio::test]
async fn test_concat_lex_sort_without_optimization() -> Result<()> {
test_concat_lex_ordering(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 test get a flag if to run optimization on this or not
/// in both cases it expect the data to be ordered correctly
async fn test_concat_lex_ordering(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,
);
// WHERE c = concat(a, b)
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<dyn PhysicalExpr>;
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,
);
let optimized = if optimize {
EnsureRequirements::new().optimize(plan, &ConfigOptions::default())?
} else {
plan
};
let plan_str = displayable(optimized.as_ref()).indent(true).to_string();
let results =
datafusion_physical_plan::collect(optimized,
Arc::new(TaskContext::default()))
.await?;
let all = arrow::compute::concat_batches(&results[0].schema(),
&results)?;
let expected = create_array!(Utf8, vec!["a", "a0"]) as ArrayRef;
assert_eq!(
all.column(0),
&expected,
"output must be sorted by (a, b); plan:\n{plan_str}"
);
Ok(())
}
```
### Expected behavior
tests should pass (they should return "a01" before "a1")
### Additional context
In a lot of places in the code it is mentioned as an example that concat
preserve lex ordering which is not true, so this should be fixed as well
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
