Struct CrossPartitionAggregateStore 

pub struct CrossPartitionAggregateStore { /* private fields */ }

Lock-free concurrent store for cross-partition partial aggregates.

Each partition publishes serialized partial aggregates under its partition_id. Readers merge partials for a given group key to produce the final aggregate.

Performance

• Write (publish partial): single papaya::HashMap::insert — lock-free
• Read (get partial): single papaya::HashMap::get — lock-free
• Merge: iterate known partitions, collect partials, caller merges

Implementations

impl CrossPartitionAggregateStore

pub fn new(num_partitions: u32) -> Self

Create a new store for the given number of partitions.

pub fn publish(&self, group_key: Bytes, partition_id: u32, partial: Bytes)

Publish a partial aggregate from a partition.

Overwrites any previous partial for this (group_key, partition_id).

pub fn get_partial(&self, group_key: &[u8], partition_id: u32) -> Option<Bytes>

Get the partial aggregate for a specific partition.

pub fn collect_partials(&self, group_key: &[u8]) -> Vec<(u32, Bytes)>

Collect all partial aggregates for a group key across all partitions.

Returns a vector of (partition_id, partial_bytes) for all partitions that have published a partial for this key.

pub fn remove_group(&self, group_key: &[u8])

Remove all partials for a group key.

pub fn num_partitions(&self) -> u32

Total number of partitions.

pub fn len(&self) -> usize

Number of entries in the map (across all partitions and groups).

pub fn is_empty(&self) -> bool

Whether the store has no entries.

pub fn snapshot(&self) -> Vec<(Vec<u8>, Vec<u8>)>

Snapshot all partial aggregates for checkpointing.

Each entry is serialized as:

• Key: group_key_len(4 bytes LE) + group_key + partition_id(4 bytes LE)
• Value: raw partial aggregate bytes

The num_partitions is stored as a sentinel entry with an empty key and value containing the partition count as 4 bytes LE.

pub fn restore(&self, snapshot: &[(Vec<u8>, Vec<u8>)])

Restore partial aggregates from a checkpoint snapshot.

Clears the current state and inserts all entries from the snapshot.

Panics

Panics if a non-sentinel entry has a key shorter than the encoded length prefix (corrupted snapshot). Malformed entries with incorrect total length are silently skipped.