jayzhan211 commented on issue #24704: URL: https://github.com/apache/datafusion/issues/24704#issuecomment-5467378543
> Have an new idea that maybe can support blocked approach in a simpler way [#15591 (comment)](https://github.com/apache/datafusion/pull/15591#issuecomment-5463126272) I like the idea of `EmitTo::Last(n)`, especially for migration toward the end state (blocked storage) — although introducing `EmitTo::Last(n)` directly would add complexity for downstream users and the existing emission mechanism. So the proposal below keeps its semantics but doesn't touch `EmitTo` at first. **Emission: the tail contract.** The contract is *emit the state of the `n` highest group indexes; indexes of all remaining groups are unchanged*. Nothing shifts and nothing is renumbered — avoiding the ~15x cost #19562 measured for incremental `First(n)` — and both layouts can serve the same request: today's contiguous storage by truncation (`split_off` on the tail), blocked storage by popping its last block in O(1) with zero copy. That symmetry is what makes it a migration path. Instead of a new enum variant, it ships as default-implemented trait methods behind a runtime flag (default off): ```rust // GroupsAccumulator — all default-implemented; no existing impl breaks fn supports_tail_emit(&self) -> bool { false } fn evaluate_tail(&mut self, n: usize) -> Result<ArrayRef> { /* default: internal_err! */ } fn state_tail(&mut self, n: usize) -> Result<Vec<ArrayRef>> { /* default: internal_err! */ } // GroupValues fn supports_tail_emit(&self) -> bool { false } fn emit_tail(&mut self, n: usize) -> Result<Vec<ArrayRef>> { /* default: internal_err! */ } ``` Zero downstream compile impact (same pattern as `supports_convert_to_state`), and the stream negotiates: tail drain only if *all* state owners in the plan support it, otherwise fall back to `All` — so unconverted UDAF/FFI accumulators keep working unchanged, and a plan mixing blocked built-ins with a merely-truncatable external accumulator still drains incrementally. The count `n` sits at the call site because the group values and every accumulator must release the *same* groups each step anyway (or output columns don't align). Scope: end-of-input drain on the unordered path only — `First(n)` and its renumbering contract stay untouched for ordered streams and the spill/merge read-back. Whether to later fold this into a real `EmitTo::Last(n)` variant (ideally together with marking `EmitTo` `#[non_exhaustive]`) can be a major-release decision after the contract has soaked. **Storage.** As sketched in this issue: fixed-capacity blocks of `B = target_batch_size` rows (power of two), never resized, allocated on demand, filling strictly front-to-back. Two refinements: block 0 is sized like today's initial capacity, so low-cardinality queries stay effectively flat (aimed at the ~10% regression observed in #15591); and byte/byte-view blocks each own their values buffer *and* their `i32` offsets, so no offset ever spans more than one block — which closes #23694 structurally rather than by widening types. **Indexing.** The group index stays a single dense `usize`: with power-of-two blocks filling front-to-back, `(block, offset) = (idx >> k, idx & (B - 1))` is a bijection. `intern`, `update_batch`, `total_num_groups`, and the hash table are all unchanged — no trait signature changes anywhere, including UDAFs and FFI. The known cost is one extra indirection per state access (the #15591 concern); hot-loop parity at low cardinality is the declared benchmark gate for the storage PRs, so this gets retired with numbers rather than argument. --- **Phases toward the end state** **1. Benchmarks first — minimum only.** For *regressions*, nothing new is built: ClickBench + the h2o groupby suite already in `bench.sh` are the gate, because they are what aggregation PRs are judged against today and h2o spans the dimensions this touches (int vs. string keys, low vs. high cardinality). For *the wins*, exactly one new artifact: a drain-phase benchmark that pre-builds ~10k and ~10M groups and records five numbers — total drain time, time-to-first-batch, max inter-batch gap (the long-poll proxy, #19906), peak reserved memory, and reserved memory at the 50%-drained mark (≈100% of peak under `All`, ≈50% under tail — a one-sample proof of incremental release). Gate numbers agreed on this issue before any behavioral PR. **2. Tail contract.** One vertical slice, targeting the refactored streams from #22710 (the declared prerequisite — happy to help review there): the default trait methods + the drain loop (drop the hash table at drain start, truncate-serve, shrink the reservation per step, one await per step) + tail arms for primitive group values, the primitive accumulators, and `count` — behind a config flag, with a fuzz mode asserting `All` ≡ tail modulo row order, and extending the aggregation fuzz framework to run under constrained memory pools (the #19649 scenario — a query that exhausts a fixed pool today and completes with incremental release — becomes a regression test). Everything else negotiates back to `All`. Flip the flag after soak; release notes call out that `GROUP BY … LIMIT` without `ORDER BY` may return different (equally valid) groups. **3. Blocked internals.** `Blocks<T>` + microbenches; blocked `PrimitiveGroupsAccumulator` and `GroupValuesPrimitive` (gate: hot-loop parity); blocked `ByteGroupValueBuilder` with per-block `i32` offsets (closes #23694, with a gated >2 GiB regression test — and worth validating against Comet's suite, since #4718 came from there); byte-view; multi-column ideally by helping land @rluvaton's `add-blocks-impl` in pieces; then the remaining accumulator families. One requirement I missed earlier: the same implementations also serve the **ordered** streams and the **spill read-back**, which emit via `First(n)` — so each blocked conversion must implement `First(n)` (and `All`) over blocks too, with the renumbering semantics intact. so the per-implementation sequence is: blocked storage → `First(n)`/`All` over blocks → delete contiguous. The cross-block compaction `First(n)` needs can be written once in `Blocks<T>` (those paths keep their tables small, so it needn't be fast), leavi ng each conversion's `First` arm thin. Default is one atomic conversion PR per implementation; if one grows too big to review, the escape hatch is a temporary sibling type — old contiguous and new blocked coexisting as separate *single-mode* implementations selected by negotiation, with a tracked deletion — which is not the dual-mode-inside-one-implementation that sank #15591. Dual storage inside an implementation still never exists (the #15591 lesson). Contiguous survives only as the negotiated compatibility tier for external accumulators — a feature kept indefinitely, not a deprecation candidate — and a small permanent coverage gap remains for types that only the `Rows`-based fallback handles: those stay on `All` unless someone later blocks that path. **4. Order-sensitive follow-ups.** Ordered and partially-ordered streams keep `First(n)` and its renumbering contract untouched, as does the spill read-back (which re-aggregates a sorted merge). Mid-stream emission under memory pressure is a separate design — the hash table is live there, so tail eviction would cost a full table sweep — and it is where the interaction with skip-partial aggregation (the pending ClickBench Q5 +61% from #22712) lives. Blocked state also opens a follow-up the epic hints at: block-granular spilling (spill whole blocks under pressure instead of emit-sort-spill of everything), which is a step toward the memory pool proactively requesting release from consumers. If a hard front-first requirement surfaces, the same blocked layout supports a front cursor instead, at the documented cost of forking `First`'s contract and the dense `0..total_num_groups` invariant — recorded as the fallback design, not the default. **5. Convert to `EmitTo::Last(n)` — the formalization release.** At the next major, once the contract has soaked through phases 2–3: fold the trait methods into a real `EmitTo::Last(n)` variant, mark `EmitTo` `#[non_exhaustive]` in the same release so this class of breakage never recurs, deprecate the tail methods for one release cycle with a mechanical migration path, and ship the upgrade guide for external accumulator authors (how to move a tail arm from the methods to the `match`). The *contract* is never deprecated — only its provisional surface is folded into the enum. If soak instead shows the methods are fine as a permanent home, this phase collapses to documenting that decision; either way it is decided on schedule, not by drift. --- **Open question: does anything require *front-first* drain?** My answer is no for the unordered path — nothing forces emitting from the front there: SQL guarantees no output order without `ORDER BY`, today's insertion order is incidental, and the paths that genuinely need ordering (ordered/partially-ordered streams, the spill read-back) keep `First(n)` and its renumbering contract untouched. The end state is indifferent either way — over blocks, with the hash table dropped at drain start, either end pops in O(1) — so a front-first requirement would only threaten the phase-2 migration slice, where the tail is the only cheap direction on contiguous buffers. If anyone knows a constraint I'm missing — spill file ordering, partial-stage eviction, assumptions in `add-blocks-impl` -- 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]
