Repository navigation
SS-492 - storage: let PostgreSQL exports progress independently during a snapshot - #39479
Conversation
8536d9c to
21d9143
Compare
21d9143 to
6fc87ee
Compare
6fc87ee to
b465c30
Compare
c39f0c9 to
7e6aab2
Compare
721d91c to
c1edc45
Compare
peterdukelarsen
left a comment
There was a problem hiding this comment.
Looks good to me overall, some more nitty small requests that don't change the logic.
| let mut data_outputs = SharedFuel::new(data_outputs); | ||
| let concurrent_replication = !STORAGE_PERSIST_SINK_DESCRIPTION_LOOKAHEAD | ||
| .get(config.config.config_set()) | ||
| .is_zero(); |
There was a problem hiding this comment.
Is there a risk of a race here where one part of the full rendered source saw a different value for this flag? Wondering if we should pull the config value up higher and wire down the flag lower.
There was a problem hiding this comment.
this is addressed, moved this out of the async operator
| let mut builder = AsyncOperatorBuilder::new(op_name, scope.clone()); | ||
|
|
||
| // Read at render time, matching the value the persist sink reads in the same dataflow build. | ||
| // A read in the async body can observe an `UpdateConfiguration` sequenced after this dataflow's |
There was a problem hiding this comment.
nit: stylistically I'd pref just wiring down a shared boolean if it's less than lets say three layers of functions between the nearest common ancestor. I buy that this is correct, but it feels complex.
There was a problem hiding this comment.
Fair. I avoid adding parameters when I can derive from the existing inputs cheaply. The shared boolean pushes render to 9 parameters, and it sets a pattern for future flags that just keeps growing that input list.
I'm down with refactoring this down the road since it definitely seems like we can group some things together.
| .await; | ||
| if concurrent_replication { | ||
| data_cap_sets[output_index] = CapabilitySet::new(); | ||
| } |
There was a problem hiding this comment.
Is there a way we can collapse these repetitive pre-continue data_cap_sets[output_index] = CapabilitySet::new();? One Idea I have would be moving the piece inside the for (&output_index, info) in outputs.iter() { into a function and then wrapping that in another function that looks structurally like:
fn handle_output_wrapping() {
handle_output_existing();
if concurrent_replication {
data_cap_sets[output_index] = CapabilitySet::new();
}
}
fn handle_output_existing() {
// all existing logic (replace continues with returns).
}
There was a problem hiding this comment.
If I understand correctly, wouldn't handle_output_existing be different in each case? Wouldn't we have to break them all up in to function (or pass them as FnMut)?
Replacing:
if concurrent_replication {
data_cap_sets[output_index] = CapabilitySet::new();
}
with this is simpler to read, but not by much:
drop_caps_if_concurrent(data_cap_sets, output_index);
There was a problem hiding this comment.
I'm not sure we're on the same page. I was recommending putting the entirety of the existing code in the for loop into a function. Then slotting in the new code for capability dropping in a wrapping function.
Definitely wasn't trying to recommend something so narrow as changing:
if concurrent_replication {
data_cap_sets[output_index] = CapabilitySet::new();
}
to:
drop_caps_if_concurrent(data_cap_sets, output_index);
There was a problem hiding this comment.
ah, gotcha! hmm.. there's probably close to 10 local variables that then need to be passed into the function, and that's a mess on its own.
An alternative is to do an async block, but this is a little weird because you have to return a result from it.
let copied = async {
if let Err(err) = verify_schema(oid, info, &upstream_info) {
// give the error
return Ok(());
}
let Some(ctid_range) = worker_ctid_range(&config, block_count, oid) else {
return Ok(());
};
// COPY
Ok::<_, TransientError>(())
}.await;
copied?;
if concurrent_replication {
data_cap_sets[output_index] = CapabilitySet::new();
}
That's worth it if we're planning to add more exit paths, but honestly, I don't think that's the case.
There was a problem hiding this comment.
I tried to rewrite this morning (11 parameters as is). A few of them were derived from existing structs, so I adjusted. Should be ok now (at least no clippy exception).
There was a problem hiding this comment.
Looks great to me now, thank you!!
martykulma
left a comment
There was a problem hiding this comment.
thanks for the review!
| let mut builder = AsyncOperatorBuilder::new(op_name, scope.clone()); | ||
|
|
||
| // Read at render time, matching the value the persist sink reads in the same dataflow build. | ||
| // A read in the async body can observe an `UpdateConfiguration` sequenced after this dataflow's |
There was a problem hiding this comment.
Fair. I avoid adding parameters when I can derive from the existing inputs cheaply. The shared boolean pushes render to 9 parameters, and it sets a pattern for future flags that just keeps growing that input list.
I'm down with refactoring this down the road since it definitely seems like we can group some things together.
| .await; | ||
| if concurrent_replication { | ||
| data_cap_sets[output_index] = CapabilitySet::new(); | ||
| } |
There was a problem hiding this comment.
If I understand correctly, wouldn't handle_output_existing be different in each case? Wouldn't we have to break them all up in to function (or pass them as FnMut)?
Replacing:
if concurrent_replication {
data_cap_sets[output_index] = CapabilitySet::new();
}
with this is simpler to read, but not by much:
drop_caps_if_concurrent(data_cap_sets, output_index);
With `storage_persist_sink_description_lookahead` > 0, each export's port holds the minimum capability only while that export has snapshot data or a pending rewind. Exports that are not snapshotting advance with the replication stream, and a snapshotted export is released as soon as its own copy finishes. Co-Authored-By: Claude <[email protected]>
… out of async operator
5457141 to
0694adc
Compare
…pshot (#39419) ### Motivation Today a PostgreSQL source snapshots all of its tables before it reads any of the replication stream. This PR lets the replication operator read the stream while the snapshot runs. #39479 then lets each export's frontier advance on its own. ### Description All new behavior is keyed off `storage_persist_sink_description_lookahead` being nonzero. That dyncfg already controls whether the persist sink groups a snapshotting export's updates into one batch, and it defaults to zero in production and to 5s in mzcompose. Concurrent replication is what makes a snapshot accumulate many timestamps, and reading the stream concurrently without the grouping keeps one batch builder per timestamp resident until each reaches the blob target size. The two are therefore one setting rather than a separate flag. With a nonzero lookahead, the snapshot operator emits its rewind requests as soon as the snapshot LSN is known, before any table is copied. The replication operator then reads the stream while tables are copied. Every export still advances together, so replication data is staged in the dataflow until the snapshot completes. This trades cluster memory during hydration for progress on the replication stream. With the lookahead at zero, rewinds are emitted after the copy completes, as before. The snapshot operator reads the setting once, since the early and the late emission must not both happen or both be skipped. Only the snapshot operator reads it in this PR, so a change takes effect for snapshots that start afterwards. parallel-workload already rolls the lookahead through zero and nonzero values. The persist sink logs the resolved lookahead per export at dataflow build time, so a hydration's logs show whether grouping and concurrent replication are on. ### Verification The mzcompose default sets the lookahead to 5s, so the existing mzcompose-based PostgreSQL source tests exercise the early rewind path. #39420 adds targeted coverage. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude <[email protected]>
Release notes for the **v26.47.0** train. One commit was added per snapshot; the `final` snapshot sets the real dates. **All six snapshots (`rc.1`–`rc.5`, `final`) are now applied — this branch is complete and ready for review.** - **Content**: 2 Feature sections (`TRY_CAST`, `EXCLUDE ALL NOT NULL CONSTRAINTS`), 7 Improvements, 13 Bug Fixes — `rc.1` contributed all of it except the top four Bug Fixes: `rc.2` added two, and `rc.3` and `rc.4` one each. `rc.5` added no new entries; it amended `rc.4`'s probe-port Bug Fix instead (see below). `final` adds nothing at all — its tracking issue holds only the unassociated `release: bump to version v26.47.0` commit ([`bed23494d`](bed2349)), and the `v26.47.0-rc.5` and `v26.47.0` tags are 14h28m apart with that bump the only thing between them. Its commit is empty and exists to record on the branch that the snapshot was applied. - **Dates (final, no longer provisional)**: Cloud **2026-10-22**, Self-Managed **2026-10-23**. Every RC estimated this same pair from the `v26.47.0-rc.1` tag, and the real dates confirmed it — the five RCs compressed into one week rather than pushing the train out — so no published date text changed in the `final` commit.⚠️ One thing worth a last look before merge: **2026-10-22 is a Wednesday**, one day off the usual Thursday Cloud date (2026-10-23 is the normal Friday Self-Managed day). That is the supplied value and it stands, but if the release owner expected Thursday 2026-10-23 / Friday 2026-10-24, this branch is where to correct it. - **Operator compatibility**: row added for `v26.47`, release date `2026-10-23`. - **Branch base**: `main`, whose newest section is `## v26.45.1` — v26.46.0's section is still on its own unmerged branch (#39690). Expect a top-of-list conflict when both merge; `v26.47.0` goes above `v26.46.0`. ✅ **Corrections already on this branch** — a `final` verification pass over the whole assembled section confirmed these and deliberately left them alone: - **`rc.1` named a syntax that does not exist.** It published "`CREATE TABLE ... FROM SOURCE` now accepts an `ALL COLUMNS NULLABLE` option", taking the name from [#39680](#39680 title and body. At the `v26.47.0` tag the option is `EXCLUDE ALL NOT NULL CONSTRAINTS` (`TableFromSourceOptionName::ExcludeAllNotNullConstraints`), and `NULLABLE` is not a lexer keyword at all, so the published syntax would have been a parse error for anyone who copied it. A human editor caught this and replaced the bullet with the `{#v26.47-exclude-not-null-constraints}` feature section now on the branch. The `rc.1` snapshot directory in mz-skills still carries the wrong wording — snapshot directories are the record of what was reviewed at the time and are left untouched. - **`rc.5`'s cross-snapshot rewrite is applied once**, including the "Deployments using password, SASL, or OIDC authentication have no separate internal SQL port and continue to be probed on the external one" clause. - **PR citations are stripped from all entries** as a deliberate house-style edit, matching how published sections read. PR links remain available in the mz-skills snapshot drafts linked below.⚠️ **Open questions for the release owner** — none of these were resolved by `final`; this branch is the last place to act on them - **Was any `rc.4` artifact handed to a customer or used for an upgrade rehearsal?** This decides a published entry. [#40036](#40036) (published as a Bug Fix by `rc.4`) made the operator probe `environmentd`'s internal SQL port unconditionally, which crashloops deployments using Password, SASL, or OIDC authentication — they have no internal SQL port. [#40046](#40046) fixed that roughly seven hours later, inside this same unreleased train, so no shipped build (`v26.46.1` probes the external port unconditionally) could have hit it. It is therefore recorded as **Borderline** rather than published as a Bug Fix of its own, and its substance folded into `rc.4`'s bullet as a scoping clause. If an `rc.4` build did reach anyone, the crashloop *was* reachable and [#40046](#40046) should be published in its own right. Note the author wrote an explicit positive release-note line for it, which is why this is flagged rather than quietly dropped. - **GA announcements — the largest judgment call in this train.** Three docs PRs remove preview markers: [#39831](#39831) (source versioning, `CREATE TABLE ... FROM SOURCE`), [#39836](#39836) (replacement materialized views), [#39830](#39830) (Kafka sink headers, noted as "GA in v26.47"). None changes code, so all three were held; drafted wording for each is in `omitted.md`. Which are actually GA in v26.47.0 and should be announced? Note v26.46.0 published Kafka sink message attribution as *private preview* — if the header goes GA here, this release should say so rather than repeat the preview framing. - **GCS blob read throughput** ([#39906](#39906), held in `rc.2`): restores the read throughput GCS deployments lost when [#39351](#39351) — already shipped — forced HTTP/2 and collapsed per-request connections into one. It is selected by `persist_blob_http_pool`, verified to default to `off` at the `v26.47.0-rc.2` tag, and the author notes "the default should eventually be 'GCS-only'". **Is `persist_blob_http_pool` being set to `gcs` for v26.47.0?** If so this is a clear win to publish for GCP users, who carry the regression today; drafted wording is in rc.2's `omitted.md`. - **`SUBSCRIBE` soft assertions** ([#37611](#37611), [#39764](#39764), both held): drafted as Bug Fixes, then pulled back. Both failures come from `soft_assert_or_log!`, which panics only where soft assertions are enabled and otherwise logs; neither PR says what a release build does. Are soft assertions on in the environments customers run? If yes, both are publishable as drafted in `omitted.md`. - **Console Freshness page** ([#39340](#39340), held): a complete new Monitoring > Freshness page, gated behind `freshness-page-CNS164`, off in both Cloud and self-managed, and the LaunchDarkly flag is not yet created. If it is turned on for v26.47.0, this likely deserves a Feature heading. - **PostgreSQL sources ingesting during a snapshot** ([#39419](#39419), [#39479](#39479), held): adding a large table to an existing source would no longer stall the tables already hydrated — but `storage_persist_sink_description_lookahead` is `Duration::ZERO` at the tag. Publish if the lookahead is being enabled. - **Coalesced materialized-view sink batches** ([#37591](#37591), held): measures ~67% fewer blob writes and ~54% smaller state diffs, at ~10 ms added p50 latency on small writes, but `compute_mv_sink_shared_batch_max_bytes` is `0` at the tag. Publish (with the latency cost stated) if the staged rollout reaches customers in this release. - **MCP 2026-07-28 conformance checks** ([#39694](#39694), [#39695](#39695), held): `enable_mcp_protocol_2026_07_28` is `false` at the tag. If the revision is being served, these belong together as one Improvement. - **Refresh strategy columns** ([#39597](#39597), published as an Improvement): this renames `aligned_to` → `aligned_to_expression` and `at` → `at_expression` and changes their type. It is a breaking change for anyone selecting them, in a schema documented as unstable — it may warrant more prominence than a bullet. - **Retained input after `ALTER SINK ... SET FROM`** (from `rc.3`'s [#37653](#37653)): the PR records a pre-existing gap it does not fix — after `SET FROM`, the *old* upstream relation stays held, and therefore un-compactable, until the sink is dropped or `environmentd` restarts. It is not something this release changed, so it is out of these notes, but it is a real footgun for the blue/green pattern that `sql/alter-sink.md` recommends `SET FROM` for. Should the `ALTER SINK` page or the blue/green guide call it out? - **Operator upgrade shows pending changes** (from `rc.4`'s [#40036](#40036)): the probe-port change alters the generated `environmentd` StatefulSet, so after upgrading the operator, existing environments report pending changes and pick up the new port on their next rollout — they are **not** restarted on their own. That is an upgrade consequence rather than something the release changed, so it is out of the notes, but a Self-Managed operator who cannot account for the pending-changes state may read it as a problem. The v26.47 upgrade guide looks like the right venue. Snapshot drafts (with PR links) in mz-skills: - rc.1: [release_notes.md](https://github.com/MaterializeInc/mz-skills/blob/main/data/mz-release-notes/v26.47.0/rc.1/release_notes.md) / [omitted.md](https://github.com/MaterializeInc/mz-skills/blob/main/data/mz-release-notes/v26.47.0/rc.1/omitted.md) - rc.2: [release_notes.md](https://github.com/MaterializeInc/mz-skills/blob/main/data/mz-release-notes/v26.47.0/rc.2/release_notes.md) / [omitted.md](https://github.com/MaterializeInc/mz-skills/blob/main/data/mz-release-notes/v26.47.0/rc.2/omitted.md) - rc.3: [release_notes.md](https://github.com/MaterializeInc/mz-skills/blob/main/data/mz-release-notes/v26.47.0/rc.3/release_notes.md) / [omitted.md](https://github.com/MaterializeInc/mz-skills/blob/main/data/mz-release-notes/v26.47.0/rc.3/omitted.md) - rc.4: [release_notes.md](https://github.com/MaterializeInc/mz-skills/blob/main/data/mz-release-notes/v26.47.0/rc.4/release_notes.md) / [omitted.md](https://github.com/MaterializeInc/mz-skills/blob/main/data/mz-release-notes/v26.47.0/rc.4/omitted.md) - rc.5: [release_notes.md](https://github.com/MaterializeInc/mz-skills/blob/main/data/mz-release-notes/v26.47.0/rc.5/release_notes.md) / [omitted.md](https://github.com/MaterializeInc/mz-skills/blob/main/data/mz-release-notes/v26.47.0/rc.5/omitted.md) - final: [release_notes.md](https://github.com/MaterializeInc/mz-skills/blob/main/data/mz-release-notes/v26.47.0/final/release_notes.md) / [omitted.md](https://github.com/MaterializeInc/mz-skills/blob/main/data/mz-release-notes/v26.47.0/final/omitted.md) **Borderline PRs omitted:** rc.1: 14 — [omitted.md#borderline](https://github.com/MaterializeInc/mz-skills/blob/main/data/mz-release-notes/v26.47.0/rc.1/omitted.md#borderline), rc.2: 2 — [omitted.md#borderline](https://github.com/MaterializeInc/mz-skills/blob/main/data/mz-release-notes/v26.47.0/rc.2/omitted.md#borderline), rc.3: 0 — [omitted.md#borderline](https://github.com/MaterializeInc/mz-skills/blob/main/data/mz-release-notes/v26.47.0/rc.3/omitted.md#borderline), rc.4: 0 — [omitted.md#borderline](https://github.com/MaterializeInc/mz-skills/blob/main/data/mz-release-notes/v26.47.0/rc.4/omitted.md#borderline), rc.5: 1 — [omitted.md#borderline](https://github.com/MaterializeInc/mz-skills/blob/main/data/mz-release-notes/v26.47.0/rc.5/omitted.md#borderline), final: 0 — [omitted.md#borderline](https://github.com/MaterializeInc/mz-skills/blob/main/data/mz-release-notes/v26.47.0/final/omitted.md#borderline) 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude <[email protected]> Co-authored-by: Valerie Chiang <[email protected]> Co-authored-by: Pranshu Maheshwari <[email protected]>
…napshot (#40207) ### Motivation Every export's frontier stays at the minimum until the last table's snapshot read finishes. Adding one large table to a MySQL source therefore still stalls ingestion for every table already hydrated in that source. #39479 fixed the same problem for PostgreSQL. ### Description With `storage_persist_sink_description_lookahead` nonzero, each export's port holds the minimum capability only while that export still has snapshot data to emit or a pending rewind. - Exports that are not being snapshotted drop their snapshot capabilities up front. They advance with the binlog. - After a worker plans its reads, it drops the ports of exports it reads no part of. This also covers exports that failed schema verification. - A snapshotted export releases its snapshot capability on each worker as soon as that worker finishes reading its table. - The replication operator downgrades every port without a pending rewind at each yield point. With the lookahead at zero, all ports advance together. Behavior then matches #40206. The replication operator now reads the setting at render time as well. Both operators read it in the same dataflow build and agree with the persist sink. ### Verification The mzcompose default sets the lookahead, so the existing mzcompose-based MySQL source tests exercise per-export progress. #40208 adds targeted coverage. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-authored-by: Claude <[email protected]>
Motivation
Every export's frontier stays at the minimum until the last table's COPY finishes. Adding one large table to a source therefore still stalls ingestion for every table already hydrated in that source.
Description
With
storage_persist_sink_description_lookaheadnonzero, each export's port holds the minimum capability only while that export still has snapshot data to emit or a pending rewind.With the lookahead at zero, all ports advance together, so behavior matches #39419.
The replication operator now reads the setting as well. The two operators read it independently at startup, and parallel-workload rolls the lookahead through zero and nonzero values, so a change can land between those two reads and leave the operators in different modes within one dataflow. That is benign in both directions. With only the snapshot operator concurrent, the replication operator holds every port until the rewinds resolve and then releases them all. With only the replication operator concurrent, it still waits for the rewind input to close, which the serial snapshot operator does after the copy.
Verification
The mzcompose default sets the lookahead, so the existing mzcompose-based PostgreSQL source tests exercise per-export progress. #39420 adds targeted coverage.
🤖 Generated with Claude Code