saurabhojha opened a new pull request, #2345: URL: https://github.com/apache/iceberg-go/pull/2345
Part of [issue 2344](https://github.com/apache/iceberg-go/issues/2344) (Phase 1: in-memory sort and streaming k-way merge. Spill to disk is designed in the issue and not implemented here.) ### What this adds Compaction can now sort instead of bin-pack. Bin-pack stays the default; nothing changes unless the new options are set. ```go cfg := compaction.DefaultConfig() cfg.RewriteAll = true // also sort files that are already well-sized cfg.MaxGroupSizeBytes = 8 << 30 // one group per partition, up to 8 GiB plan, err := compaction.Analyze(ctx, tbl, cfg) // ... result, err := txn.RewriteDataFiles(ctx, groups, table.RewriteDataFilesOptions{ GroupOptions: []table.CompactionGroupOption{table.WithCompactionSortOrder(tbl.SortOrder())}, }) ``` CLI: `iceberg compact run --sort [--rewrite-all] [--max-group-size BYTES]`, using the table's default sort order. `compact analyze` also accepts `--max-group-size`. ### Design **Planning is unchanged; the strategy is chosen per group.** As in iceberg-java, where every strategy plans with `BinPackRewriteFilePlanner` and only the runner differs (`binPack()`, `sort()`, `zOrder()`), `PlanCompaction` only selects files and packs them into size-bounded groups. `ExecuteCompactionGroup` picks an internal `rewriteStrategy` from its options: `binPackStrategy` (default) writes rows as read, `sortStrategy` orders them. Z-order would be a third implementation. **Two sort paths:** - *Merge.* When every file of a group records the order's `sort_order_id` (for example files written by an earlier sort rewrite) and was written under the current spec, each file is read as its own stream and the streams are k-way merged (`container/heap` of per-file cursors, `sort.Search` to copy whole runs as zero-copy slices). Memory holds about two batches per file instead of the group. Each loaded batch is checked, so a file that claims the order but is not sorted fails the group instead of producing wrong output. Delete files are loaded once and shared across the per-file streams. - *In memory.* Otherwise the group is decoded, split by the current spec's partitions, and each partition is stable-sorted and gathered with `compute.Take`. The whole group is held in memory, so `MaxGroupSizeBytes` should stay within what one rewrite can hold. Both paths write through the clustered writer with a presorted flag, so the writer skips its per-batch sort. **Ordering follows the spec.** One comparator serves both paths: per-field direction and null order, floats as `-NaN < -Inf < -x < -0 < 0 < x < Inf < NaN`, unsigned byte order for strings and binary. Arrow-go's sort kernel places NaN next to nulls and treats `-0 == 0`, and its row comparator is internal, so it cannot be shared with the merge. Two orderings that disagree would make the merge reject files sorted by the other. The comparator binds typed `Values()` slices once per batch; on 1M rows it runs at the speed of Arrow's kernel (about 347 ms vs 345 ms for int64 keys, 548 ms vs 570 ms for strings). **`sort_order_id` is recorded only when it holds.** Rows are ordered by each field's source column, which matches the spec's order only when every field but the last is identity and the last is order-preserving. For example, sorting `[day(ts), id]` by raw `ts` leaves `id` unsorted within a day. The order must also exist in table metadata. Otherwise output files are written sorted but record `sort_order_id = 0`. **`MaxGroupSizeBytes` separates group size from file size** (Java's `max-file-group-size-bytes`). Groups were packed to `TargetFileSizeBytes`, so each group yielded about one file. Sort orders rows only within a group, so a partition split across groups yields files that each span the whole key range: 10 GB of small files with a 512 MB target became about 20 separately sorted groups whose files all overlap, while one 10 GB group is written as about 20 files with disjoint key ranges. Zero keeps the previous behaviour; it is not raised by default because unsorted groups are held in memory. **`RewriteAll`** (Java's `rewrite-all`) selects every file regardless of size or deletes, and keeps groups below `MinInputFiles`, so a whole table can be sorted. ### Public API - New: `table.WithCompactionSortOrder(SortOrder) CompactionGroupOption`, `compaction.Config.RewriteAll`, `compaction.Config.MaxGroupSizeBytes`. - No existing signatures or defaults change. ### Limitations - A group whose files are not already sorted must fit in memory (Phase 2 in the issue). - Sort keys must be top-level columns of type boolean, int, long, float, double, date, time, timestamp, timestamptz, string, binary, fixed, decimal or uuid. - Transformed fields are ordered by their source column, not by the transformed value. - No ordering across groups (Java behaves the same). - No z-order. ### Testing - Unit tests (`table/sort_compaction_internal_test.go`): comparator float order and nulls, nested columns carried through the sort, `sort_order_id` claim rules, presorted detection, merge interleaving, ties keeping stream order, disjoint streams copied as whole batches, descending with nulls last, rejecting unsorted input, early stop releasing every stream. All use `memory.CheckedAllocator`. - End to end (`table/sort_compaction_test.go`): sort rewrite output and `sort_order_id`, splitting by partition, unsorted order rejected, `RewriteDataFiles` with sort, and a second rewrite that takes the merge path with a shared position-delete file. - Planner (`table/compaction/compaction_test.go`): `MaxGroupSizeBytes` packing and validation. - `go test ./...`, `golangci-lint run` (0 issues), and `go test -race` / `-tags=assert` on the sort and compaction tests, on top of current `main`. - Note: on current `main`, `table`'s tests do not compile because `commit_retry_test.go` and `rewrite_manifests_internal_test.go` still call `WriteFile`/`processManifests` without the `context.Context` added in #2336. I ran the suite with that fixed locally; the fix is not part of this PR. -- 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]
