Do not sort data that is already sorted

[NGA-TRAN]

Is your feature request related to a problem or challenge?

I have data that is already sorted on the columns in order by but when I run the explain, I still see the Sort operator is in the plan.

Reproducer:

This is the content of the csv file name your_data.csv

1,50,1000
2,30,1000
3,100,500
4,120,400

Create the table:

CREATE EXTERNAL TABLE test5 (
    "key"  VARCHAR NOT NULL,
    "value"  INT NOT NULL,
    "time"  BIGINT NOT NULL
)
STORED AS CSV
WITH ORDER ("key" ASC, "time" ASC)
LOCATION 'path/to/your_data.csv';

This query has order by on the sorted key, time but there is still sort operator in the plan

-- Q1
explain
select "key", "time", "value"
from test5
where "key" = '2'
order by "key" ASC, "time" ASC
limit 1;
+---------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| plan_type     | plan                                                                                                                                                                                                                                                        |
+---------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| logical_plan  | Limit: skip=0, fetch=1                                                                                                                                                                                                                                      |
|               |   Sort: test5.key ASC NULLS LAST, test5.time ASC NULLS LAST, fetch=1                                                                                                                                                                                        |
|               |     Projection: test5.key, test5.time, test5.value                                                                                                                                                                                                          |
|               |       Filter: test5.key = Utf8("2")                                                                                                                                                                                                                         |
|               |         TableScan: test5 projection=[key, value, time], partial_filters=[test5.key = Utf8("2")]                                                                                                                                                             |
| physical_plan | GlobalLimitExec: skip=0, fetch=1                                                                                                                                                                                                                            |
|               |   SortPreservingMergeExec: [key@0 ASC NULLS LAST,time@1 ASC NULLS LAST], fetch=1                                                                                                                                                                            |
|               |     ProjectionExec: expr=[key@0 as key, time@2 as time, value@1 as value]                                                                                                                                                                                   |
|               |       CoalesceBatchesExec: target_batch_size=8192                                                                                                                                                                                                           |
|               |         FilterExec: key@0 = 2                                                                                                                                                                                                                               |
|               |           RepartitionExec: partitioning=RoundRobinBatch(10), input_partitions=1                                                                                                                                                                             |
|               |             CsvExec: file_groups={1 group: [[Users/hoabinhngatran/Documents/git/ngaPlaying/commands/arrow-datafusion/sort_order/data.csv]]}, projection=[key, value, time], output_ordering=[key@0 ASC NULLS LAST, time@2 ASC NULLS LAST], has_header=false |
|               |                                                                                                                                                                                                                                                             |
+---------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+

As an advanced question, this query does not have the key column on the sort order but a filter on the key which I think can be optimized to eliminate to sort operator, too

-- Q2
explain
select "time", "value"
from test5
where "key" = '2'
order by "time" ASC
limit 1;
+---------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| plan_type     | plan                                                                                                                                                                                                                                                          |
+---------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| logical_plan  | Limit: skip=0, fetch=1                                                                                                                                                                                                                                        |
|               |   Sort: test5.time ASC NULLS LAST, fetch=1                                                                                                                                                                                                                    |
|               |     Projection: test5.time, test5.value                                                                                                                                                                                                                       |
|               |       Filter: test5.key = Utf8("2")                                                                                                                                                                                                                           |
|               |         TableScan: test5 projection=[key, value, time], partial_filters=[test5.key = Utf8("2")]                                                                                                                                                               |
| physical_plan | GlobalLimitExec: skip=0, fetch=1                                                                                                                                                                                                                              |
|               |   SortPreservingMergeExec: [time@0 ASC NULLS LAST], fetch=1                                                                                                                                                                                                   |
|               |     SortExec: fetch=1, expr=[time@0 ASC NULLS LAST]                                                                                                                                                                                                           |
|               |       ProjectionExec: expr=[time@2 as time, value@1 as value]                                                                                                                                                                                                 |
|               |         CoalesceBatchesExec: target_batch_size=8192                                                                                                                                                                                                           |
|               |           FilterExec: key@0 = 2                                                                                                                                                                                                                               |
|               |             RepartitionExec: partitioning=RoundRobinBatch(10), input_partitions=1                                                                                                                                                                             |
|               |               CsvExec: file_groups={1 group: [[Users/hoabinhngatran/Documents/git/ngaPlaying/commands/arrow-datafusion/sort_order/data.csv]]}, projection=[key, value, time], output_ordering=[key@0 ASC NULLS LAST, time@2 ASC NULLS LAST], has_header=false |
|               |                                                                                                                                                                                                                                                               |
+---------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+

I think Q2's plan can look like Q3 below (without sort)

---Q3
explain
select "time", "value"
from test5
where "key" = '2'
order by "time" ASC
limit 1;
+---------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| plan_type     | plan                                                                                                                                                                                                                                                          |
+---------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| logical_plan  | Limit: skip=0, fetch=1                                                                                                                                                                                                                                        |
|               |   Sort: test5.time ASC NULLS LAST, fetch=1                                                                                                                                                                                                                    |
|               |     Projection: test5.time, test5.value                                                                                                                                                                                                                       |
|               |       Filter: test5.key = Utf8("2")                                                                                                                                                                                                                           |
|               |         TableScan: test5 projection=[key, value, time], partial_filters=[test5.key = Utf8("2")]                                                                                                                                                               |
| physical_plan | GlobalLimitExec: skip=0, fetch=1                                                                                                                                                                                                                              |
|               |   SortPreservingMergeExec: [time@0 ASC NULLS LAST], fetch=1                                                                                                                                                                                                   |
|               |     SortExec: fetch=1, expr=[time@0 ASC NULLS LAST]                                                                                                                                                                                                           |
|               |       ProjectionExec: expr=[time@2 as time, value@1 as value]                                                                                                                                                                                                 |
|               |         CoalesceBatchesExec: target_batch_size=8192                                                                                                                                                                                                           |
|               |           FilterExec: key@0 = 2                                                                                                                                                                                                                               |
|               |             RepartitionExec: partitioning=RoundRobinBatch(10), input_partitions=1                                                                                                                                                                             |
|               |               CsvExec: file_groups={1 group: [[Users/hoabinhngatran/Documents/git/ngaPlaying/commands/arrow-datafusion/sort_order/data.csv]]}, projection=[key, value, time], output_ordering=[key@0 ASC NULLS LAST, time@2 ASC NULLS LAST], has_header=false |
|               |                                                                                                                                                                                                                                                               |
+---------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+

Describe the solution you'd like

Q1 and Q2 do not need sort operator but the plan knows data is really sorted

Describe alternatives you've considered

No response

Additional context

No response

[NGA-TRAN]

CC @alamb

[NGA-TRAN]

Extra note: I have tested on data of thousands of rows and I created a work around by not adding order by clause. It runs faster than the one with order by

[alamb]

The explain plan shown for Q1 does not appear to have a sort 🤔

For Q2 I agree the SortExec is unneeded.

[mustafasrepo]

Actually ordering defined at the source is ORDER ("key" ASC, "time" ASC), which is lexicographical ordering (tiebreaks are resolved by next expression in the vector). However, ordering asked in the second query is order by "time" ASC. Table is not globally ordered according to "time" ASC. Hence that sort is need in the second query. However if during table creation, query were

CREATE EXTERNAL TABLE test5 (
    "key"  VARCHAR NOT NULL,
    "value"  INT NOT NULL,
    "time"  BIGINT NOT NULL
)
STORED AS CSV
WITH ORDER ("key" ASC)
WITH ORDER ("time" DESC)
LOCATION 'path/to/your_data.csv';

(which according to data at your example is satisfied). Query below will not have SortExec in its plan. Because time DESC specification at the source is global.

explain select key, time, value
from test5
where key = 2
order by "time" DESC
limit 1;

[crepererum]

I think what's happening here is: DF only adds sort execs if the "pipeline" demands it (i.e. one node requires a sorted input but its input isn't sorted) or it changes existing sorts for better parallelization (mostly a change from SortExec<-Union to a SortPreservingMergeExec<-Union<-SortExec).

IIRC it does not prune sorts that are not necessary. During SQL plan construction however the SortExec is added unconditionally (because at this point you usually don't know enough about phys. sorting) and then there's no optimizer pass that removes it. I agree that DF should have such a pass (or the existing passes should take care of that). It's a bit of a design question because it doesn't really fit the pipeline constraint design. DF currently doesn't have a construct to express final output constraint. I think it would be better if SQL lowering would emit:

OutputContstraintExec:
  FilterExec:
    CsvExec

where SortConstraintExec is basically a pass-through / no-op node that just declares:

1. that it wants sorting (or not if SQL didn't ask for it)
2. that it wants a single partition

then the entire optimizer pass pipeline would add a sort if it is required and otherwise it wouldn't do anything. We remove the constraint node in a final optimizer pass, but it also doesn't hurt much.

[NGA-TRAN]

@mustafasrepo : thanks for pointing out we can add 2 WITH ORDER in the create table. It is my TIL. However, the data is not sorted on time DESC (sorry I should have made the example data clearer), the data is only sorted on key, time. Due to the fact that the query has a filter on key = a_constant, the data is actually sorted on time for that query if we optimize the plan to do so.

[mustafasrepo]

@mustafasrepo : thanks for pointing out we can add 2 WITH ORDER in the create table. It is my TIL. However, the data is not sorted on time DESC (sorry I should have made the example data clearer), the data is only sorted on key, time. Due to the fact that the query has a filter on key = a_constant, the data is actually sorted on time for that query if we optimize the plan to do so.

Now I see the problem, thanks for clarifying it.

[wolffcm]

The explain plan shown for Q1 does not appear to have a sort

That is because there is just one input partition to the RepartitionExec, as the CsvExec has just one file group. If there was more than one input partition, then I would expect a SortExec to be in there. @NGA-TRAN maybe the sort disappeared when you made a CSV reproducible from the original IOx one?

If there was more than one file group, the SortExec would go away if we turn on the optimizer config setting bounded_order_preserving_variants which I added recently; this would let RepartitionExec become a SortPreservingRepartitionExec (great recent work from @mustafasrepo and others). I plan on turning this on for IOx since I think it will be beneficial for cases like this.

[NGA-TRAN]

@wolffcm Q1 in IOx does have SortExec. See https://github.com/influxdata/idpe/issues/17962. But it does not have SortPreservingMergeExec

I would love to see what the plan looks like when you turn bounded_order_preserving_variant on

[mustafasrepo]

I plan to add this support, to datafusion. However, it might be in next week.