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. Extract & drain. Extract from the batcher just past the persist frontier (capped at the input frontier, see seal_upper) 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 (above the persist frontier, below the extract upper): 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 time-major 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.

Modules

drain 🔒

Draining sealed stash chunks against the persist feedback, one function per stash flavor.

feedback 🔒

The persist-feedback side of the upsert-v2 operator: the feedback arrangement’s types, and the UpsertValue to Row encoding it stores.

stash 🔒

The upsert-v2 source stash: the consolidating diff, the stash batcher types, and the per-flavor UpsertStashArms.

Enums

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.

Functions

build_upsert_operator 🔒

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

upsert_inner

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

Type Aliases

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.