Struct ColumnMergeBatcher 

pub struct ColumnMergeBatcher<D, T, R>
where
    D: Columnar,
    T: Columnar,
    R: Columnar,
{
    chains: Vec<VecDeque<PagedColumn<(D, T, R)>>>,
    lower: Antichain<T>,
    frontier: Antichain<T>,
    stash: Vec<Column<(D, T, R)>>,
    pager_override: Option<ColumnPager>,
    logger: Option<Logger>,
    operator_id: usize,
}

Drives the merge-batcher over Column chunks routed through a ColumnPager.

Chains hold PagedColumn entries rather than resident Columns, so each insert / merge / extract step can hand its output to the pager and store whatever the policy returns (resident, paged, or compressed). Reads during merge materialize lazily via FetchIter.

Resolves its pager lazily per call via column_pager::global_pager, so late-arriving dyncfg updates (e.g. enable_column_paged_batcher flipping on after the batcher was constructed) take effect without rebuilding the operator. Tests may override that lookup via Self::set_pager.

Fields

chains: Vec<VecDeque<PagedColumn<(D, T, R)>>>

lower: Antichain<T>

frontier: Antichain<T>

stash: Vec<Column<(D, T, R)>>

Recycled empty Column::Typed chunks. Drained heads and shipped result buffers feed in here; subsequent merge / extract calls pop from here instead of starting from a zero-capacity Column::default(). Mirrors the stash carried by the upstream differential_dataflow merge-batcher framework, which this type forks. Without it, each shipped chunk triggers a fresh per-leaf grow cycle and per-merge-round allocation dominates the inner loop.

pager_override: Option<ColumnPager>

Optional override. None means “read column_pager::global_pager fresh on every use” — the production path, so worker_config dyncfg changes that re-install the process-global pager take effect on the very next chunk this batcher processes.

logger: Option<Logger>

operator_id: usize

Implementations

impl<D, T, R> ColumnMergeBatcher<D, T, R>

where D: Columnar, T: Columnar, R: Columnar,

pub fn set_pager(&mut self, pager: ColumnPager)

Pin the pager this batcher uses, overriding the thread-local lookup. Mainly for tests; production should leave the override unset so dyncfg-driven re-installs take effect immediately.

fn pager(&self) -> ColumnPager

Current pager — override if set, else the process-global pager installed by apply_worker_config. ColumnPager is cheaply cloneable (Arc inside).

fn chain_push(&mut self, chain: VecDeque<PagedColumn<(D, T, R)>>)

Push a chain into self.chains, emitting a positive BatcherEvent covering its resident entries.

fn chain_pop(&mut self) -> Option<VecDeque<PagedColumn<(D, T, R)>>>

Pop a chain from self.chains, emitting a negative BatcherEvent retracting its resident entries.

Invariant for the retract to reconcile against the matching chain_push: chain entries are never mutated in place between push and pop. The only allowed mutation is a full pop / push pair (see insert_chain and merge_by), so each entry’s accounting category — Resident vs Paged vs Compressed — is the same at both ends. If a future change ever pages an entry out in place after push, this path silently double-counts.

fn emit_account(&self, chain: &VecDeque<PagedColumn<(D, T, R)>>, diff: isize)

Emit a single BatcherEvent summing resident accounting across chain with the given sign. No-op when no logger is attached.

impl<D, T, R> ColumnMergeBatcher<D, T, R>

where D: Columnar, for<'a> Ref<'a, D>: Copy + Ord, T: Columnar + Default + Clone + PartialOrder, for<'a> Ref<'a, T>: Copy + Ord, R: Columnar + Default + Semigroup + for<'a> Semigroup<Ref<'a, R>>,

fn insert_chain(&mut self, chain: VecDeque<PagedColumn<(D, T, R)>>)

Insert chain and rebalance: while the youngest chain is at least half the size of its predecessor, merge them.

fn merge_by( &mut self, a: VecDeque<PagedColumn<(D, T, R)>>, b: VecDeque<PagedColumn<(D, T, R)>>, ) -> VecDeque<PagedColumn<(D, T, R)>>

Merge two sorted chains. Outputs are routed through self.pager.page per chunk produced, so the result chain holds PagedColumns and the caller never sees a fully materialized merge result.

Trait Implementations

impl<D, T, R> Batcher for ColumnMergeBatcher<D, T, R>

where D: Columnar, for<'a> Ref<'a, D>: Copy + Ord, T: Columnar + Default + Timestamp + PartialOrder, for<'a> Ref<'a, T>: Copy + Ord, R: Columnar + Default + Semigroup + for<'a> Semigroup<Ref<'a, R>>, for<'a> Ref<'a, R>: Ord,

type Output = Column<(D, T, R)>

Type produced by the batcher, and also the type it consumes.

type Time = T

Times at which batches are formed.

fn new(logger: Option<Logger>, operator_id: usize) -> Self

Allocates a new empty batcher.

fn seal( &mut self, upper: Antichain<Self::Time>, ) -> (Vec<Self::Output>, Description<Self::Time>)

Returns all updates not greater or equal to an element of upper, as a sorted and consolidated chain together with the description that bounds them.

fn frontier(&mut self) -> AntichainRef<'_, Self::Time>

Returns the lower envelope of contained update times.

impl<D, T, R> Drop for ColumnMergeBatcher<D, T, R>

where D: Columnar, T: Columnar, R: Columnar,

fn drop(&mut self)

Executes the destructor for this type.

fn pin_drop(self: Pin<&mut Self>)

🔬This is a nightly-only experimental API. (pin_ergonomics)

Execute the destructor for this type, but different to Drop::drop, it requires self to be pinned.

impl<D, T, R> PushInto<Column<(D, T, R)>> for ColumnMergeBatcher<D, T, R>

where D: Columnar, for<'a> Ref<'a, D>: Copy + Ord, T: Columnar + Default + Clone + PartialOrder, for<'a> Ref<'a, T>: Copy + Ord, R: Columnar + Default + Semigroup + for<'a> Semigroup<Ref<'a, R>>,

fn push_into(&mut self, chunk: Column<(D, T, R)>)

Accept an already-consolidated chunk from the upstream chunker, route it through the pager, and insert it as a singleton chain.