postgres_cdc: batch streamed rows per transaction over a buffered channel (ENG-1344)

[squiidz]

Summary

Streaming postgres_cdc plateaued at ~81-85 MB/s regardless of cores (ENG-1344). The reader goroutine sent one row per unbuffered channel send, so reader and consumer ran in lockstep. This PR lets the reader and the batching/marshalling stages overlap rather than run in lockstep; the core-scaling claim rests on the pending Docker/AWS end-to-end numbers, not on the microbenchmark:

• batches decoded rows in the reader and flushes at commit boundaries or at a row cap of min(1000, checkpoint_limit/2) (512 by default) / 4 MiB of WAL;
• buffers the reader→consumer channel (depth 4);
• reuses the receive deadline context across rows instead of allocating one per row;
• makes the input honour a configured batching policy per message now that reader batches are multi-row (this also fixes a pre-existing overshoot on snapshot pages);
• adds an opt-in batch_transactions field (default false). With it set and no batching policy, each reader batch passes straight through as one output batch, so streaming yields transaction-sized batches. The default output shape is unchanged from main: one message per batch during streaming, one page per batch during the snapshot;
• adds BenchmarkStreamMessages, a reader benchmark against an in-process fake walsender, and four integration tests locking in the batching behaviours (one row per batch by default, transaction-sized batches with batch_transactions, an honoured count, and a 1500-row transaction splitting at the cap with no replay after restart).

LSN/ack semantics are unchanged: "emitted" still means "handed to the consumer"; the commit-LSN remap is promoted only after a successful hand-off (see streamBatch and the flush closure in streamMessages). One new config field, batch_transactions.

Why batch_transactions is opt-in

A whole-transaction batch changes the unit of failure downstream. With a multi-row sql_insert behind fallback, one row violating a constraint fails the batch with a plain (non-indexed) error, so fallback routes the whole transaction to the DLQ; without fallback, the whole transaction is retried rather than one row. The field description documents this, and batch_transactions is rejected alongside a batching policy, which would re-batch the transactions.

Numbers

Reader microbenchmark, both trees measured 2026-09-17 (Apple M4 Pro, 1200 B rows, 100 rows/txn, -benchtime 200000x -count 3; full section with raw output in docs/benchmark-results/postgres.md):

	ns/row	MB/s	allocs/row
main + benchmark only (38060f7)	3485-3564	326-333	59
this PR	1952-2024	574-595	48

The microbenchmark's consumer only counts rows, so this gain reflects fewer channel operations and context allocations, not the reader/consumer overlap the change exists to enable.

Local Docker CDC bench (cart table), MB/s at GOMAXPROCS 1/2/4/8, for the default configuration, batch_transactions: true, and batching.count: 1000: pending (requires an enterprise license on the bench host).
AWS orders-cdc rerun: pending.

Notes

• Any configured batching policy, including an explicit count: 1, is honoured exactly per message, on the snapshot path as well: a snapshot page is no longer admitted whole under a policy, so an explicit count: 1 now yields one row per batch during the snapshot (on main it was byte-identical to the default and gave a page per batch). Leave batching unset to keep page-sized snapshot batches.
• A transaction whose row count lands exactly on the streaming cap is still stamped with its commit: the caps apply before the next row is appended, so the commit record closes the batch.
• When batching.processors drop every row of a flush, the position those rows reached is still acknowledged (tracked as a row-less checkpoint behind any batch in flight), so a filter-only stretch cannot park confirmed_flush_lsn.
• The reader keeps a ring of recent (last-row LSN, commit LSN) pairs, indexed by a map, and remaps downstream acks against it, so an ack for a transaction still sitting in the channel buffer at shutdown is confirmed as its commit record rather than a mid-transaction position (commit remap survives buffering).
• Streaming reader batches are capped at half of checkpoint_limit so two can be in flight under the checkpoint cap; a snapshot page is still admitted whole, so in-flight messages can exceed the limit by up to one page (documented on the field).
• On a graceful stop, an in-flight batch that cannot be handed on is dropped silently (re-read on restart); only a non-shutdown flush failure logs at Error and restarts the stream.
• When batching.processors fail on a batch, its rows are published as the batcher held them, with the error set, instead of being silently dropped; a processor earlier in the chain may already have altered them.
• The reader's byte cap counts WAL wire bytes; buffered decoded rows (maps, schema references) can exceed the design's rough 16 MiB estimate.
• The buffered channel also applies to the snapshot path, so up to 4 snapshot pages (snapshot_batch_size rows each) can be in flight where previously it was one.
• In-flight messages can exceed checkpoint_limit by up to one batch (a 1000-row streaming batch or one snapshot page), since a batch is admitted whenever nothing is pending.
• Rows decoded but not yet handed to the consumer at shutdown are dropped and re-read on restart (at-least-once).
• With a configured batching policy, a batch flush error now restarts the stream instead of continuing past the rest of the transaction.

Testing

• streamBatch and commitRemap unit tests (commit flush, row cap, byte cap, empty suppressed commit, LSN promotion, slice independence, ring eviction and lookup).
• resolveBatchingPolicy / passesThrough table tests, including the batch_transactions + batching rejection.
• Postgres integration suite (Docker) green, including new tests: default is one row per batch; batch_transactions gives transaction-sized batches; batching.count honoured exactly; a single transaction larger than the streaming row cap (512 with the default checkpoint_limit) arrives as several batches with no rows lost or replayed across a restart on the same slot.
• Follow-ons under the epic: parallel decode/marshal; binary tuple format.

[Jeffail]

It feels a bit risky rolling this out as a default behavior for users without a batching policy already configured. Could we propose an alternative where we instead added a new field so that this would be opt-in? If we're of the opinion that the UX of an opt-in is poor then I think it's potentially fine to change the default batching behavior but I'd like to consider it.

[Leward]

An example of where this new behavior could break an existing pipeline:

input:
  postgres_cdc:
    dsn: postgres://pg:pw@localhost:15432/db?sslmode=disable
    schema: public
    tables: [src]
    slot_name: test_slot
    stream_snapshot: false
    temporary_slot: true
output:
  fallback:
    - sql_insert:
        driver: postgres
        dsn: postgres://pg:pw@localhost:15432/db?sslmode=disable
        table: sink
        columns: [id, v]
        args_mapping: root = [this.id, this.v]
    - sql_insert:
        driver: postgres
        dsn: postgres://pg:pw@localhost:15432/db?sslmode=disable
        table: dlq
        columns: [id, v, err]
        args_mapping: root = [this.id, this.v, @fallback_error]

Version	Rows landed in sink	Rows landed in dql
main	1, 3	2
this PR	none	1, 2, 3

---

Why it happens:

Setup: sink already holds id=2. One transaction inserts rows 1, 2, 3 into src. No batching policy.

main

1. No policy, so batching.Count = 1. Three 1-row batches: [1], [2], [3].
2. [1] inserts fine. [2] hits the unique constraint, sql_insert returns a plain error.
3. fallback sends the whole failed batch to the DLQ. That batch is just row 2.
4. [3] inserts fine.

this PR

1. No policy, so the whole transaction passes through as one batch: [1, 2, 3].
2. sql_insert builds one multi-row INSERT. Row 2 fails it, Postgres rolls back all three rows.
3. The error is a plain error, not an indexed *batch.Error, so fallback cannot tell which part failed. It sends the whole batch to the DLQ.
4. Rows 1 and 3 never reach sink. The batch is acked and the LSN moves on.

Neither sql_insert nor fallback changed. Only the batch shape did, and with it the blast radius of one bad row: one row on main, the whole transaction here. Without fallback the same config stalls the stream on the PR (whole transaction nacked and retried forever) where main only retries row 2.

[Leward]

Could we split this in two?

First PR: the internal changes only. The reader still groups rows per transaction, the remap ring and the checkpoint_limit cap stay, but with no batching policy we still send one row per message, like today. Nothing changes for existing configs, so this should be quick to merge.

Second PR: change the default (or add an opt-in) to send whole transactions as one batch. That one needs its own changelog entry and a discussion of the fallback/DLQ case above.

[squiidz]

@Jeffail Agreed the default should not move. Pushed as postgres_cdc: gate whole-transaction batches behind an opt-in batch_transactions field: whole-transaction batches are now behind a new batch_transactions bool, default false. With it off and no batching policy, streaming rows go through the count-1 batcher exactly as on main, one message per batch; snapshot pages still arrive whole, as they always did. With it on, each transaction (capped at min(1000, checkpoint_limit/2) rows or 4 MiB of WAL) is one batch. Setting it alongside a batching policy is a config error, since the policy would re-batch the transactions. The field description spells out the failure-unit change from the fallback/DLQ example below as the reason it is opt-in. The reader-side batching, buffered channel and commit remap stay in place either way, so the default path keeps the decode/marshal overlap.

[squiidz]

@Leward Confirmed: sql_insert.WriteBatch returns a plain error, so fallback has no index to isolate row 2 and the whole batch goes to the DLQ. One nuance: without fallback, main also stalls on this transaction (row 2 is retried forever and the LSN cannot advance past it); the difference is that rows 1 and 3 land first. Either way the default is now preserved: no policy means one message per batch during streaming. TestIntegrationPostgresDefaultBatchingIsOneRowPerMessage pins it, and the batch_transactions description carries this scenario as the reason it is off by default.

[squiidz]

@Leward Keeping this as one PR, but with the shape you describe: the internal changes (reader batching, buffered channel, remap ring, checkpoint_limit cap) land with the default output unchanged, and the transaction-sized batch is the opt-in batch_transactions field with its own changelog entry under Added.

[claude]

Only the local benchmarking phase is covered (CONTRIBUTING.md §1.3.4: "Benchmarking covers two distinct phases, both of which are required: … Real-endpoint benchmarking: benchmarks run against a real deployed server / real endpoint, exercising the connector against the actual target system rather than a local stand-in.")

This PR's performance claim (~1.75× on the reader path) rests entirely on BenchmarkStreamMessages against an in-process fake walsender over loopback, which is the local phase. The real-endpoint phase is recorded here as pending, and §1.3.5 ("benchmarks have been run at various throughput levels so that we can determine CPU and memory trendlines") is likewise unaddressed — the microbenchmark runs a single fixed workload with a counter-only consumer, which the section itself notes hides the reader/consumer overlap that the buffered channel exists to create.

The decision to land with the real-endpoint phase outstanding is the reviewers' call, but it should be an explicit one rather than deferred. If the license blocker means it genuinely cannot run before merge, consider linking a tracking issue here so the follow-up is recorded outside this file.

[squiidz]

Agreed that landing with the real-endpoint phase outstanding is the reviewers' call, and the section records it as pending rather than done. The follow-up is tracked under the epic this PR belongs to, ENG-1344, whose remaining scope is exactly the Docker and AWS orders-cdc runs for the default configuration, batch_transactions: true and batching.count: 1000.

[Leward]

We are losing the reference to xref:configuration:batching.adoc[batching policy].

Optional xref:configuration:batching.adoc[batching policy] for the emitted messages.

By default, streaming rows arrive one message at a time and snapshot rows arrive one page at a time.

To change this:

* Set a policy to control batch size and timing. It applies to both the snapshot and the stream.
* Set `batch_transactions` to receive each replication transaction as one batch. This cannot be combined with a policy.

[squiidz]

Applied, with the cross-reference restored. I kept one closing sentence on what happens when the policy's processors fail, since that behaviour is new in this PR.

[Leward]

IMO the hardest part to review of this PR is the commitRemap ring and the checkpointLimit + streamChannelDepth + 2 bound that sizes it. While the comment explains the underlying logic it seems complex and easy to make a mistake there.

I think we could get away with a simpler approach: stamp the ack target on each StreamMessage when the batch is flushed (its own LSN, or the commit LSN for the last row of a committed transaction).

flushBatch then tracks that target in the checkpointer and the reader has nothing to remap. No ring, no window to size.

[squiidz]

Done, in postgres_cdc: stamp each row's ack position at flush instead of remapping acks through a ring. The reader stamps every message it flushes with the position to acknowledge for it (StreamMessage.AckLSN: its own LSN, or the commit LSN for the last row of a transaction closed in that batch). The consumer keeps the stamp of the last change row it handed to the batcher alongside the rows it already mirrors, and flushBatch tracks that stamp as the checkpoint payload. The ring, its sizing and the CheckpointLimit plumbing into the reader are gone. The reader keeps only the last emitted stamp and the newest commit, so an ack for the last emitted row can still advance over commits that emitted nothing (heartbeats, untracked tables).

[squiidz]

Merged main, which brought incremental snapshot and the shared control-signal decoding into the same loop. How the two meet:

• The consumer keeps main's checkpoint tracker (LSN plus snapshot state), the row-less checkpoint sentinel and the blocking-snapshot barrier gate; this branch's pass-through and stamped ack positions feed it. The checkpoint payload is the stamp of the last change row added to the batcher, kept when snapshot rows follow it, which replaces the backward search over lsn metadata for a mixed batch. TestFlushBatchLSNFromMixedBatch is rewritten against that seam.
• A chunk the incremental snapshot releases at a commit is sent from inside processChange, so with transaction batching it now precedes that transaction's own rows on the channel. The consumer therefore tracks the snapshot state before the commit position, which is the safe order: the state can never be persisted as progress past a commit that was not confirmed first.
• The reader still deduplicates and dispatches signals row by row in WAL order; only the hand-off to the consumer is batched.

[Leward]

Please remove these entries. The release workflow generates CHANGELOG.md from the merged PRs (release-notes.yml#L34-L103). The last run is #4933.

The workflow also inserts the new version heading after line 5 (L96-L101). A manual ## Unreleased section would then stay below that heading as stale text.

[squiidz]

Gone: the branch now carries no changelog entries of its own, and CHANGELOG.md is byte-identical to main (the merge that brought in the 4.114.0 release regenerated it). The red docs and test jobs were the introduced-in check, since 4.114.0 shipped after the last merge from main; batch_transactions is now versioned 4.115.0.

[squiidz]

Self-review pass over the whole branch before merge (automated review plus a hand audit of the ack and LSN paths, both packages' integration suites under -race, and a throwaway restart test under a count: 7 policy). Two real gaps, both fixed with tests in the latest commit:

• Cap-aligned transactions were stamped with their last row, not their commit. The caps were checked right after a row landed, so a transaction with exactly 512 rows (default checkpoint_limit) flushed before its commit was seen. If another transaction went out before that batch was acknowledged, the slot was confirmed mid-transaction and a restart replayed all 512 rows. The caps now apply before the next row goes in, so such a transaction waits for its commit. Larger transactions still split the same way.
• A batch whose rows were all dropped by batching.processors never acknowledged its position. Pre-existing on main, but with the stamp travelling alongside the pending rows it is now a few lines: the position is tracked as a row-less checkpoint and resolved behind any batch still in flight. Without it a filter-only stretch left confirmed_flush_lsn parked and WAL retained.

Smaller: the pass-through path shares pendingRows with the batcher path instead of its own scan; the batch_transactions + batching rejection names the real conflict (processors included) and is tested with a processors-only policy; the warning for rows republished after a processor failure no longer calls them "unprocessed", since an earlier processor in the chain may already have altered them.

Findings reviewed and deliberately left as they are: explicit batching: {count: 1} is now honoured per message on snapshot pages too (the exact-honouring change this PR makes, called out in the description); channel depth 4 and the default path's per-row batcher hop are performance knobs for the benchmark follow-up under ENG-1344.