[patrickwwbutler]

Implement the encode operator and the staging half of the write loop.
Every worker encodes its share of the sink's updates as COPY BINARY
rows, tagged with the Materialize timestamp and a diff of 1 or -1. The
write operator opens one SERIALIZABLE transaction per timestamp window,
checks that the progress row still carries its epoch and the frontier it
expects, and streams the window's rows into a session temporary table
that ON COMMIT DELETE ROWS empties after every window.

Applying a staged window to the target table is an unimplemented stub.

Tests: covered by the unit tests that #39707 adds and by test/pg-sink at the top of this stack.

🤖 Generated with Claude Code

[def-]

QA LLM Review

1. HIGH -- A batch is sent as one CopyData message, so any batch over 1 GiB permanently crash-loops the sink

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

COPY_CHUNK_BYTES is only checked after a whole arrangement batch has been encoded, and send_buffered passes all of buf to CopyInSink::send as one item, which tokio-postgres frames as a single CopyData message. PostgreSQL rejects any CopyData message over 1 GiB (PQ_LARGE_MESSAGE_LIMIT), so a worker whose share of one batch encodes to more than 1 GiB fails, restarts, re-snapshots the same batch and fails again, forever. Before failing, it also holds the whole encoded batch in memory, with multiplicities expanded.

Details

The arrange operator seals one batch per capability (differential-dataflow arrangement.rs:460-482). Because the snapshot sits entirely at as_of, each worker's whole snapshot share arrives as one batch. A few GiB of data on a replica with few workers is enough to hit the limit. A single update with a large diff gets there too, because for _ in 0..n copies the row into buf n times. In CopyInSink::start_send (rust-postgres fork, copy_in.rs:149-164), any item over 4096 bytes becomes one CopyData::new(item), and anything over i32::MAX fails client-side with "message length overflow".

I reproduced this on postgres:16 with rows in the staging-table shape. One 1.1 GiB CopyData message gets invalid message length, then unexpected EOF on client connection with an open transaction and FATAL: terminating connection because protocol synchronization was lost. The same bytes sent as 1 MiB CopyData messages commit COPY 36044800.

Fix: flush inside the row loop instead of per batch. The cursor only borrows the locally owned batches, so awaiting there is fine.

                                         let n = diff.unsigned_abs();
                                         for _ in 0..n {
                                             buf.extend_from_slice(&scratch);
+                                            if buf.len() >= COPY_CHUNK_BYTES {
+                                                send_buffered(&client, &copy_stmt, &mut copy, &mut buf)
+                                                    .await?;
+                                            }
                                         }
@@
                             statistics.inc_messages_staged_by(rows);
                             statistics.inc_bytes_staged_by(bytes);
-
-                            if buf.len() >= COPY_CHUNK_BYTES {
-                                send_buffered(&client, &copy_stmt, &mut copy, &mut buf).await?;
-                            }

2. MEDIUM -- Binary COPY into a stale staging table silently reinterprets values after ALTER SINK ... SET FROM

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

The COPY encodes rows using the current from_desc, but the staging table is only CREATE TABLE IF NOT EXISTS and keeps the column types of the sink's first incarnation. Binary COPY hands each field's raw bytes to the existing column's receive function without checking the type. If a same-named column changes to a type of the same width, the values are reinterpreted bit-for-bit and land in the user's target table with no error.

Details

ALTER SINK ... SET FROM re-plans Postgres sinks without restriction (plan_alter_sink in src/sql/src/plan/statement/ddl.rs), and nothing in the alter path compares the new from_desc against the old one. On restart, create_postgres_tables leaves the existing staging table as it is and only truncates it. Example: switch the sink from a relation with x bigint to one with x double precision. Every float's bits are then accepted by int8recv. On postgres:16, binary-copying the float8 1.5 into a bigint staging column stored 4609434218613702656. The apply step then copies that into the bigint target column. Other pairs that pass silently include integer/real/date and bigint/timestamp/timestamptz. A text-format COPY would have rejected these inputs, and a width change (e.g. int4 to int8) fails loudly.

The staging table is emptied on every start, so the setup operator could drop and recreate it instead of CREATE ... IF NOT EXISTS. Alternatively, the copy operator could compare the staging columns' format_type against from_desc and halt with a clear error on a mismatch.

[def-]

QA LLM Review

1. MEDIUM -- Zero-column relations generate invalid COPY statements

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

For a zero-column source, columns is empty but the format string still emits its following comma, producing COPY pg_temp.mz_sink_stage (, mz_timestamp, mz_diff) FROM STDIN (FORMAT BINARY). PostgreSQL rejects this with a syntax error, so the writer cannot stage updates from a valid zero-column table.

Details

Materialize accepts CREATE TABLE t (), and neither plan_sink nor stage_relation_type rejects that schema. The target table and the two-column staging table can both be created successfully. For example, 60,000 copies of the empty row exceed COPY_CHUNK_BYTES, causing Writer::stage to invoke flush, which passes the malformed statement directly to copy_in at src/storage/src/sink/postgres.rs:436.

On PostgreSQL 17, the generated statement fails at the leading comma. Removing that comma accepts the same binary staging row and returns COPY 1. Build the COPY column list with one Sql::join over the source names chained with the two metadata names, as the staging-table DDL already does.