Support (order by / sort) for DataFrameWriteOptions

[zhuqi-lucas]

Which issue does this PR close?

Closes #13873

Rationale for this change

DataFrameWriteOptions is missing an order by / sort by like available in SQL.

For sql we have the option to sort, for example:

You can use the WITH ORDER clause of the CREATE EXTERNAL TABLE if your data is already ordered

https://datafusion.apache.org/user-guide/sql/ddl.html#create-external-table

CREATE EXTERNAL TABLE test (
    c1  VARCHAR NOT NULL,
    c2  INT NOT NULL,
    c3  SMALLINT NOT NULL,
    c4  SMALLINT NOT NULL,
    c5  INT NOT NULL,
    c6  BIGINT NOT NULL,
    c7  SMALLINT NOT NULL,
    c8  INT NOT NULL,
    c9  BIGINT NOT NULL,
    c10 VARCHAR NOT NULL,
    c11 FLOAT NOT NULL,
    c12 DOUBLE NOT NULL,
    c13 VARCHAR NOT NULL
)
STORED AS CSV
-- this line tells DataFusion the data in the file is already ordered by (c2 ASC)
WITH ORDER (c2 ASC)
LOCATION '/path/to/aggregate_test_100.csv'
OPTIONS ('has_header' 'true');

But for writing my parquet or other format files, we don't support it

What changes are included in this PR?

Add the sort support.

Are these changes tested?

yes

Added unit testing.

Are there any user-facing changes?

Yes, we support new order option for DataFrameWriteOptions

[alamb]

Thank you @zhuqi-lucas -- this PR is really nice in my mind -- it is well commented and well tested 🏅

cc @TheBuilderJR

[alamb]

this is a good test 👍

[Dandandan]

Shouldn't LogicalPlanBuilder::copy_to have the ordering info as well?
If we write the info to a new table, do we know avoid sorting the table again? We should expose the ordering info to the newly created table for that.

[alamb]

I think copy_to inserts the data from whatever the input plan is (if there is a redundant sort the optimizer will remove it). Is that what you are asking?

It is an interesting question of how to communicate "this table is sorted" information for newly written files. Is this what you are talking about?

[Dandandan]

Yes I mean that after this PR we seem we don't communicate "we are sorted" yet, which is I think would support usecases such as in the issue from @TheBuilderJR (write data sorted on a column, making following read query with order by on the same column fast).

[zhuqi-lucas]

Thank you @alamb @Dandandan for review, i am also interested how we can communicate "this table is sorted" information for newly written files. I will also investigate this.

[alamb]

I think the ordering information is normally handled by a higher level "catalog" rather than the parquet format itself.

This is the only thing I know of in Parquet, but I don't think it can describe the ordering

https://github.com/apache/parquet-format/blob/d784f11f4485e64fdeaa614e0bde125f5132093d/src/main/thrift/parquet.thrift#L587-L591

Maybe this is something that https://iceberg.apache.org/ could represent 🤔

It is also conceivable that DataFusion itself could write custom metadata in paquet and other formats that support that custom metadata with the ordering, but that seems like we would just be reinventing Iceberg and similar table formats

[zhuqi-lucas]

Updated the code to also include the write_table testing case which is the same with insert into sql.

[alamb]

In my opinion this PR is a step forward as at least now it is easier to configure the writer desires a particular sorted output

However, as @Dandandan points out, while this might make it easier to find ordering during writing DataFusion can already do this via the DataFrame API (which is actually what this PR does under the covers) but there is no way to communicate the ordering of a file back with the existing listing table implementation

I recommend we file a follow on issue to discuss that usecase if people want to pursue it

[zhuqi-lucas]

Thank you @alamb @Dandandan for review, it makes sense we continue investigating the solution for communicating the ordering of a file back with the existing listing table implementation.

I created a follow-up issue for this:
#13891

[Dandandan]

Thank you @zhuqi-lucas