Conversation
Convert shard_source_descs and shard_source_fetch from AsyncOperatorBuilder to synchronous OperatorBuilderRc operators paired with tokio tasks that own the persist I/O (reader/snapshot/listen and batch fetching). This drops the persist source's dependence on the timely async bridge with no behavior change. * shard_source_descs runs a listen task on the chosen worker that sends parts (split into ExchangeableBatchPart + Lease) and progress over a channel; the operator downgrades capabilities and parks leases. The former shard_source_descs_return operator is merged in as a disconnected completed_fetches input, and the listen handle (the reader's SeqNo hold) is released via a oneshot once that frontier empties. * shard_source_fetch forwards descs to a fetch task and retains a per-flight capability pair; results are matched FIFO, emitted at the data capability, with the completed-fetches capability dropped to release the lease. On a missing blob it reports through the ErrorHandler and freezes, retaining capabilities (and crucially ceasing to drain results, so a later good result cannot advance the frontier past the missing part). * Both operators reproduce builder_async's two-phase shutdown via build_reschedule + the coordinated button, so a local-only press cannot advance the downstream frontier past times other workers still feed. Adds ErrorHandler::report_and_freeze, module documentation of the consumer contract and the operator/task architecture, and regression tests for end-to-end fetch, mid-stream shutdown, listing-path error freeze, and fetch-path error freeze. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
With the persist source's listen polling moved into a tokio task, a fully caught-up DataSubscribe with no listen retry timer produces no worker activations. The stress test's read workers parked indefinitely on step_or_park(None) while polling a oneshot, so they never observed the shutdown signal. Bound the parks so they re-check between steps. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
The lease-return input now lives on the shard_source_descs operator, so introspection names that operator instead. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
The fetch operator matched results to capabilities by arrival order: it pushed a capability pair per desc into a VecDeque and popped the front for each result. That was sound only because the fetch task is a strictly sequential loop, and it would silently misassign capabilities the moment the task fetched concurrently — releasing a part's completed-fetches capability (and its lease) for a part not actually fetched. Tag each desc with the time it was minted at, echo that time back with the result, and track outstanding fetches in a per-time BTreeMap of (data cap, completed cap, count). Emit each blob at its time's capability and drop both capabilities when a time's count reaches zero. This is independent of the order results return in, and it aligns with the descs-side LeaseManager, which is also keyed by time. As a bonus the freeze-on-error is now structural: the failed time's capability is never decremented, so the frontier holds there regardless of what else drains. Per review feedback on MaterializeInc#36910. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
The source fetch task fetched one part at a time, so hydrating a shard with many small parts was bound by the blob-store round-trip. Run up to persist_source_fetch_concurrency (default 1 = serial) fetches concurrently, each on a cheap per-call clone of the BatchFetcher (shares the schema cache). The fetch operator already keys capabilities by time, so it tolerates results completing in any order, including the missing-blob freeze. In-flight bytes stay bounded by the existing fetch semaphore. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
antiguru
force-pushed
the
persist-source-deasync-design
branch
from
June 19, 2026 13:57
ffa9df5 to
b55c0d2
Compare
antiguru
pushed a commit
to MaterializeInc/materialize
that referenced
this pull request
Jul 6, 2026
Hydrating a shard with many small parts is bound by the blob-store round-trip rather than bandwidth. Let `shard_source_fetch` issue several part fetches at once so the round-trips overlap. The fetch operator stays an async timely operator. Two changes make it safe to complete fetches out of order: * Key the operator's capability bookkeeping by the time a part was minted at rather than by arrival (FIFO) order. For each time we retain a data and a completed-fetches capability and count the outstanding fetches at that time; when the count reaches zero both capabilities drop, advancing the data frontier and releasing that time's leases. Results can now complete in any order without advancing the frontier past unproduced data. * Run up to `persist_source_fetch_concurrency` (default 1, i.e. the previous serial behavior) fetches at once via a `FuturesUnordered`, each on a cheap per-call `BatchFetcher` clone that shares the schema cache. In-flight bytes stay bounded by the existing fetch semaphore inside `fetch_leased_part`. On a missing blob the operator reports the error and freezes (retains all capabilities, stops draining results) exactly as before, so the frontier never advances past a part we failed to emit. New tests cover data fetching, mid-stream shutdown, the listing- and fetch-path freezes, and the concurrent fetch and concurrent-freeze paths. Recreates antiguru#181 on top of the async operators, omitting that PR's deasync refactor of the shard_source operators (and its dependent txn-wal and introspection-naming follow-ups). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01MR6SwYEgWVmMJUGn4ZmLMX
antiguru
pushed a commit
to MaterializeInc/materialize
that referenced
this pull request
Jul 6, 2026
Hydrating a shard with many small parts is bound by the blob-store round-trip rather than bandwidth. Let `shard_source_fetch` issue several part fetches at once so the round-trips overlap. The fetch operator stays an async timely operator. Two changes make it safe to complete fetches out of order: * Key the operator's capability bookkeeping by the time a part was minted at rather than by arrival (FIFO) order. For each time we retain a data and a completed-fetches capability and count the outstanding fetches at that time; when the count reaches zero both capabilities drop, advancing the data frontier and releasing that time's leases. Results can now complete in any order without advancing the frontier past unproduced data. * Run up to `persist_source_fetch_concurrency` (default 1, i.e. the previous serial behavior) fetches at once via a `FuturesUnordered`, each on a cheap per-call `BatchFetcher` clone that shares the schema cache. In-flight bytes stay bounded by the existing fetch semaphore inside `fetch_leased_part`. On a missing blob the operator reports the error and freezes (retains all capabilities, stops draining results) exactly as before, so the frontier never advances past a part we failed to emit. New tests cover data fetching, mid-stream shutdown, the listing- and fetch-path freezes, and the concurrent fetch and concurrent-freeze paths. Recreates antiguru#181 on top of the async operators, omitting that PR's deasync refactor of the shard_source operators (and its dependent txn-wal and introspection-naming follow-ups). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01MR6SwYEgWVmMJUGn4ZmLMX
antiguru
force-pushed
the
persist-source-deasync-design
branch
from
July 20, 2026 14:51
b55c0d2 to
51e5ca4
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Motivation
The source fetch task fetched one part at a time, so hydrating a shard with many small parts is bound by the blob-store round-trip rather than bandwidth.
Stacked on MaterializeInc#36910.
Description
Run up to
persist_source_fetch_concurrencypart fetches concurrently (default1= serial), each on a cheap per-call clone of theBatchFetcherthat shares the schema cache.The fetch operator already keys capabilities by time, so it tolerates results completing in any order, including the missing-blob freeze.
In-flight bytes stay bounded by the existing fetch semaphore.
Verification
Unit tests for fetch completeness under concurrency and for the missing-blob freeze holding when a middle batch is absent and later batches complete out of order.