SS-456 - storage: give each PostgreSQL source export its own output port

[martykulma]

Motivation

The PostgreSQL snapshot and replication operators emit every export's updates on one output stream keyed by output index, with a single data capability. Because of that, no export's frontier can move independently of the others. Splitting the stream into one port per export is the groundwork for letting exports progress on their own.

Description

This PR has no functional change. Every export still observes the same frontier as before.

The snapshot and replication operators now create one data output port per source export, in output index order. Their cast steps run as one operator per export, and each operator returns one collection per export instead of a single collection keyed by output index. All ports are downgraded together, so the frontiers still move in lockstep.

give_fueled counts bytes per output handle, so with one handle per export an operator could emit up to the bound on every port before yielding. SharedFuel, added to mz_storage::source::types, owns an operator's output handles, charges each update's FuelSize against one budget, and yields once per MAX_OUTSTANDING_BYTES of their total emission. Both operators hand their data ports to a SharedFuel, so every data emission, errors included, is charged against the same budget. MAX_OUTSTANDING_BYTES becomes a module-level constant in mz_timely_util::builder_async so that give_fueled and SharedFuel share one bound.

The InvalidSnapshotLsn error path added in SS-463 (#39027) now emits on the affected export's own port.

Verification

The existing pg-cdc and testdrive suites cover the snapshot and replication paths, and their results should be unchanged.

🤖 Generated with Claude Code

[linear-code]

SS-456

[def-]

QA LLM Review

1. MEDIUM -- Per-export round-robin counters send concurrent exports' rows to the same worker

src/storage/src/source/postgres/replication.rs:635

The replication cast stage now builds a separate next_worker round-robin counter for each export, and every counter starts at worker 0. Exports that receive writes at similar rates therefore pick the same worker at the same moment, so the parallel casting that the round-robin was added for (88a1263, after an uneven-distribution OOM) collapses onto one worker at a time. The PR describes itself as having no functional change, but this changes how load spreads across workers.

Details

Before this PR there was one Exchange on the single data stream, with one global counter that moved to the next worker every 1000 records. Now each of the N exchanges routes the k-th record of its own export to (k / 1000) % W. When writes are spread evenly across tables, the counters stay aligned, so about N×1000 consecutive records go to the same worker before the next one takes over. The cast backlog, and the frontier lag it causes for every export, grows by roughly a factor of N. The same happens on every restart: during catch-up, the first 1000 records of every export are all cast on worker 0.

A simulation with W=16 and 2M uniformly distributed records measured the average share of each 20k-record window that lands on the busiest worker. Before this PR it is 10% for any N. After it, the share is 45% for N=10 tables, 93% for N=100 and 97% for N=500.

Sharing one counter across all N closures restores the old distribution. Exchange::new only needs the closure to be 'static, so an Rc works:

+    let peers = u64::cast_from(scope.peers());
+    let next_record = Rc::new(Cell::new(0u64));
     for (output_index, data_stream) in data_streams.into_iter().enumerate() {
         ...
-        let mut next_worker = (0..u64::cast_from(scope.peers()))
-            .flat_map(|w| std::iter::repeat_n(w, 1000))
-            .cycle();
-        let round_robin = Exchange::new(move |_| next_worker.next().unwrap());
+        let next_record = Rc::clone(&next_record);
+        let round_robin = Exchange::new(move |_| {
+            let n = next_record.get();
+            next_record.set(n.wrapping_add(1));
+            (n / 1000) % peers
+        });

[martykulma]

1. MEDIUM -- Per-export round-robin counters send concurrent exports' rows to the same worker

confirmed and addressed

[peterdukelarsen]

Looks very reasonable to me at a high level. I have a little bit of concern that this could change performance characteristics -- called out places that look potentially different inline.

[peterdukelarsen]

Any concerns with performance changes in timely due to a more complex dataflow graph and more capabilities to track/downgrade?

I think we touched on this before when talking through the scoping, but want to double-check it still feels reasonable to you. IIRC we had some top level number of operators timely/materialize can support (maybe 10s or 100s of thousands) before we have to worry.

[martykulma]

None that I'm aware of - see https://materializeinc.slack.com/archives/CMNT2QSBZ/p1786389233719329?thread_ts=1786389144.337239&cid=CMNT2QSBZ

[peterdukelarsen]

This is exactly the reference I was hoping for, thanks. Sounds like we have pretty strong conviction 10k+ is well-tolerated in Materialize now and are committed to scaling past.

[peterdukelarsen]

this feels somewhat duplicative of give_fueled. Wondering if this should be in timely-util and whether some aspect of this could be combined?

Broader question about give_fueled is why it previously had the 128 * 1024 * 1024 limit per output? I'm wondering if there's any efficiency we're losing for a given output by changing how much we accrue before sleeping?

[martykulma]

Indeed, it serves the same function, but it does the accounting across multiple exports. give_fueled covers the case where the CDC stream is constantly producing rows, such that awaits never return Poll::Pending (which would mean that the operator would never yield).

Setting it too small would cause excessive yields, but if we doubled it, I would be surprised if we could distinguish the difference.

[martykulma]

We could move it to timely-util, we'd either need to change it's signature to take the size (and remove D: FuelSizeO), or move SharedFuel and FuelSize trait (plus the impls) to timely-util.. but then things would need to pull in timely-util as a dep, and there are probably other knock on effects there.

I didn't like either of those compared to the current approach. If there's some reason that another consumer needs this, we can refactor if it makes sense.

[peterdukelarsen]

Setting it too small would cause excessive yields, but if we doubled it, I would be surprised if we could distinguish the difference.

Is there any concern about splitting the limit up across multiple exports though? Why do we normally set up give_fueled to be associated with one output rather than tracked as a count of data emitted in total inside the current operator?

[peterdukelarsen]

We discussed some concern that this will result in far more tiny containers if there are many exports. In our conversation we discussed this would only ever run on a single pod so it's probably not a likely bottleneck. Curious if you could help point me to where sources are limited to single-pod replicas.

[martykulma]

I can't! I remembered that incorrect - we can technically run on a multi-process replica. I would be surprised if anyone used it for OLTP. Parallel snapshotting from 124 workers would create a substantial load on the upstream, potentially disrupting production. And there is only a single replication stream reader, which seems like it's the more likely bottleneck.

I think this is slightly misleading. There is no guarantee that each timestamp will see 1000 updates. If the unary operator emits 5 rows at ts(10) and yields, the container will only have 5 rows (to the best of my knowledge). This optimization is much the like the 128MB above, we're trying to avoid some edge cases: don't send everything to one worker and don't send one row to each worker.

[peterdukelarsen]

Nothing blocking at this point