parthchandra opened a new pull request, #5331:
URL: https://github.com/apache/datafusion-comet/pull/5331
## Which issue does this PR close?
Closes #5323.
## Rationale for this change
Currently when Comet reads a sorted Iceberg table, the scan throws away
the ordering to achieve parallelism. As a result Spark can't tell the data is
already sorted, and it re-sorts on every read — in joins, aggregates, windows,
and order-by queries — even though the work was already done at write time.
This PR modifies the native scan to preserve and report that ordering, so
Spark can drop the redundant sorts. It builds on Iceberg's own
`SupportsReportOrdering` (apache/iceberg#14948): when Iceberg reports a sort
order, merge the sorted files per partition, and tell Spark the result is
sorted.
The scan also reports Iceberg's key-group partitioning. (For Spark to
eliminate shuffle in SMJ, the scan must also report how the data is grouped by
the join key (storage-partitioned join)).
## What changes are included in this PR?
For every Spark partition, the scan now reads each sorted file as its own
stream and k-way-merges them into one sorted stream using DataFusion's
`SortPreservingMergeExec`.
Summary of changes -
- **Proto**: a new `table_sort_orders` field on `IcebergScanCommon`
carries the reported sort order to the native side.
- **Scala (serde + scan exec)**: report the sort order and the key-grouped
partitioning to Spark. Ordering reporting is limited to the safe case for v1 —
identity-transform sort fields on top-level columns that are
actually in the projection. Anything else (transforms, a sort key that
isn't selected) falls back to today's unordered read and reports nothing, so
it's always correct.
- **Native (Rust)**: `IcebergScanExec` becomes multi-partition when an
ordering is present (one sorted stream per file), and the planner wraps it in
`SortPreservingMergeExec`. No changes to iceberg-rust — we just
call its existing reader once per file instead of once for the whole batch.
The PR also adds two config flags for the Iceberg scan:
- `spark.comet.scan.icebergNative.sortMerge.enabled` (default **on**) —
report the sort order and do the per-partition merge. Only does anything when
Iceberg's `spark.sql.iceberg.planning.preserve-data-ordering` is
on (off by default).
- `spark.comet.scan.icebergNative.reportPartitioning.enabled` (default
**off**) — report key-grouped partitioning for storage-partitioned joins. Off
by default while we build out coverage for the adaptive-execution
partition-pushdown path.
Note: A global `ORDER BY` still keeps its final sort — a per-partition
merge isn't a cluster-wide order — so that case is unchanged.
## How are these changes tested?
- **Native unit tests** in `iceberg_scan.rs`: multi-partition with a
reported ordering, single-partition without one.
- **End-to-end suite** `CometIcebergSortMergeReadSuite` over real Iceberg
tables (local Hadoop catalog, sort order set via the Iceberg Java API, one file
per insert).
--
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]
