Eliminating multi-column sort when major column is a one-to-one and monotonic expression

[suremarc]

Describe the bug

DataFusion is unable to eliminate multi-column sorts when the major column is a one-to-one and monotonic expression of a sorted input column:

• In certain cases when the expression is not one-to-one, eliminating the sort is invalid (e.g. sorting by floor(x), y is not equivalent to sorting by x, y, even though floor is monotonic).
• However, casting from Int32 to Int64 is one-to-one and monotonic, so DataFusion should be able to avoid sorting in such a case. See below.

To Reproduce

Data acquired from parquet_testing/data/delta_encoding_required_column_expect.csv (I couldn't get datafusion-cli to work with the parquet file for some reason).

CREATE EXTERNAL TABLE delta_encoding_required_column (
    c_customer_sk INT NOT NULL,
    c_current_cdemo_sk INT NOT NULL
)
STORED AS CSV WITH HEADER ROW
WITH ORDER (
    c_customer_sk DESC,
    c_current_cdemo_sk DESC
)
LOCATION 'delta_encoding_required_column_expect.csv';

-- sort required here
EXPLAIN
SELECT 
    CAST(c_customer_sk AS BIGINT) AS c_customer_sk_big,
    c_current_cdemo_sk
FROM delta_encoding_required_column
ORDER BY c_customer_sk_big DESC, c_current_cdemo_sk DESC;

Resulting physical plan has a SortExec:

+---------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| plan_type     | plan                                                                                                                                                                                                                                            |
+---------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| logical_plan  | Sort: c_customer_sk_big DESC NULLS FIRST, delta_encoding_required_column.c_current_cdemo_sk DESC NULLS FIRST                                                                                                                                    |
|               |   Projection: CAST(delta_encoding_required_column.c_customer_sk AS Int64) AS c_customer_sk_big, delta_encoding_required_column.c_current_cdemo_sk                                                                                               |
|               |     TableScan: delta_encoding_required_column projection=[c_customer_sk, c_current_cdemo_sk]                                                                                                                                                    |
| physical_plan | SortPreservingMergeExec: [c_customer_sk_big@0 DESC,c_current_cdemo_sk@1 DESC]                                                                                                                                                                   |
|               |   SortExec: expr=[c_customer_sk_big@0 DESC,c_current_cdemo_sk@1 DESC]                                                                                                                                                                           |
|               |     ProjectionExec: expr=[CAST(c_customer_sk@0 AS Int64) as c_customer_sk_big, c_current_cdemo_sk@1 as c_current_cdemo_sk]                                                                                                                      |
|               |       RepartitionExec: partitioning=RoundRobinBatch(32), input_partitions=1                                                                                                                                                                     |
|               |         CsvExec: file_groups={1 group: [[home/suremarc/rust-app-atlas/delta_encoding_required_column.csv]]}, projection=[c_customer_sk, c_current_cdemo_sk], output_ordering=[c_customer_sk@0 DESC, c_current_cdemo_sk@1 DESC], has_header=true |
|               |                                                                                                                                                                                                                                                 |
+---------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+

-- sort NOT required here
-- notice we have a single sorting column
EXPLAIN
SELECT 
    CAST(c_customer_sk AS BIGINT) AS c_customer_sk_big
FROM delta_encoding_required_column
ORDER BY c_customer_sk_big DESC;

Resulting physical plan has no SortExec:

+---------------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| plan_type     | plan                                                                                                                                                                                           |
+---------------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| logical_plan  | Sort: c_customer_sk_big DESC NULLS FIRST                                                                                                                                                       |
|               |   Projection: CAST(delta_encoding_required_column.c_customer_sk AS Int64) AS c_customer_sk_big                                                                                                 |
|               |     TableScan: delta_encoding_required_column projection=[c_customer_sk]                                                                                                                       |
| physical_plan | SortPreservingMergeExec: [c_customer_sk_big@0 DESC]                                                                                                                                            |
|               |   ProjectionExec: expr=[CAST(c_customer_sk@0 AS Int64) as c_customer_sk_big]                                                                                                                   |
|               |     RepartitionExec: partitioning=RoundRobinBatch(32), input_partitions=1                                                                                                                      |
|               |       CsvExec: file_groups={1 group: [[home/suremarc/rust-app-atlas/delta_encoding_required_column.csv]]}, projection=[c_customer_sk], output_ordering=[c_customer_sk@0 DESC], has_header=true |
|               |                                                                                                                                                                                                |
+---------------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+

Expected behavior

The first query provided above should not require a sort

Additional context

I encountered this bug when trying to re-cast the timezone of a table ordered by both timestamp and a secondary ticker column.

[alamb]

Thank you @suremarc -- I agree DataFusion should be able to remove the sort in this case.

I marked this as a enhancement rather than a bug because I think it would be new functionality rather than something that used to work (though the distinction may be a bit arbitrary)

[Lordworms]

I want to try this

[alamb]

I suggest looking for monotonic rather than disabling by expression -- if the expressions are monotonic functions of the first column then I think that is sufficient

[suremarc]

Just to make sure we're all on the same page, I think the first case (single order by expression) already works due to monotonicity, and I think CAST is considered monotonic. In the original post I mentioned that the following query does not require a sort, due to having only one sorting column:

EXPLAIN
SELECT 
    CAST(c_customer_sk AS BIGINT) AS c_customer_sk_big
FROM delta_encoding_required_column
ORDER BY c_customer_sk_big DESC;

The second case (multiple order by expressions) is the troublesome one, because I think monotonicity alone is not sufficient (that's where one-to-one-ness becomes relevant).

[Lordworms]

No, the reason the second one works and the first one fails is that when two projections come, the first one would build a wrong dependency graph

then leads to a wrong projection_ordering. and now I have two ways to fix it

1. changing the generation of the dependency graph
2. change the generation of project ordering
   but the two ways all need to specify the Expr, in this case, it generated a CAST expr, I know in buildinscalarFunction has an is_monotonic function which could show whether it is a monotonic function or not, but I think there are some other functions not covered and we need a way to figure out whether the func(multi-expression) is monotonic or not. But I don't know the correct way to do it. Still struggling with it