Trait UpsertStashArm 

trait UpsertStashArm<T, O>
where
    T: Timestamp + Lattice + 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: TraceReader<Time = T> + 'static;
    type Batcher: Batcher<Time = T> + 'static;

    // Required methods
    fn new_batcher() -> Self::Batcher;
    fn push_chunk(
        batcher: &mut Self::Batcher,
        chunk: Column<(UpsertKey, T, UpsertDiff<O>)>,
    );
    async fn drain(
        sealed: Vec<<Self::Batcher as Batcher>::Output>,
        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;

    // Provided method
    fn flush(
        updates: &mut Vec<(UpsertKey, T, UpsertDiff<O>)>,
        chunker: &mut ColumnChunker<(UpsertKey, T, UpsertDiff<O>)>,
        batcher: &mut Self::Batcher,
    ) { ... }
}

The flavor-specific pieces of the upsert-v2 operator: the stash batcher, the feedback arrangement’s spine, and how the drain reads prior state. build_upsert_operator holds the flavor-independent operator loop and calls through this trait at the few points where the flavors diverge.

Required Associated Types

type Spine: TraceReader<Time = T> + 'static

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

type Batcher: Batcher<Time = T> + 'static

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

Required Methods

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<<Self::Batcher as Batcher>::Output>, 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.

Provided Methods

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.

Dyn Compatibility

This trait is not dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety".

Implementors

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>>>>>

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

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

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<ArcBatch<OrdValBatch<ValRowLayout<((UpsertKey, Row), T, Overflowing<i64>)>>>>

type Batcher = ColumnMergeBatcher<UpsertKey, T, UpsertDiff<O>>