parthchandra opened a new issue, #5323: URL: https://github.com/apache/datafusion-comet/issues/5323
### What is the problem the feature request solves? ## What are we trying to do? When an Iceberg table is already sorted, we want Comet to *know* that and stop re-sorting data that is already in order. More precisely, when Comet's native scan reads a sorted Iceberg table, we want each Spark partition to come out already sorted, and we want to tell Spark's optimiser that it is sorted — so redundant `Sort` and shuffle stages in sort-merge joins, grouped aggregates, and windows over that data simply disappear (and an `ORDER BY ... LIMIT` becomes a cheap bounded take). In short - sort the data once when you write it, and never pay to sort it again on the way out. ## How is it done today, and what are the limits? Upstream Iceberg is building exactly this pattern: - [apache/iceberg#14948](https://github.com/apache/iceberg/pull/14948) implements the Spark DSv2 `SupportsReportOrdering` API and a k-way merge reader, so a partition that holds several already-sorted files still hands Spark one globally sorted stream. Spark can then drop the sort. - [apache/iceberg#16305](https://github.com/apache/iceberg/pull/16305) adds a `k-way-merge` `RewriteDataFiles` strategy that keeps files sorted over time cheaply — no shuffle, no spill — instead of re-running a full sort compaction. The two reinforce each other: #16305 keeps the table cheaply sorted, and #14948 makes every read benefit from it. Comet's native Iceberg scan does not take part in any of this yet. Today it hands *all* of a partition's files to iceberg-rust in one call that reads them **unordered** and concatenates the result, and it hard-codes the scan's reported ordering to "none". So the moment Comet takes over a scan from Spark, the ordering Iceberg worked out is thrown away, and Spark has to re-sort — even though the data on disk was already in order. We get the correctness, but we pay for a sort we should not need. ## What is new in our approach, and why do we think it will work? We will replicate Iceberg's read-side behaviour natively, using a building block DataFusion already has: its order-preserving k-way merge (`SortPreservingMergeExec`, a loser-tree merge over already-sorted streams). So for each Spark partition: - read each sorted file in the partition as its own stream (instead of one unordered lump); - feed those streams into `SortPreservingMergeExec`, which merges them into a single sorted stream; - advertise that ordering out of the scan so Spark eliminates the sort above it. Why we are confident: - We are not inventing a merge — we reuse DataFusion's tested loser-tree merge, its memory accounting, and its ordering propagation. There is no new merge algorithm to get wrong. - We do not re-check whether the files are really sorted. If Iceberg reports an ordering, it has already verified (via `SortOrderAnalyzer`) that every file carries a sort-order id matching the table's current sort order, and that files are not split mid-file. So each file is genuinely sorted, and merging sorted files yields a sorted result. We simply trust Iceberg's report. - The native merge and the ordering we report to Spark are driven by the *same* signal, so they cannot disagree — we never promise Spark an ordering that the native plan does not actually produce. - It fits Comet's execution model cleanly. Comet drives one partition at a time; the merge is purely within a single partition, so there is no cross-partition coordination to design. There is a second half to this for joins, and it matters. A sort-merge join needs both sides lined up the same way: sorted on the join key *and* already grouped so that matching keys sit together. If Comet told Spark only about the sorting, Spark would still shuffle the data across the cluster to group it for the join — and a shuffle moves rows between partitions, which throws the sorted order away, so the sort would come straight back. We would have paid for the merge and got nothing. So to actually remove the sort in a join, we also tell Spark how the table's files are already grouped by the partition key, letting it join the two sides directly without a shuffle. Iceberg already works this grouping out; Comet just needs to pass it along, exactly the way it now passes along the sort order. (This is the "storage-partitioned join" path.) There is one case we want to be careful about. When Spark runs adaptively, it can work out the exact set of partition keys at runtime and push them down into the scan. It knows how to push them into its own scan node, but not into Comet's native one — so we are not yet sure the grouping lines up correctly in that case. Rather than guess, we would keep the grouping report behind its own switch, off by default, and add a test that drives that adaptive path before turning it on. The sorting half does not depend on any of this, so it is unaffected either way. We would keep the whole thing safe with switches, layered like this: - The sort-order reporting and the merge sit behind `spark.comet.scan.icebergNative.sortMerge.enabled` (we would default it on), which itself only does anything when Iceberg's own `spark.sql.iceberg.planning.preserve-data-ordering` is on (off by default). - The grouping report sits behind a separate `spark.comet.scan.icebergNative.reportPartitioning.enabled` (we would default it off, until that adaptive path is verified). With any of these off, behaviour is exactly as it is today. **And crucially — no changes to iceberg-rust.** We get per-file streams by calling iceberg-rust's existing reader once per file rather than once for the whole batch. All the new logic lives in Comet. ## Who cares, and what difference does it make? Anyone running Comet over sorted Iceberg tables — which is a common, deliberately-maintained layout for large fact tables. Sort elimination is not a micro-optimisation: a removed sort stage removes a full pass over the data, its memory pressure, and its spill risk, and a removed shuffle takes a cluster-wide data movement out of the plan. It directly speeds up sort-merge joins, grouped aggregates, and windows over these tables — exactly the heavy queries people run against them. And it lets Comet match, rather than regress, the behaviour Spark itself gets from #14948 once that lands. ## What are the risks, and how do we contain them? - **Reporting an ordering that is not real.** This is the one that matters, because a wrong ordering means silently wrong query results. We contain it by tying the reported ordering to the same signal that turns on the native merge, and by trusting Iceberg's validated report rather than guessing. We also start with the narrowest, safest case (below). - **Comparison-semantics mismatches.** Comet's comparison of a value must match how Iceberg sorted it on write, or the merge is subtly out of order. #14948 already flagged one of these (UUIDs order differently in Iceberg than in Spark). We de-risk by starting with identity-sorted, top-level, primitive columns only. - **Losing read parallelism.** Today's unordered read is aggressively concurrent; a merge reads more lazily. We treat this as a benchmark question, not a blocker — details in the closing section. - **Mismatched grouping.** The grouping we hand to Spark for joins must exactly match how Comet actually lays the data out across partitions, or a join could pair up the wrong rows. We contain this by deriving the reported grouping from the very same partition list Comet reads from, and by leaning on Iceberg's own grouping rather than recomputing it. ## How hard is it, and how long? It is a focused, single change touching five parts: 1. **Wire format** — carry the table's sort order in the scan's protobuf message. 2. **Serialisation (Scala)** — translate Iceberg's reported ordering into that message, reusing Comet's existing sort-order serialisation. 3. **Ordering report (Scala)** — return the ordering from the scan operator instead of "none", so Spark can drop a redundant sort. 4. **Grouping report (Scala)** — also pass along how the table's files are grouped by the partition key, so joins skip the shuffle. Without this, joins reshuffle and lose the ordering, and the merge cost buys nothing — so it belongs in the same change. It ships behind its own switch, off by default, until the adaptive-execution corner above is verified. 5. **Native execution (Rust)** — make the scan expose one sorted stream per file and merge them. We propose shipping it as a single end-to-end change, behind two switches: `spark.comet.scan.icebergNative.sortMerge.enabled` (on by default) for the sort-order reporting and merge, and `spark.comet.scan.icebergNative.reportPartitioning.enabled` (off by default) for the grouping report. **v1 scope (deliberately narrow):** sort orders on identity-transform, top-level, primitive columns that are already in the query's projection, with exact ascending/descending and nulls-first/last handling. Anything outside that falls back to today's unordered read and reports no ordering — so it is always safe, never wrong. ## How would we know it works? We would check it in two layers. - **Correctness, on every relevant query.** Compare Comet's output against plain Spark's on the same data. Any merge slip — a dropped, duplicated, or mis-ordered row, or a scan that claims an order or grouping it does not really produce — shows up as a mismatch. This does not lean on the Iceberg build reporting anything, so it works everywhere, including the published Iceberg in CI. - **The payoff, where the build supports it.** Check that the extra sort and the shuffle are gone from the plan for a sort-merge join, a grouped aggregate, and a window over co-partitioned sorted inputs. Since the published Iceberg does not yet report the ordering, these plan checks would run on a build that does and skip elsewhere, so they never raise a false alarm. We would cover the awkward cases too: several files merged into one order, the same key spread across files, nulls-first and nulls-last, descending and multi-column orders, a single file (no merge), the sort key not being selected (which falls back to an unordered read and must still be correct), and the feature on-versus-off giving identical results. A global `ORDER BY` is a deliberate non-target — it keeps its final sort, because a per-partition merge is not a cluster-wide order. We would also benchmark the merge cost against the sort it removes, and add a test for the adaptive-execution grouping case before enabling the grouping switch by default. --- ## One more thing Two things we are intentionally leaving out of v1 are worth documenting here so we can follow up appropriately. ### Sort transforms Iceberg sort orders are not always on raw columns. A sort field can be on a *transform* of a column: `bucket[N]`, `truncate[W]`, `year`/`month`/`day`/`hour`, or `void`, as well as plain `identity`. The files are physically sorted by the *transformed* value, so to merge them correctly we would have to recompute the same transform and compare on it — not on the raw column. This is more tricky than it sounds: - `bucket[N]` sorts by a hash bucket ordinal (Murmur3, mod N), which has nothing to do with the raw value order. We would have to reproduce Iceberg's exact hashing, including its type-specific encoding, to compare correctly. - `truncate[W]` is type-dependent — strings truncate by code points, decimals by scaling, integers by flooring to a multiple of W — and we compare on the truncated value. - `year`/`month`/`day`/`hour` map a date or timestamp to an integer ordinal; ordering is on that. - `void` contributes no ordering and must be dropped from the key entirely. - UUID (flagged in #14948) orders differently in Iceberg than in Spark's row comparator, so even the identity case for UUID has to follow Iceberg's byte ordering. There is also a payoff question. Even where we *can* merge on a transformed value, Spark only removes a sort when the sort it wants matches the ordering we report. A normal `ORDER BY col` wants the raw column, so a `bucket(col)` ordering will not eliminate it — transform orderings mostly help operators whose requirement is itself on the transform (e.g. storage-partitioned/bucketed joins). So transforms are more work and more risk for a narrower benefit. That is precisely why v1 restricts to identity, top-level, primitive columns. If a transform sort field arrives as a non-column expression, we fall back to the unordered path. When we do add transforms, the plan is to build native transform kernels, validate each type against Iceberg's own comparators on a fuzz corpus, and only then start reporting orderings for them. We can potentially use these transforms from iceberg-rust (if they are available) ### Read parallelism Today's read is deliberately concurrent: iceberg-rust reads up to N files at once and interleaves their batches (bounded by a concurrency limit), which is fast but throws ordering away. That is to say there is a trade-off between concurrency and ordering. With the merge, each file becomes its own stream and `SortPreservingMergeExec` pulls from them *cooperatively*: it primes one batch from every file at the start, then advances lazily, re-reading a file only when the merge has consumed its current batch. So cross-file read overlap is reduced compared with today's eager approach. There are two things to note here: - Within a single file, iceberg-rust still reads row groups concurrently — so for a **few large files** per partition, latency is largely hidden anyway and the difference should be small. - For **many small files** per partition, the lost eager overlap is where a regression, if any, would show up. We are treating this as a measurement, not an assumption. We ship v1 on the simple cooperative model, benchmark the few-large-files and many-small-files cases separately, and only if the latter regresses do we plan to address it. One approach to address this is to add a read-ahead: spawn each file's reader onto the runtime with a small bounded buffer (similar to Comet's repartition) so every file is read ahead while the merge operates. That restores the concurrency at the cost of a little extra memory and code complexity. ### Describe the potential solution _No response_ ### Additional context _No response_ -- 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]
