Module upsert_continual_feedback_v2 

Implementation of the feedback UPSERT operator.

Architecture

The operator converts a stream of upsert commands (key, Option<value>) into a differential collection of (key, value) pairs, using a feedback loop through persist to maintain the “previous value” state needed for computing retractions.

Dataflow topology

  Source input ──► ┌──────────┐ ──► Output ──► Persist
                   │  Upsert  │
  Persist read ──► └──────────┘
      ▲                                           │
      └───────────── feedback ────────────────────┘

Operator loop (each iteration)

1. Ingest source data. Read upsert commands from the source input, wrap each in an UpsertDiff (carrying a columnar order key projected from FromTime via UpsertSourceTime for dedup), and push into the source-stash batcher. The batcher consolidates entries for the same (key, time) via the UpsertDiff Semigroup, keeping the update with the highest order key (latest source offset), through amortized geometric merging as data is pushed in, and cold stash state leaves RSS through the flavor’s spill path (see below). This bounds resident memory even during large source snapshots.

2. Read persist frontier. Check the probe on the persist arrangement to learn which times have been committed. When the persist frontier reaches the resume upper, rehydration is complete.

3. Seal & drain. Call batcher.seal(input_upper) to extract all source-finalized entries as sorted, consolidated chunks. Each entry is classified:

   • Eligible (at the persist frontier): the persist trace has the correct “before” state for this time. Look up the old value in the feedback arrangement, emit a retraction if present, and emit the new value.
   • Ineligible (between persist and input frontiers): persist hasn’t caught up yet. Push back into the batcher for the next iteration.
   • Already persisted (below the persist frontier): some writer has already advanced the shard past this time, so it is dropped. See the drain functions for why re-stashing it would strand the data and pin the output frontier below the shard upper.

4. Capability management. Downgrade the output capability to the minimum time of any remaining buffered data (in the batcher or pushed back as ineligible). Drop the capability entirely when the batcher is empty.

Stash flavors

UpsertStashFlavor, resolved from enable_upsert_chunked_stash at operator construction, selects between two instantiations of the same loop:

• Chunked: the stash is differential’s chunk merge batcher over ColumnChunks and the feedback arrangement is a spine of chunk batches. Committed chunk bodies spill to the process buffer pool, the drain loads sealed chunks one at a time, and prior state comes back through bulk probes of the trace’s batches.
• Paged: the stash is the paged columnar merge batcher and the feedback arrangement is a ValRowSpine. Cold chains page out of RSS through the storage-owned column pager, and prior state comes back through a trace cursor.

Both flavors’ spill paths are gated by enable_upsert_paged_spill.

Eligibility condition (total order)

For a total-order timestamp with input_upper = {i} and persist_upper = {p}, an entry at time ts is eligible when ts == p < i — the source has finalized it and persist is exactly at that time, so the feedback trace holds the correct prior state. An entry with p < ts is ineligible (persist hasn’t caught up), and one with ts < p is already persisted and dropped.

Structs

ChunkedArm 🔒

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

DrainStats 🔒

Counts from a single call to drain_sealed_input_chunked or drain_sealed_input_paged, used to update metrics.

PagedArm 🔒

The paged flavor: UpsertPagedBatcher stash, ValRowSpine feedback arrangement, cursor drain. Cold chains page out of RSS through the storage-owned column pager, captured once per batcher at construction.

UpsertDiff 🔒

UpsertDiffContainer 🔒

Derived columnar container for a struct.

UpsertDiffReference 🔒

Derived columnar reference for a struct.

UpsertFeedbackBatcher 🔒

The paged flavor’s persist-feedback batcher, wrapping Col2ValPagedBatcher only to capture the storage upsert-stash pager at construction.

Enums

TimeClass 🔒

Where a stashed entry’s time falls relative to the feedback frontier.

UpsertStashFlavor

Which stash and feedback-arrangement representation the upsert-v2 operator instantiates. The two flavors run the same operator loop; they differ in the batcher, the feedback trace, and how the drain reads prior state. See the module docs for the comparison.

Traits

UpsertStashArm 🔒

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.

Functions

build_upsert_operator 🔒

The flavor-independent upsert-v2 operator loop, generic over an UpsertStashArm.

classify_time 🔒

Classify ts against persist_upper: the single spelling of the drain’s eligibility test. The chunked drain’s probe-collection pass and its classification pass, and the paged drain’s cursor walk, all call this, so the probe set and the classification cannot disagree. Under the operator’s total order, Eligible means ts equals the frontier’s one element.

decode_upsert_value 🔒

Decode an UpsertValue produced by [upsert_value_to_row] from any datum iterator, whether over a columnar Row reference or an owned Row.

drain_sealed_input_chunked 🔒

Process sealed chunks from the batcher, classifying each entry by its timestamp relative to persist_upper:

drain_sealed_input_paged 🔒

drain_sealed_input_chunked’s counterpart for the paged flavor, classifying entries the same way but reading prior state through a trace cursor.

encode_feedback 🔒

Key the persist feedback by UpsertKey, record source statistics, and encode (UpsertKey, UpsertValue) as (UpsertKey, Row) Column chunks, the input both flavors’ feedback arrangements consume. Built with Pipeline downstream of an UpsertKey::hashed exchange, so the arrangement keeps that locality.

upsert_inner

Transforms a stream of upserts (key-value updates) into a differential collection.

upsert_value_byte_len 🔒

Heap-size estimate for an emitted UpsertValue, used to drive give_fueled yielding on the output edge.

Type Aliases

FeedbackChunk 🔒

One feedback-arrangement chunk: a sorted, consolidated run of updates, resident or spilled to the buffer pool. Both batcher chains and sealed spine batches are sequences of these, so the arrangement’s state pages out of RSS under the pool’s budget, and the drain reads it back through the bulk UnloadChunk surface: copy-out probes, no cursor borrows.

FeedbackSpine 🔒

The feedback arrangement’s trace: a spine of Rc-shared chunk batches.

FeedbackUpdate 🔒

One persist-feedback update: a key, its current value row, the time, and an additive count.

UpsertChunk 🔒

One stash chunk: a sorted, consolidated run of updates, resident or spilled to the buffer pool.

UpsertChunkBatcher 🔒

The chunked flavor’s stash: differential’s chunk merge batcher over ColumnChunks. Data is pushed in unsorted. The batcher maintains geometrically-sized sorted chains and consolidates via the UpsertDiff Semigroup automatically. Committed chunks spill their bodies to the process buffer pool (see mz_timely_util::columnar::chunk), so the not-yet-eligible backlog (the snapshot / persist-lag window) pages out of RSS instead of growing it.

UpsertChunker 🔒

The chunker that sorts and consolidates raw input into the Column chunks both stash batchers consume.

UpsertOutputHandle 🔒

The operator’s data-output handle. A fueled Vec builder so the drain can give_fueled each emitted update and yield to timely under large snapshot drains instead of monopolizing the worker.

UpsertPagedBatcher 🔒

The paged flavor’s stash: the paged columnar merge batcher, consolidating like UpsertChunkBatcher but storing each chain entry as a Column routed through the storage-owned pager, which pages cold chains out of RSS.

UpsertUpdate 🔒

One source-stash update: a key, its dataflow time, and the payload diff. O is the columnar order key projected from the source FromTime (see UpsertSourceTime).