compute: pack accumulable reduce accumulators at their natural width

[djahandarie]

Motivation

The accumulable reduce (sum, count, any, all) keeps one Accum per aggregate in the diff of its input arrangement. Accum is an enum sized for its widest variant (Numeric: a 64 byte Decimal<27> plus four i64 counters) and aligned to 16 by the i128 in its sibling variants, so size_of::<Accum>() == 112 regardless of which variant a slot holds:

variant	needs	pays
SimpleNumber (integer sums, count)	24 B	112 B
Float	48 B	112 B
Bool (any/all)	16 B	112 B
Numeric	96 B	112 B

A reduce with seven sums over integer and double precision columns pays about 816 bytes per group for the diff alone. This change came out of a production dataflow with exactly that shape over roughly half a billion groups, whose accumulable arrangement holds about 400 GB, most of the dataflow's memory; at the per-record saving measured below, this layout should take well over 200 GB of that back.

Description

Adds Accums, a newtype over Vec<u8> that stores the same accumulators as a self-delimiting sequence of slots, each a one-byte tag followed by that variant's fixed-size payload (16/24/48/95 bytes), and lets the new enable_packed_accumulable_diff flag select it as the arrangement diff type.

Non-obvious decisions:

• Accum is untouched and remains the arithmetic type. Every operation on Accums decodes a slot to Accum, uses the existing Semigroup/IsZero/Multiply<Diff> impls, and re-encodes. The accumulation domain (Overflowing<i128>, NumericAgg = Decimal<27>) is stored at full width and finalize_accum and all output types are unchanged, so this is a storage-layout change only.
• Numeric slots go through Decimal::to_raw_parts/from_raw_parts, written field by field, so no struct padding enters the encoding.
• Rendering is generic over a private AccumulableDiff trait implemented for both Vec<Accum> and Accums. The Vec<Accum> body was moved into the generic function without behavioral edits. The trait carries a per-plan Layout (slot byte offsets for Accums, () for Vec<Accum>) computed once next to the zero diff, so per-row slot writes in explode_one stay O(1).
• decode_slot bounds its reader to the slot, so a field list that ever disagrees with slot_len panics in every build profile instead of reading into the neighbouring slot.
• Flag defaults off in production and on in CI (mzcompose VariableSystemParameter, also randomized under CI_SYSTEM_PARAMETERS=random). It is listed in KNOWN_MISSING_FROM_LD, so enabling it in cloud needs a LaunchDarkly flag first.

Size note: about half of the insertions are tests (reduce/tests.rs); the accumulable-reduce test module moved out of line to satisfy the mz-compute crate's inline-test-module limit.

Verification

New unit tests in src/compute/src/render/reduce/tests.rs:

• accums_slot_sizes pins the encoded size of every variant.
• accums_roundtrip round-trips every variant through encode/decode at extremes (i128::MIN/MAX, negative counters, large and small numeric exponents).
• accums_region_roundtrip drives AccumsRegion, the only unsafe in the change, through a ColumnStack: copy, read back, heap_size, clear, copy again.
• accums_empty_is_identity checks the empty value behaves as DD's Vec<R> zero.
• accums_agree_with_vec_accum applies seeded random plus_equals/multiply sequences over every accumulable AggregateFunc (nulls, float and numeric specials, negative diffs, mixed multi-slot layouts) to a Vec<Accum> and an Accums and asserts they agree at every step, reading the packed side through as_accums with a scratch buffer that is never cleared by the test.

Manual: bin/environmentd with 100k distinct groups and sum over four int and three double columns. mz_arrangement_sizes for ArrangeAccumulable [val: empty]:

flag	records	size	bytes/record
off	100,000	83.5 MB	835
on	100,000	29.8 MB	298

The 537 bytes saved per record is exactly seven 112-byte Accums becoming four 24-byte and three 48-byte payloads plus a tag byte each (784 → 247); the rest of the record (key, time, offsets, Diff) is unchanged, and the ReduceAccumulable output arrangement is byte-identical in both runs.

Possible follow-ups, not in this PR: is_zero decodes a full Accum per slot on the consolidation path and could read the counters directly if it shows up in benchmarks; the output arrangement is unchanged by this PR.

🤖 Generated with Claude Code

[antiguru]

Reviewed at 5b11372. Relaying the Buildkite results first, since they are not visible outside the org.

Clippy fails on both linux and macOS, and it is the only genuine failure in the change. Both errors sit in the new test file, and neither surfaces under cargo check or cargo test because they are clippy-only lints on a test target.

error: use of a disallowed method `std::iter::Iterator::zip`
  --> src/compute/src/render/reduce/tests.rs:377:46
377 |     for (idx, (aggr, datum)) in aggrs.iter().zip(datums).enumerate() {
    = note: use `Itertools::zip_eq` instead

error: this let-binding has unit value
  --> src/compute/src/render/reduce/tests.rs:468:9
468 |         let vec_layout = vec_state.layout();

error: could not compile `mz-compute` (lib test) due to 2 previous errors

The second one fires because accums_agree_with_vec_accum instantiates D as Vec<Accum>, so Layout resolves to (). Clippy's own suggestion, passing &() at each call site, is worse than the cause. Lifting the two halves of that test into one helper generic over D: AccumulableDiff keeps Layout opaque, silences the lint, and drops the duplicated step sequence.

Restart 2 also failed, and it is unrelated to this change. The assertion is test fixture did not produce an MV with its persist-sink worker off worker 0 in hydration-history-survives-restart. This branch's merge base with main is d4861fb (2026-09-01), so it predates #38636 (2026-09-03), which reworked exactly that fixture to stop requiring a worker-0 separation on every trial. A rebase clears it.

Everything else passed, including rust-cargo-test, all five sqllogictest shards, testdrive, cluster, and platform checks. The packed layout is therefore exercised end to end with the flag on.

On the change itself. The design reads right for the constraint it works under, because Semigroup::plus_equals, IsZero::is_zero and Multiply receive no Layout, so the value has to carry its own tags even though the plan fixes them. That reasoning is worth recording next to the type, as it turns the byte per slot from apparent slack into a documented consequence of the trait signatures. Four things I would want addressed or answered:

• There is no CPU number to go with the memory result. plus_equals and is_zero now decode and re-encode a slot at a time on the merge path, which is where a reduce over half a billion groups spends its time. Two things point the other way, since explode_one's per-row clone drops from 784 to 247 bytes and Accums::multiply mutates in place where DD's Multiply for Vec<R> allocates through collect(). The AccumulateReductions feature benchmark has the right shape and reports both wall clock and memory, so on and off numbers from it would settle this before the flag graduates.
• The soft-panic paths lose results where the current path panics. Accums::plus_equals returns mid-loop, leaving the slots before the offset added and the rest not, and set drops the accumulator outright. Under --optimized both produce a permanently wrong aggregate, whereas Vec<Accum> reaches Accum::plus_equals's unreachable! for the same violation. The invariant is structural, a diff from another plan reaching this arrangement, so a replica restart is the better outcome.
• plus_equals does not detect a longer rhs. The loop bounds on self.0.len(), so trailing rhs slots are dropped with no soft panic at all, which is the one misalignment the NOTE: promises to catch. DD's Vec<R> impl pushes those elements onto self instead. A single length comparison after the two empty cases catches it and subsumes the per-slot check.
• Nit: Accums reads close to Vec<Accum>, and as_accums returning &[Accum] from an Accums compounds it. PackedAccums would keep the two apart at every use site.

Net improvement to the codebase. Good to merge once clippy is green.

🤖 Posted by Claude Code

[djahandarie]

Thank you for the review! It looks like #38718 (using columnar for Accum) is more on strategy, so I'll assume further work will continue there, but if something about that PR doesn't work out I'm happy to pick this one back up.