Skip to content

SS-492 - storage: let PostgreSQL exports progress independently during a snapshot - #39479

Merged
martykulma merged 3 commits into
maz-pg-concurrent-replicationfrom
maz-pg-per-export-progress
Oct 14, 2026
Merged

martykulma merged 3 commits into
maz-pg-concurrent-replicationfrom
maz-pg-per-export-progress

Conversation

@martykulma

@martykulma martykulma commented Oct 2, 2026 •

Copy link
Copy Markdown
Contributor

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_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, so they advance with the replication stream.
  • A snapshotted export releases its snapshot capability as soon as its own copy finishes on each worker.
  • The replication operator downgrades every port without a pending rewind at each yield point.

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

@martykulma
martykulma added this pull request to stack #39480 October 2, 2026 11:39
@martykulma martykulma changed the title storage: let PostgreSQL exports progress independently during a snapshot SS-492 - storage: let PostgreSQL exports progress independently during a snapshot Oct 2, 2026
@linear-code

linear-code Bot commented Oct 2, 2026

Copy link
Copy Markdown

SS-492

@martykulma
martykulma force-pushed the maz-pg-per-export-progress branch from 8536d9c to 21d9143 Compare October 7, 2026 00:38
@martykulma
martykulma force-pushed the maz-pg-per-export-progress branch from 21d9143 to 6fc87ee Compare October 7, 2026 13:45
@martykulma
martykulma marked this pull request as ready for review October 7, 2026 14:20
@martykulma
martykulma requested a review from a team as a code owner October 7, 2026 14:20
@martykulma
martykulma force-pushed the maz-pg-per-export-progress branch from 6fc87ee to b465c30 Compare October 7, 2026 17:01
@martykulma
martykulma force-pushed the maz-pg-per-export-progress branch 2 times, most recently from c39f0c9 to 7e6aab2 Compare October 9, 2026 02:37
@martykulma
martykulma force-pushed the maz-pg-per-export-progress branch 3 times, most recently from 721d91c to c1edc45 Compare October 9, 2026 14:51

@peterdukelarsen peterdukelarsen left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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();

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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();
}

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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).
}

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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);

@peterdukelarsen peterdukelarsen Oct 13, 2026 •

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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);

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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).

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Looks great to me now, thank you!!

@martykulma martykulma left a comment

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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();
}

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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);

martykulma and others added 3 commits October 14, 2026 11:06
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]>
@martykulma
martykulma force-pushed the maz-pg-per-export-progress branch from 5457141 to 0694adc Compare October 14, 2026 15:06
@martykulma
martykulma merged commit 653e079 into main Oct 14, 2026
83 checks passed
@martykulma
martykulma deleted the maz-pg-per-export-progress branch October 14, 2026 16:40
martykulma added a commit that referenced this pull request Oct 14, 2026
…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]>
@bosconi bosconi mentioned this pull request Oct 16, 2026
maheshwarip added a commit that referenced this pull request Oct 23, 2026
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]>
martykulma added a commit that referenced this pull request Nov 5, 2026
…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]>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants