adapter: serve SUBSCRIBE from the persist shard without a dataflow

[jubrad]

Motivation

A SUBSCRIBE on a table, source, or materialized view installs a dataflow on a cluster whose only job is to read that collection's persist shard and ship it back. Thousands of browser or distributed clients reconnecting after a network blip therefore create thousands of dataflows, each with its own persist reader, and the cluster spends its time installing them rather than serving anyone. Materialized views already have a shard in persist with a stream pushing to it; a subscribe on one should be able to read that directly.

This is a proof of concept, behind enable_subscribe_persist_fast_path, which defaults off in production and on in the test configuration. It is meant to be measured and argued about, not shipped as is.

What it does

A subscribe whose source is a bare reference to an existing storage collection is served by reading the shard from environmentd. A query, a view, an index, or a log source builds the same dataflow as before.

• StorageCollections::subscribe(id, as_of, with_snapshot, max_buffered_bytes) returns the collection's snapshot at the subscribe's timestamp followed by every later update and strictly advancing progress. It hides persist and txn-wal behind one seam: for txn-wal backed tables the shard's physical upper only moves when a write is applied, so the stream combines a persist listen with the txns shard's remap entries the way txns_progress does, driven from a task instead of a dataflow.
• A SharedTail per collection owns one listen reader and one snapshot reader and fans decoded batches out to per-subscriber queues. Registering a persist reader is a compare-and-set on the shard's state, so a storm of private readers serializes on it; sharing also decodes each part once instead of once per subscriber. The tail retains a short window of past batches so a subscriber whose as_of trails the shared listen still gets a complete stream.
• Formatting moves out of ActiveSubscribe into a SubscribeFormatter so it can run where the rows are needed. For this path that is the session: PersistTailStream is the row stream the session polls during FETCH, and it reads, batches, and formats only when polled. The coordinator's loop is off the data path.
• The snapshot is read through persist's consolidating cursor and handed out in pieces as the client fetches, so a large snapshot is never held whole and FETCH 10 does not wait for it. Persist merges the shard's runs in key order, so each piece is final and can ship without the rest. An output that orders or groups a timestamp, meaning WITHIN TIMESTAMP ORDER BY and the envelopes, still buffers the whole timestamp.
• A client that stops fetching costs only its queue of decoded events, counted against the existing subscribe_max_buffered_bytes budget. Over budget the tail drops the queue rather than the client, which resumes below its delivered frontier from the retained window or from the shard itself. The collection's RETAIN HISTORY is therefore what bounds how far behind a client may fall. Nothing new holds back compaction.

What is unchanged for clients

Same statements, same rows, same progress messages, same cancellation, dependency drops, mz_subscriptions, and statement logging. SUBSCRIBE (SELECT ...) is untouched. A materialized view without RETAIN HISTORY behaves as today, is more tolerant of short stalls, and produces the same error and SQLSTATE for long ones.

Measurements

Local bin/environmentd --optimized on a laptop, one quickstart replica, a materialized view over 10,000 rows of 64 bytes, N clients connecting at once over eight load-generator processes. Median of two runs, no errors at any size.

Clients	Path	Stmt to snapshot p50 / p95	Storm	CPU s, envd + clusterd
256	dataflow	2.21 s / 3.10 s	3.23 s	2.5 + 1.8
256	persist	0.56 s / 0.63 s	0.69 s	4.5 + 0.1
1024	dataflow	12.8 s / 18.9 s	19.5 s	10.5 + 14.0
1024	persist	1.50 s / 2.03 s	3.84 s	18.7 + 0.6
2048	dataflow	30.3 s / 45.9 s	47.9 s	23.6 + 45.1
2048	persist	3.49 s / 4.19 s	5.81 s	39.1 + 1.6

Total CPU falls 41% at 2048 clients and the cluster does nothing, but environmentd rises, because decode and consolidate move there and are still paid per subscriber. Sharing that work between subscribers at the same timestamp is the obvious next step and is listed in the design doc.

For one client on a materialized view of 1,000,000 rows, with one subscribe run first on each path so neither pays for the other's cold cache:

One client, 1,000,000 rows	Dataflow	Persist
FETCH 10 from a new cursor	0.97 s	0.22 s
First row over COPY	1.00 s	0.10 s
Whole snapshot delivered	2.50 s	1.59 s

Tests

• test/testdrive/subscribe-persist-fast-path.td runs tables, materialized views, envelopes, WITHIN TIMESTAMP ORDER BY, SNAPSHOT false, AS OF, UP TO, a dropped dependency, a stalled client that resumes, a snapshot spanning several chunks, and a snapshot needing consolidation, then repeats the statements with the flag off and expects identical output.
• test/testdrive/github-6335.td pins the flag off, since it asserts on subscribe dataflows.
• src/adapter/src/coord/persist_tail.rs carries unit tests for the batcher's frontier, snapshot, UP TO and error semantics, for chunked snapshot delivery, and for the stream's detach, resume, and termination paths.
• The feature benchmark gains SubscribeFirstFetchLargeSnapshot, and the parallel subscribe scenarios gain dataflow-pinned counterparts. Those counterparts no longer inherit from the scenarios they mirror, which had made the originals non-leaf and silently dropped them from the run.

Reviewing

This is larger than one reviewable change and is a draft for that reason. It splits along these lines if that is easier: the persist and txn-wal visibility changes; the storage seam and SharedTail; the formatter extraction out of ActiveSubscribe; the sequencing, flag, and PersistTailStream; chunked snapshot delivery; and the tests and benchmark scenarios.

Release notes

None. The change is behind a feature flag that defaults off, and no user-visible behavior changes with it off.

🤖 Generated with Claude Code