saurabhojha opened a new issue, #2344: URL: https://github.com/apache/iceberg-go/issues/2344
### Feature Request / Improvement ### Problem `RewriteDataFiles` / `ExecuteCompactionGroup` only bin-pack: rows are written in the order they are read, so compaction never produces files sorted by the table's sort order and never records a `sort_order_id`. iceberg-java offers `sort()` (and `zOrder()`) next to the default `binPack()` on `RewriteDataFilesSparkAction`; iceberg-go has no equivalent and no issue tracking it. Readers rely on sorted, non-overlapping files for min/max pruning, so tables written by streaming ingestion (many small files with random keys) get no benefit from compaction today beyond fewer files. ### Proposal Add sort as a second rewrite strategy, selected per group, with bin-pack staying the default. Planning (`compaction.Config.PlanCompaction`) stays strategy-independent, matching Java, where every strategy plans with `BinPackRewriteFilePlanner` and only the runner differs. **Phase 1 (proposed PR): in-memory sort + streaming k-way merge** - `table.WithCompactionSortOrder(order)` group option; CLI `iceberg compact run --sort` uses the table's default order. - Groups whose files already record the order's `sort_order_id` are streamed file by file and k-way merged, so memory holds about two batches per file instead of the whole group. A file that claims the order but is not sorted fails the group. - Other groups are decoded and sorted in memory. - Ordering follows the spec, including floats (`-NaN < -Inf < -x < -0 < 0 < x < Inf < NaN`), per-field null order, and unsigned byte order for strings and binary. - Output files record `sort_order_id` only when the claim holds: the order exists in table metadata, each field has one source column and an order-preserving transform, and every field but the last is identity. - `compaction.Config.MaxGroupSizeBytes` (Java's `max-file-group-size-bytes`). Today a group is packed to `TargetFileSizeBytes`, so each group yields about one file; because sort orders rows only within a group, files of different groups would overlap on the sort key. Zero keeps today's behaviour. - `compaction.Config.RewriteAll` (Java's `rewrite-all`), so right-sized files are sorted too. **Phase 2 (design only): external sort / spill to disk** Spark gets external sorting from its shuffle; without it, a group larger than memory needs spilling. 1. *Budget.* A group option such as `WithCompactionSortMemoryBytes(n)`; zero keeps Phase 1 behaviour. Batch sizes are measured from Arrow buffer sizes. 2. *Run generation.* Accumulate decoded batches per partition of the current spec; when the budget is reached, sort the largest partition's buffered rows with the Phase 1 comparator and write them as a sorted run. Input files that already carry the order's `sort_order_id` are runs as they are and are never re-sorted. 3. *Run format.* Arrow IPC with ZSTD/LZ4 buffer compression: no Parquet encode/decode cost, and it round-trips every Arrow type the scan produces without going through the Iceberg schema. Runs are scratch data, never table files. 4. *Location and cleanup.* A local scratch directory (default `os.TempDir()`, configurable; an `emptyDir` on Kubernetes). Runs are deleted when the group finishes, fails or is cancelled; after a crash only scratch space is left behind, never the table location. 5. *Merge.* Phase 1's k-way merge consumes `iter.Seq2[arrow.RecordBatch, error]` streams, so spilled runs read back with `ipc.Reader` plug in unchanged, partition by partition, keeping the stream clustered for the clustered writer. 6. *Fan-in.* Cap open streams (e.g. 64) to bound merge memory and file descriptors; with more runs, merge groups of runs into larger runs first (multi-pass). This also bounds Phase 1's merge of many presorted small files. 7. *Scope.* Entirely inside `ExecuteCompactionGroup`; no change to `CompactionGroupResult`, commit or conflict validation, so the distributed coordinator pattern is unaffected. 8. *Tests.* A tiny budget that spills every batch must produce output identical to the in-memory path; `CheckedAllocator` leak checks; scratch files removed on error and cancellation. **Out of scope / later** - Z-order (interleaved ordering); it fits as a third strategy behind the same seam. - Transform-aware sort keys (sorting by `bucket`/`truncate` values, as Java does); Phase 1 sorts by each field's source column. - Nested and multi-argument sort fields. - Ordering across groups (Java also sorts each file group independently). ### Open questions 1. Should planning defaults depend on the strategy (e.g. a larger `MaxGroupSizeBytes` default when sorting)? Phase 1 keeps zero because unsorted groups are held in memory. 2. Arrow-go's sort kernel places NaN next to nulls and treats `-0 == 0`, which is not the spec's order, so Phase 1 uses its own comparator. Worth an arrow-go option for spec-compatible float ordering? 3. A middle step between Phase 1 and 2: in a group that mixes already-sorted files with new ones, sort only the new files in memory and merge them with the streamed sorted files. Memory then holds the new files rather than the whole group. Worth doing before spilling? I'd like to work on this; a Phase 1 PR is ready. -- 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]
