Generalize merge-mode streaming toward the full reducer surface (parity with spill)

[espg]

🤖 from Claude

Motivation

The #280 reducer-coverage discussion surfaced an asymmetry: spill applies to all reducers (single-block is byte-identical to pooled, so even non-mergeable reducers work; only multi-block spill is mergeable-only), whereas merge-mode streaming is mergeable-only up front — validate_streaming rejects expressions, non-len scalars, and non-tdigest ragged before any reads. Today the only wired merge laws are len/count (summation) and the two tdigest reducers (#279).

This issue is to decide how (and how far) to generalize merge-mode toward the full reducer surface, so a config isn't silently forced onto spill/pooled just because it uses a common reduction.

The math — three tiers of reducer

1. Trivially mergeable (associative), not yet wired. These have exact incremental laws and just need wiring:
   • min, max (elementwise), sum, count
   • mean (as running sum + count)
   • var / std (via streaming moments — Welford M2 / Chan parallel-variance combine)
   • bounding box / extent, any linear combination of the above
2. Approximately mergeable. Exact form is non-mergeable but a mergeable sketch exists:
   • quantiles / median → already covered by t-digest (t-digest agg hot path is unvectorized: ~15× build + ~6× fold available in pure numpy #279); the answer is "use the tdigest reducer," documented.
   • distinct-count → HLL (out of scope unless wanted).
3. Genuinely non-mergeable. No incremental law over flushes:
   • arbitrary expression: fields whose intermediate isn't a function of mergeable aggregates (e.g. np.median(h) directly, order-dependent ops)
   • histograms with data-dependent bins

Options

• (a) Wire the tier-1 associative laws (min/max/sum/mean/moments) into the streaming/spill fold. Biggest coverage win for least risk — these are exact. Mirrors how len and tdigest are already wired.
• (b) Policy for tier-3 (non-mergeable) fields. Choose one:
  • reject (status quo) — clear but forces manual backend choice.
  • auto-fallback — a config with any non-mergeable field auto-routes to spill (single-block exact) or pooled, instead of erroring. Best "just works" UX; needs care that the fallback's memory is acceptable.
  • partial-stream — fold the mergeable fields, buffer only the non-mergeable fields' raw obs. Hybrid; the non-mergeable fields lose the memory bound, so probably only sensible when they're few/small.
• (c) Expression decomposition (stretch). Detect when an expression: is a pure function of mergeable aggregates (e.g. mean = sum/count) and stream it by folding the parts. Powerful but fiddly; likely a later phase.

Open questions

1. Is the goal true parity with spill (accept every config, fall back as needed), or just widening the mergeable set (tier 1) and keeping an explicit reject for tier 3?
2. For tier-3 fallback — auto-route to spill, auto-route to pooled, or keep rejecting with a clearer "use mode: spill / drop streaming" message?
3. Should the associative laws live in one place shared by merge-mode and spill (they both fold), so a reducer's merge law is declared once? (Today len and tdigest are special-cased in both StreamingAggregator and SpillAggregator.)

Related: #279 (tdigest merge laws, the current template for wiring one), #280 (parallel spill reducer — Flavor A is reducer-agnostic, so it composes with whatever laws this issue adds). Labeled discuss — no code until the policy (esp. the tier-3 fallback) is settled.

[espg]

Is the goal true parity with spill (accept every config, fall back as needed)

yes, pretty much

For tier-3 fallback — auto-route to spill, auto-route to pooled, or keep rejecting with a clearer "use mode: spill / drop streaming" message?

leans, thoughts, pros / cons?

Should the associative laws live in one place shared by merge-mode and spill (they both fold), so a reducer's merge law is declared once? (Today len and tdigest are special-cased in both StreamingAggregator and SpillAggregator.)

this sounds correct / nice -- any reason not to do it?

[espg]

🤖 from Claude

Picking up the two you turned back.

Tier-3 fallback — leaning: auto-route, but to spill, not pooled, and only behind an explicit escape hatch. Reasoning:

• Auto-route to spill is the closest thing to true parity: spill's single-block path is byte-identical to pooled and reducer-agnostic, so a non-mergeable field just runs exact there. The multi-block spill path stays mergeable-only (the existing contract) — so a tier-3 field caps spill at one block, bounded by _default_block_bytes, with no new correctness surface.
• Auto-route to pooled gives up the memory bound entirely (full per-cell dict), which is the thing streaming/spill exist to avoid — worst fit for the large-cell case the module docstring calls out (the 88S tangent-granule that motivated the streaming path).
• Keep rejecting is safest but is exactly the status-quo friction this issue is trying to remove.

The cost of auto-routing is silence: a user asks for mode: streaming, gets spill (or a degraded single-block), and their memory profile changes invisibly. So the concrete proposal is auto-route with a loud log line + a strict escape hatch — default routes, streaming: {strict: true} restores today's hard reject. That keeps "just works" as the default without hiding a footprint change from someone who set mode deliberately.

Unifying the merge laws — no reason not to, with one caveat. A single registry (e.g. field → {fold, finalize, empty} keyed off function:/kind:) is the right shape and would delete the parallel special-casing in both StreamingAggregator (the _count_fields + inline build_tdigest/merge_tdigests block) and SpillAggregator. The caveat is that the two folders don't have identical shapes: streaming folds k-way order-independent into a live dict, while spill folds block-by-block and has a pairwise variant (build_tdigest_pairwise) precisely because order matters there. So a registry entry can't be a single merge(a, b) — it must distinguish the order-independent combine from the pairwise/sequential one, or tdigest's associativity assumption gets silently applied to a reducer that isn't order-independent. Declaring associative: bool (or separate combine/accumulate hooks) per reducer covers it.

Two clarifying questions before wiring tier-1:

1. For var/std, do you want Welford/Chan streaming moments carried as a (count, mean, M2) triple in the store schema, or is var expected to be a derived expression: over stored sum/sumsq/count fields? That decides whether tier-1 adds a new stored intermediate or just new fold laws over existing ones — and it interacts with the expression-decomposition path in Document user-side t-digest via declarative numpy expression + define the user-side-code policy #281.
2. Should the registry live in stats/ (next to tdigest.py, where the one existing non-trivial law already sits) or in processing/ (next to the two folders that consume it)? The former keeps a reducer's math in one namespace; the latter keeps it next to the fold loops.

Risk to flag: mean/var fold laws are exact, but floating-point summation order differs between the k-way (streaming) and block-sequential (spill) folders, so the two backends can disagree in the last ULPs on the same config. Numerically fine, but it will surface as flaky "streaming vs spill parity" assertions if a test compares them for exact equality — worth deciding now that parity tests use tolerance, not ==.

---

Generated by Claude Code