Core, Spark 4.1: Add K-way merge rewrite strategy for pre-sorted data files

[anuragmantri]

This PR adds a K-way merge compaction strategy to the RewriteDataFiles action. K-way merge rewrites pre-sorted data files by streaming-merging them in sort-key order without shuffle, preserving the table's sort order in output files.

K-way merge is intended for tables that are already sorted but have accumulated multiple overlapping files per partition (e.g., after daily ingestion into a previously sort-compacted table). It achieves the same output as SORT but eliminates shuffle and spill entirely.

Implementation summary:

1. API: New kWayMerge() default method on RewriteDataFiles interface. The table must have a defined sort order.
2. Planner (KWayMergeRewriteFilePlanner): Extends SizeBasedFileRewritePlanner with sort-key-ordered file grouping. Files are sorted by their lower bounds on the first sort field before bin-packing, ensuring groups cover contiguous key ranges. Includes includeColumnStats() on the table scan to load file bounds for planning.
3. Runner (SparkKWayMergeFileRewriteRunner): Opens all files in a group as streaming iterators, applies GenericDeleteFilter for position/equality deletes, then merges using SortedMerge (priority queue). Output is written via GenericAppenderFactory with size-based file rotation and partition-change detection.
4. Range parallelism: For large groups, the runner splits files into ranges by total size and processes ranges in parallel via jsc.parallelize(). Range assignments are broadcast to executors. Controlled by range-parallelism-enabled (default true), ranges-per-group (default 25), and min-files-for-range-parallelism (default 10).

Constraints:

1. All input files must have a valid sort_order_id > 0 matching the table's sort order. Files without sort metadata are rejected with a clear error.
2. Uses Iceberg's generic reader/writer stack (row-by-row), not Spark's vectorized path. This means higher per-record overhead than sort, but zero shuffle I/O.
3. Output files are individually sorted. When range parallelism is enabled, files from different ranges may have overlapping key ranges (consistent with how SORT behaves).

Usage:

-- Via procedure 
CALL catalog.system.rewrite_data_files( table =>'db.my_table', strategy =>'k-way-merge', options => map('max-concurrent-file-group-rewrites', '100') ) 

// Via action API SparkActions.get() .rewriteDataFiles(table) .kWayMerge() .option("max-concurrent-file-group-rewrites", "100") .execute(); 

AI Usage: I used Claude Opus 4.7 for code generation, test writing, and review. I manually reviewed and validated all generated code.

[RussellSpitzer]

What makes this distinct from the planning processes in the other planners? I'm wondering if it's right to have a completely independent planner or or whether this shouldn't just be some special things or flags on the Sort planner? In general I'm not sure this need to be a completely separate hierarchy and maybe should be part of sort?

[anuragmantri]

You're right. The planning logic is structurally the same as BinPack/Sort. The key differences are that k-way merge needs includeColumnStats() on the scan and sorts files by lower bounds before bin-packing (so groups cover contiguous key ranges). I refactored to extend BinPackRewriteFilePlanner and made the scan overridable for k-way merge. Let me know what you think.

[RussellSpitzer]

This is basically the same as the parent code so i'm wondering if we have the inheritance off here

[anuragmantri]

I now inherited from the BinPackRewriteFilePlanner. That reduced a lot of duplication but I had to override planFileGroups() because we sort by lower bounds to reduce the overlap.

[github-actions]

This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@iceberg.apache.org list. Thank you for your contributions.

[anuragmantri]

This is not stale.

[nssalian]

@anuragmantri this needs a rebase to pick up some of the changes that went in for CVEs

[gimgit]

I've been working in the same area (#17504, overlap measurement/selection on the sort key), so I read this with interest. K-way merge for pre-sorted files fills a real gap — sort compaction re-shuffles data that is already 99% ordered, and that cost is what pushes people to skip maintenance. A few questions and one coordination note:

1. File-level sortedness. The streaming merge assumes each input file is internally sorted, but table sort order is advisory — an engine or a plain append can write an unsorted file into a sorted table, and nothing in the metadata records per-file sortedness. If one input file is unsorted, the merge output is silently mis-ordered. Is there a validation pass, or is this documented as a precondition? A cheap partial guard could be checking that each file's bounds don't contradict its neighbors, though that can't see inside a file.

2. Selection. The planner extends size-based selection, but the population this strategy targets — pre-sorted, size-healthy files that overlap on the sort key — is exactly the set that size-based selection can't see (reproduction in Sort/z-order compaction permanently no-ops on size-healthy but badly-clustered tables — no way to detect or select by overlap #17489). With size thresholds alone, the files that most need a k-way merge may never be picked. The min-overlap-depth option proposed in Core, Spark 4.1: Measure and select sort-key file overlap for rewrite #17504 measures that state from the same bounds you already load; the two would compose naturally (detect by overlap, rewrite by k-way merge). Happy to coordinate if that's useful.

3. Lower-bound-only ordering. Sorting by lower bound before bin-packing keeps adjacent-key files together, but nested ranges break it: a file spanning [0, 1000] sorts next to [1, 2], [3, 4]..., and the wide file drags overlap into whichever group it lands in. Considering the upper bound too (e.g., sorting by midpoint, or splitting wide files into their own group) might reduce cross-group overlap. We hit the same co-grouping problem on the selection side, so no clean answer here either — flagging it as a shared open question.

4. Coordination note. This PR adds a buildTableScan() hook on BinPackRewriteFilePlanner to call includeColumnStats(); Core, Spark 4.1: Measure and select sort-key file overlap for rewrite #17504 adds a columnsToIncludeStats() hook on the same class for the same reason. One of us will conflict with the other. Also, the no-arg includeColumnStats() retains bounds for every column of every task through planning — on a wide schema that gets heavy. includeColumnStats(Collection) scoped to the sort column avoids it; that's what Core, Spark 4.1: Measure and select sort-key file overlap for rewrite #17504 ended up doing after hitting the memory question.