Incorrect ORDER BY results with mismatched Parquet timestamp units

[cdelmonte-zg]

Describe the bug

I have Parquet files that store ts as Timestamp(Microsecond) and a table declared as ts TIMESTAMP (which is Timestamp(Nanosecond) in DataFusion) with WITH ORDER (ts ASC).

SELECT ts, v FROM t ORDER BY ts returns the rows out of order: the files are read in listing order and the SortExec is removed. Declaring the same table with ts TIMESTAMP(6) gives the correct order.

The same mismatch also disables file-level pruning on ts (files_ranges_pruned_statistics=3 total --> 3 matched instead of 3 total --> 1 matched for a one-day range predicate), although results of filtering queries stay correct.

To Reproduce

datafusion-cli 55.1.0 built from main at 871058c. Run this in a fresh directory containing no other Parquet files. Each file is individually sorted by ts; the file listing is not.

-- three microsecond files; listing order (a, b, c) is not time order (day 2, day 0, day 1)
-- 1710201600000000 us = 2024-03-12T00:00:00Z, one row every 5 hours

COPY (SELECT arrow_cast(1710374400000000 + i * 18000000000, 'Timestamp(Microsecond, None)') AS ts, 20 + i AS v FROM generate_series(0, 4) AS t(i) ORDER BY ts) TO '/tmp/repro/data/a.parquet';

COPY (SELECT arrow_cast(1710201600000000 + i * 18000000000, 'Timestamp(Microsecond, None)') AS ts,      i AS v FROM generate_series(0, 4) AS t(i) ORDER BY ts) TO '/tmp/repro/data/b.parquet';

COPY (SELECT arrow_cast(1710288000000000 + i * 18000000000, 'Timestamp(Microsecond, None)') AS ts, 10 + i AS v FROM generate_series(0, 4) AS t(i) ORDER BY ts) TO '/tmp/repro/data/c.parquet';

 -- put the three files in one file group
SET datafusion.execution.target_partitions = 1;

CREATE EXTERNAL TABLE t_ns (ts TIMESTAMP, v BIGINT) STORED AS PARQUET WITH ORDER (ts ASC) LOCATION '/tmp/repro/data/';

CREATE EXTERNAL TABLE t_us (ts TIMESTAMP(6), v BIGINT) STORED AS PARQUET WITH ORDER (ts ASC) LOCATION '/tmp/repro/data/';

SELECT ts, v FROM t_ns ORDER BY ts;

SELECT ts, v FROM t_us ORDER BY ts;

Observed for t_ns:

+---------------------+----+
| ts                  | v  |
+---------------------+----+
| 2024-03-14T00:00:00 | 20 |
| 2024-03-14T05:00:00 | 21 |
| 2024-03-14T10:00:00 | 22 |
| 2024-03-14T15:00:00 | 23 |
| 2024-03-14T20:00:00 | 24 |
| 2024-03-12T00:00:00 | 0  |
| 2024-03-12T05:00:00 | 1  |
| 2024-03-12T10:00:00 | 2  |
| 2024-03-12T15:00:00 | 3  |
| 2024-03-12T20:00:00 | 4  |
| 2024-03-13T00:00:00 | 10 |
| 2024-03-13T05:00:00 | 11 |
| 2024-03-13T10:00:00 | 12 |
| 2024-03-13T15:00:00 | 13 |
| 2024-03-13T20:00:00 | 14 |
+---------------------+----+

EXPLAIN FORMAT INDENT shows both plans without a SortExec and with the declared output_ordering; only t_us reordered the files by statistics:

t_ns: DataSourceExec: file_groups={1 group: [[.../a.parquet, .../b.parquet, .../c.parquet]]}, projection=[ts, v], output_ordering=[ts@0 ASC NULLS LAST], file_type=parquet

t_us: DataSourceExec: file_groups={1 group: [[.../b.parquet, .../c.parquet, .../a.parquet]]}, projection=[ts, v], output_ordering=[ts@0 ASC NULLS LAST], file_type=parquet

Expected behavior

ORDER BY ts returns ordered rows regardless of the declared unit, and file statistics for ts are the same instants expressed in the declared type.

Additional context

As far as I can tell, in DFParquetMetadata::statistics_from_parquet_metadata the min/max accumulators are created from the table type (nanoseconds) while StatisticsConverter produces the per-row-group bounds in the file type (microseconds).

update_batch fails on the type mismatch, the error is discarded (summarize_column_statistics(...).ok()), and the never-updated accumulator is reported as min = max = Exact(TimestampNanosecond(NULL)) with null_count = Absent.

MinMaxStatistics::new_from_files accepts these bounds and is_sorted() accepts the equal NULL bounds as ordered, so validated_output_ordering keeps the declared ordering for files that are actually out of order.