Skip to content

Add arbitrary group retention for grouped aggregation state #24603

Description

@rkrishn7

Is your feature request related to a problem or challenge?

Long-lived aggregation state needs to delete groups that are no longer active. A streaming engine may evict keys after a TTL or watermark while keeping every other group available for future updates.

EmitTo::First(n) can remove only a prefix, and it emits values as part of the operation. Arbitrary cleanup currently requires draining all G group keys and accumulator states, selecting the K survivors, rebuilding the group interner, and merging every survivor back into each accumulator. This is expensive and temporarily materializes the complete state.

For example, expiring groups {1, 4, 7} from ten live groups cannot be expressed without rebuilding all remaining groups.

Describe the solution you'd like

Add retain_groups to both state-owning traits:

pub trait GroupValues: Send {
    fn retain_groups(&mut self, retained_group_indices: &[usize]) -> Result<()>;
}

pub trait GroupsAccumulator: Send + Any {
    fn retain_groups(
        &mut self,
        retained_group_indices: &[usize],
        total_num_groups: usize,
    ) -> Result<()>;
}

The input order defines the new dense group IDs. Retaining [5, 2] maps old group 5 to new group 0 and old group 2 to new group 1. Indices must be unique and in bounds. An empty selection clears the state; retaining every group in order is a no-op.

The operation destroys the complement and returns no output. A caller applies the same selection to GroupValues and every accumulator before using the new IDs. This keeps retention separate from EmitTo, whose contract combines output with prefix removal.

Universal defaults make the API usable by custom implementations:

  • GroupsAccumulator can call state(EmitTo::All), use Arrow take, and merge the retained state under new group IDs.
  • GroupValues can materialize the selected keys using the preserving-read API, clear its lookup state, and re-intern the survivors. An emit-and-take fallback can support implementations without preserving reads.

This work should follow the implementation of #24602 so this fallback can avoid materializing discarded keys when possible.

Built-in implementations can likely do better most of the time. Vector and bitmap-backed accumulators can move or copy only retained entries. Hash tables and collision lists can remap IDs in place or rebuild at retained size. Variable-width and collection accumulators can drop unreferenced buffers and batches. Optimized implementations should update memory accounting and release excess capacity, although exact capacity is not part of the logical API contract.

Describe alternatives you've considered

  • Extending EmitTo with arbitrary indices would make its removal and output semantics harder to reason about. Retention does not need to produce aggregate results.
  • A free list or tombstone design would avoid immediate compaction, but it would give up dense IDs and require changes throughout vector-backed accumulators. Block-based state may make some removals cheaper in the future, but it does not define how arbitrary survivors are remapped.

Additional context

Activity

  1. added theissue type on Aug 24, 2026
  2. rkrishn7 commented on Aug 24, 2026

    @rkrishn7
    ContributorAuthor

    take

  3. zhuqi-lucas commented on Aug 24, 2026

    @zhuqi-lucas
    Contributor

    The prefix-only limitation of EmitTo::First(n) is real, so +1 on separating retention from emission.

    One question on the contract: retain_groups has to apply the same permutation to GroupValues and to every accumulator for the new dense IDs to line up, and it returns Result. What's the contract if accumulator k fails after 0..k-1 have already remapped? At that point GroupValues and the accumulators disagree on what a group ID means, and there's no way to roll back.

    Worth pinning down one way or the other — either document it as poisoning (caller must discard the whole aggregation state on Err), or make it two-phase so validation/allocation can fail before anything is mutated. Silent divergence across accumulators would be nasty to debug.

  4. 2010YOUY01 commented on Sep 7, 2026

    @2010YOUY01
    Contributor

    Long-lived aggregation state needs to delete groups that are no longer active. A streaming engine may evict keys after a TTL or watermark while keeping every other group available for future updates.

    I think the current groups accumulator physical design is only optimized for throughput, so it's really hard to make eviction efficient. Reusing the GroupsAccumulator s can make both batched and streaming execution harder to evolve. The existing Accumulator interface seems even more suitable than GroupsAccumulator to extend for streaming use cases.

    I'm wondering whether you've considered extending Accumulator, or adding a parallel accumulator implementation optimized for streaming use cases, and what the blockers are.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

Labels

enhancementNew feature or request

Projects

No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions