Table sync with merge when the target already holds rows

docs/configuring.md

@@ -252,6 +252,36 @@ liveness detection. Default: `300` (5 minutes).
 spock.apply_idle_timeout = 300
 ```
 
+### `spock.sync_stage_and_merge`
+
+Controls how the initial table copy behaves when the table on the subscriber
+already holds rows. Default: `off`.
+
+With this `off`, the copy is a direct `COPY` into the table, which is what
+Spock has always done. If the table is not empty, that `COPY` aborts on the
+first duplicate key, the table's entry in `spock.local_sync_status` is left at
+`failed`, and from that point the apply worker discards every change for the
+table - so it stops replicating and silently diverges.
+
+With this `on`, a table that already holds rows is copied into a temporary
+staging table instead and then merged with `ON CONFLICT DO NOTHING`. Rows
+already present locally are kept rather than overwritten, so a sync of data the
+node already has converges instead of wedging.
+
+This matters most when adding an already-populated table to a replication set
+with `synchronize_data := true`, which asks every peer to copy rows it may
+already hold.
+
+The copy runs in Spock's sync worker rather than in your session, so a
+session-level `SET` has no effect on it. Set it in `postgresql.conf`, or with
+`ALTER SYSTEM SET spock.sync_stage_and_merge = on` followed by
+`SELECT pg_reload_conf()`, and set it on the subscriber - the node receiving
+the copy.
+
+```
+spock.sync_stage_and_merge = off
+```
+
 ### `spock.sync_timeout`
 
 Overrides the time (in seconds) budgeted for a single synchronisation wait in

include/spock.h

@@ -51,6 +51,7 @@ extern int	restart_delay_on_exception;
 extern int	spock_replay_queue_size;
 extern int	spock_pause_timeout;
 extern int	spock_sync_timeout;
+extern bool spock_sync_stage_and_merge;
 extern int	spock_read_retry_count;
 extern bool check_all_uc_indexes;
 extern bool spock_enable_quiet_mode;

src/spock.c

@@ -178,6 +178,9 @@ int			spock_pause_timeout = 10;	/* seconds to wait for apply workers
 										 * to pause */
 int			spock_sync_timeout = 0; /* seconds per sync wait; 0 = routine's
 									 * own default */
+bool		spock_sync_stage_and_merge = false;	/* stage the initial COPY
+												 * when the target is not
+												 * empty */
 int			spock_read_retry_count = 5; /* heap update/delete: retries when
 										 * local tuple is missing */
 bool		check_all_uc_indexes = false;
@@ -1324,6 +1327,19 @@ _PG_init(void)
 							 0,
 							 NULL, NULL, NULL);
 
+	DefineCustomBoolVariable("spock.sync_stage_and_merge",
+							 "Load the initial table copy through a staging table when the target already holds rows.",
+							 "Off by default, which keeps the direct COPY: a target that is not "
+							 "empty aborts on the first duplicate key and the table's sync status "
+							 "stays failed. On, such a table is copied into a temporary staging "
+							 "table and merged with ON CONFLICT DO NOTHING, so rows already "
+							 "present are kept and the sync converges instead.",
+							 &spock_sync_stage_and_merge,
+							 false,
+							 PGC_USERSET,
+							 0,
+							 NULL, NULL, NULL);
+
 	DefineCustomIntVariable("spock.restart_delay_default",
 							"Default apply-worker restart delay in ms",
 							NULL,

src/spock_apply.c

@@ -4471,7 +4471,33 @@ 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)),
+					/*
+					 * 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;
 			}