Skip to main content

UpsertStashArm

mz_storage::upsert_continual_feedback_v2::stash

Trait UpsertStashArm 

Source 
pub(super) trait UpsertStashArm<T, O>
where
    T: Timestamp + Lattice + PoolColumnar + Default,
    for<'a> Ref<'a, T>: Copy + Ord,
    O: PoolColumnar + Default + Ord + Clone + Send + Sync + 'static,
    for<'a> Ref<'a, O>: Ord + Copy,
{
    type Spine: TraceReader<Time = T, Batch: SpineBatch> + 'static;
    type Batcher: 'static;
    type Chunk;
    type Chunker: Default + 'static;

    // Required methods
    fn new_batcher() -> Self::Batcher;
    fn extract(
        batcher: &mut Self::Batcher,
        upper: AntichainRef<'_, T>,
    ) -> (Vec<Self::Chunk>, Antichain<T>);
    fn flush(
        updates: &mut Vec<(UpsertKey, T, UpsertDiff<O>)>,
        chunker: &mut Self::Chunker,
        batcher: &mut Self::Batcher,
    );
    async fn drain(
        sealed: Vec<Self::Chunk>,
        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;
}
Expand description

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§

Source

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

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

Source

type Batcher: 'static

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

Source

type Chunk

The chunk representation the stash batcher accumulates into.

Source

type Chunker: Default + 'static

Sorts and consolidates raw updates into the batcher’s chunks.

Required Methods§

Source

fn new_batcher() -> Self::Batcher

A new stash batcher for one source dataflow.

Source

fn extract( batcher: &mut Self::Batcher, upper: AntichainRef<'_, T>, ) -> (Vec<Self::Chunk>, Antichain<T>)

Carve the chain of updates upper unblocks out of the stash.

Returns the chunks and a lower bound on the times of what stays behind. The stash keeps the chunks as they are: a caller that wanted a trace batch would have to take it apart again.

Source

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

Consolidate updates through chunker into chunks and push them into batcher, emptying updates (keeping its capacity). The chunker readies a fully-consolidated chunk per push_into, so nothing stays pending in it between calls.

Source

async fn drain( sealed: Vec<Self::Chunk>, 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.

Dyn Compatibility§

This trait is not dyn compatible.

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

Implementors§