Line data Source code
1 : /*-------------------------------------------------------------------------
2 : * tablesync.c
3 : * PostgreSQL logical replication: initial table data synchronization
4 : *
5 : * Copyright (c) 2012-2025, PostgreSQL Global Development Group
6 : *
7 : * IDENTIFICATION
8 : * src/backend/replication/logical/tablesync.c
9 : *
10 : * NOTES
11 : * This file contains code for initial table data synchronization for
12 : * logical replication.
13 : *
14 : * The initial data synchronization is done separately for each table,
15 : * in a separate apply worker that only fetches the initial snapshot data
16 : * from the publisher and then synchronizes the position in the stream with
17 : * the leader apply worker.
18 : *
19 : * There are several reasons for doing the synchronization this way:
20 : * - It allows us to parallelize the initial data synchronization
21 : * which lowers the time needed for it to happen.
22 : * - The initial synchronization does not have to hold the xid and LSN
23 : * for the time it takes to copy data of all tables, causing less
24 : * bloat and lower disk consumption compared to doing the
25 : * synchronization in a single process for the whole database.
26 : * - It allows us to synchronize any tables added after the initial
27 : * synchronization has finished.
28 : *
29 : * The stream position synchronization works in multiple steps:
30 : * - Apply worker requests a tablesync worker to start, setting the new
31 : * table state to INIT.
32 : * - Tablesync worker starts; changes table state from INIT to DATASYNC while
33 : * copying.
34 : * - Tablesync worker does initial table copy; there is a FINISHEDCOPY (sync
35 : * worker specific) state to indicate when the copy phase has completed, so
36 : * if the worker crashes with this (non-memory) state then the copy will not
37 : * be re-attempted.
38 : * - Tablesync worker then sets table state to SYNCWAIT; waits for state change.
39 : * - Apply worker periodically checks for tables in SYNCWAIT state. When
40 : * any appear, it sets the table state to CATCHUP and starts loop-waiting
41 : * until either the table state is set to SYNCDONE or the sync worker
42 : * exits.
43 : * - After the sync worker has seen the state change to CATCHUP, it will
44 : * read the stream and apply changes (acting like an apply worker) until
45 : * it catches up to the specified stream position. Then it sets the
46 : * state to SYNCDONE. There might be zero changes applied between
47 : * CATCHUP and SYNCDONE, because the sync worker might be ahead of the
48 : * apply worker.
49 : * - Once the state is set to SYNCDONE, the apply will continue tracking
50 : * the table until it reaches the SYNCDONE stream position, at which
51 : * point it sets state to READY and stops tracking. Again, there might
52 : * be zero changes in between.
53 : *
54 : * So the state progression is always: INIT -> DATASYNC -> FINISHEDCOPY
55 : * -> SYNCWAIT -> CATCHUP -> SYNCDONE -> READY.
56 : *
57 : * The catalog pg_subscription_rel is used to keep information about
58 : * subscribed tables and their state. The catalog holds all states
59 : * except SYNCWAIT and CATCHUP which are only in shared memory.
60 : *
61 : * Example flows look like this:
62 : * - Apply is in front:
63 : * sync:8
64 : * -> set in catalog FINISHEDCOPY
65 : * -> set in memory SYNCWAIT
66 : * apply:10
67 : * -> set in memory CATCHUP
68 : * -> enter wait-loop
69 : * sync:10
70 : * -> set in catalog SYNCDONE
71 : * -> exit
72 : * apply:10
73 : * -> exit wait-loop
74 : * -> continue rep
75 : * apply:11
76 : * -> set in catalog READY
77 : *
78 : * - Sync is in front:
79 : * sync:10
80 : * -> set in catalog FINISHEDCOPY
81 : * -> set in memory SYNCWAIT
82 : * apply:8
83 : * -> set in memory CATCHUP
84 : * -> continue per-table filtering
85 : * sync:10
86 : * -> set in catalog SYNCDONE
87 : * -> exit
88 : * apply:10
89 : * -> set in catalog READY
90 : * -> stop per-table filtering
91 : * -> continue rep
92 : *-------------------------------------------------------------------------
93 : */
94 :
95 : #include "postgres.h"
96 :
97 : #include "access/table.h"
98 : #include "access/xact.h"
99 : #include "catalog/indexing.h"
100 : #include "catalog/pg_subscription_rel.h"
101 : #include "catalog/pg_type.h"
102 : #include "commands/copy.h"
103 : #include "miscadmin.h"
104 : #include "nodes/makefuncs.h"
105 : #include "parser/parse_relation.h"
106 : #include "pgstat.h"
107 : #include "replication/logicallauncher.h"
108 : #include "replication/logicalrelation.h"
109 : #include "replication/logicalworker.h"
110 : #include "replication/origin.h"
111 : #include "replication/slot.h"
112 : #include "replication/walreceiver.h"
113 : #include "replication/worker_internal.h"
114 : #include "storage/ipc.h"
115 : #include "storage/lmgr.h"
116 : #include "utils/acl.h"
117 : #include "utils/array.h"
118 : #include "utils/builtins.h"
119 : #include "utils/lsyscache.h"
120 : #include "utils/rls.h"
121 : #include "utils/snapmgr.h"
122 : #include "utils/syscache.h"
123 : #include "utils/usercontext.h"
124 :
125 : List *table_states_not_ready = NIL;
126 :
127 : static StringInfo copybuf = NULL;
128 :
129 : /*
130 : * Wait until the relation sync state is set in the catalog to the expected
131 : * one; return true when it happens.
132 : *
133 : * Returns false if the table sync worker or the table itself have
134 : * disappeared, or the table state has been reset.
135 : *
136 : * Currently, this is used in the apply worker when transitioning from
137 : * CATCHUP state to SYNCDONE.
138 : */
139 : static bool
140 354 : wait_for_table_state_change(Oid relid, char expected_state)
141 : {
142 : char state;
143 :
144 : for (;;)
145 424 : {
146 : LogicalRepWorker *worker;
147 : XLogRecPtr statelsn;
148 :
149 778 : CHECK_FOR_INTERRUPTS();
150 :
151 778 : InvalidateCatalogSnapshot();
152 778 : state = GetSubscriptionRelState(MyLogicalRepWorker->subid,
153 : relid, &statelsn);
154 :
155 778 : if (state == SUBREL_STATE_UNKNOWN)
156 0 : break;
157 :
158 778 : if (state == expected_state)
159 354 : return true;
160 :
161 : /* Check if the sync worker is still running and bail if not. */
162 424 : LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
163 424 : worker = logicalrep_worker_find(WORKERTYPE_TABLESYNC,
164 424 : MyLogicalRepWorker->subid, relid,
165 : false);
166 424 : LWLockRelease(LogicalRepWorkerLock);
167 424 : if (!worker)
168 0 : break;
169 :
170 424 : (void) WaitLatch(MyLatch,
171 : WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
172 : 1000L, WAIT_EVENT_LOGICAL_SYNC_STATE_CHANGE);
173 :
174 424 : ResetLatch(MyLatch);
175 : }
176 :
177 0 : return false;
178 : }
179 :
180 : /*
181 : * Wait until the apply worker changes the state of our synchronization
182 : * worker to the expected one.
183 : *
184 : * Used when transitioning from SYNCWAIT state to CATCHUP.
185 : *
186 : * Returns false if the apply worker has disappeared.
187 : */
188 : static bool
189 368 : wait_for_worker_state_change(char expected_state)
190 : {
191 : int rc;
192 :
193 : for (;;)
194 370 : {
195 : LogicalRepWorker *worker;
196 :
197 738 : CHECK_FOR_INTERRUPTS();
198 :
199 : /*
200 : * Done if already in correct state. (We assume this fetch is atomic
201 : * enough to not give a misleading answer if we do it with no lock.)
202 : */
203 738 : if (MyLogicalRepWorker->relstate == expected_state)
204 368 : return true;
205 :
206 : /*
207 : * Bail out if the apply worker has died, else signal it we're
208 : * waiting.
209 : */
210 370 : LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
211 370 : worker = logicalrep_worker_find(WORKERTYPE_APPLY,
212 370 : MyLogicalRepWorker->subid, InvalidOid,
213 : false);
214 370 : if (worker && worker->proc)
215 370 : logicalrep_worker_wakeup_ptr(worker);
216 370 : LWLockRelease(LogicalRepWorkerLock);
217 370 : if (!worker)
218 0 : break;
219 :
220 : /*
221 : * Wait. We expect to get a latch signal back from the apply worker,
222 : * but use a timeout in case it dies without sending one.
223 : */
224 370 : rc = WaitLatch(MyLatch,
225 : WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
226 : 1000L, WAIT_EVENT_LOGICAL_SYNC_STATE_CHANGE);
227 :
228 370 : if (rc & WL_LATCH_SET)
229 370 : ResetLatch(MyLatch);
230 : }
231 :
232 0 : return false;
233 : }
234 :
235 : /*
236 : * Handle table synchronization cooperation from the synchronization
237 : * worker.
238 : *
239 : * If the sync worker is in CATCHUP state and reached (or passed) the
240 : * predetermined synchronization point in the WAL stream, mark the table as
241 : * SYNCDONE and finish.
242 : */
243 : void
244 448 : ProcessSyncingTablesForSync(XLogRecPtr current_lsn)
245 : {
246 448 : SpinLockAcquire(&MyLogicalRepWorker->relmutex);
247 :
248 448 : if (MyLogicalRepWorker->relstate == SUBREL_STATE_CATCHUP &&
249 448 : current_lsn >= MyLogicalRepWorker->relstate_lsn)
250 : {
251 : TimeLineID tli;
252 368 : char syncslotname[NAMEDATALEN] = {0};
253 368 : char originname[NAMEDATALEN] = {0};
254 :
255 368 : MyLogicalRepWorker->relstate = SUBREL_STATE_SYNCDONE;
256 368 : MyLogicalRepWorker->relstate_lsn = current_lsn;
257 :
258 368 : SpinLockRelease(&MyLogicalRepWorker->relmutex);
259 :
260 : /*
261 : * UpdateSubscriptionRelState must be called within a transaction.
262 : */
263 368 : if (!IsTransactionState())
264 368 : StartTransactionCommand();
265 :
266 368 : UpdateSubscriptionRelState(MyLogicalRepWorker->subid,
267 368 : MyLogicalRepWorker->relid,
268 368 : MyLogicalRepWorker->relstate,
269 368 : MyLogicalRepWorker->relstate_lsn,
270 : false);
271 :
272 : /*
273 : * End streaming so that LogRepWorkerWalRcvConn can be used to drop
274 : * the slot.
275 : */
276 368 : walrcv_endstreaming(LogRepWorkerWalRcvConn, &tli);
277 :
278 : /*
279 : * Cleanup the tablesync slot.
280 : *
281 : * This has to be done after updating the state because otherwise if
282 : * there is an error while doing the database operations we won't be
283 : * able to rollback dropped slot.
284 : */
285 368 : ReplicationSlotNameForTablesync(MyLogicalRepWorker->subid,
286 368 : MyLogicalRepWorker->relid,
287 : syncslotname,
288 : sizeof(syncslotname));
289 :
290 : /*
291 : * It is important to give an error if we are unable to drop the slot,
292 : * otherwise, it won't be dropped till the corresponding subscription
293 : * is dropped. So passing missing_ok = false.
294 : */
295 368 : ReplicationSlotDropAtPubNode(LogRepWorkerWalRcvConn, syncslotname, false);
296 :
297 368 : CommitTransactionCommand();
298 368 : pgstat_report_stat(false);
299 :
300 : /*
301 : * Start a new transaction to clean up the tablesync origin tracking.
302 : * This transaction will be ended within the FinishSyncWorker(). Now,
303 : * even, if we fail to remove this here, the apply worker will ensure
304 : * to clean it up afterward.
305 : *
306 : * We need to do this after the table state is set to SYNCDONE.
307 : * Otherwise, if an error occurs while performing the database
308 : * operation, the worker will be restarted and the in-memory state of
309 : * replication progress (remote_lsn) won't be rolled-back which would
310 : * have been cleared before restart. So, the restarted worker will use
311 : * invalid replication progress state resulting in replay of
312 : * transactions that have already been applied.
313 : */
314 368 : StartTransactionCommand();
315 :
316 368 : ReplicationOriginNameForLogicalRep(MyLogicalRepWorker->subid,
317 368 : MyLogicalRepWorker->relid,
318 : originname,
319 : sizeof(originname));
320 :
321 : /*
322 : * Resetting the origin session removes the ownership of the slot.
323 : * This is needed to allow the origin to be dropped.
324 : */
325 368 : replorigin_session_reset();
326 368 : replorigin_session_origin = InvalidRepOriginId;
327 368 : replorigin_session_origin_lsn = InvalidXLogRecPtr;
328 368 : replorigin_session_origin_timestamp = 0;
329 :
330 : /*
331 : * Drop the tablesync's origin tracking if exists.
332 : *
333 : * There is a chance that the user is concurrently performing refresh
334 : * for the subscription where we remove the table state and its origin
335 : * or the apply worker would have removed this origin. So passing
336 : * missing_ok = true.
337 : */
338 368 : replorigin_drop_by_name(originname, true, false);
339 :
340 368 : FinishSyncWorker();
341 : }
342 : else
343 80 : SpinLockRelease(&MyLogicalRepWorker->relmutex);
344 80 : }
345 :
346 : /*
347 : * Handle table synchronization cooperation from the apply worker.
348 : *
349 : * Walk over all subscription tables that are individually tracked by the
350 : * apply process (currently, all that have state other than
351 : * SUBREL_STATE_READY) and manage synchronization for them.
352 : *
353 : * If there are tables that need synchronizing and are not being synchronized
354 : * yet, start sync workers for them (if there are free slots for sync
355 : * workers). To prevent starting the sync worker for the same relation at a
356 : * high frequency after a failure, we store its last start time with each sync
357 : * state info. We start the sync worker for the same relation after waiting
358 : * at least wal_retrieve_retry_interval.
359 : *
360 : * For tables that are being synchronized already, check if sync workers
361 : * either need action from the apply worker or have finished. This is the
362 : * SYNCWAIT to CATCHUP transition.
363 : *
364 : * If the synchronization position is reached (SYNCDONE), then the table can
365 : * be marked as READY and is no longer tracked.
366 : */
367 : void
368 16504 : ProcessSyncingTablesForApply(XLogRecPtr current_lsn)
369 : {
370 : struct tablesync_start_time_mapping
371 : {
372 : Oid relid;
373 : TimestampTz last_start_time;
374 : };
375 : static HTAB *last_start_times = NULL;
376 : ListCell *lc;
377 16504 : bool started_tx = false;
378 16504 : bool should_exit = false;
379 16504 : Relation rel = NULL;
380 :
381 : Assert(!IsTransactionState());
382 :
383 : /* We need up-to-date sync state info for subscription tables here. */
384 16504 : FetchRelationStates(&started_tx);
385 :
386 : /*
387 : * Prepare a hash table for tracking last start times of workers, to avoid
388 : * immediate restarts. We don't need it if there are no tables that need
389 : * syncing.
390 : */
391 16504 : if (table_states_not_ready != NIL && !last_start_times)
392 238 : {
393 : HASHCTL ctl;
394 :
395 238 : ctl.keysize = sizeof(Oid);
396 238 : ctl.entrysize = sizeof(struct tablesync_start_time_mapping);
397 238 : last_start_times = hash_create("Logical replication table sync worker start times",
398 : 256, &ctl, HASH_ELEM | HASH_BLOBS);
399 : }
400 :
401 : /*
402 : * Clean up the hash table when we're done with all tables (just to
403 : * release the bit of memory).
404 : */
405 16266 : else if (table_states_not_ready == NIL && last_start_times)
406 : {
407 176 : hash_destroy(last_start_times);
408 176 : last_start_times = NULL;
409 : }
410 :
411 : /*
412 : * Process all tables that are being synchronized.
413 : */
414 19876 : foreach(lc, table_states_not_ready)
415 : {
416 3374 : SubscriptionRelState *rstate = (SubscriptionRelState *) lfirst(lc);
417 :
418 3374 : if (rstate->state == SUBREL_STATE_SYNCDONE)
419 : {
420 : /*
421 : * Apply has caught up to the position where the table sync has
422 : * finished. Mark the table as ready so that the apply will just
423 : * continue to replicate it normally.
424 : */
425 354 : if (current_lsn >= rstate->lsn)
426 : {
427 : char originname[NAMEDATALEN];
428 :
429 352 : rstate->state = SUBREL_STATE_READY;
430 352 : rstate->lsn = current_lsn;
431 352 : if (!started_tx)
432 : {
433 24 : StartTransactionCommand();
434 24 : started_tx = true;
435 : }
436 :
437 : /*
438 : * Remove the tablesync origin tracking if exists.
439 : *
440 : * There is a chance that the user is concurrently performing
441 : * refresh for the subscription where we remove the table
442 : * state and its origin or the tablesync worker would have
443 : * already removed this origin. We can't rely on tablesync
444 : * worker to remove the origin tracking as if there is any
445 : * error while dropping we won't restart it to drop the
446 : * origin. So passing missing_ok = true.
447 : *
448 : * Lock the subscription and origin in the same order as we
449 : * are doing during DDL commands to avoid deadlocks. See
450 : * AlterSubscription_refresh.
451 : */
452 352 : LockSharedObject(SubscriptionRelationId, MyLogicalRepWorker->subid,
453 : 0, AccessShareLock);
454 :
455 352 : if (!rel)
456 352 : rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);
457 :
458 352 : ReplicationOriginNameForLogicalRep(MyLogicalRepWorker->subid,
459 : rstate->relid,
460 : originname,
461 : sizeof(originname));
462 352 : replorigin_drop_by_name(originname, true, false);
463 :
464 : /*
465 : * Update the state to READY only after the origin cleanup.
466 : */
467 352 : UpdateSubscriptionRelState(MyLogicalRepWorker->subid,
468 352 : rstate->relid, rstate->state,
469 : rstate->lsn, true);
470 : }
471 : }
472 : else
473 : {
474 : LogicalRepWorker *syncworker;
475 :
476 : /*
477 : * Look for a sync worker for this relation.
478 : */
479 3020 : LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
480 :
481 3020 : syncworker = logicalrep_worker_find(WORKERTYPE_TABLESYNC,
482 3020 : MyLogicalRepWorker->subid,
483 : rstate->relid, false);
484 :
485 3020 : if (syncworker)
486 : {
487 : /* Found one, update our copy of its state */
488 1320 : SpinLockAcquire(&syncworker->relmutex);
489 1320 : rstate->state = syncworker->relstate;
490 1320 : rstate->lsn = syncworker->relstate_lsn;
491 1320 : if (rstate->state == SUBREL_STATE_SYNCWAIT)
492 : {
493 : /*
494 : * Sync worker is waiting for apply. Tell sync worker it
495 : * can catchup now.
496 : */
497 354 : syncworker->relstate = SUBREL_STATE_CATCHUP;
498 354 : syncworker->relstate_lsn =
499 354 : Max(syncworker->relstate_lsn, current_lsn);
500 : }
501 1320 : SpinLockRelease(&syncworker->relmutex);
502 :
503 : /* If we told worker to catch up, wait for it. */
504 1320 : if (rstate->state == SUBREL_STATE_SYNCWAIT)
505 : {
506 : /* Signal the sync worker, as it may be waiting for us. */
507 354 : if (syncworker->proc)
508 354 : logicalrep_worker_wakeup_ptr(syncworker);
509 :
510 : /* Now safe to release the LWLock */
511 354 : LWLockRelease(LogicalRepWorkerLock);
512 :
513 354 : if (started_tx)
514 : {
515 : /*
516 : * We must commit the existing transaction to release
517 : * the existing locks before entering a busy loop.
518 : * This is required to avoid any undetected deadlocks
519 : * due to any existing lock as deadlock detector won't
520 : * be able to detect the waits on the latch.
521 : *
522 : * Also close any tables prior to the commit.
523 : */
524 354 : if (rel)
525 : {
526 52 : table_close(rel, NoLock);
527 52 : rel = NULL;
528 : }
529 354 : CommitTransactionCommand();
530 354 : pgstat_report_stat(false);
531 : }
532 :
533 : /*
534 : * Enter busy loop and wait for synchronization worker to
535 : * reach expected state (or die trying).
536 : */
537 354 : StartTransactionCommand();
538 354 : started_tx = true;
539 :
540 354 : wait_for_table_state_change(rstate->relid,
541 : SUBREL_STATE_SYNCDONE);
542 : }
543 : else
544 966 : LWLockRelease(LogicalRepWorkerLock);
545 : }
546 : else
547 : {
548 : /*
549 : * If there is no sync worker for this table yet, count
550 : * running sync workers for this subscription, while we have
551 : * the lock.
552 : */
553 : int nsyncworkers =
554 1700 : logicalrep_sync_worker_count(MyLogicalRepWorker->subid);
555 :
556 : /* Now safe to release the LWLock */
557 1700 : LWLockRelease(LogicalRepWorkerLock);
558 :
559 : /*
560 : * If there are free sync worker slot(s), start a new sync
561 : * worker for the table.
562 : */
563 1700 : if (nsyncworkers < max_sync_workers_per_subscription)
564 : {
565 512 : TimestampTz now = GetCurrentTimestamp();
566 : struct tablesync_start_time_mapping *hentry;
567 : bool found;
568 :
569 512 : hentry = hash_search(last_start_times, &rstate->relid,
570 : HASH_ENTER, &found);
571 :
572 652 : if (!found ||
573 140 : TimestampDifferenceExceeds(hentry->last_start_time, now,
574 : wal_retrieve_retry_interval))
575 : {
576 : /*
577 : * Set the last_start_time even if we fail to start
578 : * the worker, so that we won't retry until
579 : * wal_retrieve_retry_interval has elapsed.
580 : */
581 396 : hentry->last_start_time = now;
582 396 : (void) logicalrep_worker_launch(WORKERTYPE_TABLESYNC,
583 396 : MyLogicalRepWorker->dbid,
584 396 : MySubscription->oid,
585 396 : MySubscription->name,
586 396 : MyLogicalRepWorker->userid,
587 : rstate->relid,
588 : DSM_HANDLE_INVALID,
589 : false);
590 : }
591 : }
592 : }
593 : }
594 : }
595 :
596 : /* Close table if opened */
597 16502 : if (rel)
598 300 : table_close(rel, NoLock);
599 :
600 :
601 16502 : if (started_tx)
602 : {
603 : /*
604 : * Even when the two_phase mode is requested by the user, it remains
605 : * as 'pending' until all tablesyncs have reached READY state.
606 : *
607 : * When this happens, we restart the apply worker and (if the
608 : * conditions are still ok) then the two_phase tri-state will become
609 : * 'enabled' at that time.
610 : *
611 : * Note: If the subscription has no tables then leave the state as
612 : * PENDING, which allows ALTER SUBSCRIPTION ... REFRESH PUBLICATION to
613 : * work.
614 : */
615 1646 : if (MySubscription->twophasestate == LOGICALREP_TWOPHASE_STATE_PENDING)
616 : {
617 50 : CommandCounterIncrement(); /* make updates visible */
618 50 : if (AllTablesyncsReady())
619 : {
620 12 : ereport(LOG,
621 : (errmsg("logical replication apply worker for subscription \"%s\" will restart so that two_phase can be enabled",
622 : MySubscription->name)));
623 12 : should_exit = true;
624 : }
625 : }
626 :
627 1646 : CommitTransactionCommand();
628 1646 : pgstat_report_stat(true);
629 : }
630 :
631 16502 : if (should_exit)
632 : {
633 : /*
634 : * Reset the last-start time for this worker so that the launcher will
635 : * restart it without waiting for wal_retrieve_retry_interval.
636 : */
637 12 : ApplyLauncherForgetWorkerStartTime(MySubscription->oid);
638 :
639 12 : proc_exit(0);
640 : }
641 16490 : }
642 :
643 : /*
644 : * Create list of columns for COPY based on logical relation mapping.
645 : */
646 : static List *
647 384 : make_copy_attnamelist(LogicalRepRelMapEntry *rel)
648 : {
649 384 : List *attnamelist = NIL;
650 : int i;
651 :
652 1032 : for (i = 0; i < rel->remoterel.natts; i++)
653 : {
654 648 : attnamelist = lappend(attnamelist,
655 648 : makeString(rel->remoterel.attnames[i]));
656 : }
657 :
658 :
659 384 : return attnamelist;
660 : }
661 :
662 : /*
663 : * Data source callback for the COPY FROM, which reads from the remote
664 : * connection and passes the data back to our local COPY.
665 : */
666 : static int
667 27970 : copy_read_data(void *outbuf, int minread, int maxread)
668 : {
669 27970 : int bytesread = 0;
670 : int avail;
671 :
672 : /* If there are some leftover data from previous read, use it. */
673 27970 : avail = copybuf->len - copybuf->cursor;
674 27970 : if (avail)
675 : {
676 0 : if (avail > maxread)
677 0 : avail = maxread;
678 0 : memcpy(outbuf, ©buf->data[copybuf->cursor], avail);
679 0 : copybuf->cursor += avail;
680 0 : maxread -= avail;
681 0 : bytesread += avail;
682 : }
683 :
684 27972 : while (maxread > 0 && bytesread < minread)
685 : {
686 27972 : pgsocket fd = PGINVALID_SOCKET;
687 : int len;
688 27972 : char *buf = NULL;
689 :
690 : for (;;)
691 : {
692 : /* Try read the data. */
693 27972 : len = walrcv_receive(LogRepWorkerWalRcvConn, &buf, &fd);
694 :
695 27972 : CHECK_FOR_INTERRUPTS();
696 :
697 27972 : if (len == 0)
698 2 : break;
699 27970 : else if (len < 0)
700 27970 : return bytesread;
701 : else
702 : {
703 : /* Process the data */
704 27590 : copybuf->data = buf;
705 27590 : copybuf->len = len;
706 27590 : copybuf->cursor = 0;
707 :
708 27590 : avail = copybuf->len - copybuf->cursor;
709 27590 : if (avail > maxread)
710 0 : avail = maxread;
711 27590 : memcpy(outbuf, ©buf->data[copybuf->cursor], avail);
712 27590 : outbuf = (char *) outbuf + avail;
713 27590 : copybuf->cursor += avail;
714 27590 : maxread -= avail;
715 27590 : bytesread += avail;
716 : }
717 :
718 27590 : if (maxread <= 0 || bytesread >= minread)
719 27590 : return bytesread;
720 : }
721 :
722 : /*
723 : * Wait for more data or latch.
724 : */
725 2 : (void) WaitLatchOrSocket(MyLatch,
726 : WL_SOCKET_READABLE | WL_LATCH_SET |
727 : WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
728 : fd, 1000L, WAIT_EVENT_LOGICAL_SYNC_DATA);
729 :
730 2 : ResetLatch(MyLatch);
731 : }
732 :
733 0 : return bytesread;
734 : }
735 :
736 :
737 : /*
738 : * Get information about remote relation in similar fashion the RELATION
739 : * message provides during replication.
740 : *
741 : * This function also returns (a) the relation qualifications to be used in
742 : * the COPY command, and (b) whether the remote relation has published any
743 : * generated column.
744 : */
745 : static void
746 388 : fetch_remote_table_info(char *nspname, char *relname, LogicalRepRelation *lrel,
747 : List **qual, bool *gencol_published)
748 : {
749 : WalRcvExecResult *res;
750 : StringInfoData cmd;
751 : TupleTableSlot *slot;
752 388 : Oid tableRow[] = {OIDOID, CHAROID, CHAROID};
753 388 : Oid attrRow[] = {INT2OID, TEXTOID, OIDOID, BOOLOID, BOOLOID};
754 388 : Oid qualRow[] = {TEXTOID};
755 : bool isnull;
756 : int natt;
757 388 : StringInfo pub_names = NULL;
758 388 : Bitmapset *included_cols = NULL;
759 388 : int server_version = walrcv_server_version(LogRepWorkerWalRcvConn);
760 :
761 388 : lrel->nspname = nspname;
762 388 : lrel->relname = relname;
763 :
764 : /* First fetch Oid and replica identity. */
765 388 : initStringInfo(&cmd);
766 388 : appendStringInfo(&cmd, "SELECT c.oid, c.relreplident, c.relkind"
767 : " FROM pg_catalog.pg_class c"
768 : " INNER JOIN pg_catalog.pg_namespace n"
769 : " ON (c.relnamespace = n.oid)"
770 : " WHERE n.nspname = %s"
771 : " AND c.relname = %s",
772 : quote_literal_cstr(nspname),
773 : quote_literal_cstr(relname));
774 388 : res = walrcv_exec(LogRepWorkerWalRcvConn, cmd.data,
775 : lengthof(tableRow), tableRow);
776 :
777 388 : if (res->status != WALRCV_OK_TUPLES)
778 0 : ereport(ERROR,
779 : (errcode(ERRCODE_CONNECTION_FAILURE),
780 : errmsg("could not fetch table info for table \"%s.%s\" from publisher: %s",
781 : nspname, relname, res->err)));
782 :
783 388 : slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple);
784 388 : if (!tuplestore_gettupleslot(res->tuplestore, true, false, slot))
785 0 : ereport(ERROR,
786 : (errcode(ERRCODE_UNDEFINED_OBJECT),
787 : errmsg("table \"%s.%s\" not found on publisher",
788 : nspname, relname)));
789 :
790 388 : lrel->remoteid = DatumGetObjectId(slot_getattr(slot, 1, &isnull));
791 : Assert(!isnull);
792 388 : lrel->replident = DatumGetChar(slot_getattr(slot, 2, &isnull));
793 : Assert(!isnull);
794 388 : lrel->relkind = DatumGetChar(slot_getattr(slot, 3, &isnull));
795 : Assert(!isnull);
796 :
797 388 : ExecDropSingleTupleTableSlot(slot);
798 388 : walrcv_clear_result(res);
799 :
800 :
801 : /*
802 : * Get column lists for each relation.
803 : *
804 : * We need to do this before fetching info about column names and types,
805 : * so that we can skip columns that should not be replicated.
806 : */
807 388 : if (server_version >= 150000)
808 : {
809 : WalRcvExecResult *pubres;
810 : TupleTableSlot *tslot;
811 388 : Oid attrsRow[] = {INT2VECTOROID};
812 :
813 : /* Build the pub_names comma-separated string. */
814 388 : pub_names = makeStringInfo();
815 388 : GetPublicationsStr(MySubscription->publications, pub_names, true);
816 :
817 : /*
818 : * Fetch info about column lists for the relation (from all the
819 : * publications).
820 : */
821 388 : resetStringInfo(&cmd);
822 388 : appendStringInfo(&cmd,
823 : "SELECT DISTINCT"
824 : " (CASE WHEN (array_length(gpt.attrs, 1) = c.relnatts)"
825 : " THEN NULL ELSE gpt.attrs END)"
826 : " FROM pg_publication p,"
827 : " LATERAL pg_get_publication_tables(p.pubname) gpt,"
828 : " pg_class c"
829 : " WHERE gpt.relid = %u AND c.oid = gpt.relid"
830 : " AND p.pubname IN ( %s )",
831 : lrel->remoteid,
832 : pub_names->data);
833 :
834 388 : pubres = walrcv_exec(LogRepWorkerWalRcvConn, cmd.data,
835 : lengthof(attrsRow), attrsRow);
836 :
837 388 : if (pubres->status != WALRCV_OK_TUPLES)
838 0 : ereport(ERROR,
839 : (errcode(ERRCODE_CONNECTION_FAILURE),
840 : errmsg("could not fetch column list info for table \"%s.%s\" from publisher: %s",
841 : nspname, relname, pubres->err)));
842 :
843 : /*
844 : * We don't support the case where the column list is different for
845 : * the same table when combining publications. See comments atop
846 : * fetch_relation_list. So there should be only one row returned.
847 : * Although we already checked this when creating the subscription, we
848 : * still need to check here in case the column list was changed after
849 : * creating the subscription and before the sync worker is started.
850 : */
851 388 : if (tuplestore_tuple_count(pubres->tuplestore) > 1)
852 0 : ereport(ERROR,
853 : errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
854 : errmsg("cannot use different column lists for table \"%s.%s\" in different publications",
855 : nspname, relname));
856 :
857 : /*
858 : * Get the column list and build a single bitmap with the attnums.
859 : *
860 : * If we find a NULL value, it means all the columns should be
861 : * replicated.
862 : */
863 388 : tslot = MakeSingleTupleTableSlot(pubres->tupledesc, &TTSOpsMinimalTuple);
864 388 : if (tuplestore_gettupleslot(pubres->tuplestore, true, false, tslot))
865 : {
866 388 : Datum cfval = slot_getattr(tslot, 1, &isnull);
867 :
868 388 : if (!isnull)
869 : {
870 : ArrayType *arr;
871 : int nelems;
872 : int16 *elems;
873 :
874 44 : arr = DatumGetArrayTypeP(cfval);
875 44 : nelems = ARR_DIMS(arr)[0];
876 44 : elems = (int16 *) ARR_DATA_PTR(arr);
877 :
878 118 : for (natt = 0; natt < nelems; natt++)
879 74 : included_cols = bms_add_member(included_cols, elems[natt]);
880 : }
881 :
882 388 : ExecClearTuple(tslot);
883 : }
884 388 : ExecDropSingleTupleTableSlot(tslot);
885 :
886 388 : walrcv_clear_result(pubres);
887 : }
888 :
889 : /*
890 : * Now fetch column names and types.
891 : */
892 388 : resetStringInfo(&cmd);
893 388 : appendStringInfoString(&cmd,
894 : "SELECT a.attnum,"
895 : " a.attname,"
896 : " a.atttypid,"
897 : " a.attnum = ANY(i.indkey)");
898 :
899 : /* Generated columns can be replicated since version 18. */
900 388 : if (server_version >= 180000)
901 388 : appendStringInfoString(&cmd, ", a.attgenerated != ''");
902 :
903 776 : appendStringInfo(&cmd,
904 : " FROM pg_catalog.pg_attribute a"
905 : " LEFT JOIN pg_catalog.pg_index i"
906 : " ON (i.indexrelid = pg_get_replica_identity_index(%u))"
907 : " WHERE a.attnum > 0::pg_catalog.int2"
908 : " AND NOT a.attisdropped %s"
909 : " AND a.attrelid = %u"
910 : " ORDER BY a.attnum",
911 : lrel->remoteid,
912 388 : (server_version >= 120000 && server_version < 180000 ?
913 : "AND a.attgenerated = ''" : ""),
914 : lrel->remoteid);
915 388 : res = walrcv_exec(LogRepWorkerWalRcvConn, cmd.data,
916 : server_version >= 180000 ? lengthof(attrRow) : lengthof(attrRow) - 1, attrRow);
917 :
918 388 : if (res->status != WALRCV_OK_TUPLES)
919 0 : ereport(ERROR,
920 : (errcode(ERRCODE_CONNECTION_FAILURE),
921 : errmsg("could not fetch table info for table \"%s.%s\" from publisher: %s",
922 : nspname, relname, res->err)));
923 :
924 : /* We don't know the number of rows coming, so allocate enough space. */
925 388 : lrel->attnames = palloc0(MaxTupleAttributeNumber * sizeof(char *));
926 388 : lrel->atttyps = palloc0(MaxTupleAttributeNumber * sizeof(Oid));
927 388 : lrel->attkeys = NULL;
928 :
929 : /*
930 : * Store the columns as a list of names. Ignore those that are not
931 : * present in the column list, if there is one.
932 : */
933 388 : natt = 0;
934 388 : slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple);
935 1110 : while (tuplestore_gettupleslot(res->tuplestore, true, false, slot))
936 : {
937 : char *rel_colname;
938 : AttrNumber attnum;
939 :
940 722 : attnum = DatumGetInt16(slot_getattr(slot, 1, &isnull));
941 : Assert(!isnull);
942 :
943 : /* If the column is not in the column list, skip it. */
944 722 : if (included_cols != NULL && !bms_is_member(attnum, included_cols))
945 : {
946 62 : ExecClearTuple(slot);
947 62 : continue;
948 : }
949 :
950 660 : rel_colname = TextDatumGetCString(slot_getattr(slot, 2, &isnull));
951 : Assert(!isnull);
952 :
953 660 : lrel->attnames[natt] = rel_colname;
954 660 : lrel->atttyps[natt] = DatumGetObjectId(slot_getattr(slot, 3, &isnull));
955 : Assert(!isnull);
956 :
957 660 : if (DatumGetBool(slot_getattr(slot, 4, &isnull)))
958 216 : lrel->attkeys = bms_add_member(lrel->attkeys, natt);
959 :
960 : /* Remember if the remote table has published any generated column. */
961 660 : if (server_version >= 180000 && !(*gencol_published))
962 : {
963 660 : *gencol_published = DatumGetBool(slot_getattr(slot, 5, &isnull));
964 : Assert(!isnull);
965 : }
966 :
967 : /* Should never happen. */
968 660 : if (++natt >= MaxTupleAttributeNumber)
969 0 : elog(ERROR, "too many columns in remote table \"%s.%s\"",
970 : nspname, relname);
971 :
972 660 : ExecClearTuple(slot);
973 : }
974 388 : ExecDropSingleTupleTableSlot(slot);
975 :
976 388 : lrel->natts = natt;
977 :
978 388 : walrcv_clear_result(res);
979 :
980 : /*
981 : * Get relation's row filter expressions. DISTINCT avoids the same
982 : * expression of a table in multiple publications from being included
983 : * multiple times in the final expression.
984 : *
985 : * We need to copy the row even if it matches just one of the
986 : * publications, so we later combine all the quals with OR.
987 : *
988 : * For initial synchronization, row filtering can be ignored in following
989 : * cases:
990 : *
991 : * 1) one of the subscribed publications for the table hasn't specified
992 : * any row filter
993 : *
994 : * 2) one of the subscribed publications has puballtables set to true
995 : *
996 : * 3) one of the subscribed publications is declared as TABLES IN SCHEMA
997 : * that includes this relation
998 : */
999 388 : if (server_version >= 150000)
1000 : {
1001 : /* Reuse the already-built pub_names. */
1002 : Assert(pub_names != NULL);
1003 :
1004 : /* Check for row filters. */
1005 388 : resetStringInfo(&cmd);
1006 388 : appendStringInfo(&cmd,
1007 : "SELECT DISTINCT pg_get_expr(gpt.qual, gpt.relid)"
1008 : " FROM pg_publication p,"
1009 : " LATERAL pg_get_publication_tables(p.pubname) gpt"
1010 : " WHERE gpt.relid = %u"
1011 : " AND p.pubname IN ( %s )",
1012 : lrel->remoteid,
1013 : pub_names->data);
1014 :
1015 388 : res = walrcv_exec(LogRepWorkerWalRcvConn, cmd.data, 1, qualRow);
1016 :
1017 388 : if (res->status != WALRCV_OK_TUPLES)
1018 0 : ereport(ERROR,
1019 : (errmsg("could not fetch table WHERE clause info for table \"%s.%s\" from publisher: %s",
1020 : nspname, relname, res->err)));
1021 :
1022 : /*
1023 : * Multiple row filter expressions for the same table will be combined
1024 : * by COPY using OR. If any of the filter expressions for this table
1025 : * are null, it means the whole table will be copied. In this case it
1026 : * is not necessary to construct a unified row filter expression at
1027 : * all.
1028 : */
1029 388 : slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple);
1030 418 : while (tuplestore_gettupleslot(res->tuplestore, true, false, slot))
1031 : {
1032 396 : Datum rf = slot_getattr(slot, 1, &isnull);
1033 :
1034 396 : if (!isnull)
1035 30 : *qual = lappend(*qual, makeString(TextDatumGetCString(rf)));
1036 : else
1037 : {
1038 : /* Ignore filters and cleanup as necessary. */
1039 366 : if (*qual)
1040 : {
1041 6 : list_free_deep(*qual);
1042 6 : *qual = NIL;
1043 : }
1044 366 : break;
1045 : }
1046 :
1047 30 : ExecClearTuple(slot);
1048 : }
1049 388 : ExecDropSingleTupleTableSlot(slot);
1050 :
1051 388 : walrcv_clear_result(res);
1052 388 : destroyStringInfo(pub_names);
1053 : }
1054 :
1055 388 : pfree(cmd.data);
1056 388 : }
1057 :
1058 : /*
1059 : * Copy existing data of a table from publisher.
1060 : *
1061 : * Caller is responsible for locking the local relation.
1062 : */
1063 : static void
1064 388 : copy_table(Relation rel)
1065 : {
1066 : LogicalRepRelMapEntry *relmapentry;
1067 : LogicalRepRelation lrel;
1068 388 : List *qual = NIL;
1069 : WalRcvExecResult *res;
1070 : StringInfoData cmd;
1071 : CopyFromState cstate;
1072 : List *attnamelist;
1073 : ParseState *pstate;
1074 388 : List *options = NIL;
1075 388 : bool gencol_published = false;
1076 :
1077 : /* Get the publisher relation info. */
1078 388 : fetch_remote_table_info(get_namespace_name(RelationGetNamespace(rel)),
1079 388 : RelationGetRelationName(rel), &lrel, &qual,
1080 : &gencol_published);
1081 :
1082 : /* Put the relation into relmap. */
1083 388 : logicalrep_relmap_update(&lrel);
1084 :
1085 : /* Map the publisher relation to local one. */
1086 388 : relmapentry = logicalrep_rel_open(lrel.remoteid, NoLock);
1087 : Assert(rel == relmapentry->localrel);
1088 :
1089 : /* Start copy on the publisher. */
1090 384 : initStringInfo(&cmd);
1091 :
1092 : /* Regular table with no row filter or generated columns */
1093 384 : if (lrel.relkind == RELKIND_RELATION && qual == NIL && !gencol_published)
1094 : {
1095 328 : appendStringInfo(&cmd, "COPY %s",
1096 328 : quote_qualified_identifier(lrel.nspname, lrel.relname));
1097 :
1098 : /* If the table has columns, then specify the columns */
1099 328 : if (lrel.natts)
1100 : {
1101 326 : appendStringInfoString(&cmd, " (");
1102 :
1103 : /*
1104 : * XXX Do we need to list the columns in all cases? Maybe we're
1105 : * replicating all columns?
1106 : */
1107 890 : for (int i = 0; i < lrel.natts; i++)
1108 : {
1109 564 : if (i > 0)
1110 238 : appendStringInfoString(&cmd, ", ");
1111 :
1112 564 : appendStringInfoString(&cmd, quote_identifier(lrel.attnames[i]));
1113 : }
1114 :
1115 326 : appendStringInfoChar(&cmd, ')');
1116 : }
1117 :
1118 328 : appendStringInfoString(&cmd, " TO STDOUT");
1119 : }
1120 : else
1121 : {
1122 : /*
1123 : * For non-tables and tables with row filters, we need to do COPY
1124 : * (SELECT ...), but we can't just do SELECT * because we may need to
1125 : * copy only subset of columns including generated columns. For tables
1126 : * with any row filters, build a SELECT query with OR'ed row filters
1127 : * for COPY.
1128 : *
1129 : * We also need to use this same COPY (SELECT ...) syntax when
1130 : * generated columns are published, because copy of generated columns
1131 : * is not supported by the normal COPY.
1132 : */
1133 56 : appendStringInfoString(&cmd, "COPY (SELECT ");
1134 140 : for (int i = 0; i < lrel.natts; i++)
1135 : {
1136 84 : appendStringInfoString(&cmd, quote_identifier(lrel.attnames[i]));
1137 84 : if (i < lrel.natts - 1)
1138 28 : appendStringInfoString(&cmd, ", ");
1139 : }
1140 :
1141 56 : appendStringInfoString(&cmd, " FROM ");
1142 :
1143 : /*
1144 : * For regular tables, make sure we don't copy data from a child that
1145 : * inherits the named table as those will be copied separately.
1146 : */
1147 56 : if (lrel.relkind == RELKIND_RELATION)
1148 22 : appendStringInfoString(&cmd, "ONLY ");
1149 :
1150 56 : appendStringInfoString(&cmd, quote_qualified_identifier(lrel.nspname, lrel.relname));
1151 : /* list of OR'ed filters */
1152 56 : if (qual != NIL)
1153 : {
1154 : ListCell *lc;
1155 22 : char *q = strVal(linitial(qual));
1156 :
1157 22 : appendStringInfo(&cmd, " WHERE %s", q);
1158 24 : for_each_from(lc, qual, 1)
1159 : {
1160 2 : q = strVal(lfirst(lc));
1161 2 : appendStringInfo(&cmd, " OR %s", q);
1162 : }
1163 22 : list_free_deep(qual);
1164 : }
1165 :
1166 56 : appendStringInfoString(&cmd, ") TO STDOUT");
1167 : }
1168 :
1169 : /*
1170 : * Prior to v16, initial table synchronization will use text format even
1171 : * if the binary option is enabled for a subscription.
1172 : */
1173 384 : if (walrcv_server_version(LogRepWorkerWalRcvConn) >= 160000 &&
1174 384 : MySubscription->binary)
1175 : {
1176 10 : appendStringInfoString(&cmd, " WITH (FORMAT binary)");
1177 10 : options = list_make1(makeDefElem("format",
1178 : (Node *) makeString("binary"), -1));
1179 : }
1180 :
1181 384 : res = walrcv_exec(LogRepWorkerWalRcvConn, cmd.data, 0, NULL);
1182 384 : pfree(cmd.data);
1183 384 : if (res->status != WALRCV_OK_COPY_OUT)
1184 0 : ereport(ERROR,
1185 : (errcode(ERRCODE_CONNECTION_FAILURE),
1186 : errmsg("could not start initial contents copy for table \"%s.%s\": %s",
1187 : lrel.nspname, lrel.relname, res->err)));
1188 384 : walrcv_clear_result(res);
1189 :
1190 384 : copybuf = makeStringInfo();
1191 :
1192 384 : pstate = make_parsestate(NULL);
1193 384 : (void) addRangeTableEntryForRelation(pstate, rel, AccessShareLock,
1194 : NULL, false, false);
1195 :
1196 384 : attnamelist = make_copy_attnamelist(relmapentry);
1197 384 : cstate = BeginCopyFrom(pstate, rel, NULL, NULL, false, copy_read_data, attnamelist, options);
1198 :
1199 : /* Do the copy */
1200 382 : (void) CopyFrom(cstate);
1201 :
1202 368 : logicalrep_rel_close(relmapentry, NoLock);
1203 368 : }
1204 :
1205 : /*
1206 : * Determine the tablesync slot name.
1207 : *
1208 : * The name must not exceed NAMEDATALEN - 1 because of remote node constraints
1209 : * on slot name length. We append system_identifier to avoid slot_name
1210 : * collision with subscriptions in other clusters. With the current scheme
1211 : * pg_%u_sync_%u_UINT64_FORMAT (3 + 10 + 6 + 10 + 20 + '\0'), the maximum
1212 : * length of slot_name will be 50.
1213 : *
1214 : * The returned slot name is stored in the supplied buffer (syncslotname) with
1215 : * the given size.
1216 : *
1217 : * Note: We don't use the subscription slot name as part of tablesync slot name
1218 : * because we are responsible for cleaning up these slots and it could become
1219 : * impossible to recalculate what name to cleanup if the subscription slot name
1220 : * had changed.
1221 : */
1222 : void
1223 772 : ReplicationSlotNameForTablesync(Oid suboid, Oid relid,
1224 : char *syncslotname, Size szslot)
1225 : {
1226 772 : snprintf(syncslotname, szslot, "pg_%u_sync_%u_" UINT64_FORMAT, suboid,
1227 : relid, GetSystemIdentifier());
1228 772 : }
1229 :
1230 : /*
1231 : * Start syncing the table in the sync worker.
1232 : *
1233 : * If nothing needs to be done to sync the table, we exit the worker without
1234 : * any further action.
1235 : *
1236 : * The returned slot name is palloc'ed in current memory context.
1237 : */
1238 : static char *
1239 394 : LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
1240 : {
1241 : char *slotname;
1242 : char *err;
1243 : char relstate;
1244 : XLogRecPtr relstate_lsn;
1245 : Relation rel;
1246 : AclResult aclresult;
1247 : WalRcvExecResult *res;
1248 : char originname[NAMEDATALEN];
1249 : RepOriginId originid;
1250 : UserContext ucxt;
1251 : bool must_use_password;
1252 : bool run_as_owner;
1253 :
1254 : /* Check the state of the table synchronization. */
1255 394 : StartTransactionCommand();
1256 394 : relstate = GetSubscriptionRelState(MyLogicalRepWorker->subid,
1257 394 : MyLogicalRepWorker->relid,
1258 : &relstate_lsn);
1259 394 : CommitTransactionCommand();
1260 :
1261 : /* Is the use of a password mandatory? */
1262 780 : must_use_password = MySubscription->passwordrequired &&
1263 386 : !MySubscription->ownersuperuser;
1264 :
1265 394 : SpinLockAcquire(&MyLogicalRepWorker->relmutex);
1266 394 : MyLogicalRepWorker->relstate = relstate;
1267 394 : MyLogicalRepWorker->relstate_lsn = relstate_lsn;
1268 394 : SpinLockRelease(&MyLogicalRepWorker->relmutex);
1269 :
1270 : /*
1271 : * If synchronization is already done or no longer necessary, exit now
1272 : * that we've updated shared memory state.
1273 : */
1274 394 : switch (relstate)
1275 : {
1276 0 : case SUBREL_STATE_SYNCDONE:
1277 : case SUBREL_STATE_READY:
1278 : case SUBREL_STATE_UNKNOWN:
1279 0 : FinishSyncWorker(); /* doesn't return */
1280 : }
1281 :
1282 : /* Calculate the name of the tablesync slot. */
1283 394 : slotname = (char *) palloc(NAMEDATALEN);
1284 394 : ReplicationSlotNameForTablesync(MySubscription->oid,
1285 394 : MyLogicalRepWorker->relid,
1286 : slotname,
1287 : NAMEDATALEN);
1288 :
1289 : /*
1290 : * Here we use the slot name instead of the subscription name as the
1291 : * application_name, so that it is different from the leader apply worker,
1292 : * so that synchronous replication can distinguish them.
1293 : */
1294 394 : LogRepWorkerWalRcvConn =
1295 394 : walrcv_connect(MySubscription->conninfo, true, true,
1296 : must_use_password,
1297 : slotname, &err);
1298 394 : if (LogRepWorkerWalRcvConn == NULL)
1299 0 : ereport(ERROR,
1300 : (errcode(ERRCODE_CONNECTION_FAILURE),
1301 : errmsg("table synchronization worker for subscription \"%s\" could not connect to the publisher: %s",
1302 : MySubscription->name, err)));
1303 :
1304 : Assert(MyLogicalRepWorker->relstate == SUBREL_STATE_INIT ||
1305 : MyLogicalRepWorker->relstate == SUBREL_STATE_DATASYNC ||
1306 : MyLogicalRepWorker->relstate == SUBREL_STATE_FINISHEDCOPY);
1307 :
1308 : /* Assign the origin tracking record name. */
1309 394 : ReplicationOriginNameForLogicalRep(MySubscription->oid,
1310 394 : MyLogicalRepWorker->relid,
1311 : originname,
1312 : sizeof(originname));
1313 :
1314 394 : if (MyLogicalRepWorker->relstate == SUBREL_STATE_DATASYNC)
1315 : {
1316 : /*
1317 : * We have previously errored out before finishing the copy so the
1318 : * replication slot might exist. We want to remove the slot if it
1319 : * already exists and proceed.
1320 : *
1321 : * XXX We could also instead try to drop the slot, last time we failed
1322 : * but for that, we might need to clean up the copy state as it might
1323 : * be in the middle of fetching the rows. Also, if there is a network
1324 : * breakdown then it wouldn't have succeeded so trying it next time
1325 : * seems like a better bet.
1326 : */
1327 14 : ReplicationSlotDropAtPubNode(LogRepWorkerWalRcvConn, slotname, true);
1328 : }
1329 380 : else if (MyLogicalRepWorker->relstate == SUBREL_STATE_FINISHEDCOPY)
1330 : {
1331 : /*
1332 : * The COPY phase was previously done, but tablesync then crashed
1333 : * before it was able to finish normally.
1334 : */
1335 0 : StartTransactionCommand();
1336 :
1337 : /*
1338 : * The origin tracking name must already exist. It was created first
1339 : * time this tablesync was launched.
1340 : */
1341 0 : originid = replorigin_by_name(originname, false);
1342 0 : replorigin_session_setup(originid, 0);
1343 0 : replorigin_session_origin = originid;
1344 0 : *origin_startpos = replorigin_session_get_progress(false);
1345 :
1346 0 : CommitTransactionCommand();
1347 :
1348 0 : goto copy_table_done;
1349 : }
1350 :
1351 394 : SpinLockAcquire(&MyLogicalRepWorker->relmutex);
1352 394 : MyLogicalRepWorker->relstate = SUBREL_STATE_DATASYNC;
1353 394 : MyLogicalRepWorker->relstate_lsn = InvalidXLogRecPtr;
1354 394 : SpinLockRelease(&MyLogicalRepWorker->relmutex);
1355 :
1356 : /* Update the state and make it visible to others. */
1357 394 : StartTransactionCommand();
1358 394 : UpdateSubscriptionRelState(MyLogicalRepWorker->subid,
1359 394 : MyLogicalRepWorker->relid,
1360 394 : MyLogicalRepWorker->relstate,
1361 394 : MyLogicalRepWorker->relstate_lsn,
1362 : false);
1363 390 : CommitTransactionCommand();
1364 390 : pgstat_report_stat(true);
1365 :
1366 390 : StartTransactionCommand();
1367 :
1368 : /*
1369 : * Use a standard write lock here. It might be better to disallow access
1370 : * to the table while it's being synchronized. But we don't want to block
1371 : * the main apply process from working and it has to open the relation in
1372 : * RowExclusiveLock when remapping remote relation id to local one.
1373 : */
1374 390 : rel = table_open(MyLogicalRepWorker->relid, RowExclusiveLock);
1375 :
1376 : /*
1377 : * Start a transaction in the remote node in REPEATABLE READ mode. This
1378 : * ensures that both the replication slot we create (see below) and the
1379 : * COPY are consistent with each other.
1380 : */
1381 390 : res = walrcv_exec(LogRepWorkerWalRcvConn,
1382 : "BEGIN READ ONLY ISOLATION LEVEL REPEATABLE READ",
1383 : 0, NULL);
1384 390 : if (res->status != WALRCV_OK_COMMAND)
1385 0 : ereport(ERROR,
1386 : (errcode(ERRCODE_CONNECTION_FAILURE),
1387 : errmsg("table copy could not start transaction on publisher: %s",
1388 : res->err)));
1389 390 : walrcv_clear_result(res);
1390 :
1391 : /*
1392 : * Create a new permanent logical decoding slot. This slot will be used
1393 : * for the catchup phase after COPY is done, so tell it to use the
1394 : * snapshot to make the final data consistent.
1395 : */
1396 390 : walrcv_create_slot(LogRepWorkerWalRcvConn,
1397 : slotname, false /* permanent */ , false /* two_phase */ ,
1398 : MySubscription->failover,
1399 : CRS_USE_SNAPSHOT, origin_startpos);
1400 :
1401 : /*
1402 : * Setup replication origin tracking. The purpose of doing this before the
1403 : * copy is to avoid doing the copy again due to any error in setting up
1404 : * origin tracking.
1405 : */
1406 388 : originid = replorigin_by_name(originname, true);
1407 388 : if (!OidIsValid(originid))
1408 : {
1409 : /*
1410 : * Origin tracking does not exist, so create it now.
1411 : *
1412 : * Then advance to the LSN got from walrcv_create_slot. This is WAL
1413 : * logged for the purpose of recovery. Locks are to prevent the
1414 : * replication origin from vanishing while advancing.
1415 : */
1416 388 : originid = replorigin_create(originname);
1417 :
1418 388 : LockRelationOid(ReplicationOriginRelationId, RowExclusiveLock);
1419 388 : replorigin_advance(originid, *origin_startpos, InvalidXLogRecPtr,
1420 : true /* go backward */ , true /* WAL log */ );
1421 388 : UnlockRelationOid(ReplicationOriginRelationId, RowExclusiveLock);
1422 :
1423 388 : replorigin_session_setup(originid, 0);
1424 388 : replorigin_session_origin = originid;
1425 : }
1426 : else
1427 : {
1428 0 : ereport(ERROR,
1429 : (errcode(ERRCODE_DUPLICATE_OBJECT),
1430 : errmsg("replication origin \"%s\" already exists",
1431 : originname)));
1432 : }
1433 :
1434 : /*
1435 : * Make sure that the copy command runs as the table owner, unless the
1436 : * user has opted out of that behaviour.
1437 : */
1438 388 : run_as_owner = MySubscription->runasowner;
1439 388 : if (!run_as_owner)
1440 386 : SwitchToUntrustedUser(rel->rd_rel->relowner, &ucxt);
1441 :
1442 : /*
1443 : * Check that our table sync worker has permission to insert into the
1444 : * target table.
1445 : */
1446 388 : aclresult = pg_class_aclcheck(RelationGetRelid(rel), GetUserId(),
1447 : ACL_INSERT);
1448 388 : if (aclresult != ACLCHECK_OK)
1449 0 : aclcheck_error(aclresult,
1450 0 : get_relkind_objtype(rel->rd_rel->relkind),
1451 0 : RelationGetRelationName(rel));
1452 :
1453 : /*
1454 : * COPY FROM does not honor RLS policies. That is not a problem for
1455 : * subscriptions owned by roles with BYPASSRLS privilege (or superuser,
1456 : * who has it implicitly), but other roles should not be able to
1457 : * circumvent RLS. Disallow logical replication into RLS enabled
1458 : * relations for such roles.
1459 : */
1460 388 : if (check_enable_rls(RelationGetRelid(rel), InvalidOid, false) == RLS_ENABLED)
1461 0 : ereport(ERROR,
1462 : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1463 : errmsg("user \"%s\" cannot replicate into relation with row-level security enabled: \"%s\"",
1464 : GetUserNameFromId(GetUserId(), true),
1465 : RelationGetRelationName(rel))));
1466 :
1467 : /* Now do the initial data copy */
1468 388 : PushActiveSnapshot(GetTransactionSnapshot());
1469 388 : copy_table(rel);
1470 368 : PopActiveSnapshot();
1471 :
1472 368 : res = walrcv_exec(LogRepWorkerWalRcvConn, "COMMIT", 0, NULL);
1473 368 : if (res->status != WALRCV_OK_COMMAND)
1474 0 : ereport(ERROR,
1475 : (errcode(ERRCODE_CONNECTION_FAILURE),
1476 : errmsg("table copy could not finish transaction on publisher: %s",
1477 : res->err)));
1478 368 : walrcv_clear_result(res);
1479 :
1480 368 : if (!run_as_owner)
1481 366 : RestoreUserContext(&ucxt);
1482 :
1483 368 : table_close(rel, NoLock);
1484 :
1485 : /* Make the copy visible. */
1486 368 : CommandCounterIncrement();
1487 :
1488 : /*
1489 : * Update the persisted state to indicate the COPY phase is done; make it
1490 : * visible to others.
1491 : */
1492 368 : UpdateSubscriptionRelState(MyLogicalRepWorker->subid,
1493 368 : MyLogicalRepWorker->relid,
1494 : SUBREL_STATE_FINISHEDCOPY,
1495 368 : MyLogicalRepWorker->relstate_lsn,
1496 : false);
1497 :
1498 368 : CommitTransactionCommand();
1499 :
1500 368 : copy_table_done:
1501 :
1502 368 : elog(DEBUG1,
1503 : "LogicalRepSyncTableStart: '%s' origin_startpos lsn %X/%08X",
1504 : originname, LSN_FORMAT_ARGS(*origin_startpos));
1505 :
1506 : /*
1507 : * We are done with the initial data synchronization, update the state.
1508 : */
1509 368 : SpinLockAcquire(&MyLogicalRepWorker->relmutex);
1510 368 : MyLogicalRepWorker->relstate = SUBREL_STATE_SYNCWAIT;
1511 368 : MyLogicalRepWorker->relstate_lsn = *origin_startpos;
1512 368 : SpinLockRelease(&MyLogicalRepWorker->relmutex);
1513 :
1514 : /*
1515 : * Finally, wait until the leader apply worker tells us to catch up and
1516 : * then return to let LogicalRepApplyLoop do it.
1517 : */
1518 368 : wait_for_worker_state_change(SUBREL_STATE_CATCHUP);
1519 368 : return slotname;
1520 : }
1521 :
1522 : /*
1523 : * Execute the initial sync with error handling. Disable the subscription,
1524 : * if it's required.
1525 : *
1526 : * Allocate the slot name in long-lived context on return. Note that we don't
1527 : * handle FATAL errors which are probably because of system resource error and
1528 : * are not repeatable.
1529 : */
1530 : static void
1531 394 : start_table_sync(XLogRecPtr *origin_startpos, char **slotname)
1532 : {
1533 394 : char *sync_slotname = NULL;
1534 :
1535 : Assert(am_tablesync_worker());
1536 :
1537 394 : PG_TRY();
1538 : {
1539 : /* Call initial sync. */
1540 394 : sync_slotname = LogicalRepSyncTableStart(origin_startpos);
1541 : }
1542 20 : PG_CATCH();
1543 : {
1544 20 : if (MySubscription->disableonerr)
1545 2 : DisableSubscriptionAndExit();
1546 : else
1547 : {
1548 : /*
1549 : * Report the worker failed during table synchronization. Abort
1550 : * the current transaction so that the stats message is sent in an
1551 : * idle state.
1552 : */
1553 18 : AbortOutOfAnyTransaction();
1554 18 : pgstat_report_subscription_error(MySubscription->oid, false);
1555 :
1556 18 : PG_RE_THROW();
1557 : }
1558 : }
1559 368 : PG_END_TRY();
1560 :
1561 : /* allocate slot name in long-lived context */
1562 368 : *slotname = MemoryContextStrdup(ApplyContext, sync_slotname);
1563 368 : pfree(sync_slotname);
1564 368 : }
1565 :
1566 : /*
1567 : * Runs the tablesync worker.
1568 : *
1569 : * It starts syncing tables. After a successful sync, sets streaming options
1570 : * and starts streaming to catchup with apply worker.
1571 : */
1572 : static void
1573 394 : run_tablesync_worker()
1574 : {
1575 : char originname[NAMEDATALEN];
1576 394 : XLogRecPtr origin_startpos = InvalidXLogRecPtr;
1577 394 : char *slotname = NULL;
1578 : WalRcvStreamOptions options;
1579 :
1580 394 : start_table_sync(&origin_startpos, &slotname);
1581 :
1582 368 : ReplicationOriginNameForLogicalRep(MySubscription->oid,
1583 368 : MyLogicalRepWorker->relid,
1584 : originname,
1585 : sizeof(originname));
1586 :
1587 368 : set_apply_error_context_origin(originname);
1588 :
1589 368 : set_stream_options(&options, slotname, &origin_startpos);
1590 :
1591 368 : walrcv_startstreaming(LogRepWorkerWalRcvConn, &options);
1592 :
1593 : /* Apply the changes till we catchup with the apply worker. */
1594 368 : start_apply(origin_startpos);
1595 0 : }
1596 :
1597 : /* Logical Replication Tablesync worker entry point */
1598 : void
1599 396 : TablesyncWorkerMain(Datum main_arg)
1600 : {
1601 396 : int worker_slot = DatumGetInt32(main_arg);
1602 :
1603 396 : SetupApplyOrSyncWorker(worker_slot);
1604 :
1605 394 : run_tablesync_worker();
1606 :
1607 0 : FinishSyncWorker();
1608 : }
1609 :
1610 : /*
1611 : * If the subscription has no tables then return false.
1612 : *
1613 : * Otherwise, are all tablesyncs READY?
1614 : *
1615 : * Note: This function is not suitable to be called from outside of apply or
1616 : * tablesync workers because MySubscription needs to be already initialized.
1617 : */
1618 : bool
1619 390 : AllTablesyncsReady(void)
1620 : {
1621 390 : bool started_tx = false;
1622 390 : bool has_subrels = false;
1623 :
1624 : /* We need up-to-date sync state info for subscription tables here. */
1625 390 : has_subrels = FetchRelationStates(&started_tx);
1626 :
1627 390 : if (started_tx)
1628 : {
1629 32 : CommitTransactionCommand();
1630 32 : pgstat_report_stat(true);
1631 : }
1632 :
1633 : /*
1634 : * Return false when there are no tables in subscription or not all tables
1635 : * are in ready state; true otherwise.
1636 : */
1637 390 : return has_subrels && (table_states_not_ready == NIL);
1638 : }
1639 :
1640 : /*
1641 : * Return whether the subscription currently has any tables.
1642 : *
1643 : * Note: Unlike HasSubscriptionTables(), this function relies on cached
1644 : * information for subscription tables. Additionally, it should not be
1645 : * invoked outside of apply or tablesync workers, as MySubscription must be
1646 : * initialized first.
1647 : */
1648 : bool
1649 258 : HasSubscriptionTablesCached(void)
1650 : {
1651 : bool started_tx;
1652 : bool has_subrels;
1653 :
1654 : /* We need up-to-date subscription tables info here */
1655 258 : has_subrels = FetchRelationStates(&started_tx);
1656 :
1657 258 : if (started_tx)
1658 : {
1659 4 : CommitTransactionCommand();
1660 4 : pgstat_report_stat(true);
1661 : }
1662 :
1663 258 : return has_subrels;
1664 : }
1665 :
1666 : /*
1667 : * Update the two_phase state of the specified subscription in pg_subscription.
1668 : */
1669 : void
1670 20 : UpdateTwoPhaseState(Oid suboid, char new_state)
1671 : {
1672 : Relation rel;
1673 : HeapTuple tup;
1674 : bool nulls[Natts_pg_subscription];
1675 : bool replaces[Natts_pg_subscription];
1676 : Datum values[Natts_pg_subscription];
1677 :
1678 : Assert(new_state == LOGICALREP_TWOPHASE_STATE_DISABLED ||
1679 : new_state == LOGICALREP_TWOPHASE_STATE_PENDING ||
1680 : new_state == LOGICALREP_TWOPHASE_STATE_ENABLED);
1681 :
1682 20 : rel = table_open(SubscriptionRelationId, RowExclusiveLock);
1683 20 : tup = SearchSysCacheCopy1(SUBSCRIPTIONOID, ObjectIdGetDatum(suboid));
1684 20 : if (!HeapTupleIsValid(tup))
1685 0 : elog(ERROR,
1686 : "cache lookup failed for subscription oid %u",
1687 : suboid);
1688 :
1689 : /* Form a new tuple. */
1690 20 : memset(values, 0, sizeof(values));
1691 20 : memset(nulls, false, sizeof(nulls));
1692 20 : memset(replaces, false, sizeof(replaces));
1693 :
1694 : /* And update/set two_phase state */
1695 20 : values[Anum_pg_subscription_subtwophasestate - 1] = CharGetDatum(new_state);
1696 20 : replaces[Anum_pg_subscription_subtwophasestate - 1] = true;
1697 :
1698 20 : tup = heap_modify_tuple(tup, RelationGetDescr(rel),
1699 : values, nulls, replaces);
1700 20 : CatalogTupleUpdate(rel, &tup->t_self, tup);
1701 :
1702 20 : heap_freetuple(tup);
1703 20 : table_close(rel, RowExclusiveLock);
1704 20 : }
|
