compute: sort snapshot arrangements once instead of merging them

[frankmcsherry]

Motivation

Hydration time for an index is dominated by arrangement formation, in every shape measured: 57% of a plain index over a 10M-row table, 42% of an index on a SELECT DISTINCT view, 47% of a fact-to-dimension join, and eight copies of it in a hierarchical max or a TopK. Inside the merge batcher that time is the per-chunk sort, then a geometric ladder of two-way merges that copies every row once per level (about thirteen levels at 10M rows) and compares full rows at every step, then the builder copy.

During hydration every update carries the snapshot time. Nothing can cancel across times, and the chain seal emits is simply every held update in (data, time) order, consolidated. The ladder buys nothing for that case.

Change

SnapshotBatcher (new, in mz_row_spine) holds incoming chunks untouched while they all share one time. At seal it builds a compact (key prefix, chunk, position) index (16 bytes per update), sorts it comparing 8-byte prefixes and falling back to the full (data, time) comparison only on equal prefixes, and emits the sorted, consolidated chain in one copy pass. Copies per row drop from about fifteen to three (chunker, seal, builder) and most comparisons become integer compares over a cache-friendly array.

The first chunk that carries a second time hands everything held to a MergeBatcher, and the batcher stays on that path from then on, so steady-state behaviour is exactly the merge batcher's. A batch whose single time is not yet below upper is kept and reported through frontier(), as the merge batcher does.

RowRef::sort_prefix provides the prefix (length saturated at u16::MAX, then the first six bytes) next to the Ord impl whose length-first order it must agree with, so the contract has one owner. The batcher is generic over any data type led by a Row through a small SortPrefix trait, which tuples implement by delegating to their first component.

Row-keyed arrangement sites pick it up through the RowRowBatcher, RowValBatcher and RowBatcher aliases, the two direct Col2ValBatcher sites in render/context.rs and render/join/linear_join.rs, and a new mz_row_spine::KeyBatcher used by the five Row-led consolidate_named sites in top_k.rs and reduce.rs (the TopK final consolidate alone is 2.5s of a 10M-row TopK hydration).

Measurements

One worker, 10M rows, scrambled keys, five repetitions each, the same tree with and without this change:

shape	before	after	arrangement operator
index on a table	3.53s	2.81s	2.00s to 1.27s
index on SELECT DISTINCT	5.40s	4.76s	2.25s to 1.68s
fact join dim, index on output	8.3 to 11.5s	7.25s	input 2.0s to 1.6s, output 1.7s to 1.1s
count(*), sum(v) GROUP BY with 1.25M groups	7.36s	6.96s	4.07s to 3.16s

Profile of the index after the change: the index sort is 0.17s of the 1.27s arrangement, the emission copy 0.25s, the builder about 0.3s, the chunker's per-container sort 0.15s.

Memory: pending chunks are the same 64 KiB chunker chunks the merge batcher would hold in its chains, plus the transient 16-byte-per-update index at seal, plus the output chain while both are live. The merge batcher's final merge holds a comparable transient.

Testing

Unit tests in snapshot_batcher.rs cover sorting and consolidation across chunks, a seal below the held time keeping data and reporting the frontier, the handover to the general path on a second time, later single-time batches staying on the fast path, and sort_prefix agreeing with row order across lengths and types. Every sqllogictest that arranges data exercises the path.

A second commit adds UnsortedChunker: the chunkers used to sort and consolidate every input container before the batcher saw it, which is wasted work ahead of a batcher that sorts everything at seal. The unsorted chunker packs updates in arrival order, and the batcher sorts and consolidates each chunk itself only on the fallback path, so the merge batcher still receives what it requires. Error arrangements keep their chunker.

Measured on top of the other PRs of this series (identity key, prefix split, decode builder), one worker, 10M rows, three repetitions: index 2.40s to 2.17s (arrangement 1.40s to 1.15s), distinct 3.81s to 3.13s (its input arrangement 1.71s to 1.06s: the reduce-side chunker was sorting Row pairs with full comparisons), join 6.43s to 6.07s.

Open questions for review

• This is a draft for design review.
• sort_unstable_by on the index is 0.17s at 10M; a radix pass on the prefix would remove most of that but was not needed to show the effect.
• Whether to keep the fallback "for good" or return to the fast path once the general batcher drains.

Checklist

• This PR has adequate test coverage / QA involvement has been duly considered. (trigger-ci for additional test/nightly runs)
• This PR has an associated up-to-date design doc, is a design doc (template), or is sufficiently small to not require a design.
• If this PR evolves an existing $T ⇔ Proto$T mapping (possibly in a backwards-incompatible way), then it is tagged with a T-proto label.
• If this PR will require changes to cloud orchestration or tests, there is a companion cloud PR to account for those changes that is tagged with the release-blocker label.
• If this PR includes major user-facing behavior changes, I have pinged the relevant PM to schedule a changelog post.

Release notes

This release will not include user-visible changes.

🤖 Generated with Claude Code

Sorting the index by radix, and settling equal prefixes per run

Two further changes to the seal, measured on the same 10M-row hydration:

• The (prefix, chunk, position) index is sorted with an LSD radix sort on the u64 prefix, one histogram pass then one scatter pass per byte position on which prefixes differ. Row prefixes agree on their length and tag bytes, so three or four of the eight passes run. Inputs below 65536 entries keep the comparison sort.
• The gather walks runs of equal prefix. Only within such a run can two rows compare equal, or the prefix order be undecided, so the full row comparison, a random read into the held chunks, happens only there. With unique keys every run is one entry.
• The gather touches the row 16 entries ahead so that its cache miss overlaps the current row's work.

Hydration at 10M rows, one worker, median of three, on top of the earlier commits of this PR and the other hydration PRs in flight:

shape	before	after	arrangement operator
index	2.19s	2.00s	ArrangeBy 1162ms to 963ms
distinct	3.16s	3.01s	Arranged DistinctBy 1150ms to 887ms
join	5.78s	5.14s	two ArrangeBy 1453ms to 1107ms and 1046ms to 815ms
count_by	5.26s	4.42s	ArrangeAccumulable 1790ms to 1073ms
max_by with hint	5.2s	4.72s
topk with hint	8.9s	8.10s

count_by gains most because its key repeats eight times per group, so equal prefixes were common and the comparator resolved each tie with two random row reads.

In the batcher's benchmark (bench_snapshot_path, ignored by default), the seal of 10M rows goes from 691ms to 483ms: sort 229ms to 103ms, gather 420ms to 341ms. New tests cover the radix sort against a comparison sort and the ordering of rows that share a prefix.