Repository navigation
Conversation
Design notes for an opt-in subscribe mode whose output is a sequence of consolidated, unordered batches delimited by progress, with large batches carried through the peek stash. The document guides the implementation stack and is not part of it. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
|
|
||
| One worker mints batch descriptions `(lower, upper)` and broadcasts them, so every worker writes its share under the same bounds. The shape follows the materialized view sink's `mint` operator (`src/compute/src/sink/materialized_view.rs:520`), which runs on one worker and reaches the others through timely's `broadcast()`. | ||
|
|
||
| * The minting worker is worker 0. The controller sends `AllowCompaction` for a write-only collection to shard 0 only (`PartitionedComputeState::split_command`, `src/compute-client/src/service.rs:371`), which reaches worker 0 of process 0. Running mint there lets it read the feedback described below without forwarding it. |
There was a problem hiding this comment.
AllowCompaction is broadcasted to all workers, so any selected worker could mint descriptions.
There was a problem hiding this comment.
Right, worker 0 receives it and broadcasts it over the command channel (compute/src/server.rs:558). I've removed the worker-0 reasoning in f8b25e7. Minting now runs on the worker picked by hashing the sink id, as in the MV sink.
Posted by Claude Code.
|
|
||
| One worker mints batch descriptions `(lower, upper)` and broadcasts them, so every worker writes its share under the same bounds. The shape follows the materialized view sink's `mint` operator (`src/compute/src/sink/materialized_view.rs:520`), which runs on one worker and reaches the others through timely's `broadcast()`. | ||
|
|
||
| * The minting worker is worker 0. The controller sends `AllowCompaction` for a write-only collection to shard 0 only (`PartitionedComputeState::split_command`, `src/compute-client/src/service.rs:371`), which reaches worker 0 of process 0. Running mint there lets it read the feedback described below without forwarding it. |
There was a problem hiding this comment.
I don't understand why AllocCompaction has anything to do with minting. Compaction moves the since forward, which in this case is really something the reader would do. The writer shouldn't hold it back at all.
There was a problem hiding this comment.
Agreed, and removed. Feedback of the accepted upper is now its own compute command, kept in the command history so a replayed dataflow learns it. The doc also states that compaction is the reader's progress, not the writer's.
Posted by Claude Code.
| ### Replica: minting | ||
|
|
||
| One worker mints batch descriptions `(lower, upper)` and broadcasts them, so every worker writes its share under the same bounds. The shape follows the materialized view sink's `mint` operator (`src/compute/src/sink/materialized_view.rs:520`), which runs on one worker and reaches the others through timely's `broadcast()`. | ||
|
|
||
| * The minting worker is worker 0. The controller sends `AllowCompaction` for a write-only collection to shard 0 only (`PartitionedComputeState::split_command`, `src/compute-client/src/service.rs:371`), which reaches worker 0 of process 0. Running mint there lets it read the feedback described below without forwarding it. | ||
| * `lower` starts at the as-of. Each description's `lower` is the previous `upper`, advanced to the accepted upper when feedback is ahead of it. | ||
| * `upper` is the meet of the consolidated ok and error frontiers at minting time. Mint emits a description only when `upper` is beyond `lower`. | ||
| * Mint emits at most one description per frontier advance and keeps `lower` strictly increasing. Receivers skip a description whose `lower` is not the next expected one, as `write::absorb_batch_description` does in the materialized view sink (`materialized_view.rs:1146`). | ||
|
|
||
| ### Replica: writing | ||
|
|
||
| Each worker buffers its consolidated updates and, for each description it receives, extracts the updates with times in `[lower, upper)` once its own frontier has passed `upper`. | ||
|
|
||
| * A share whose encoded size is below a threshold travels inline as an `UpdateCollection`. | ||
| * A larger share goes through a generalized `StashUpload` (`src/compute/src/compute_state/peek_stash.rs`). The upload takes a shard id derived from `stash_id`, the description's bounds, and real update times, and accepts negative diffs. Writing happens in an async operator, since it awaits persist. | ||
| * Every worker emits a share for every description, including an empty one, because the partitioned merge counts shares to detect completion. | ||
| * An error in the error stream poisons the subscribe: the share carries the error, and later shares repeat it, matching today's sink. | ||
| * When the replayed as-of is beyond the original as-of, the sink renders without a snapshot. `rehydrate_replica` replays the command history with the as-of advanced to the allowed compaction, and the existing sink keeps `with_snapshot` set and resends the full state. | ||
| * The writer applies `up_to` the way today's sink does (`subscribe.rs:115`) and reports completion with an empty upper. |
There was a problem hiding this comment.
I'm not sure this interpretation is correct. What should happen is deterministic on each replica, and essentially is writing down batches as time advances. The batches have descriptions that in sequence cover time progressively without gaps, i.e., they advance upper to upper', filling the time between with data. The lower of the new batch needs to have the upper of the previous batch. Two replicas can disagree about how far they push upper, and a write gets filtered somewhere that is less than upper. I think we get this guarantee by changing lower <= U < upper to lower = U < upper below.
There was a problem hiding this comment.
Adopted lower = U < upper, and the read filter is gone. The new "Protocol and guarantees" section writes this up as obligations R1-R4 and C1-C3.
With strict equality, liveness needs two more things. Replicas have to learn U and rewind to it when their own boundaries straddle it. They also have to keep data from the last U they learned, rather than from their own last upper. Without the rewind, a replica cutting at 7, 12, ... never matches a U of 10, and the subscribe stalls once the replica that set 10 fails.
Cutover.lean checks this in Veil: 135 checks, plus four mutants that each fail. One accepts straddling batches, one retains from the replica's own upper, one has no rewind, and one lets feedback run ahead of U.
Posted by Claude Code.
| Each worker buffers its consolidated updates and, for each description it receives, extracts the updates with times in `[lower, upper)` once its own frontier has passed `upper`. | ||
|
|
||
| * A share whose encoded size is below a threshold travels inline as an `UpdateCollection`. | ||
| * A larger share goes through a generalized `StashUpload` (`src/compute/src/compute_state/peek_stash.rs`). The upload takes a shard id derived from `stash_id`, the description's bounds, and real update times, and accepts negative diffs. Writing happens in an async operator, since it awaits persist. |
There was a problem hiding this comment.
Note that we likely need the shared batch abstraction here, too, to avoid writing too many batches.
There was a problem hiding this comment.
Added. Workers of a process coalesce stashed parts for one description through the SharedBatches core, keyed by the description. Every worker still reports a share for every description, because the merge counts shares to detect completion.
Posted by Claude Code.
| * When the replayed as-of is beyond the original as-of, the sink renders without a snapshot. `rehydrate_replica` replays the command history with the as-of advanced to the allowed compaction, and the existing sink keeps `with_snapshot` set and resends the full state. | ||
| * The writer applies `up_to` the way today's sink does (`subscribe.rs:115`) and reports completion with an empty upper. | ||
|
|
||
| ### Compute protocol |
There was a problem hiding this comment.
What we're missing here is how much of the peek infrastructure we could re-use here. The intention of the new peek infra was that it could be re-used for subscribes too, so we should either use it, or figure out why we can't.
There was a problem hiding this comment.
The sink now holds arrangement batches (ArcBatch, which is Send + Sync) instead of buffering rows. It writes each description with a range scan built from PeekResultIterator::from_cursor over a CursorList of the batches that overlap it. The scan runs off the worker through the offload driver and PeekPermits, and writes through a generalized StashUpload.
The new "Peek infrastructure reuse" table lists what changes. Mainly the time predicate becomes per description and per time, and retractions are kept. It also lists what doesn't transfer: the answer path and the finishing.
Posted by Claude Code.
State the protocol as obligations on replicas and the controller: contiguous descriptions per replica, acceptance iff a batch's lower equals the accepted upper, feedback of that upper to replicas, retention from the last learned upper, and rewinding on feedback. A Veil model checks that the forwarded batches form one gapless chain with the right content, and mutants confirm which obligations are load-bearing. Replace row buffering in the sink with held arrangement batches and a range scan that reuses the peek scan, offload driver, stash, and shared batches. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
The stashed read needs a time-preserving persist read, because read_batches_consolidated uses the snapshot filter. CursorList::map_times visits a position's updates across held batches, so the range scan consolidates per position. On the arranged path the query MFP travels in the sink connection, since lowering owns from_arrangement and from_desc is the schema of the indexed collection. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Drop feedback from the controller to replicas. Replicas send contiguous batches and keep no state after sending. The controller processes each replica's batches in order, drops batches that end at or before its accepted upper, and accepts a straddling batch trimmed to start there. The Veil model checks the new protocol, and mutants confirm that strict lower equality stalls, an untrimmed batch overlaps, out-of-order processing breaks content, and rehydrating beyond the accepted upper leaves a gap. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Render every replica with the snapshot at its as-of and let the controller's trim remove the accumulation where it is not wanted, which fixes a lost snapshot after a restart before the first accept. State the obligations the protocol relied on implicitly: a replica that loses batches in flight re-renders, which needs the new sink variant in subscribe_ids, and it reports no Frontiers. Add the per-worker frontier wait, replace description skipping with an error, consolidate shares after the MFP, and specify the merge's completion and DroppedAt rules, blob deletion at every hop, and the trim of unordered shares. The model now checks the read-hold argument instead of assuming it, and models lost batches in flight. All mutants fail with counterexamples. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Small subscribe shares travel inline through the compute protocol, so they need no coalescing. Reconciliation must replace the batched subscribe response queue, or batches queued before a reconnect start the re-rendered chain. A new mutant of the cutover model confirms it. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
… bound Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Design notes for an opt-in subscribe mode whose output is a sequence of consolidated, unordered batches, each closed by a progress row. Large batches travel through the peek stash. Any replica can serve any batch, so batch boundaries are where one replica takes over from another.
This document guides the implementation and stays a draft. The implementation stack branches from
mainon its own and does not include it.Related: CPU-318, #38468.
🤖 Generated with Claude Code