Replace compaction's in-memory sort with an external merge sort

[jghoman]

The sorted rewrite reads a whole group into an ArrayList<Group> and calls sortedWith (ParquetRewriter.kt:685). An in-memory sort means the group must fit in heap, which means the group must be small — #133 measured SimpleGroup at 1,668 bytes of heap per 70-byte row (151.6 B/node, budgeted as 192), so a 512 MiB group of zstd event data wants ~11.9 GiB.

#133 bounds that rather than fixing it: HOGLAKE_COMPACTION_SORTED_HEAP_BYTES is converted to a row ceiling (sortedRowCeiling = sortedHeapBytes / (192 × nodes)), and effectiveTargetBytes derates the compaction target for sorted tables to the byte size of that many rows at the table's observed density. Bin packing (#151) then packs to the derated target with a minimum that scales down with the largest input, so sorted tables still form groups — they are just small ones. Groups that cannot fit are refused with heap_budget_exceeded, uncharged to maxGroupsPerRun, and re-planned every sweep.

What it costs today. Sorted tables compact to tens of MB instead of the 512 MiB target: ~34 MiB of zstd input per GiB of sort buffer for a ten-column table. In production, one events.pageviews group (5,318,495 rows, 19 nodes/row) has been refused on every sweep against a 4,709,393-row ceiling; admitting it needs ≥18.1 GiB of sort heap on a pod configured for 16. No setting of the knob fixes this; it only moves it, and it moves the pod's memory request with it.

The way out

Compaction is unusually well set up for an external merge sort, because a group's inputs come in exactly two kinds:

• Previous compaction outputs are already sorted runs. The rewriter sorts what it writes by the live sort spec (schema.sql: the spec is ADVISORY for writers and BINDING for compaction rewrites), and every such file is identifiable in the catalog: hog_data_file.explicit_row_ids = true is set only by compaction. Merging k sorted runs needs one row per run in a priority queue — O(files) live rows, not O(group). Since bin packing packs by size and outputs approach the target, these are where almost all the bytes in a group are.
• Client-written files cannot be assumed sorted. The server never verifies file sortedness, and millpond, the bench stream and Trino all write them. But they are the smaller inputs by construction (bin packing only ever merges files below the target, and never-compacted files are the ones that start small), so sorting one alone is bounded by that file, not by the group. Sort each on its own, spill it as a temp run, and stream-merge the runs like any other.

Do that and group size stops being a heap question entirely: HOGLAKE_COMPACTION_SORTED_HEAP_BYTES, sortedRowCeiling, effectiveTargetBytes's derate, and the heap_budget_exceeded refusal all go away, and sorted tables get the same 512 MiB outputs as everything else.

Worth settling while designing it

• A run is only a run for the spec it was sorted by. The sort spec is versioned, but hog_data_file does not record which spec version an output was sorted under. An output written before set_sort_order changed the spec is sorted, but not by the live spec, and must take the sort-and-spill path. Either record the sort spec version on compaction outputs, or treat explicit_row_ids files as runs only when the table's sort spec has not changed since the file's begin_snapshot (derivable from hog_sort_spec history).
• Can a client-written file declare its own sortedness? Compaction marking its own outputs (above) unlocks the run path without trusting anyone. Letting writers assert it is a separate, untrusted-input question with the usual forged-footer shape — millpond and the bench stream both pre-sort, so the information exists, but so does the incentive to lie.
• Merging with deletion vectors. The existing path applies DVs while materializing; a streaming merge has to apply them per run without breaking the row-id carrier's guarantees.
• Ordering ties. sortedWith is stable, so ties currently keep row-id order — a k-way merge must reproduce that, since Sorted tables must get more prunable with every compaction round #100's clustering argument depends on the output ordering being deterministic.
• Where the spill goes. Compaction streams both ends since compaction: stream inputs and outputs instead of staging to disk #146 (S3InputFile/S3OutputFile) and touches no local disk. The spill is the first thing that would need scratch space again, so whoever builds it owns that decision — including whether an emptyDir whose overrun evicts the pod is the right place. Client-written inputs are small, so the spill's footprint is bounded by the sum of those inputs, not the group.
• Row cap stays. One pathological ROW (a million-element list) still materializes whole on either path; that remains maxNodesPerRow's job, not this issue's.

Related

#133 (the bound this replaces), #151 (bin packing: the scaled minimum and derated target that keep sorted tables compacting at all under the bound), #146 (streaming IO; removed the whole-object heap copies that used to sit next to the sort), #100 (sorted tables should get more prunable with each round — which needs compaction to preserve sortedness at a useful group size, not at 34 MiB), #115 (the compression that made the old proxy 1.7× worse and exposed all of this), #118 (the OOM this bound turned into a counted refusal).