From db74c6970408586a91c0741f3826550640e41bbf Mon Sep 17 00:00:00 2001 From: Ibrar Ahmed Date: Thu, 13 Aug 2026 21:09:07 +0500 Subject: [PATCH 1/2] Do not wedge table sync when the target already holds rows. A COPY into a populated table aborts on the first duplicate key and leaves the sync status failed, after which apply silently discards every later change for that table. Stage the load and merge, and warn on the failure. --- src/spock_apply.c | 17 +++++ src/spock_sync.c | 159 ++++++++++++++++++++++++++++++++++++++++++++-- 2 files changed, 169 insertions(+), 7 deletions(-) diff --git a/src/spock_apply.c b/src/spock_apply.c index 693b3a28..939e0080 100644 --- a/src/spock_apply.c +++ b/src/spock_apply.c @@ -4387,7 +4387,24 @@ process_syncing_tables(XLogRecPtr end_lsn) /* * Failed SYNC operation should be ignored until someone processes * the error and changes the status. + * + * Say so once, on the transition. From here on every change + * for this table is dropped by should_apply_changes_for_rel(), + * so the table stops replicating and diverges; without this + * the only trace is a status column in + * spock.local_sync_status that nobody thinks to read. */ + if (sync->status != SYNC_STATUS_FAILED) + ereport(WARNING, + (errmsg("SPOCK %s: synchronization of table %s.%s failed, changes for it are no longer applied", + MySubscription->name, + NameStr(sync->nspname), + NameStr(sync->relname)), + errhint("Re-synchronize with spock.sub_resync_table('%s', '%s.%s') once the cause is fixed.", + MySubscription->name, + NameStr(sync->nspname), + NameStr(sync->relname)))); + sync->status = SYNC_STATUS_FAILED; sync->statuslsn = InvalidXLogRecPtr; } diff --git a/src/spock_sync.c b/src/spock_sync.c index da7656a1..f58f3171 100644 --- a/src/spock_sync.c +++ b/src/spock_sync.c @@ -72,6 +72,12 @@ #define PGDUMP_BINARY "pg_dump" #define PGRESTORE_BINARY "pg_restore" +/* + * Staging table used by copy_table_data() when the target already holds rows. + * Lives in pg_temp on the target connection for the duration of the COPY. + */ +#define SPOCK_SYNC_STAGE_RELNAME "spock_sync_stage" + #define Natts_local_sync_state 6 #define Anum_sync_kind 1 #define Anum_sync_subid 2 @@ -1001,8 +1007,12 @@ copy_table_data(PGconn *origin_conn, PGconn *target_conn, List *attnamelist; ListCell *lc; bool first; + bool stage_load; + bool override_identity = false; + char *merged = NULL; StringInfoData query; StringInfoData attlist; + StringInfoData relident; MemoryContext curctx = CurrentMemoryContext, oldctx; @@ -1019,6 +1029,25 @@ copy_table_data(PGconn *origin_conn, PGconn *target_conn, attnamelist = make_copy_attnamelist(rel); + /* + * COPY may write GENERATED ALWAYS AS IDENTITY columns, an INSERT may not + * without OVERRIDING SYSTEM VALUE. Remember whether we need it for the + * staged-load path below. + */ + { + TupleDesc desc = RelationGetDescr(rel->rel); + int attnum; + + for (attnum = 0; attnum < desc->natts; attnum++) + { + if (TupleDescAttr(desc, attnum)->attidentity == ATTRIBUTE_IDENTITY_ALWAYS) + { + override_identity = true; + break; + } + } + } + initStringInfo(&attlist); first = true; foreach(lc, attnamelist) @@ -1118,13 +1147,71 @@ copy_table_data(PGconn *origin_conn, PGconn *target_conn, PQerrorMessage(origin_conn)))); } - /* Build COPY FROM query. */ - resetStringInfo(&query); - appendStringInfo(&query, "COPY %s.%s ", - PQescapeIdentifier(origin_conn, remoterel->nspname, + /* + * Decide whether to load straight into the table or through a staging + * table. + * + * A direct COPY into a table that already holds rows aborts on the first + * key collision, and that failure is not recoverable: the table's sync + * status ends up SYNC_STATUS_FAILED, and from then on the apply worker + * drops every change for it (see should_apply_changes_for_rel), silently + * and permanently. In a mesh this is the normal case rather than an edge + * case, because adding an already-populated table to a replication set + * with synchronize_data := true asks every peer to copy rows it already + * has. Load those tables into an unconstrained staging table and merge, + * so a sync of data we already hold converges instead of wedging. + */ + initStringInfo(&relident); + appendStringInfo(&relident, "%s.%s", + PQescapeIdentifier(target_conn, remoterel->nspname, strlen(remoterel->nspname)), - PQescapeIdentifier(origin_conn, remoterel->relname, + PQescapeIdentifier(target_conn, remoterel->relname, strlen(remoterel->relname))); + + resetStringInfo(&query); + appendStringInfo(&query, "SELECT 1 FROM %s LIMIT 1", relident.data); + res = PQexec(target_conn, query.data); + if (PQresultStatus(res) != PGRES_TUPLES_OK) + { + char *msg = pstrdup(PQerrorMessage(target_conn)); + + PQclear(res); + ereport(ERROR, + (errmsg("could not check whether target table %s.%s is empty", + remoterel->nspname, remoterel->relname), + errdetail("destination connection reported: %s", msg))); + } + stage_load = PQntuples(res) > 0; + PQclear(res); + + if (stage_load) + { + resetStringInfo(&query); + appendStringInfo(&query, + "DROP TABLE IF EXISTS pg_temp.%s;" + "CREATE TEMP TABLE %s (LIKE %s)", + SPOCK_SYNC_STAGE_RELNAME, SPOCK_SYNC_STAGE_RELNAME, + relident.data); + res = PQexec(target_conn, query.data); + if (PQresultStatus(res) != PGRES_COMMAND_OK) + { + char *msg = pstrdup(PQerrorMessage(target_conn)); + + PQclear(res); + ereport(ERROR, + (errmsg("could not create staging table for %s.%s", + remoterel->nspname, remoterel->relname), + errdetail("destination connection reported: %s", msg))); + } + PQclear(res); + } + + /* Build COPY FROM query. */ + resetStringInfo(&query); + if (stage_load) + appendStringInfo(&query, "COPY pg_temp.%s ", SPOCK_SYNC_STAGE_RELNAME); + else + appendStringInfo(&query, "COPY %s ", relident.data); if (list_length(attnamelist)) appendStringInfo(&query, "(%s) ", attlist.data); appendStringInfoString(&query, "FROM stdin"); @@ -1190,8 +1277,66 @@ copy_table_data(PGconn *origin_conn, PGconn *target_conn, } PQclear(res); - elog(INFO, "finished synchronization of data for table %s.%s", - remoterel->nspname, remoterel->relname); + /* + * Merge the staged rows. Rows we already have are left alone rather than + * overwritten: the local copy is the one the rest of the cluster has + * already replicated from us, so keeping it is the conservative choice. + */ + if (stage_load) + { + resetStringInfo(&query); + if (list_length(attnamelist)) + appendStringInfo(&query, + "INSERT INTO %s (%s) %sSELECT %s FROM pg_temp.%s " + "ON CONFLICT DO NOTHING", + relident.data, attlist.data, + override_identity ? "OVERRIDING SYSTEM VALUE " : "", + attlist.data, SPOCK_SYNC_STAGE_RELNAME); + else + appendStringInfo(&query, + "INSERT INTO %s %sSELECT * FROM pg_temp.%s " + "ON CONFLICT DO NOTHING", + relident.data, + override_identity ? "OVERRIDING SYSTEM VALUE " : "", + SPOCK_SYNC_STAGE_RELNAME); + + res = PQexec(target_conn, query.data); + if (PQresultStatus(res) != PGRES_COMMAND_OK) + { + char *msg = pstrdup(PQerrorMessage(target_conn)); + + PQclear(res); + ereport(ERROR, + (errmsg("merging synchronized data into %s.%s failed", + remoterel->nspname, remoterel->relname), + errdetail("destination connection reported: %s", msg))); + } + merged = pstrdup(PQcmdTuples(res)); + PQclear(res); + + resetStringInfo(&query); + appendStringInfo(&query, "DROP TABLE pg_temp.%s", + SPOCK_SYNC_STAGE_RELNAME); + res = PQexec(target_conn, query.data); + if (PQresultStatus(res) != PGRES_COMMAND_OK) + { + char *msg = pstrdup(PQerrorMessage(target_conn)); + + PQclear(res); + ereport(ERROR, + (errmsg("could not drop staging table for %s.%s", + remoterel->nspname, remoterel->relname), + errdetail("destination connection reported: %s", msg))); + } + PQclear(res); + } + + if (stage_load) + elog(INFO, "finished synchronization of data for table %s.%s, %s row(s) added to existing data", + remoterel->nspname, remoterel->relname, merged); + else + elog(INFO, "finished synchronization of data for table %s.%s", + remoterel->nspname, remoterel->relname); } /* From d35d6036314aa33f59acccd0c4cc1c66a3ba63e4 Mon Sep 17 00:00:00 2001 From: Ibrar Ahmed Date: Tue, 18 Aug 2026 00:52:24 +0500 Subject: [PATCH 2/2] Quote the resync hint and settle the empty-table check under a lock The hint is pasted into psql, so escape both arguments. The emptiness probe takes no lock, so re-check it under EXCLUSIVE before loading directly: a row committed after the first probe would abort the COPY and wedge the table. --- src/spock_apply.c | 17 +++++++--- src/spock_sync.c | 80 ++++++++++++++++++++++++++++++++++++++++------- 2 files changed, 82 insertions(+), 15 deletions(-) diff --git a/src/spock_apply.c b/src/spock_apply.c index 939e0080..c5d93781 100644 --- a/src/spock_apply.c +++ b/src/spock_apply.c @@ -4400,10 +4400,19 @@ process_syncing_tables(XLogRecPtr end_lsn) MySubscription->name, NameStr(sync->nspname), NameStr(sync->relname)), - errhint("Re-synchronize with spock.sub_resync_table('%s', '%s.%s') once the cause is fixed.", - MySubscription->name, - NameStr(sync->nspname), - NameStr(sync->relname)))); + /* + * The hint is meant to be pasted into psql, so both + * arguments have to survive names that need quoting: the + * relation is a regclass, which for a mixed-case or + * dotted name resolves to the wrong table (or nothing) + * unqualified, and an apostrophe in either name would + * truncate the literal. + */ + errhint("Re-synchronize with spock.sub_resync_table(%s, %s) once the cause is fixed.", + quote_literal_cstr(MySubscription->name), + quote_literal_cstr( + quote_qualified_identifier(NameStr(sync->nspname), + NameStr(sync->relname)))))); sync->status = SYNC_STATUS_FAILED; sync->statuslsn = InvalidXLogRecPtr; diff --git a/src/spock_sync.c b/src/spock_sync.c index f58f3171..c6f46732 100644 --- a/src/spock_sync.c +++ b/src/spock_sync.c @@ -993,6 +993,42 @@ make_copy_attnamelist(SpockRelation *rel) return attnamelist; } +/* + * Does the target table already hold rows? + * + * Chooses between the direct COPY and the staging path in copy_table_data(). + * Errors out rather than guessing, because getting this wrong in the "empty" + * direction is what wedges the table's sync status. + */ +static bool +target_table_has_rows(PGconn *target_conn, SpockRemoteRel *remoterel, + const char *relident) +{ + PGresult *res; + bool has_rows; + StringInfoData query; + + initStringInfo(&query); + appendStringInfo(&query, "SELECT 1 FROM %s LIMIT 1", relident); + res = PQexec(target_conn, query.data); + pfree(query.data); + + if (PQresultStatus(res) != PGRES_TUPLES_OK) + { + char *msg = pstrdup(PQerrorMessage(target_conn)); + + PQclear(res); + ereport(ERROR, + (errmsg("could not check whether target table %s.%s is empty", + remoterel->nspname, remoterel->relname), + errdetail("destination connection reported: %s", msg))); + } + has_rows = PQntuples(res) > 0; + PQclear(res); + + return has_rows; +} + /* * COPY single table over wire. */ @@ -1168,21 +1204,43 @@ copy_table_data(PGconn *origin_conn, PGconn *target_conn, PQescapeIdentifier(target_conn, remoterel->relname, strlen(remoterel->relname))); - resetStringInfo(&query); - appendStringInfo(&query, "SELECT 1 FROM %s LIMIT 1", relident.data); - res = PQexec(target_conn, query.data); - if (PQresultStatus(res) != PGRES_TUPLES_OK) + stage_load = target_table_has_rows(target_conn, remoterel, relident.data); + + if (!stage_load) { - char *msg = pstrdup(PQerrorMessage(target_conn)); + /* + * That probe took no lock, so on its own it does not settle anything: + * another transaction can commit a row between it and the COPY, and + * the COPY then aborts on the duplicate key and wedges the table's + * sync status, which is the exact failure the staging path exists to + * avoid. Lock writers out and ask again; the second answer holds for + * the rest of the copy transaction. + * + * The lock is taken only on this path. Here the table is empty, so + * nothing should be contending for it, and blocking writes to a table + * that is mid initial load is what we want anyway. Locking before the + * first probe would instead hold EXCLUSIVE on a populated table for + * the whole sync, which on a live node is a real availability cost. + */ + resetStringInfo(&query); + appendStringInfo(&query, "LOCK TABLE %s IN EXCLUSIVE MODE", + relident.data); + res = PQexec(target_conn, query.data); + if (PQresultStatus(res) != PGRES_COMMAND_OK) + { + char *msg = pstrdup(PQerrorMessage(target_conn)); + PQclear(res); + ereport(ERROR, + (errmsg("could not lock target table %s.%s for synchronization", + remoterel->nspname, remoterel->relname), + errdetail("destination connection reported: %s", msg))); + } PQclear(res); - ereport(ERROR, - (errmsg("could not check whether target table %s.%s is empty", - remoterel->nspname, remoterel->relname), - errdetail("destination connection reported: %s", msg))); + + stage_load = target_table_has_rows(target_conn, remoterel, + relident.data); } - stage_load = PQntuples(res) > 0; - PQclear(res); if (stage_load) {