From 5eaf3b452a5db84c4f93d6187b12a5d35153eca0 Mon Sep 17 00:00:00 2001 From: Daniel Sutton Date: Sat, 25 Jul 2026 11:12:32 +0100 Subject: [PATCH 1/2] Bound the libpq pipeline backlog with byte-based syncs The CDC apply pipelines statements and only synced at COMMIT/KEEPALIVE boundaries (1s timer) or at 512 MB of EXECUTE parameter bytes. Inside a large transaction nothing bounds libpq's buffers, and libpq memmoves its whole outstanding buffer contents on every partial send and read, making apply quadratic in the backlog size: a production migration measured 99% of apply CPU in __memmove_avx512_unaligned_erms at ~350 statements/s. Count every applied statement's bytes and sync the pipeline each time 1 MB has been sent since the last sync, mid-transaction included. The 1-second timer sync at transaction boundaries remains as the latency bound. Applying a single 100k-statement transaction takes 2.71s before and 0.74s after this change; at 25k statements the two builds are equal (0.23s vs 0.20s), the gap being quadratic growth in transaction size. --- src/bin/pgcopydb/ld_apply.c | 84 +++++++++++++---------------------- src/bin/pgcopydb/ld_replay.c | 4 ++ tests/cdc-low-level/copydb.sh | 29 ++++++++++++ tests/cdc-low-level/ddl.sql | 3 ++ 4 files changed, 67 insertions(+), 53 deletions(-) diff --git a/src/bin/pgcopydb/ld_apply.c b/src/bin/pgcopydb/ld_apply.c index 0a9c1dda4..109a9d3af 100644 --- a/src/bin/pgcopydb/ld_apply.c +++ b/src/bin/pgcopydb/ld_apply.c @@ -32,14 +32,14 @@ #include "summary.h" /* - * libpq's output buffer uses a signed int for its size, so it can only grow - * to ~1 GB before the doubling arithmetic overflows. When the CDC apply - * pipelines many EXECUTE statements with large parameter values (e.g. rows - * containing 300 MB email bodies), the accumulated data can exceed that - * limit. We force a pipeline sync at transaction boundaries once the - * estimated parameter data reaches this threshold. + * libpq memmoves its whole outstanding buffer contents on every partial send + * and read, making apply quadratic in the pipeline backlog size. Sync the + * pipeline every time this many bytes were sent, mid-transaction included. */ -#define PIPELINE_BYTES_SYNC_THRESHOLD (512ULL * 1024 * 1024) +#define PIPELINE_BYTES_SYNC_THRESHOLD (1024 * 1024) + +/* per-statement allowance for protocol framing and result traffic */ +#define PIPELINE_STMT_OVERHEAD 64 GUC applySettingsSync[] = { COMMON_GUC_SETTINGS, @@ -583,14 +583,9 @@ stream_apply_file(StreamApplyContext *context) /* * Sync the pipeline at transaction boundaries (COMMIT or - * KEEPALIVE) when either the 1-second timer has elapsed or - * when accumulated parameter data approaches libpq's output - * buffer limit. - * - * libpq's output buffer size is tracked with a signed int, - * so the buffer can grow to ~1 GB before the doubling logic - * overflows. We sync well before that at 512 MB to leave - * headroom for wire-protocol framing and PREPARE overhead. + * KEEPALIVE) when the 1-second timer has elapsed, bounding + * replay latency. The byte-based sync that bounds the + * pipeline backlog size lives in stream_apply_sql. */ if (metadata->action == STREAM_ACTION_COMMIT || metadata->action == STREAM_ACTION_KEEPALIVE) @@ -598,10 +593,7 @@ stream_apply_file(StreamApplyContext *context) bool timeToSync = 1 < (time(NULL) - context->applyPgConn.pipelineSyncTime); - bool bufferNearFull = - context->pipelineBytes >= PIPELINE_BYTES_SYNC_THRESHOLD; - - if (timeToSync || bufferNearFull) + if (timeToSync) { /* fetch results until done */ if (!pgsql_sync_pipeline(&(context->applyPgConn))) @@ -1294,40 +1286,6 @@ stream_apply_sql(StreamApplyContext *context, /* errors have already been logged */ return false; } - - /* - * Track accumulated parameter bytes in the pipeline. - * libpq output buffer uses int (~1 GB effective max - * due to doubling), force sync before overflow. - */ - for (int j = 0; j < count; j++) - { - if (paramValues[j] != NULL) - { - context->pipelineBytes += strlen(paramValues[j]); - } - } - - /* - * When a single transaction contains many large rows - * (e.g. 330 MB email bodies), the pipeline buffer can - * exceed libpq's ~1 GB limit before reaching COMMIT. - * Sync mid-transaction to drain the buffer. This is - * safe: we use explicit BEGIN/COMMIT, so a pipeline - * sync only flushes pending results without affecting - * the transaction. - */ - if (context->pipelineBytes >= PIPELINE_BYTES_SYNC_THRESHOLD) - { - if (!pgsql_sync_pipeline(applyPgConn)) - { - log_error("Failed to sync the pipeline, " - "see previous error for details"); - return false; - } - - context->pipelineBytes = 0; - } } @@ -1379,6 +1337,26 @@ stream_apply_sql(StreamApplyContext *context, } } + /* + * Sync the pipeline as soon as the data sent since the last sync + * exceeds the threshold, even in the middle of a transaction: a + * pipeline sync only fetches pending results, the explicit + * BEGIN/COMMIT transaction is unaffected. + */ + context->pipelineBytes += strlen(sql) + PIPELINE_STMT_OVERHEAD; + + if (context->pipelineBytes >= PIPELINE_BYTES_SYNC_THRESHOLD) + { + if (!pgsql_sync_pipeline(applyPgConn)) + { + log_error("Failed to sync the pipeline, " + "see previous error for details"); + return false; + } + + context->pipelineBytes = 0; + } + return true; } diff --git a/src/bin/pgcopydb/ld_replay.c b/src/bin/pgcopydb/ld_replay.c index b7a526389..bb6d2036a 100644 --- a/src/bin/pgcopydb/ld_replay.c +++ b/src/bin/pgcopydb/ld_replay.c @@ -418,6 +418,8 @@ stream_replay_line(void *ctx, const char *line, bool *stop) "error for details"); return false; } + + context->pipelineBytes = 0; } break; } @@ -482,6 +484,8 @@ stream_replay_line(void *ctx, const char *line, bool *stop) "details"); return false; } + + context->pipelineBytes = 0; } return true; diff --git a/tests/cdc-low-level/copydb.sh b/tests/cdc-low-level/copydb.sh index 656fcbddc..3b6444449 100755 --- a/tests/cdc-low-level/copydb.sh +++ b/tests/cdc-low-level/copydb.sh @@ -129,5 +129,34 @@ pgcopydb stream replay --resume --endpos "${lsn}" psql -At -d ${PGCOPYDB_TARGET_PGURI} -c "select pg_replication_origin_advance('pgcopydb', '0/0');" pgcopydb stream apply /usr/src/pgcopydb/pipeline-deadlock.sql +# bounded pipeline sync test: a single multi-MB transaction must trigger +# byte-based pipeline syncs before its COMMIT, bounding libpq's buffers +BOUNDED=/tmp/bounded-sync.sql + +awk 'BEGIN { + printf "BEGIN; -- {\"xid\":99000001,\"lsn\":\"E0/10000000\",\"timestamp\":\"2026-07-25 10:00:00.000000+0000\",\"commit_lsn\":\"E0/20000000\"}\n"; + for (i = 1; i <= 25000; i++) { + printf "PREPARE ab12cd34 AS INSERT INTO \"public\".\"bounded_sync\" (\"id\", \"payload\") overriding system value VALUES ($1, $2);\n"; + printf "EXECUTE ab12cd34[\"%d\",\"payload-%d-0123456789abcdefghijklmnopqrstuvwxyz0123456789abcdefghijklmnopqrstuvwxyz\"];\n", i, i; + } + printf "COMMIT; -- {\"xid\":99000001,\"lsn\":\"E0/20000000\",\"timestamp\":\"2026-07-25 10:00:00.000000+0000\"}\n"; + printf "-- SWITCH WAL {\"lsn\":\"E0/21000000\"}\n"; +}' > ${BOUNDED} + +psql -At -d ${PGCOPYDB_TARGET_PGURI} -c "select pg_replication_origin_advance('pgcopydb', '0/0');" + +pgcopydb stream apply --trace ${BOUNDED} > /tmp/bounded-sync.log 2>&1 || \ + { tail -50 /tmp/bounded-sync.log; exit 1; } + +syncs=$(grep -c "Start pipeline sync" /tmp/bounded-sync.log || true) +rows=$(psql -At -d ${PGCOPYDB_TARGET_PGURI} -c "select count(*) from bounded_sync") + +echo "bounded sync test: ${syncs} pipeline syncs, ${rows} rows applied" + +test "${rows}" -eq 25000 + +# unpatched pgcopydb only syncs at COMMIT/EOF, which fails this assertion +test "${syncs}" -ge 5 + # cleanup pgcopydb stream cleanup --verbose diff --git a/tests/cdc-low-level/ddl.sql b/tests/cdc-low-level/ddl.sql index a26c5cfa2..3de3e7e27 100644 --- a/tests/cdc-low-level/ddl.sql +++ b/tests/cdc-low-level/ddl.sql @@ -37,4 +37,7 @@ EXECUTE FUNCTION notify_on_insert(); -- Lets enable the trigger for the replica mode ALTER TABLE metrics ENABLE REPLICA TRIGGER insert_notice_trigger; +-- bounded pipeline sync test table; keep trigger-free (no NOTICE traffic) +create table bounded_sync(id int, payload text); + commit; From 0dd4d14e47135ac35e763cdcfd685bd4b45240ee Mon Sep 17 00:00:00 2001 From: Daniel Sutton Date: Sat, 25 Jul 2026 12:19:04 +0100 Subject: [PATCH 2/2] fix: count only bytes actually sent toward the pipeline sync threshold stream_apply_sql charged strlen(sql) for every line reaching the end of the switch, including repeated PREPARE lines that are skipped once the statement is prepared, and SWITCH records that send nothing at all. The transform emits one PREPARE per EXECUTE and those lines are slightly longer, so roughly half the counted bytes never reached the wire and the sync fired at about 500 KB of real backlog rather than the 1 MB PIPELINE_BYTES_SYNC_THRESHOLD documents. Track the bytes actually handed to libpq instead. On the 25k-row single transaction fixture in tests/cdc-low-level this takes the pipeline sync count from 9 down to 5. --- src/bin/pgcopydb/ld_apply.c | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/src/bin/pgcopydb/ld_apply.c b/src/bin/pgcopydb/ld_apply.c index 109a9d3af..f17343b17 100644 --- a/src/bin/pgcopydb/ld_apply.c +++ b/src/bin/pgcopydb/ld_apply.c @@ -647,6 +647,9 @@ stream_apply_sql(StreamApplyContext *context, { PGSQL *applyPgConn = &(context->applyPgConn); + /* bytes this line hands to libpq; branches that send nothing zero it */ + uint64_t sentBytes = strlen(sql); + switch (metadata->action) { case STREAM_ACTION_SWITCH: @@ -660,6 +663,7 @@ stream_apply_sql(StreamApplyContext *context, * .sql file to apply. */ context->switchLSN = metadata->lsn; + sentBytes = 0; break; } @@ -1196,6 +1200,8 @@ stream_apply_sql(StreamApplyContext *context, return true; } + sentBytes = 0; + /* Only prepare if we haven't already */ if (stmt != NULL && !stmt->prepared) { @@ -1210,6 +1216,7 @@ stream_apply_sql(StreamApplyContext *context, } stmt->prepared = true; + sentBytes = strlen(metadata->stmt); } break; @@ -1343,7 +1350,10 @@ stream_apply_sql(StreamApplyContext *context, * pipeline sync only fetches pending results, the explicit * BEGIN/COMMIT transaction is unaffected. */ - context->pipelineBytes += strlen(sql) + PIPELINE_STMT_OVERHEAD; + if (sentBytes > 0) + { + context->pipelineBytes += sentBytes + PIPELINE_STMT_OVERHEAD; + } if (context->pipelineBytes >= PIPELINE_BYTES_SYNC_THRESHOLD) {