#include "postgres.h"
#include "access/table.h"
#include "access/xact.h"
#include "catalog/indexing.h"
#include "catalog/pg_subscription_rel.h"
#include "catalog/pg_type.h"
#include "commands/copy.h"
#include "miscadmin.h"
#include "parser/parse_relation.h"
#include "pgstat.h"
#include "replication/logicallauncher.h"
#include "replication/logicalrelation.h"
#include "replication/walreceiver.h"
#include "replication/worker_internal.h"
#include "replication/slot.h"
#include "replication/origin.h"
#include "storage/ipc.h"
#include "storage/lmgr.h"
#include "utils/acl.h"
#include "utils/array.h"
#include "utils/builtins.h"
#include "utils/lsyscache.h"
#include "utils/memutils.h"
#include "utils/rls.h"
#include "utils/snapmgr.h"
#include "utils/syscache.h"

Include dependency graph for tablesync.c:

Functions
static bool	FetchTableStates (bool *started_tx)

static void	pg_attribute_noreturn () finish_sync_worker(void)

static bool	wait_for_relation_state_change (Oid relid, char expected_state)

static bool	wait_for_worker_state_change (char expected_state)

void	invalidate_syncing_table_states (Datum arg, int cacheid, uint32 hashvalue)

static void	process_syncing_tables_for_sync (XLogRecPtr current_lsn)

static void	process_syncing_tables_for_apply (XLogRecPtr current_lsn)

void	process_syncing_tables (XLogRecPtr current_lsn)

static List *	make_copy_attnamelist (LogicalRepRelMapEntry *rel)

static int	copy_read_data (void *outbuf, int minread, int maxread)

static void	fetch_remote_table_info (char nspname, char relname, LogicalRepRelation lrel, List *qual)

static void	copy_table (Relation rel)

void	ReplicationSlotNameForTablesync (Oid suboid, Oid relid, char *syncslotname, Size szslot)

char *	LogicalRepSyncTableStart (XLogRecPtr *origin_startpos)

bool	AllTablesyncsReady (void)

void	UpdateTwoPhaseState (Oid suboid, char new_state)

Variables
static bool	table_states_valid = false

static List *	table_states_not_ready = NIL

static StringInfo	copybuf = NULL

Function Documentation

◆ AllTablesyncsReady()

bool AllTablesyncsReady ( void )

Definition at line 1558 of file tablesync.c.

 {
     bool        started_tx = false;
     bool        has_subrels = false;
  
     /* We need up-to-date sync state info for subscription tables here. */
     has_subrels = FetchTableStates(&started_tx);
  
     if (started_tx)
     {
         CommitTransactionCommand();
         pgstat_report_stat(true);
     }
  
     /*
      * Return false when there are no tables in subscription or not all tables
      * are in ready state; true otherwise.
      */
     return has_subrels && (table_states_not_ready == NIL);
 }

References CommitTransactionCommand(), FetchTableStates(), NIL, pgstat_report_stat(), and table_states_not_ready.

Referenced by ApplyWorkerMain(), pa_can_start(), and process_syncing_tables_for_apply().

◆ copy_read_data()

static int copy_read_data	(	void *	outbuf,
		int	minread,
		int	maxread
	)

static

Definition at line 686 of file tablesync.c.

 {
     int         bytesread = 0;
     int         avail;
  
     /* If there are some leftover data from previous read, use it. */
     avail = copybuf->len - copybuf->cursor;
     if (avail)
     {
         if (avail > maxread)
             avail = maxread;
         memcpy(outbuf, &copybuf->data[copybuf->cursor], avail);
         copybuf->cursor += avail;
         maxread -= avail;
         bytesread += avail;
     }
  
     while (maxread > 0 && bytesread < minread)
     {
         pgsocket    fd = PGINVALID_SOCKET;
         int         len;
         char       *buf = NULL;
  
         for (;;)
         {
             /* Try read the data. */
             len = walrcv_receive(LogRepWorkerWalRcvConn, &buf, &fd);
  
             CHECK_FOR_INTERRUPTS();
  
             if (len == 0)
                 break;
             else if (len < 0)
                 return bytesread;
             else
             {
                 /* Process the data */
                 copybuf->data = buf;
                 copybuf->len = len;
                 copybuf->cursor = 0;
  
                 avail = copybuf->len - copybuf->cursor;
                 if (avail > maxread)
                     avail = maxread;
                 memcpy(outbuf, &copybuf->data[copybuf->cursor], avail);
                 outbuf = (void *) ((char *) outbuf + avail);
                 copybuf->cursor += avail;
                 maxread -= avail;
                 bytesread += avail;
             }
  
             if (maxread <= 0 || bytesread >= minread)
                 return bytesread;
         }
  
         /*
          * Wait for more data or latch.
          */
         (void) WaitLatchOrSocket(MyLatch,
                                  WL_SOCKET_READABLE | WL_LATCH_SET |
                                  WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
                                  fd, 1000L, WAIT_EVENT_LOGICAL_SYNC_DATA);
  
         ResetLatch(MyLatch);
     }
  
     return bytesread;
 }

References buf, CHECK_FOR_INTERRUPTS, copybuf, StringInfoData::cursor, StringInfoData::data, fd(), StringInfoData::len, len, LogRepWorkerWalRcvConn, MyLatch, PGINVALID_SOCKET, ResetLatch(), WAIT_EVENT_LOGICAL_SYNC_DATA, WaitLatchOrSocket(), walrcv_receive, WL_EXIT_ON_PM_DEATH, WL_LATCH_SET, WL_SOCKET_READABLE, and WL_TIMEOUT.

Referenced by copy_table().

◆ copy_table()

static void copy_table ( Relation rel )

static

Definition at line 1083 of file tablesync.c.

 {
     LogicalRepRelMapEntry *relmapentry;
     LogicalRepRelation lrel;
     List       *qual = NIL;
     WalRcvExecResult *res;
     StringInfoData cmd;
     CopyFromState cstate;
     List       *attnamelist;
     ParseState *pstate;
  
     /* Get the publisher relation info. */
     fetch_remote_table_info(get_namespace_name(RelationGetNamespace(rel)),
                             RelationGetRelationName(rel), &lrel, &qual);
  
     /* Put the relation into relmap. */
     logicalrep_relmap_update(&lrel);
  
     /* Map the publisher relation to local one. */
     relmapentry = logicalrep_rel_open(lrel.remoteid, NoLock);
     Assert(rel == relmapentry->localrel);
  
     /* Start copy on the publisher. */
     initStringInfo(&cmd);
  
     /* Regular table with no row filter */
     if (lrel.relkind == RELKIND_RELATION && qual == NIL)
     {
         appendStringInfo(&cmd, "COPY %s (",
                          quote_qualified_identifier(lrel.nspname, lrel.relname));
  
         /*
          * XXX Do we need to list the columns in all cases? Maybe we're
          * replicating all columns?
          */
         for (int i = 0; i < lrel.natts; i++)
         {
             if (i > 0)
                 appendStringInfoString(&cmd, ", ");
  
             appendStringInfoString(&cmd, quote_identifier(lrel.attnames[i]));
         }
  
         appendStringInfoString(&cmd, ") TO STDOUT");
     }
     else
     {
         /*
          * For non-tables and tables with row filters, we need to do COPY
          * (SELECT ...), but we can't just do SELECT * because we need to not
          * copy generated columns. For tables with any row filters, build a
          * SELECT query with OR'ed row filters for COPY.
          */
         appendStringInfoString(&cmd, "COPY (SELECT ");
         for (int i = 0; i < lrel.natts; i++)
         {
             appendStringInfoString(&cmd, quote_identifier(lrel.attnames[i]));
             if (i < lrel.natts - 1)
                 appendStringInfoString(&cmd, ", ");
         }
  
         appendStringInfoString(&cmd, " FROM ");
  
         /*
          * For regular tables, make sure we don't copy data from a child that
          * inherits the named table as those will be copied separately.
          */
         if (lrel.relkind == RELKIND_RELATION)
             appendStringInfoString(&cmd, "ONLY ");
  
         appendStringInfoString(&cmd, quote_qualified_identifier(lrel.nspname, lrel.relname));
         /* list of OR'ed filters */
         if (qual != NIL)
         {
             ListCell   *lc;
             char       *q = strVal(linitial(qual));
  
             appendStringInfo(&cmd, " WHERE %s", q);
             for_each_from(lc, qual, 1)
             {
                 q = strVal(lfirst(lc));
                 appendStringInfo(&cmd, " OR %s", q);
             }
             list_free_deep(qual);
         }
  
         appendStringInfoString(&cmd, ") TO STDOUT");
     }
     res = walrcv_exec(LogRepWorkerWalRcvConn, cmd.data, 0, NULL);
     pfree(cmd.data);
     if (res->status != WALRCV_OK_COPY_OUT)
         ereport(ERROR,
                 (errcode(ERRCODE_CONNECTION_FAILURE),
                  errmsg("could not start initial contents copy for table \"%s.%s\": %s",
                         lrel.nspname, lrel.relname, res->err)));
     walrcv_clear_result(res);
  
     copybuf = makeStringInfo();
  
     pstate = make_parsestate(NULL);
     (void) addRangeTableEntryForRelation(pstate, rel, AccessShareLock,
                                          NULL, false, false);
  
     attnamelist = make_copy_attnamelist(relmapentry);
     cstate = BeginCopyFrom(pstate, rel, NULL, NULL, false, copy_read_data, attnamelist, NIL);
  
     /* Do the copy */
     (void) CopyFrom(cstate);
  
     logicalrep_rel_close(relmapentry, NoLock);
 }

References AccessShareLock, addRangeTableEntryForRelation(), appendStringInfo(), appendStringInfoString(), Assert(), LogicalRepRelation::attnames, BeginCopyFrom(), copy_read_data(), copybuf, CopyFrom(), StringInfoData::data, ereport, errcode(), errmsg(), ERROR, fetch_remote_table_info(), for_each_from, get_namespace_name(), i, initStringInfo(), lfirst, linitial, list_free_deep(), LogicalRepRelMapEntry::localrel, logicalrep_rel_close(), logicalrep_rel_open(), logicalrep_relmap_update(), LogRepWorkerWalRcvConn, make_copy_attnamelist(), make_parsestate(), makeStringInfo(), LogicalRepRelation::natts, NIL, NoLock, LogicalRepRelation::nspname, pfree(), quote_identifier(), quote_qualified_identifier(), RelationGetNamespace, RelationGetRelationName, LogicalRepRelation::relkind, LogicalRepRelation::relname, LogicalRepRelation::remoteid, res, strVal, walrcv_clear_result(), walrcv_exec, and WALRCV_OK_COPY_OUT.

Referenced by LogicalRepSyncTableStart().

◆ fetch_remote_table_info()

static void fetch_remote_table_info	(	char *	nspname,
		char *	relname,
		LogicalRepRelation *	lrel,
		List **	qual
	)

static

Definition at line 762 of file tablesync.c.

 {
     WalRcvExecResult *res;
     StringInfoData cmd;
     TupleTableSlot *slot;
     Oid         tableRow[] = {OIDOID, CHAROID, CHAROID};
     Oid         attrRow[] = {INT2OID, TEXTOID, OIDOID, BOOLOID};
     Oid         qualRow[] = {TEXTOID};
     bool        isnull;
     int         natt;
     ListCell   *lc;
     Bitmapset  *included_cols = NULL;
  
     lrel->nspname = nspname;
     lrel->relname = relname;
  
     /* First fetch Oid and replica identity. */
     initStringInfo(&cmd);
     appendStringInfo(&cmd, "SELECT c.oid, c.relreplident, c.relkind"
                      "  FROM pg_catalog.pg_class c"
                      "  INNER JOIN pg_catalog.pg_namespace n"
                      "        ON (c.relnamespace = n.oid)"
                      " WHERE n.nspname = %s"
                      "   AND c.relname = %s",
                      quote_literal_cstr(nspname),
                      quote_literal_cstr(relname));
     res = walrcv_exec(LogRepWorkerWalRcvConn, cmd.data,
                       lengthof(tableRow), tableRow);
  
     if (res->status != WALRCV_OK_TUPLES)
         ereport(ERROR,
                 (errcode(ERRCODE_CONNECTION_FAILURE),
                  errmsg("could not fetch table info for table \"%s.%s\" from publisher: %s",
                         nspname, relname, res->err)));
  
     slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple);
     if (!tuplestore_gettupleslot(res->tuplestore, true, false, slot))
         ereport(ERROR,
                 (errcode(ERRCODE_UNDEFINED_OBJECT),
                  errmsg("table \"%s.%s\" not found on publisher",
                         nspname, relname)));
  
     lrel->remoteid = DatumGetObjectId(slot_getattr(slot, 1, &isnull));
     Assert(!isnull);
     lrel->replident = DatumGetChar(slot_getattr(slot, 2, &isnull));
     Assert(!isnull);
     lrel->relkind = DatumGetChar(slot_getattr(slot, 3, &isnull));
     Assert(!isnull);
  
     ExecDropSingleTupleTableSlot(slot);
     walrcv_clear_result(res);
  
  
     /*
      * Get column lists for each relation.
      *
      * We need to do this before fetching info about column names and types,
      * so that we can skip columns that should not be replicated.
      */
     if (walrcv_server_version(LogRepWorkerWalRcvConn) >= 150000)
     {
         WalRcvExecResult *pubres;
         TupleTableSlot *tslot;
         Oid         attrsRow[] = {INT2VECTOROID};
         StringInfoData pub_names;
  
         initStringInfo(&pub_names);
         foreach(lc, MySubscription->publications)
         {
             if (foreach_current_index(lc) > 0)
                 appendStringInfoString(&pub_names, ", ");
             appendStringInfoString(&pub_names, quote_literal_cstr(strVal(lfirst(lc))));
         }
  
         /*
          * Fetch info about column lists for the relation (from all the
          * publications).
          */
         resetStringInfo(&cmd);
         appendStringInfo(&cmd,
                          "SELECT DISTINCT"
                          "  (CASE WHEN (array_length(gpt.attrs, 1) = c.relnatts)"
                          "   THEN NULL ELSE gpt.attrs END)"
                          "  FROM pg_publication p,"
                          "  LATERAL pg_get_publication_tables(p.pubname) gpt,"
                          "  pg_class c"
                          " WHERE gpt.relid = %u AND c.oid = gpt.relid"
                          "   AND p.pubname IN ( %s )",
                          lrel->remoteid,
                          pub_names.data);
  
         pubres = walrcv_exec(LogRepWorkerWalRcvConn, cmd.data,
                              lengthof(attrsRow), attrsRow);
  
         if (pubres->status != WALRCV_OK_TUPLES)
             ereport(ERROR,
                     (errcode(ERRCODE_CONNECTION_FAILURE),
                      errmsg("could not fetch column list info for table \"%s.%s\" from publisher: %s",
                             nspname, relname, pubres->err)));
  
         /*
          * We don't support the case where the column list is different for
          * the same table when combining publications. See comments atop
          * fetch_table_list. So there should be only one row returned.
          * Although we already checked this when creating the subscription, we
          * still need to check here in case the column list was changed after
          * creating the subscription and before the sync worker is started.
          */
         if (tuplestore_tuple_count(pubres->tuplestore) > 1)
             ereport(ERROR,
                     errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                     errmsg("cannot use different column lists for table \"%s.%s\" in different publications",
                            nspname, relname));
  
         /*
          * Get the column list and build a single bitmap with the attnums.
          *
          * If we find a NULL value, it means all the columns should be
          * replicated.
          */
         tslot = MakeSingleTupleTableSlot(pubres->tupledesc, &TTSOpsMinimalTuple);
         if (tuplestore_gettupleslot(pubres->tuplestore, true, false, tslot))
         {
             Datum       cfval = slot_getattr(tslot, 1, &isnull);
  
             if (!isnull)
             {
                 ArrayType  *arr;
                 int         nelems;
                 int16      *elems;
  
                 arr = DatumGetArrayTypeP(cfval);
                 nelems = ARR_DIMS(arr)[0];
                 elems = (int16 *) ARR_DATA_PTR(arr);
  
                 for (natt = 0; natt < nelems; natt++)
                     included_cols = bms_add_member(included_cols, elems[natt]);
             }
  
             ExecClearTuple(tslot);
         }
         ExecDropSingleTupleTableSlot(tslot);
  
         walrcv_clear_result(pubres);
  
         pfree(pub_names.data);
     }
  
     /*
      * Now fetch column names and types.
      */
     resetStringInfo(&cmd);
     appendStringInfo(&cmd,
                      "SELECT a.attnum,"
                      "       a.attname,"
                      "       a.atttypid,"
                      "       a.attnum = ANY(i.indkey)"
                      "  FROM pg_catalog.pg_attribute a"
                      "  LEFT JOIN pg_catalog.pg_index i"
                      "       ON (i.indexrelid = pg_get_replica_identity_index(%u))"
                      " WHERE a.attnum > 0::pg_catalog.int2"
                      "   AND NOT a.attisdropped %s"
                      "   AND a.attrelid = %u"
                      " ORDER BY a.attnum",
                      lrel->remoteid,
                      (walrcv_server_version(LogRepWorkerWalRcvConn) >= 120000 ?
                       "AND a.attgenerated = ''" : ""),
                      lrel->remoteid);
     res = walrcv_exec(LogRepWorkerWalRcvConn, cmd.data,
                       lengthof(attrRow), attrRow);
  
     if (res->status != WALRCV_OK_TUPLES)
         ereport(ERROR,
                 (errcode(ERRCODE_CONNECTION_FAILURE),
                  errmsg("could not fetch table info for table \"%s.%s\" from publisher: %s",
                         nspname, relname, res->err)));
  
     /* We don't know the number of rows coming, so allocate enough space. */
     lrel->attnames = palloc0(MaxTupleAttributeNumber * sizeof(char *));
     lrel->atttyps = palloc0(MaxTupleAttributeNumber * sizeof(Oid));
     lrel->attkeys = NULL;
  
     /*
      * Store the columns as a list of names.  Ignore those that are not
      * present in the column list, if there is one.
      */
     natt = 0;
     slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple);
     while (tuplestore_gettupleslot(res->tuplestore, true, false, slot))
     {
         char       *rel_colname;
         AttrNumber  attnum;
  
         attnum = DatumGetInt16(slot_getattr(slot, 1, &isnull));
         Assert(!isnull);
  
         /* If the column is not in the column list, skip it. */
         if (included_cols != NULL && !bms_is_member(attnum, included_cols))
         {
             ExecClearTuple(slot);
             continue;
         }
  
         rel_colname = TextDatumGetCString(slot_getattr(slot, 2, &isnull));
         Assert(!isnull);
  
         lrel->attnames[natt] = rel_colname;
         lrel->atttyps[natt] = DatumGetObjectId(slot_getattr(slot, 3, &isnull));
         Assert(!isnull);
  
         if (DatumGetBool(slot_getattr(slot, 4, &isnull)))
             lrel->attkeys = bms_add_member(lrel->attkeys, natt);
  
         /* Should never happen. */
         if (++natt >= MaxTupleAttributeNumber)
             elog(ERROR, "too many columns in remote table \"%s.%s\"",
                  nspname, relname);
  
         ExecClearTuple(slot);
     }
     ExecDropSingleTupleTableSlot(slot);
  
     lrel->natts = natt;
  
     walrcv_clear_result(res);
  
     /*
      * Get relation's row filter expressions. DISTINCT avoids the same
      * expression of a table in multiple publications from being included
      * multiple times in the final expression.
      *
      * We need to copy the row even if it matches just one of the
      * publications, so we later combine all the quals with OR.
      *
      * For initial synchronization, row filtering can be ignored in following
      * cases:
      *
      * 1) one of the subscribed publications for the table hasn't specified
      * any row filter
      *
      * 2) one of the subscribed publications has puballtables set to true
      *
      * 3) one of the subscribed publications is declared as TABLES IN SCHEMA
      * that includes this relation
      */
     if (walrcv_server_version(LogRepWorkerWalRcvConn) >= 150000)
     {
         StringInfoData pub_names;
  
         /* Build the pubname list. */
         initStringInfo(&pub_names);
         foreach(lc, MySubscription->publications)
         {
             char       *pubname = strVal(lfirst(lc));
  
             if (foreach_current_index(lc) > 0)
                 appendStringInfoString(&pub_names, ", ");
  
             appendStringInfoString(&pub_names, quote_literal_cstr(pubname));
         }
  
         /* Check for row filters. */
         resetStringInfo(&cmd);
         appendStringInfo(&cmd,
                          "SELECT DISTINCT pg_get_expr(gpt.qual, gpt.relid)"
                          "  FROM pg_publication p,"
                          "  LATERAL pg_get_publication_tables(p.pubname) gpt"
                          " WHERE gpt.relid = %u"
                          "   AND p.pubname IN ( %s )",
                          lrel->remoteid,
                          pub_names.data);
  
         res = walrcv_exec(LogRepWorkerWalRcvConn, cmd.data, 1, qualRow);
  
         if (res->status != WALRCV_OK_TUPLES)
             ereport(ERROR,
                     (errmsg("could not fetch table WHERE clause info for table \"%s.%s\" from publisher: %s",
                             nspname, relname, res->err)));
  
         /*
          * Multiple row filter expressions for the same table will be combined
          * by COPY using OR. If any of the filter expressions for this table
          * are null, it means the whole table will be copied. In this case it
          * is not necessary to construct a unified row filter expression at
          * all.
          */
         slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple);
         while (tuplestore_gettupleslot(res->tuplestore, true, false, slot))
         {
             Datum       rf = slot_getattr(slot, 1, &isnull);
  
             if (!isnull)
                 *qual = lappend(*qual, makeString(TextDatumGetCString(rf)));
             else
             {
                 /* Ignore filters and cleanup as necessary. */
                 if (*qual)
                 {
                     list_free_deep(*qual);
                     *qual = NIL;
                 }
                 break;
             }
  
             ExecClearTuple(slot);
         }
         ExecDropSingleTupleTableSlot(slot);
  
         walrcv_clear_result(res);
     }
  
     pfree(cmd.data);
 }

Referenced by copy_table().

◆ FetchTableStates()

static bool FetchTableStates ( bool * started_tx )

static

Definition at line 1497 of file tablesync.c.

 {
     static bool has_subrels = false;
  
     *started_tx = false;
  
     if (!table_states_valid)
     {
         MemoryContext oldctx;
         List       *rstates;
         ListCell   *lc;
         SubscriptionRelState *rstate;
  
         /* Clean the old lists. */
         list_free_deep(table_states_not_ready);
         table_states_not_ready = NIL;
  
         if (!IsTransactionState())
         {
             StartTransactionCommand();
             *started_tx = true;
         }
  
         /* Fetch all non-ready tables. */
         rstates = GetSubscriptionRelations(MySubscription->oid, true);
  
         /* Allocate the tracking info in a permanent memory context. */
         oldctx = MemoryContextSwitchTo(CacheMemoryContext);
         foreach(lc, rstates)
         {
             rstate = palloc(sizeof(SubscriptionRelState));
             memcpy(rstate, lfirst(lc), sizeof(SubscriptionRelState));
             table_states_not_ready = lappend(table_states_not_ready, rstate);
         }
         MemoryContextSwitchTo(oldctx);
  
         /*
          * Does the subscription have tables?
          *
          * If there were not-READY relations found then we know it does. But
          * if table_state_not_ready was empty we still need to check again to
          * see if there are 0 tables.
          */
         has_subrels = (table_states_not_ready != NIL) ||
             HasSubscriptionRelations(MySubscription->oid);
  
         table_states_valid = true;
     }
  
     return has_subrels;
 }

References CacheMemoryContext, GetSubscriptionRelations(), HasSubscriptionRelations(), IsTransactionState(), lappend(), lfirst, list_free_deep(), MemoryContextSwitchTo(), MySubscription, NIL, Subscription::oid, palloc(), StartTransactionCommand(), table_states_not_ready, and table_states_valid.

Referenced by AllTablesyncsReady(), and process_syncing_tables_for_apply().

◆ invalidate_syncing_table_states()

void invalidate_syncing_table_states	(	Datum	arg,
		int	cacheid,
		uint32	hashvalue
	)

Definition at line 271 of file tablesync.c.

 {
     table_states_valid = false;
 }

References table_states_valid.

Referenced by ApplyWorkerMain(), and ParallelApplyWorkerMain().

◆ LogicalRepSyncTableStart()

char* LogicalRepSyncTableStart ( XLogRecPtr * origin_startpos )

Definition at line 1229 of file tablesync.c.

 {
     char       *slotname;
     char       *err;
     char        relstate;
     XLogRecPtr  relstate_lsn;
     Relation    rel;
     AclResult   aclresult;
     WalRcvExecResult *res;
     char        originname[NAMEDATALEN];
     RepOriginId originid;
  
     /* Check the state of the table synchronization. */
     StartTransactionCommand();
     relstate = GetSubscriptionRelState(MyLogicalRepWorker->subid,
                                        MyLogicalRepWorker->relid,
                                        &relstate_lsn);
     CommitTransactionCommand();
  
     SpinLockAcquire(&MyLogicalRepWorker->relmutex);
     MyLogicalRepWorker->relstate = relstate;
     MyLogicalRepWorker->relstate_lsn = relstate_lsn;
     SpinLockRelease(&MyLogicalRepWorker->relmutex);
  
     /*
      * If synchronization is already done or no longer necessary, exit now
      * that we've updated shared memory state.
      */
     switch (relstate)
     {
         case SUBREL_STATE_SYNCDONE:
         case SUBREL_STATE_READY:
         case SUBREL_STATE_UNKNOWN:
             finish_sync_worker();   /* doesn't return */
     }
  
     /* Calculate the name of the tablesync slot. */
     slotname = (char *) palloc(NAMEDATALEN);
     ReplicationSlotNameForTablesync(MySubscription->oid,
                                     MyLogicalRepWorker->relid,
                                     slotname,
                                     NAMEDATALEN);
  
     /*
      * Here we use the slot name instead of the subscription name as the
      * application_name, so that it is different from the leader apply worker,
      * so that synchronous replication can distinguish them.
      */
     LogRepWorkerWalRcvConn =
         walrcv_connect(MySubscription->conninfo, true, slotname, &err);
     if (LogRepWorkerWalRcvConn == NULL)
         ereport(ERROR,
                 (errcode(ERRCODE_CONNECTION_FAILURE),
                  errmsg("could not connect to the publisher: %s", err)));
  
     Assert(MyLogicalRepWorker->relstate == SUBREL_STATE_INIT ||
            MyLogicalRepWorker->relstate == SUBREL_STATE_DATASYNC ||
            MyLogicalRepWorker->relstate == SUBREL_STATE_FINISHEDCOPY);
  
     /* Assign the origin tracking record name. */
     ReplicationOriginNameForLogicalRep(MySubscription->oid,
                                        MyLogicalRepWorker->relid,
                                        originname,
                                        sizeof(originname));
  
     if (MyLogicalRepWorker->relstate == SUBREL_STATE_DATASYNC)
     {
         /*
          * We have previously errored out before finishing the copy so the
          * replication slot might exist. We want to remove the slot if it
          * already exists and proceed.
          *
          * XXX We could also instead try to drop the slot, last time we failed
          * but for that, we might need to clean up the copy state as it might
          * be in the middle of fetching the rows. Also, if there is a network
          * breakdown then it wouldn't have succeeded so trying it next time
          * seems like a better bet.
          */
         ReplicationSlotDropAtPubNode(LogRepWorkerWalRcvConn, slotname, true);
     }
     else if (MyLogicalRepWorker->relstate == SUBREL_STATE_FINISHEDCOPY)
     {
         /*
          * The COPY phase was previously done, but tablesync then crashed
          * before it was able to finish normally.
          */
         StartTransactionCommand();
  
         /*
          * The origin tracking name must already exist. It was created first
          * time this tablesync was launched.
          */
         originid = replorigin_by_name(originname, false);
         replorigin_session_setup(originid, 0);
         replorigin_session_origin = originid;
         *origin_startpos = replorigin_session_get_progress(false);
  
         CommitTransactionCommand();
  
         goto copy_table_done;
     }
  
     SpinLockAcquire(&MyLogicalRepWorker->relmutex);
     MyLogicalRepWorker->relstate = SUBREL_STATE_DATASYNC;
     MyLogicalRepWorker->relstate_lsn = InvalidXLogRecPtr;
     SpinLockRelease(&MyLogicalRepWorker->relmutex);
  
     /* Update the state and make it visible to others. */
     StartTransactionCommand();
     UpdateSubscriptionRelState(MyLogicalRepWorker->subid,
                                MyLogicalRepWorker->relid,
                                MyLogicalRepWorker->relstate,
                                MyLogicalRepWorker->relstate_lsn);
     CommitTransactionCommand();
     pgstat_report_stat(true);
  
     StartTransactionCommand();
  
     /*
      * Use a standard write lock here. It might be better to disallow access
      * to the table while it's being synchronized. But we don't want to block
      * the main apply process from working and it has to open the relation in
      * RowExclusiveLock when remapping remote relation id to local one.
      */
     rel = table_open(MyLogicalRepWorker->relid, RowExclusiveLock);
  
     /*
      * Check that our table sync worker has permission to insert into the
      * target table.
      */
     aclresult = pg_class_aclcheck(RelationGetRelid(rel), GetUserId(),
                                   ACL_INSERT);
     if (aclresult != ACLCHECK_OK)
         aclcheck_error(aclresult,
                        get_relkind_objtype(rel->rd_rel->relkind),
                        RelationGetRelationName(rel));
  
     /*
      * COPY FROM does not honor RLS policies.  That is not a problem for
      * subscriptions owned by roles with BYPASSRLS privilege (or superuser,
      * who has it implicitly), but other roles should not be able to
      * circumvent RLS.  Disallow logical replication into RLS enabled
      * relations for such roles.
      */
     if (check_enable_rls(RelationGetRelid(rel), InvalidOid, false) == RLS_ENABLED)
         ereport(ERROR,
                 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                  errmsg("user \"%s\" cannot replicate into relation with row-level security enabled: \"%s\"",
                         GetUserNameFromId(GetUserId(), true),
                         RelationGetRelationName(rel))));
  
     /*
      * Start a transaction in the remote node in REPEATABLE READ mode.  This
      * ensures that both the replication slot we create (see below) and the
      * COPY are consistent with each other.
      */
     res = walrcv_exec(LogRepWorkerWalRcvConn,
                       "BEGIN READ ONLY ISOLATION LEVEL REPEATABLE READ",
                       0, NULL);
     if (res->status != WALRCV_OK_COMMAND)
         ereport(ERROR,
                 (errcode(ERRCODE_CONNECTION_FAILURE),
                  errmsg("table copy could not start transaction on publisher: %s",
                         res->err)));
     walrcv_clear_result(res);
  
     /*
      * Create a new permanent logical decoding slot. This slot will be used
      * for the catchup phase after COPY is done, so tell it to use the
      * snapshot to make the final data consistent.
      */
     walrcv_create_slot(LogRepWorkerWalRcvConn,
                        slotname, false /* permanent */ , false /* two_phase */ ,
                        CRS_USE_SNAPSHOT, origin_startpos);
  
     /*
      * Setup replication origin tracking. The purpose of doing this before the
      * copy is to avoid doing the copy again due to any error in setting up
      * origin tracking.
      */
     originid = replorigin_by_name(originname, true);
     if (!OidIsValid(originid))
     {
         /*
          * Origin tracking does not exist, so create it now.
          *
          * Then advance to the LSN got from walrcv_create_slot. This is WAL
          * logged for the purpose of recovery. Locks are to prevent the
          * replication origin from vanishing while advancing.
          */
         originid = replorigin_create(originname);
  
         LockRelationOid(ReplicationOriginRelationId, RowExclusiveLock);
         replorigin_advance(originid, *origin_startpos, InvalidXLogRecPtr,
                            true /* go backward */ , true /* WAL log */ );
         UnlockRelationOid(ReplicationOriginRelationId, RowExclusiveLock);
  
         replorigin_session_setup(originid, 0);
         replorigin_session_origin = originid;
     }
     else
     {
         ereport(ERROR,
                 (errcode(ERRCODE_DUPLICATE_OBJECT),
                  errmsg("replication origin \"%s\" already exists",
                         originname)));
     }
  
     /* Now do the initial data copy */
     PushActiveSnapshot(GetTransactionSnapshot());
     copy_table(rel);
     PopActiveSnapshot();
  
     res = walrcv_exec(LogRepWorkerWalRcvConn, "COMMIT", 0, NULL);
     if (res->status != WALRCV_OK_COMMAND)
         ereport(ERROR,
                 (errcode(ERRCODE_CONNECTION_FAILURE),
                  errmsg("table copy could not finish transaction on publisher: %s",
                         res->err)));
     walrcv_clear_result(res);
  
     table_close(rel, NoLock);
  
     /* Make the copy visible. */
     CommandCounterIncrement();
  
     /*
      * Update the persisted state to indicate the COPY phase is done; make it
      * visible to others.
      */
     UpdateSubscriptionRelState(MyLogicalRepWorker->subid,
                                MyLogicalRepWorker->relid,
                                SUBREL_STATE_FINISHEDCOPY,
                                MyLogicalRepWorker->relstate_lsn);
  
     CommitTransactionCommand();
  
 copy_table_done:
  
     elog(DEBUG1,
          "LogicalRepSyncTableStart: '%s' origin_startpos lsn %X/%X",
          originname, LSN_FORMAT_ARGS(*origin_startpos));
  
     /*
      * We are done with the initial data synchronization, update the state.
      */
     SpinLockAcquire(&MyLogicalRepWorker->relmutex);
     MyLogicalRepWorker->relstate = SUBREL_STATE_SYNCWAIT;
     MyLogicalRepWorker->relstate_lsn = *origin_startpos;
     SpinLockRelease(&MyLogicalRepWorker->relmutex);
  
     /*
      * Finally, wait until the leader apply worker tells us to catch up and
      * then return to let LogicalRepApplyLoop do it.
      */
     wait_for_worker_state_change(SUBREL_STATE_CATCHUP);
     return slotname;
 }

References ACL_INSERT, aclcheck_error(), ACLCHECK_OK, Assert(), check_enable_rls(), CommandCounterIncrement(), CommitTransactionCommand(), Subscription::conninfo, copy_table(), CRS_USE_SNAPSHOT, DEBUG1, elog(), ereport, err(), errcode(), ERRCODE_DUPLICATE_OBJECT, errmsg(), ERROR, get_relkind_objtype(), GetSubscriptionRelState(), GetTransactionSnapshot(), GetUserId(), GetUserNameFromId(), InvalidOid, InvalidXLogRecPtr, LockRelationOid(), LogRepWorkerWalRcvConn, LSN_FORMAT_ARGS, MyLogicalRepWorker, MySubscription, NAMEDATALEN, NoLock, Subscription::oid, OidIsValid, palloc(), pg_class_aclcheck(), pgstat_report_stat(), PopActiveSnapshot(), PushActiveSnapshot(), RelationData::rd_rel, RelationGetRelationName, RelationGetRelid, LogicalRepWorker::relid, LogicalRepWorker::relmutex, LogicalRepWorker::relstate, LogicalRepWorker::relstate_lsn, ReplicationOriginNameForLogicalRep(), ReplicationSlotDropAtPubNode(), ReplicationSlotNameForTablesync(), replorigin_advance(), replorigin_by_name(), replorigin_create(), replorigin_session_get_progress(), replorigin_session_origin, replorigin_session_setup(), res, RLS_ENABLED, RowExclusiveLock, SpinLockAcquire, SpinLockRelease, StartTransactionCommand(), LogicalRepWorker::subid, table_close(), table_open(), UnlockRelationOid(), UpdateSubscriptionRelState(), wait_for_worker_state_change(), walrcv_clear_result(), walrcv_connect, walrcv_create_slot, walrcv_exec, and WALRCV_OK_COMMAND.

Referenced by start_table_sync().

◆ make_copy_attnamelist()

static List* make_copy_attnamelist ( LogicalRepRelMapEntry * rel )

static

Definition at line 666 of file tablesync.c.

 {
     List       *attnamelist = NIL;
     int         i;
  
     for (i = 0; i < rel->remoterel.natts; i++)
     {
         attnamelist = lappend(attnamelist,
                               makeString(rel->remoterel.attnames[i]));
     }
  
  
     return attnamelist;
 }

References LogicalRepRelation::attnames, i, lappend(), makeString(), LogicalRepRelation::natts, NIL, and LogicalRepRelMapEntry::remoterel.

Referenced by copy_table().

◆ pg_attribute_noreturn()

static void pg_attribute_noreturn ( )

static

Definition at line 133 of file tablesync.c.

 {
     /*
      * Commit any outstanding transaction. This is the usual case, unless
      * there was nothing to do for the table.
      */
     if (IsTransactionState())
     {
         CommitTransactionCommand();
         pgstat_report_stat(true);
     }
  
     /* And flush all writes. */
     XLogFlush(GetXLogWriteRecPtr());
  
     StartTransactionCommand();
     ereport(LOG,
             (errmsg("logical replication table synchronization worker for subscription \"%s\", table \"%s\" has finished",
                     MySubscription->name,
                     get_rel_name(MyLogicalRepWorker->relid))));
     CommitTransactionCommand();
  
     /* Find the leader apply worker and signal it. */
     logicalrep_worker_wakeup(MyLogicalRepWorker->subid, InvalidOid);
  
     /* Stop gracefully */
     proc_exit(0);
 }

References CommitTransactionCommand(), ereport, errmsg(), get_rel_name(), GetXLogWriteRecPtr(), InvalidOid, IsTransactionState(), LOG, logicalrep_worker_wakeup(), MyLogicalRepWorker, MySubscription, Subscription::name, pgstat_report_stat(), proc_exit(), LogicalRepWorker::relid, StartTransactionCommand(), LogicalRepWorker::subid, and XLogFlush().

◆ process_syncing_tables()

void process_syncing_tables ( XLogRecPtr current_lsn )

Definition at line 646 of file tablesync.c.

 {
     /*
      * Skip for parallel apply workers because they only operate on tables
      * that are in a READY state. See pa_can_start() and
      * should_apply_changes_for_rel().
      */
     if (am_parallel_apply_worker())
         return;
  
     if (am_tablesync_worker())
         process_syncing_tables_for_sync(current_lsn);
     else
         process_syncing_tables_for_apply(current_lsn);
 }

References am_parallel_apply_worker(), am_tablesync_worker(), process_syncing_tables_for_apply(), and process_syncing_tables_for_sync().

Referenced by apply_handle_commit(), apply_handle_commit_prepared(), apply_handle_prepare(), apply_handle_rollback_prepared(), apply_handle_stream_commit(), apply_handle_stream_prepare(), and LogicalRepApplyLoop().

◆ process_syncing_tables_for_apply()

static void process_syncing_tables_for_apply ( XLogRecPtr current_lsn )

static

Definition at line 408 of file tablesync.c.

 {
     struct tablesync_start_time_mapping
     {
         Oid         relid;
         TimestampTz last_start_time;
     };
     static HTAB *last_start_times = NULL;
     ListCell   *lc;
     bool        started_tx = false;
     bool        should_exit = false;
  
     Assert(!IsTransactionState());
  
     /* We need up-to-date sync state info for subscription tables here. */
     FetchTableStates(&started_tx);
  
     /*
      * Prepare a hash table for tracking last start times of workers, to avoid
      * immediate restarts.  We don't need it if there are no tables that need
      * syncing.
      */
     if (table_states_not_ready != NIL && !last_start_times)
     {
         HASHCTL     ctl;
  
         ctl.keysize = sizeof(Oid);
         ctl.entrysize = sizeof(struct tablesync_start_time_mapping);
         last_start_times = hash_create("Logical replication table sync worker start times",
                                        256, &ctl, HASH_ELEM | HASH_BLOBS);
     }
  
     /*
      * Clean up the hash table when we're done with all tables (just to
      * release the bit of memory).
      */
     else if (table_states_not_ready == NIL && last_start_times)
     {
         hash_destroy(last_start_times);
         last_start_times = NULL;
     }
  
     /*
      * Process all tables that are being synchronized.
      */
     foreach(lc, table_states_not_ready)
     {
         SubscriptionRelState *rstate = (SubscriptionRelState *) lfirst(lc);
  
         if (rstate->state == SUBREL_STATE_SYNCDONE)
         {
             /*
              * Apply has caught up to the position where the table sync has
              * finished.  Mark the table as ready so that the apply will just
              * continue to replicate it normally.
              */
             if (current_lsn >= rstate->lsn)
             {
                 char        originname[NAMEDATALEN];
  
                 rstate->state = SUBREL_STATE_READY;
                 rstate->lsn = current_lsn;
                 if (!started_tx)
                 {
                     StartTransactionCommand();
                     started_tx = true;
                 }
  
                 /*
                  * Remove the tablesync origin tracking if exists.
                  *
                  * There is a chance that the user is concurrently performing
                  * refresh for the subscription where we remove the table
                  * state and its origin or the tablesync worker would have
                  * already removed this origin. We can't rely on tablesync
                  * worker to remove the origin tracking as if there is any
                  * error while dropping we won't restart it to drop the
                  * origin. So passing missing_ok = true.
                  */
                 ReplicationOriginNameForLogicalRep(MyLogicalRepWorker->subid,
                                                    rstate->relid,
                                                    originname,
                                                    sizeof(originname));
                 replorigin_drop_by_name(originname, true, false);
  
                 /*
                  * Update the state to READY only after the origin cleanup.
                  */
                 UpdateSubscriptionRelState(MyLogicalRepWorker->subid,
                                            rstate->relid, rstate->state,
                                            rstate->lsn);
             }
         }
         else
         {
             LogicalRepWorker *syncworker;
  
             /*
              * Look for a sync worker for this relation.
              */
             LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
  
             syncworker = logicalrep_worker_find(MyLogicalRepWorker->subid,
                                                 rstate->relid, false);
  
             if (syncworker)
             {
                 /* Found one, update our copy of its state */
                 SpinLockAcquire(&syncworker->relmutex);
                 rstate->state = syncworker->relstate;
                 rstate->lsn = syncworker->relstate_lsn;
                 if (rstate->state == SUBREL_STATE_SYNCWAIT)
                 {
                     /*
                      * Sync worker is waiting for apply.  Tell sync worker it
                      * can catchup now.
                      */
                     syncworker->relstate = SUBREL_STATE_CATCHUP;
                     syncworker->relstate_lsn =
                         Max(syncworker->relstate_lsn, current_lsn);
                 }
                 SpinLockRelease(&syncworker->relmutex);
  
                 /* If we told worker to catch up, wait for it. */
                 if (rstate->state == SUBREL_STATE_SYNCWAIT)
                 {
                     /* Signal the sync worker, as it may be waiting for us. */
                     if (syncworker->proc)
                         logicalrep_worker_wakeup_ptr(syncworker);
  
                     /* Now safe to release the LWLock */
                     LWLockRelease(LogicalRepWorkerLock);
  
                     /*
                      * Enter busy loop and wait for synchronization worker to
                      * reach expected state (or die trying).
                      */
                     if (!started_tx)
                     {
                         StartTransactionCommand();
                         started_tx = true;
                     }
  
                     wait_for_relation_state_change(rstate->relid,
                                                    SUBREL_STATE_SYNCDONE);
                 }
                 else
                     LWLockRelease(LogicalRepWorkerLock);
             }
             else
             {
                 /*
                  * If there is no sync worker for this table yet, count
                  * running sync workers for this subscription, while we have
                  * the lock.
                  */
                 int         nsyncworkers =
                 logicalrep_sync_worker_count(MyLogicalRepWorker->subid);
  
                 /* Now safe to release the LWLock */
                 LWLockRelease(LogicalRepWorkerLock);
  
                 /*
                  * If there are free sync worker slot(s), start a new sync
                  * worker for the table.
                  */
                 if (nsyncworkers < max_sync_workers_per_subscription)
                 {
                     TimestampTz now = GetCurrentTimestamp();
                     struct tablesync_start_time_mapping *hentry;
                     bool        found;
  
                     hentry = hash_search(last_start_times, &rstate->relid,
                                          HASH_ENTER, &found);
  
                     if (!found ||
                         TimestampDifferenceExceeds(hentry->last_start_time, now,
                                                    wal_retrieve_retry_interval))
                     {
                         logicalrep_worker_launch(MyLogicalRepWorker->dbid,
                                                  MySubscription->oid,
                                                  MySubscription->name,
                                                  MyLogicalRepWorker->userid,
                                                  rstate->relid,
                                                  DSM_HANDLE_INVALID);
                         hentry->last_start_time = now;
                     }
                 }
             }
         }
     }
  
     if (started_tx)
     {
         /*
          * Even when the two_phase mode is requested by the user, it remains
          * as 'pending' until all tablesyncs have reached READY state.
          *
          * When this happens, we restart the apply worker and (if the
          * conditions are still ok) then the two_phase tri-state will become
          * 'enabled' at that time.
          *
          * Note: If the subscription has no tables then leave the state as
          * PENDING, which allows ALTER SUBSCRIPTION ... REFRESH PUBLICATION to
          * work.
          */
         if (MySubscription->twophasestate == LOGICALREP_TWOPHASE_STATE_PENDING)
         {
             CommandCounterIncrement();  /* make updates visible */
             if (AllTablesyncsReady())
             {
                 ereport(LOG,
                         (errmsg("logical replication apply worker for subscription \"%s\" will restart so that two_phase can be enabled",
                                 MySubscription->name)));
                 should_exit = true;
             }
         }
  
         CommitTransactionCommand();
         pgstat_report_stat(true);
     }
  
     if (should_exit)
     {
         /*
          * Reset the last-start time for this worker so that the launcher will
          * restart it without waiting for wal_retrieve_retry_interval.
          */
         ApplyLauncherForgetWorkerStartTime(MySubscription->oid);
  
         proc_exit(0);
     }
 }

References AllTablesyncsReady(), ApplyLauncherForgetWorkerStartTime(), Assert(), CommandCounterIncrement(), CommitTransactionCommand(), LogicalRepWorker::dbid, DSM_HANDLE_INVALID, HASHCTL::entrysize, ereport, errmsg(), FetchTableStates(), GetCurrentTimestamp(), HASH_BLOBS, hash_create(), hash_destroy(), HASH_ELEM, HASH_ENTER, hash_search(), IsTransactionState(), HASHCTL::keysize, last_start_times, lfirst, LOG, logicalrep_sync_worker_count(), LOGICALREP_TWOPHASE_STATE_PENDING, logicalrep_worker_find(), logicalrep_worker_launch(), logicalrep_worker_wakeup_ptr(), SubscriptionRelState::lsn, LW_SHARED, LWLockAcquire(), LWLockRelease(), Max, max_sync_workers_per_subscription, MyLogicalRepWorker, MySubscription, Subscription::name, NAMEDATALEN, NIL, now(), Subscription::oid, pgstat_report_stat(), LogicalRepWorker::proc, proc_exit(), SubscriptionRelState::relid, LogicalRepWorker::relmutex, LogicalRepWorker::relstate, LogicalRepWorker::relstate_lsn, ReplicationOriginNameForLogicalRep(), replorigin_drop_by_name(), SpinLockAcquire, SpinLockRelease, StartTransactionCommand(), SubscriptionRelState::state, LogicalRepWorker::subid, table_states_not_ready, TimestampDifferenceExceeds(), Subscription::twophasestate, UpdateSubscriptionRelState(), LogicalRepWorker::userid, wait_for_relation_state_change(), and wal_retrieve_retry_interval.

Referenced by process_syncing_tables().

◆ process_syncing_tables_for_sync()

static void process_syncing_tables_for_sync ( XLogRecPtr current_lsn )

static

Definition at line 285 of file tablesync.c.

 {
     SpinLockAcquire(&MyLogicalRepWorker->relmutex);
  
     if (MyLogicalRepWorker->relstate == SUBREL_STATE_CATCHUP &&
         current_lsn >= MyLogicalRepWorker->relstate_lsn)
     {
         TimeLineID  tli;
         char        syncslotname[NAMEDATALEN] = {0};
         char        originname[NAMEDATALEN] = {0};
  
         MyLogicalRepWorker->relstate = SUBREL_STATE_SYNCDONE;
         MyLogicalRepWorker->relstate_lsn = current_lsn;
  
         SpinLockRelease(&MyLogicalRepWorker->relmutex);
  
         /*
          * UpdateSubscriptionRelState must be called within a transaction.
          */
         if (!IsTransactionState())
             StartTransactionCommand();
  
         UpdateSubscriptionRelState(MyLogicalRepWorker->subid,
                                    MyLogicalRepWorker->relid,
                                    MyLogicalRepWorker->relstate,
                                    MyLogicalRepWorker->relstate_lsn);
  
         /*
          * End streaming so that LogRepWorkerWalRcvConn can be used to drop
          * the slot.
          */
         walrcv_endstreaming(LogRepWorkerWalRcvConn, &tli);
  
         /*
          * Cleanup the tablesync slot.
          *
          * This has to be done after updating the state because otherwise if
          * there is an error while doing the database operations we won't be
          * able to rollback dropped slot.
          */
         ReplicationSlotNameForTablesync(MyLogicalRepWorker->subid,
                                         MyLogicalRepWorker->relid,
                                         syncslotname,
                                         sizeof(syncslotname));
  
         /*
          * It is important to give an error if we are unable to drop the slot,
          * otherwise, it won't be dropped till the corresponding subscription
          * is dropped. So passing missing_ok = false.
          */
         ReplicationSlotDropAtPubNode(LogRepWorkerWalRcvConn, syncslotname, false);
  
         CommitTransactionCommand();
         pgstat_report_stat(false);
  
         /*
          * Start a new transaction to clean up the tablesync origin tracking.
          * This transaction will be ended within the finish_sync_worker().
          * Now, even, if we fail to remove this here, the apply worker will
          * ensure to clean it up afterward.
          *
          * We need to do this after the table state is set to SYNCDONE.
          * Otherwise, if an error occurs while performing the database
          * operation, the worker will be restarted and the in-memory state of
          * replication progress (remote_lsn) won't be rolled-back which would
          * have been cleared before restart. So, the restarted worker will use
          * invalid replication progress state resulting in replay of
          * transactions that have already been applied.
          */
         StartTransactionCommand();
  
         ReplicationOriginNameForLogicalRep(MyLogicalRepWorker->subid,
                                            MyLogicalRepWorker->relid,
                                            originname,
                                            sizeof(originname));
  
         /*
          * Resetting the origin session removes the ownership of the slot.
          * This is needed to allow the origin to be dropped.
          */
         replorigin_session_reset();
         replorigin_session_origin = InvalidRepOriginId;
         replorigin_session_origin_lsn = InvalidXLogRecPtr;
         replorigin_session_origin_timestamp = 0;
  
         /*
          * Drop the tablesync's origin tracking if exists.
          *
          * There is a chance that the user is concurrently performing refresh
          * for the subscription where we remove the table state and its origin
          * or the apply worker would have removed this origin. So passing
          * missing_ok = true.
          */
         replorigin_drop_by_name(originname, true, false);
  
         finish_sync_worker();
     }
     else
         SpinLockRelease(&MyLogicalRepWorker->relmutex);
 }

References CommitTransactionCommand(), InvalidRepOriginId, InvalidXLogRecPtr, IsTransactionState(), LogRepWorkerWalRcvConn, MyLogicalRepWorker, NAMEDATALEN, pgstat_report_stat(), LogicalRepWorker::relid, LogicalRepWorker::relmutex, LogicalRepWorker::relstate, LogicalRepWorker::relstate_lsn, ReplicationOriginNameForLogicalRep(), ReplicationSlotDropAtPubNode(), ReplicationSlotNameForTablesync(), replorigin_drop_by_name(), replorigin_session_origin, replorigin_session_origin_lsn, replorigin_session_origin_timestamp, replorigin_session_reset(), SpinLockAcquire, SpinLockRelease, StartTransactionCommand(), LogicalRepWorker::subid, UpdateSubscriptionRelState(), and walrcv_endstreaming.

Referenced by process_syncing_tables().

◆ ReplicationSlotNameForTablesync()

void ReplicationSlotNameForTablesync	(	Oid	suboid,
		Oid	relid,
		char *	syncslotname,
		Size	szslot
	)

Definition at line 1213 of file tablesync.c.

 {
     snprintf(syncslotname, szslot, "pg_%u_sync_%u_" UINT64_FORMAT, suboid,
              relid, GetSystemIdentifier());
 }

References GetSystemIdentifier(), snprintf, and UINT64_FORMAT.

Referenced by AlterSubscription_refresh(), DropSubscription(), LogicalRepSyncTableStart(), process_syncing_tables_for_sync(), and ReportSlotConnectionError().

◆ UpdateTwoPhaseState()

void UpdateTwoPhaseState	(	Oid	suboid,
		char	new_state
	)

Definition at line 1583 of file tablesync.c.

 {
     Relation    rel;
     HeapTuple   tup;
     bool        nulls[Natts_pg_subscription];
     bool        replaces[Natts_pg_subscription];
     Datum       values[Natts_pg_subscription];
  
     Assert(new_state == LOGICALREP_TWOPHASE_STATE_DISABLED ||
            new_state == LOGICALREP_TWOPHASE_STATE_PENDING ||
            new_state == LOGICALREP_TWOPHASE_STATE_ENABLED);
  
     rel = table_open(SubscriptionRelationId, RowExclusiveLock);
     tup = SearchSysCacheCopy1(SUBSCRIPTIONOID, ObjectIdGetDatum(suboid));
     if (!HeapTupleIsValid(tup))
         elog(ERROR,
              "cache lookup failed for subscription oid %u",
              suboid);
  
     /* Form a new tuple. */
     memset(values, 0, sizeof(values));
     memset(nulls, false, sizeof(nulls));
     memset(replaces, false, sizeof(replaces));
  
     /* And update/set two_phase state */
     values[Anum_pg_subscription_subtwophasestate - 1] = CharGetDatum(new_state);
     replaces[Anum_pg_subscription_subtwophasestate - 1] = true;
  
     tup = heap_modify_tuple(tup, RelationGetDescr(rel),
                             values, nulls, replaces);
     CatalogTupleUpdate(rel, &tup->t_self, tup);
  
     heap_freetuple(tup);
     table_close(rel, RowExclusiveLock);
 }

References Assert(), CatalogTupleUpdate(), CharGetDatum(), elog(), ERROR, heap_freetuple(), heap_modify_tuple(), HeapTupleIsValid, LOGICALREP_TWOPHASE_STATE_DISABLED, LOGICALREP_TWOPHASE_STATE_ENABLED, LOGICALREP_TWOPHASE_STATE_PENDING, ObjectIdGetDatum(), RelationGetDescr, RowExclusiveLock, SearchSysCacheCopy1, SUBSCRIPTIONOID, HeapTupleData::t_self, table_close(), table_open(), and values.

Referenced by ApplyWorkerMain(), and CreateSubscription().

◆ wait_for_relation_state_change()

static bool wait_for_relation_state_change	(	Oid	relid,
		char	expected_state
	)

static

Definition at line 174 of file tablesync.c.

 {
     char        state;
  
     for (;;)
     {
         LogicalRepWorker *worker;
         XLogRecPtr  statelsn;
  
         CHECK_FOR_INTERRUPTS();
  
         InvalidateCatalogSnapshot();
         state = GetSubscriptionRelState(MyLogicalRepWorker->subid,
                                         relid, &statelsn);
  
         if (state == SUBREL_STATE_UNKNOWN)
             break;
  
         if (state == expected_state)
             return true;
  
         /* Check if the sync worker is still running and bail if not. */
         LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
         worker = logicalrep_worker_find(MyLogicalRepWorker->subid, relid,
                                         false);
         LWLockRelease(LogicalRepWorkerLock);
         if (!worker)
             break;
  
         (void) WaitLatch(MyLatch,
                          WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
                          1000L, WAIT_EVENT_LOGICAL_SYNC_STATE_CHANGE);
  
         ResetLatch(MyLatch);
     }
  
     return false;
 }

References CHECK_FOR_INTERRUPTS, GetSubscriptionRelState(), InvalidateCatalogSnapshot(), logicalrep_worker_find(), LW_SHARED, LWLockAcquire(), LWLockRelease(), MyLatch, MyLogicalRepWorker, ResetLatch(), LogicalRepWorker::subid, WAIT_EVENT_LOGICAL_SYNC_STATE_CHANGE, WaitLatch(), WL_EXIT_ON_PM_DEATH, WL_LATCH_SET, and WL_TIMEOUT.

Referenced by process_syncing_tables_for_apply().

◆ wait_for_worker_state_change()

static bool wait_for_worker_state_change ( char expected_state )

static

Definition at line 222 of file tablesync.c.

 {
     int         rc;
  
     for (;;)
     {
         LogicalRepWorker *worker;
  
         CHECK_FOR_INTERRUPTS();
  
         /*
          * Done if already in correct state.  (We assume this fetch is atomic
          * enough to not give a misleading answer if we do it with no lock.)
          */
         if (MyLogicalRepWorker->relstate == expected_state)
             return true;
  
         /*
          * Bail out if the apply worker has died, else signal it we're
          * waiting.
          */
         LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
         worker = logicalrep_worker_find(MyLogicalRepWorker->subid,
                                         InvalidOid, false);
         if (worker && worker->proc)
             logicalrep_worker_wakeup_ptr(worker);
         LWLockRelease(LogicalRepWorkerLock);
         if (!worker)
             break;
  
         /*
          * Wait.  We expect to get a latch signal back from the apply worker,
          * but use a timeout in case it dies without sending one.
          */
         rc = WaitLatch(MyLatch,
                        WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
                        1000L, WAIT_EVENT_LOGICAL_SYNC_STATE_CHANGE);
  
         if (rc & WL_LATCH_SET)
             ResetLatch(MyLatch);
     }
  
     return false;
 }

References CHECK_FOR_INTERRUPTS, InvalidOid, logicalrep_worker_find(), logicalrep_worker_wakeup_ptr(), LW_SHARED, LWLockAcquire(), LWLockRelease(), MyLatch, MyLogicalRepWorker, LogicalRepWorker::proc, LogicalRepWorker::relstate, ResetLatch(), LogicalRepWorker::subid, WAIT_EVENT_LOGICAL_SYNC_STATE_CHANGE, WaitLatch(), WL_EXIT_ON_PM_DEATH, WL_LATCH_SET, and WL_TIMEOUT.

Referenced by LogicalRepSyncTableStart().

Variable Documentation

◆ copybuf

StringInfo copybuf = NULL

static

◆ table_states_not_ready

List* table_states_not_ready = NIL

static

Definition at line 124 of file tablesync.c.

Referenced by AllTablesyncsReady(), FetchTableStates(), and process_syncing_tables_for_apply().

◆ table_states_valid

bool table_states_valid = false

static

Definition at line 123 of file tablesync.c.

Referenced by FetchTableStates(), and invalidate_syncing_table_states().

Functions

Variables

Function Documentation

◆ AllTablesyncsReady()

◆ copy_read_data()

◆ copy_table()

◆ fetch_remote_table_info()

◆ FetchTableStates()

◆ invalidate_syncing_table_states()

◆ LogicalRepSyncTableStart()

◆ make_copy_attnamelist()

◆ pg_attribute_noreturn()

◆ process_syncing_tables()

◆ process_syncing_tables_for_apply()

◆ process_syncing_tables_for_sync()

◆ ReplicationSlotNameForTablesync()

◆ UpdateTwoPhaseState()

◆ wait_for_relation_state_change()

◆ wait_for_worker_state_change()

Variable Documentation

◆ copybuf

◆ table_states_not_ready

◆ table_states_valid