Struct ChunkedArm 

struct ChunkedArm;

The chunked flavor: UpsertChunkBatcher stash, FeedbackSpine feedback arrangement, bulk-probe drain.

NOTE: the seal’s chain merge loads the bodies of the chunks it actually merges, while untouched survivors keep their spilled bodies. The partition against the upper passes chunks whose resident time bounds fall entirely on one side through without loading them, so only chunks the upper splits round-trip the pool codec. The drain then loads sealed chunks one at a time, so a large drain (a frontier advance releasing a snapshot’s worth of stash at once) holds at most one chunk resident rather than the whole backlog.

Trait Implementations

impl<T, O> UpsertStashArm<T, O> for ChunkedArm

where T: Timestamp + TotalOrder + Lattice + Sync + Columnation + Columnar + Default, for<'a> Ref<'a, T>: Copy + Ord, O: Columnar + Default + Ord + Clone + Send + Sync + 'static, for<'a> Ref<'a, O>: Ord + Copy,

type Spine = Spine<Rc<ChunkBatch<ColumnChunk<(UpsertKey, Row), T, Overflowing<i64>>>>>

The feedback arrangement’s spine. 'static because the operator future owns a trace agent for it.

type Batcher = MergeBatcher<ChunkMerger<ColumnChunk<UpsertKey, T, UpsertDiff<O>>>>

The source-stash batcher. 'static because the operator future owns it.

fn new_batcher() -> Self::Batcher

A new stash batcher for one source dataflow.

fn push_chunk( batcher: &mut Self::Batcher, chunk: Column<(UpsertKey, T, UpsertDiff<O>)>, )

Push one sorted, consolidated Column chunk into the batcher, in the batcher’s chunk representation.

async fn drain( sealed: Vec<ColumnChunk<UpsertKey, T, UpsertDiff<O>>>, ineligible: &mut Vec<(UpsertKey, T, UpsertDiff<O>)>, output_handle: &AsyncOutputHandle<T, FueledBuilder<CapacityContainerBuilder<Vec<(UpsertValue, T, Diff)>>>>, output_cap: &Capability<T>, persist_upper: &Antichain<T>, trace: &mut TraceAgent<Self::Spine>, worker_id: usize, source_id: GlobalId, ) -> DrainStats

Classify one sealed stash against persist_upper and emit eligible output; see DrainStats.

fn flush( updates: &mut Vec<(UpsertKey, T, UpsertDiff<O>)>, chunker: &mut ColumnChunker<(UpsertKey, T, UpsertDiff<O>)>, batcher: &mut Self::Batcher, )

Consolidate updates through chunker into Column chunks and push them into batcher, emptying updates (keeping its capacity). The chunker readies a fully-consolidated chunk per push_into, so the extract loop drains everything it produced.