[patrickwwbutler]

Complete the write loop. Each completed timestamp window is applied to
the target table in the transaction that staged it, and the new frontier
is recorded in materialize.sink_progress only while the row still
carries this incarnation's epoch. The persist progress shard is brought
up to that frontier after the commit, and reconciled forward at startup.

A keyed sink takes every key a window touches straight from its state
before the window to its state after it, with at most one INSERT, UPDATE
or DELETE per key, so a window holding several timestamps for one key
never puts the key in the target table twice. A keyless sink inserts and
retracts exact row multiplicities.

Tests: adds unit tests in src/storage/src/sink/postgres.rs, run against the database at POSTGRES_URL, for key replacement, one operation per key, rejected windows, keyless multiplicities and fencing. Also covered end to end by test/pg-sink at the top of this stack.

🤖 Generated with Claude Code

[def-]

QA LLM Review

1. MEDIUM -- Keyless sinks with a column named n fail every window

src/storage/src/sink/postgres.rs:849

The retract statement aliases count(*) AS n in the counts CTE alongside the user's columns. If the sunk relation has a column named n, every reference to c.n becomes ambiguous and every window fails with column reference "n" is ambiguous, so the sink halts and restarts forever without advancing.

Details

Postgres rejects the statement during analysis, before it runs. That means it fails on every window, including windows with no retractions. The keyless insert runs first and succeeds, then the retract errors and the transaction rolls back. Reproduced with columns (a, n). The same statement with columns (a, b) applies correctly.

staging_relation_type only reserves mz_timestamp and mz_diff, and the planner adds no check of its own. A sink is keyless whenever it has no KEY and the relation has no natural key, which covers most sinks from plain tables. n is a common column name (count(*) AS n, generate_series(...) AS n).

Fix: give the count an mz_-prefixed alias (for example mz_n) and add it to the names staging_relation_type rejects. Alternatively, keep the user columns out of the CTE's flat namespace.

2. MEDIUM -- IS NOT DISTINCT FROM key match forces a nested loop over the whole target table on every window

src/storage/src/sink/postgres.rs:766

PostgreSQL cannot use an index, a hash join or a merge join for IS NOT DISTINCT FROM. The keyed deleted CTE therefore rescans the entire target table once for every key the window touches, so each window costs (target rows × keys in the window). This holds even when the target has a primary key on the key columns, so keyed sinks fall behind once the table reaches a moderate size.

Details

Test on PostgreSQL 16: a target with 200k rows and PRIMARY KEY (id), and a window that touches 200 keys. The generated statement plans as Nested Loop with Join Filter: (NOT (t.id IS DISTINCT FROM l.id)) and Rows Removed by Join Filter: 39999800, and takes 1.35 s. With t.id = l.id instead, the plan becomes Index Scan using tgt_pkey and takes 0.8 ms.

A sink that lags gets windows that touch more keys, which makes each window slower still, so it does not catch up on its own.

Fix: emit t.k = l.k for key columns that are non-nullable in from_desc, which covers natural keys and most user KEYs. Only nullable key columns need a null-safe form. The keyless retract join at line 832 has the same problem. Its NOTE blames the missing key for the cost, but with = on non-nullable columns Postgres could at least use a hash join there too.

[patrickwwbutler]

Suggested change

	if !PartialOrder::less_equal(&resume_upper, &progress) {
	if !PartialOrder::less_equal(&lower, &progress) {

resume upper is not the right thing to check here, and may not even exist

[patrickwwbutler]

verify that this works with updates containing multiple ticks - potentially many updates to a single key and we need to make sure we only emit the correct retraction, and only insert the single correct value

[def-]

QA LLM Review

1. HIGH -- Applying unclosed timestamps breaks exactly-once replay

src/storage/src/sink/postgres.rs:374

The new commit path applies every staged row, including timestamps at or beyond the progress frontier. Those rows are committed with an earlier resume frontier, so a restart replays them, silently duplicating keyless output or blocking keyed sinks on duplicate-key errors.

Details

write_collection stages every received timestamp at or above upper (src/storage/src/sink/postgres.rs:349). Rows from different workers can arrive ahead of the global frontier. Neither apply path filters by progress, and committing empties the entire staging table.

For example, staging an insertion at time 10 and committing frontier 5 inserts the row immediately. Restarting from 5 then applies the insertion again. The generated SQL on PostgreSQL 17 produces two copies for a keyless target and a primary-key violation for a keyed target.

Apply only timestamps strictly below progress and retain later updates for subsequent windows, either before staging or in staging storage that survives commits.

2. MEDIUM -- Keyless retractions fail on partitioned targets with overlapping ctids

src/storage/src/sink/postgres.rs:1023

The keyless delete identifies rows only by ctid, which is unique within an individual physical table. On a partitioned target, a retraction can also select unrelated rows in another partition, causing the row-count check to reject the window and leaving the sink unable to advance.

Details

Existing partitioned targets pass the CREATE TABLE IF NOT EXISTS setup (src/storage/src/sink/postgres.rs:710). With one row in each of two fresh partitions, both rows have ctid = '(0,1)'. Retracting only the first row makes ranked select that ctid, but the outer delete scans both partitions and deletes both rows. On PostgreSQL 17, the generated statement returns (retracted, deleted) = (1, 2), so Writer::commit rejects and rolls back the window (src/storage/src/sink/postgres.rs:522).

Carry tableoid through ranked and match (tableoid, ctid) in the delete. This restricts the same reproduction to the intended row and returns (1, 1).