Bound the libpq pipeline backlog with byte-based syncs

[d-cs]

Problem

During a production CDC migration, apply fell behind the source and never converged: each 64 MiB WAL segment (~130 MB of transformed SQL, ~388k statements) took ~18 minutes to apply while the source produced one every ~13.6 minutes. perf on the apply process showed 99.3% of CPU inside __memmove_avx512_unaligned_erms with the call graph stream_apply_file → pgsql_sync_pipeline → PQconsumeInput → memmove, ~350 statements/s reaching the target (which was idle at 0.06 ms/statement), and apply RSS ballooned to 1.13 GB.

Apply pipelines every statement on a nonblocking connection and only drains the pipeline at COMMIT/KEEPALIVE boundaries behind a 1-second timer, or mid-transaction after 512 MB of accumulated EXECUTE parameter bytes. The timer bounds time, not bytes — the loop can enqueue tens of MB into libpq's buffers before the first sync fires — and the parameter-only 512 MB threshold effectively never triggers. libpq shifts the entire outstanding contents of its send/receive buffers with memmove on every partial send and read (pqSendSome / pqReadData in fe-misc.c), so per-statement cost grows with the backlog and applying a batch is quadratic in its size. At the observed ~65 MB mean backlog that is ~25 TB of memory traffic per 130 MB file: one core pinned at memory bandwidth.

Fix

Count every applied statement's bytes (strlen(sql) plus a small framing allowance) and sync the pipeline each time 1 MB has been sent since the last sync, mid-transaction included. A pipeline sync only fetches pending results; the explicit BEGIN/COMMIT transaction is unaffected — the same argument as the existing 512 MB mid-transaction sync this replaces. The 1-second timer sync at transaction boundaries remains as the latency bound. The EXECUTE-parameter-only accounting is removed (parameters are part of the counted statement line), and the two replay-path sync sites that didn't reset the byte counter now do.

With the backlog capped at 1 MB, each libpq buffer shift is KB-scale and apply becomes target-bound instead of memmove-bound.

Testing

New regression test in tests/cdc-low-level: generates a single-transaction, 25k-statement (~5.6 MB) .sql file, applies it with --trace, and asserts the full row count plus ≥5 pipeline syncs. Pre-fix the assertion fails with exactly 2 syncs (final COMMIT + end-of-file, zero mid-transaction); post-fix it passes with 9.

Benchmarks on a production-shaped fixture (85 MB, 420k lines, 35k transactions of ~10 statements), unpatched vs patched, same databases:

• wall clock: 64 s → 4 s; apply peak RSS 430 MB → 288 MB (the remainder is the whole-file buffer plus per-line metadata that file-mode apply always allocates)
• gdb stack-sampling during apply: 97% of samples executing memcpy under PQconsumeInput before; 0% after, with every sample idle in select() waiting on the target
• scaling 25k → 100k statements: 0.23 s → 2.71 s before (11.8×, super-linear); 0.20 s → 0.74 s after (3.7×, linear)

Check	Result
make build (docker, PG16 image)	clean
make tests/cdc-low-level (includes new test)	pass
make tests/cdc-endpos-between-transaction	pass
tests/cdc-filtering (run directly, see note)	pass
tests/cdc-message-handling (run directly, see note)	pass
ci/banned.h.sh	pass
citus_indent --check	not run locally (tool unavailable); CI runs it

Note: make tests/cdc-filtering is silently a no-op — including in CI — because the directory name shadows the missing make target, and cdc-message-handling is wired into neither tests/Makefile nor the CI matrix. Both suites were run directly here via make -C tests/<dir> test DOCKER=docker. Worth fixing in a separate PR.