From 9771a648e4d1398713c02d778aff196f55556ba0 Mon Sep 17 00:00:00 2001 From: Ibrar Ahmed Date: Thu, 13 Aug 2026 21:09:07 +0500 Subject: [PATCH 1/5] 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 c3331db7..f14c6962 100644 --- a/src/spock_apply.c +++ b/src/spock_apply.c @@ -4447,7 +4447,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 be566cb9..a35a8f5f 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 6d0cc722c18be5f451251b18a19425a0360e28fb Mon Sep 17 00:00:00 2001 From: Ibrar Ahmed Date: Tue, 18 Aug 2026 00:52:24 +0500 Subject: [PATCH 2/5] 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 f14c6962..5fc17dff 100644 --- a/src/spock_apply.c +++ b/src/spock_apply.c @@ -4460,10 +4460,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 a35a8f5f..215116da 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) { From 166c596853eda950629fc5e0820ba73df5862811 Mon Sep 17 00:00:00 2001 From: Ibrar Ahmed Date: Thu, 20 Aug 2026 18:46:50 +0500 Subject: [PATCH 3/5] Fix the staged sync path and bound its table lock. A bare LIKE copies NOT NULL but not the default, generated expression or identity that fills it, so staging a copy aborted on the constraint. Bound the empty-table lock under a savepoint so it falls back instead of hanging. --- src/spock_sync.c | 147 ++++++++++++++++++++++++++++++++--------------- 1 file changed, 102 insertions(+), 45 deletions(-) diff --git a/src/spock_sync.c b/src/spock_sync.c index 215116da..6af5990f 100644 --- a/src/spock_sync.c +++ b/src/spock_sync.c @@ -78,6 +78,10 @@ */ #define SPOCK_SYNC_STAGE_RELNAME "spock_sync_stage" +/* Savepoint and bound for the empty-table lock in copy_table_data(). */ +#define SPOCK_SYNC_LOCK_SAVEPOINT "spock_sync_lock" +#define SPOCK_SYNC_LOCK_TIMEOUT_MS 5000 + #define Natts_local_sync_state 6 #define Anum_sync_kind 1 #define Anum_sync_subid 2 @@ -1029,6 +1033,31 @@ target_table_has_rows(PGconn *target_conn, SpockRemoteRel *remoterel, return has_rows; } +/* + * Run one command on the target connection during a table copy. + * + * Never returns on failure. On success the caller owns the result. + */ +static PGresult * +sync_target_cmd(PGconn *target_conn, const char *sql, + SpockRemoteRel *remoterel, const char *what) +{ + PGresult *res = PQexec(target_conn, sql); + + if (PQresultStatus(res) != PGRES_COMMAND_OK) + { + char *msg = pstrdup(PQerrorMessage(target_conn)); + + PQclear(res); + ereport(ERROR, + (errmsg("could not %s for %s.%s", what, + remoterel->nspname, remoterel->relname), + errdetail("destination connection reported: %s", msg))); + } + + return res; +} + /* * COPY single table over wire. */ @@ -1221,47 +1250,94 @@ copy_table_data(PGconn *origin_conn, PGconn *target_conn, * 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. + * + * The wait is bounded, because every table is copied in one + * transaction: this lock is held until the last table is done, and an + * apply worker holding ROW EXCLUSIVE on a table this sync has not + * reached yet would deadlock against it. A savepoint keeps the failure + * recoverable, since an error would otherwise abort the copy + * transaction. If the lock does not arrive, fall back to the staging + * path, which is correct whether or not the table is empty. */ + PQclear(sync_target_cmd(target_conn, + "SAVEPOINT " SPOCK_SYNC_LOCK_SAVEPOINT, + remoterel, "open a savepoint")); + resetStringInfo(&query); - appendStringInfo(&query, "LOCK TABLE %s IN EXCLUSIVE MODE", - relident.data); + appendStringInfo(&query, + "SET LOCAL lock_timeout = %d;" + "LOCK TABLE %s IN EXCLUSIVE MODE", + SPOCK_SYNC_LOCK_TIMEOUT_MS, relident.data); res = PQexec(target_conn, query.data); + if (PQresultStatus(res) != PGRES_COMMAND_OK) { + char *sqlstate = PQresultErrorField(res, PG_DIAG_SQLSTATE); + bool busy = sqlstate != NULL && + strcmp(sqlstate, "55P03" /*ERRCODE_LOCK_NOT_AVAILABLE*/) == 0; 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))); + + /* + * Rolling back leaves the savepoint live, so release it too. Every + * table in the sync shares this transaction, and one dangling + * subtransaction per table would push a large sync past the 64 the + * snapshot can track without overflowing. + */ + PQclear(sync_target_cmd(target_conn, + "ROLLBACK TO SAVEPOINT " SPOCK_SYNC_LOCK_SAVEPOINT, + remoterel, "roll back to the lock savepoint")); + PQclear(sync_target_cmd(target_conn, + "RELEASE SAVEPOINT " SPOCK_SYNC_LOCK_SAVEPOINT, + remoterel, "release the lock savepoint")); + + if (!busy) + ereport(ERROR, + (errmsg("could not lock target table %s.%s for synchronization", + remoterel->nspname, remoterel->relname), + errdetail("destination connection reported: %s", msg))); + + elog(LOG, "SPOCK: could not lock %s.%s within %dms, staging the copy instead", + remoterel->nspname, remoterel->relname, + SPOCK_SYNC_LOCK_TIMEOUT_MS); + stage_load = true; } - PQclear(res); + else + { + PQclear(res); + + /* Restore what start_copy_target_tx() set for the rest of the copy. */ + PQclear(sync_target_cmd(target_conn, "SET LOCAL lock_timeout = 0", + remoterel, "reset lock_timeout")); + PQclear(sync_target_cmd(target_conn, + "RELEASE SAVEPOINT " SPOCK_SYNC_LOCK_SAVEPOINT, + remoterel, "release the lock savepoint")); - stage_load = target_table_has_rows(target_conn, remoterel, - relident.data); + stage_load = target_table_has_rows(target_conn, remoterel, + relident.data); + } } if (stage_load) { + /* + * A bare LIKE copies NOT NULL but not the defaults, generation + * expressions or identity behind it, and make_copy_attnamelist() + * leaves generated and provider-absent columns out of the COPY. The + * staging table would then take a NULL where the real table computes + * a value, and the COPY would abort on the not-null constraint. + */ resetStringInfo(&query); appendStringInfo(&query, "DROP TABLE IF EXISTS pg_temp.%s;" - "CREATE TEMP TABLE %s (LIKE %s)", + "CREATE TEMP TABLE %s (LIKE %s" + " INCLUDING DEFAULTS INCLUDING GENERATED" + " INCLUDING IDENTITY)", 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); + PQclear(sync_target_cmd(target_conn, query.data, remoterel, + "create the staging table")); } /* Build COPY FROM query. */ @@ -1358,35 +1434,16 @@ copy_table_data(PGconn *origin_conn, PGconn *target_conn, 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))); - } + res = sync_target_cmd(target_conn, query.data, remoterel, + "merge the staged rows"); 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); + PQclear(sync_target_cmd(target_conn, query.data, remoterel, + "drop the staging table")); } if (stage_load) From f5852453688202d18862ba1dfcb6ec0f0e3353cf Mon Sep 17 00:00:00 2001 From: Ibrar Ahmed Date: Fri, 21 Aug 2026 08:59:30 +0500 Subject: [PATCH 4/5] Drop the sync lock when staging, and name the merge columns. Rolling back to the savepoint releases the EXCLUSIVE lock, so a table found non-empty after locking no longer blocks writers for the whole copy. The merge now names its columns; SELECT * would pass back a generated one. --- src/spock_sync.c | 77 +++++++++++++++++++++++++++++++++++------------- 1 file changed, 57 insertions(+), 20 deletions(-) diff --git a/src/spock_sync.c b/src/spock_sync.c index 6af5990f..46d6023e 100644 --- a/src/spock_sync.c +++ b/src/spock_sync.c @@ -1307,15 +1307,33 @@ copy_table_data(PGconn *origin_conn, PGconn *target_conn, { PQclear(res); - /* Restore what start_copy_target_tx() set for the rest of the copy. */ - PQclear(sync_target_cmd(target_conn, "SET LOCAL lock_timeout = 0", - remoterel, "reset lock_timeout")); + stage_load = target_table_has_rows(target_conn, remoterel, + relident.data); + + if (stage_load) + { + /* + * Rows landed between the first probe and the lock. Staging + * copes with that, and holding EXCLUSIVE on a populated table + * for the rest of the sync is the availability cost this path + * exists to avoid, so drop the lock again. Rolling back to the + * savepoint releases locks taken inside it, and undoes the + * lock_timeout with them. + */ + PQclear(sync_target_cmd(target_conn, + "ROLLBACK TO SAVEPOINT " SPOCK_SYNC_LOCK_SAVEPOINT, + remoterel, "roll back to the lock savepoint")); + } + else + { + /* Keep the lock, restore what start_copy_target_tx() set. */ + PQclear(sync_target_cmd(target_conn, "SET LOCAL lock_timeout = 0", + remoterel, "reset lock_timeout")); + } + PQclear(sync_target_cmd(target_conn, "RELEASE SAVEPOINT " SPOCK_SYNC_LOCK_SAVEPOINT, remoterel, "release the lock savepoint")); - - stage_load = target_table_has_rows(target_conn, remoterel, - relident.data); } } @@ -1418,21 +1436,40 @@ copy_table_data(PGconn *origin_conn, PGconn *target_conn, */ if (stage_load) { + const char *mergelist = attlist.data; + + /* + * The copy had no column list, so it moved every non-generated column. + * Name them for the merge rather than using SELECT *, which would hand + * the target a generated column and be rejected. + */ + if (!list_length(attnamelist)) + { + res = sync_target_cmd(target_conn, + "SELECT string_agg(quote_ident(attname), ', ' ORDER BY attnum)" + " FROM pg_attribute" + " WHERE attrelid = 'pg_temp." SPOCK_SYNC_STAGE_RELNAME "'::regclass" + " AND attnum > 0 AND NOT attisdropped AND attgenerated = ''", + remoterel, "list the staging table columns"); + + if (PQntuples(res) != 1 || PQgetisnull(res, 0, 0)) + { + PQclear(res); + ereport(ERROR, + (errmsg("staging table for %s.%s has no columns to merge", + remoterel->nspname, remoterel->relname))); + } + mergelist = pstrdup(PQgetvalue(res, 0, 0)); + PQclear(res); + } + 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); + appendStringInfo(&query, + "INSERT INTO %s (%s) %sSELECT %s FROM pg_temp.%s " + "ON CONFLICT DO NOTHING", + relident.data, mergelist, + override_identity ? "OVERRIDING SYSTEM VALUE " : "", + mergelist, SPOCK_SYNC_STAGE_RELNAME); res = sync_target_cmd(target_conn, query.data, remoterel, "merge the staged rows"); From 8e012bb7452fa7a485aafcaccaff1743dd6cb1c4 Mon Sep 17 00:00:00 2001 From: Ibrar Ahmed Date: Fri, 21 Aug 2026 08:59:36 +0500 Subject: [PATCH 5/5] Use done_testing() in 014_pgdump_restore_conflict. The file planned 20 tests but ran 28, so the last eight were reported as failures. The suite only runs nightly, so pull request CI never showed it. --- tests/tap/t/014_pgdump_restore_conflict.pl | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/tests/tap/t/014_pgdump_restore_conflict.pl b/tests/tap/t/014_pgdump_restore_conflict.pl index 7c11a6da..1b6111ec 100755 --- a/tests/tap/t/014_pgdump_restore_conflict.pl +++ b/tests/tap/t/014_pgdump_restore_conflict.pl @@ -1,6 +1,6 @@ use strict; use warnings; -use Test::More tests => 20; +use Test::More; use lib '.'; use SpockTest qw(create_cluster destroy_cluster system_or_bail command_ok get_test_config scalar_query psql_or_bail); @@ -341,3 +341,5 @@ sub wait_for_value { unlink $dump_file if -e $dump_file; destroy_cluster('Cleanup pg_dump/restore conflict test cluster'); + +done_testing();