diff --git a/src/bin/pgcopydb/ld_apply.c b/src/bin/pgcopydb/ld_apply.c index 0a9c1dda4..f17343b17 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))) @@ -655,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: @@ -668,6 +663,7 @@ stream_apply_sql(StreamApplyContext *context, * .sql file to apply. */ context->switchLSN = metadata->lsn; + sentBytes = 0; break; } @@ -1204,6 +1200,8 @@ stream_apply_sql(StreamApplyContext *context, return true; } + sentBytes = 0; + /* Only prepare if we haven't already */ if (stmt != NULL && !stmt->prepared) { @@ -1218,6 +1216,7 @@ stream_apply_sql(StreamApplyContext *context, } stmt->prepared = true; + sentBytes = strlen(metadata->stmt); } break; @@ -1294,40 +1293,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 +1344,29 @@ 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. + */ + if (sentBytes > 0) + { + context->pipelineBytes += sentBytes + 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;