Repository navigation
Add selective, non-destructive reads for grouped aggregation state #24602
Description
Is your feature request related to a problem or challenge?
GroupValues and GroupsAccumulator expose grouped output through destructive APIs:
GroupValues::emit(EmitTo)removes group keys.GroupsAccumulator::evaluate(EmitTo)removes final values.GroupsAccumulator::state(EmitTo)removes intermediate state.
This works well when a batch aggregate emits each group once. It is expensive for consumers that keep grouped state alive and periodically read it, such as a streaming engine emitting changed results or checkpointing dirty groups.
Suppose the state contains G groups and only K changed since the previous flush. To read K groups while keeping the state live, the current APIs require materializing and restoring all G groups. Evaluation may require another drain and restore. The work and scratch memory remain proportional to G, even when K is small.
Describe the solution you'd like
Add a shared selection type and preserving read methods:
pub enum GroupSelection<'a> {
All,
Indices(&'a [usize]),
}
pub trait GroupValues: Send {
fn values_preserving(
&mut self,
_selection: GroupSelection<'_>,
) -> Result<Vec<ArrayRef>> {
not_impl_err!("Preserving group values are not implemented")
}
fn supports_values_preserving(&self) -> bool {
false
}
}
pub trait GroupsAccumulator: Send + Any {
fn evaluate_preserving(
&mut self,
_selection: GroupSelection<'_>,
) -> Result<ArrayRef> {
not_impl_err!("Preserving grouped evaluation is not implemented")
}
fn supports_evaluate_preserving(&self) -> bool {
false
}
fn state_preserving(
&mut self,
_selection: GroupSelection<'_>,
) -> Result<Vec<ArrayRef>> {
not_impl_err!("Preserving grouped state is not implemented")
}
fn supports_state_preserving(&self) -> bool {
false
}
}Proposed semantics:
Indicesreturns rows in request order and may contain duplicates.- Reads do not remove or reorder groups. Existing group IDs remain valid for later updates and merges.
- Logical accumulator state does not change. Implementations may still take
&mut selfto reuse builders or update caches. - The API does not promise zero-copy output.
- Callers with untrusted indices can validate them explicitly; hot paths need not repeat a bounds-checking scan.
Default methods can return NotImplemented. Separate capability methods let callers choose preserving reads before starting a multi-aggregate snapshot. This also avoids breaking custom accumulators and FFI implementations.
Built-in vector and bitmap-backed implementations can gather selected entries directly, making the selected path proportional to K where the representation permits it. Primitive, byte, row, and multi-column group values can likewise materialize selected keys without rebuilding their lookup tables. Existing destructive methods remain the preferred path when the state is no longer needed.
The GroupsAccumulatorAdapter can preserve final evaluation because Accumulator::evaluate is repeatable. Preserving intermediate state may remain unsupported there because Accumulator::state consumes its state.
Describe alternatives you've considered
The current drain, select, and restore workaround needs O(G) work and scratch memory for every read.
Adding arbitrary indices to EmitTo would mix two different contracts. EmitTo removes groups and shifts IDs; this operation only reads state.
Additional context
Related work:
- Support for observing intermediate aggregation results #18773 proposes observing intermediate aggregation results.
- Track long-term replacement for destructive EmitTo::First in aggregation #23251 tracks the longer-term problems with destructive
EmitTo::First. - A separate retention proposal will cover deleting arbitrary groups and remapping survivors.
Activity
rkrishn7 commentedon Aug 24, 2026
take
zhuqi-lucas commentedon Aug 24, 2026
Nice write-up — the K≪G motivation is clear, and I agree
EmitTois the wrong place for this (its contract is output + removal; this is read-only).Two questions on the API shape:
Trait surface. This adds 6 methods to
GroupsAccumulator(3 preserving + 3 capability), on top of 13 in-tree impls plus FFI/external ones. Defaults keep it non-breaking — but is there appetite for collapsing it, e.g. onepreserving_read(kind, selection), or splitting the preserving methods into their own sub-trait that impls opt into? Curious whether you considered and rejected that.&mut selfon a "preserving" read. Makes sense for builder/cache reuse, but it means a snapshot still needs exclusive access to every accumulator — no concurrent reads across aggregates. Worth stating explicitly in the doc comment so callers don't assume otherwise from "non-destructive".xudong963 commentedon Aug 24, 2026
+1 on separating preserving reads from EmitTo; the “select vs. drain” distinction makes the contract much easier to reason about.
One point worth tightening is index validation and safety.
GroupSelection::Indiceswill be passed through GroupValues and every GroupsAccumulator, and optimized implementations will likely use direct vector indexing or Arrow take. Since these are safe trait methods, an out-of-bounds index must not be able to cause UB, while validating the same K indices in every implementation would add O(K × number_of_accumulators) work.Would it make sense to introduce a validated selection object, constructed once against
total_num_groups, or otherwise specify the required error/panic behavior explicitly? The contract should also define Indices(&[]). A reusable validated selection could then be shared by the group keys and all accumulators participating in the same snapshot.rkrishn7 commentedon Aug 24, 2026
Hmm, I might not be fully understanding the proposed
preserving_read(kind, selection)suggestion. But regarding the separate subtrait, my biggest concern is with calling code. Callers generally holdBox<dyn GroupsAccumulator>, and a plainPreservingGroupsAccumulator: GroupsAccumulatorcannot be recovered from that trait object without adding anas_preserving() -> Option<&mut dyn PreservingGroupsAccumulator>hook. Which returns some complexity backed to the base trait. Additionally, the proposed approach lets callers check support across all accumulators, then choose either the preserving or destructive path.Definitely, I agree it is weird for a logically non-mutating API to be
&mut. This should be documented clearly. There seems to be some precedent, withAccumulator::evaluate().rkrishn7 commentedon Aug 24, 2026
💯 I agree that validating on each call is undesirable. Validating once and reusing is better. Definitely makes sense to me
feat: preserving grouped state reads (apache#24714)