PostgreSQL Source Code  git master
tablesync.c File Reference
#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:

Go to the source code of this file.

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 Listmake_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 Listtable_states_not_ready = NIL
 
static StringInfo copybuf = NULL
 

Function Documentation

◆ AllTablesyncsReady()

bool AllTablesyncsReady ( void  )

Definition at line 1558 of file tablesync.c.

1559 {
1560  bool started_tx = false;
1561  bool has_subrels = false;
1562 
1563  /* We need up-to-date sync state info for subscription tables here. */
1564  has_subrels = FetchTableStates(&started_tx);
1565 
1566  if (started_tx)
1567  {
1569  pgstat_report_stat(true);
1570  }
1571 
1572  /*
1573  * Return false when there are no tables in subscription or not all tables
1574  * are in ready state; true otherwise.
1575  */
1576  return has_subrels && (table_states_not_ready == NIL);
1577 }
#define NIL
Definition: pg_list.h:68
long pgstat_report_stat(bool force)
Definition: pgstat.c:575
static List * table_states_not_ready
Definition: tablesync.c:124
static bool FetchTableStates(bool *started_tx)
Definition: tablesync.c:1497
void CommitTransactionCommand(void)
Definition: xact.c:3041

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.

687 {
688  int bytesread = 0;
689  int avail;
690 
691  /* If there are some leftover data from previous read, use it. */
692  avail = copybuf->len - copybuf->cursor;
693  if (avail)
694  {
695  if (avail > maxread)
696  avail = maxread;
697  memcpy(outbuf, &copybuf->data[copybuf->cursor], avail);
698  copybuf->cursor += avail;
699  maxread -= avail;
700  bytesread += avail;
701  }
702 
703  while (maxread > 0 && bytesread < minread)
704  {
706  int len;
707  char *buf = NULL;
708 
709  for (;;)
710  {
711  /* Try read the data. */
713 
715 
716  if (len == 0)
717  break;
718  else if (len < 0)
719  return bytesread;
720  else
721  {
722  /* Process the data */
723  copybuf->data = buf;
724  copybuf->len = len;
725  copybuf->cursor = 0;
726 
727  avail = copybuf->len - copybuf->cursor;
728  if (avail > maxread)
729  avail = maxread;
730  memcpy(outbuf, &copybuf->data[copybuf->cursor], avail);
731  outbuf = (void *) ((char *) outbuf + avail);
732  copybuf->cursor += avail;
733  maxread -= avail;
734  bytesread += avail;
735  }
736 
737  if (maxread <= 0 || bytesread >= minread)
738  return bytesread;
739  }
740 
741  /*
742  * Wait for more data or latch.
743  */
744  (void) WaitLatchOrSocket(MyLatch,
748 
750  }
751 
752  return bytesread;
753 }
WalReceiverConn * LogRepWorkerWalRcvConn
Definition: worker.c:312
struct Latch * MyLatch
Definition: globals.c:58
int WaitLatchOrSocket(Latch *latch, int wakeEvents, pgsocket sock, long timeout, uint32 wait_event_info)
Definition: latch.c:540
void ResetLatch(Latch *latch)
Definition: latch.c:699
#define WL_SOCKET_READABLE
Definition: latch.h:126
#define WL_TIMEOUT
Definition: latch.h:128
#define WL_EXIT_ON_PM_DEATH
Definition: latch.h:130
#define WL_LATCH_SET
Definition: latch.h:125
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:121
const void size_t len
static char * buf
Definition: pg_test_fsync.c:67
int pgsocket
Definition: port.h:29
#define PGINVALID_SOCKET
Definition: port.h:31
static int fd(const char *x, int i)
Definition: preproc-init.c:105
static StringInfo copybuf
Definition: tablesync.c:127
@ WAIT_EVENT_LOGICAL_SYNC_DATA
Definition: wait_event.h:111
#define walrcv_receive(conn, buffer, wait_fd)
Definition: walreceiver.h:426

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.

1084 {
1085  LogicalRepRelMapEntry *relmapentry;
1086  LogicalRepRelation lrel;
1087  List *qual = NIL;
1089  StringInfoData cmd;
1090  CopyFromState cstate;
1091  List *attnamelist;
1092  ParseState *pstate;
1093 
1094  /* Get the publisher relation info. */
1096  RelationGetRelationName(rel), &lrel, &qual);
1097 
1098  /* Put the relation into relmap. */
1099  logicalrep_relmap_update(&lrel);
1100 
1101  /* Map the publisher relation to local one. */
1102  relmapentry = logicalrep_rel_open(lrel.remoteid, NoLock);
1103  Assert(rel == relmapentry->localrel);
1104 
1105  /* Start copy on the publisher. */
1106  initStringInfo(&cmd);
1107 
1108  /* Regular table with no row filter */
1109  if (lrel.relkind == RELKIND_RELATION && qual == NIL)
1110  {
1111  appendStringInfo(&cmd, "COPY %s (",
1113 
1114  /*
1115  * XXX Do we need to list the columns in all cases? Maybe we're
1116  * replicating all columns?
1117  */
1118  for (int i = 0; i < lrel.natts; i++)
1119  {
1120  if (i > 0)
1121  appendStringInfoString(&cmd, ", ");
1122 
1124  }
1125 
1126  appendStringInfoString(&cmd, ") TO STDOUT");
1127  }
1128  else
1129  {
1130  /*
1131  * For non-tables and tables with row filters, we need to do COPY
1132  * (SELECT ...), but we can't just do SELECT * because we need to not
1133  * copy generated columns. For tables with any row filters, build a
1134  * SELECT query with OR'ed row filters for COPY.
1135  */
1136  appendStringInfoString(&cmd, "COPY (SELECT ");
1137  for (int i = 0; i < lrel.natts; i++)
1138  {
1140  if (i < lrel.natts - 1)
1141  appendStringInfoString(&cmd, ", ");
1142  }
1143 
1144  appendStringInfoString(&cmd, " FROM ");
1145 
1146  /*
1147  * For regular tables, make sure we don't copy data from a child that
1148  * inherits the named table as those will be copied separately.
1149  */
1150  if (lrel.relkind == RELKIND_RELATION)
1151  appendStringInfoString(&cmd, "ONLY ");
1152 
1154  /* list of OR'ed filters */
1155  if (qual != NIL)
1156  {
1157  ListCell *lc;
1158  char *q = strVal(linitial(qual));
1159 
1160  appendStringInfo(&cmd, " WHERE %s", q);
1161  for_each_from(lc, qual, 1)
1162  {
1163  q = strVal(lfirst(lc));
1164  appendStringInfo(&cmd, " OR %s", q);
1165  }
1166  list_free_deep(qual);
1167  }
1168 
1169  appendStringInfoString(&cmd, ") TO STDOUT");
1170  }
1171  res = walrcv_exec(LogRepWorkerWalRcvConn, cmd.data, 0, NULL);
1172  pfree(cmd.data);
1173  if (res->status != WALRCV_OK_COPY_OUT)
1174  ereport(ERROR,
1175  (errcode(ERRCODE_CONNECTION_FAILURE),
1176  errmsg("could not start initial contents copy for table \"%s.%s\": %s",
1177  lrel.nspname, lrel.relname, res->err)));
1179 
1180  copybuf = makeStringInfo();
1181 
1182  pstate = make_parsestate(NULL);
1183  (void) addRangeTableEntryForRelation(pstate, rel, AccessShareLock,
1184  NULL, false, false);
1185 
1186  attnamelist = make_copy_attnamelist(relmapentry);
1187  cstate = BeginCopyFrom(pstate, rel, NULL, NULL, false, copy_read_data, attnamelist, NIL);
1188 
1189  /* Do the copy */
1190  (void) CopyFrom(cstate);
1191 
1192  logicalrep_rel_close(relmapentry, NoLock);
1193 }
CopyFromState BeginCopyFrom(ParseState *pstate, Relation rel, Node *whereClause, const char *filename, bool is_program, copy_data_source_cb data_source_cb, List *attnamelist, List *options)
Definition: copyfrom.c:1334
uint64 CopyFrom(CopyFromState cstate)
Definition: copyfrom.c:632
int errcode(int sqlerrcode)
Definition: elog.c:858
int errmsg(const char *fmt,...)
Definition: elog.c:1069
#define ERROR
Definition: elog.h:39
#define ereport(elevel,...)
Definition: elog.h:149
int i
Definition: isn.c:73
Assert(fmt[strlen(fmt) - 1] !='\n')
void list_free_deep(List *list)
Definition: list.c:1559
#define NoLock
Definition: lockdefs.h:34
#define AccessShareLock
Definition: lockdefs.h:36
char * get_namespace_name(Oid nspid)
Definition: lsyscache.c:3331
void pfree(void *pointer)
Definition: mcxt.c:1436
ParseState * make_parsestate(ParseState *parentParseState)
Definition: parse_node.c:44
ParseNamespaceItem * addRangeTableEntryForRelation(ParseState *pstate, Relation rel, int lockmode, Alias *alias, bool inh, bool inFromCl)
#define lfirst(lc)
Definition: pg_list.h:172
#define for_each_from(cell, lst, N)
Definition: pg_list.h:414
#define linitial(l)
Definition: pg_list.h:178
#define RelationGetRelationName(relation)
Definition: rel.h:537
#define RelationGetNamespace(relation)
Definition: rel.h:544
const char * quote_identifier(const char *ident)
Definition: ruleutils.c:11551
char * quote_qualified_identifier(const char *qualifier, const char *ident)
Definition: ruleutils.c:11635
void logicalrep_relmap_update(LogicalRepRelation *remoterel)
Definition: relation.c:162
LogicalRepRelMapEntry * logicalrep_rel_open(LogicalRepRelId remoteid, LOCKMODE lockmode)
Definition: relation.c:325
void logicalrep_rel_close(LogicalRepRelMapEntry *rel, LOCKMODE lockmode)
Definition: relation.c:471
StringInfo makeStringInfo(void)
Definition: stringinfo.c:41
void appendStringInfo(StringInfo str, const char *fmt,...)
Definition: stringinfo.c:91
void appendStringInfoString(StringInfo str, const char *s)
Definition: stringinfo.c:176
void initStringInfo(StringInfo str)
Definition: stringinfo.c:59
Definition: pg_list.h:54
LogicalRepRelId remoteid
Definition: logicalproto.h:107
static int copy_read_data(void *outbuf, int minread, int maxread)
Definition: tablesync.c:686
static void fetch_remote_table_info(char *nspname, char *relname, LogicalRepRelation *lrel, List **qual)
Definition: tablesync.c:762
static List * make_copy_attnamelist(LogicalRepRelMapEntry *rel)
Definition: tablesync.c:666
#define strVal(v)
Definition: value.h:82
@ WALRCV_OK_COPY_OUT
Definition: walreceiver.h:209
static void walrcv_clear_result(WalRcvExecResult *walres)
Definition: walreceiver.h:440
#define walrcv_exec(conn, exec, nRetTypes, retTypes)
Definition: walreceiver.h:434

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.

764 {
766  StringInfoData cmd;
767  TupleTableSlot *slot;
768  Oid tableRow[] = {OIDOID, CHAROID, CHAROID};
769  Oid attrRow[] = {INT2OID, TEXTOID, OIDOID, BOOLOID};
770  Oid qualRow[] = {TEXTOID};
771  bool isnull;
772  int natt;
773  ListCell *lc;
774  Bitmapset *included_cols = NULL;
775 
776  lrel->nspname = nspname;
777  lrel->relname = relname;
778 
779  /* First fetch Oid and replica identity. */
780  initStringInfo(&cmd);
781  appendStringInfo(&cmd, "SELECT c.oid, c.relreplident, c.relkind"
782  " FROM pg_catalog.pg_class c"
783  " INNER JOIN pg_catalog.pg_namespace n"
784  " ON (c.relnamespace = n.oid)"
785  " WHERE n.nspname = %s"
786  " AND c.relname = %s",
787  quote_literal_cstr(nspname),
790  lengthof(tableRow), tableRow);
791 
792  if (res->status != WALRCV_OK_TUPLES)
793  ereport(ERROR,
794  (errcode(ERRCODE_CONNECTION_FAILURE),
795  errmsg("could not fetch table info for table \"%s.%s\" from publisher: %s",
796  nspname, relname, res->err)));
797 
798  slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple);
799  if (!tuplestore_gettupleslot(res->tuplestore, true, false, slot))
800  ereport(ERROR,
801  (errcode(ERRCODE_UNDEFINED_OBJECT),
802  errmsg("table \"%s.%s\" not found on publisher",
803  nspname, relname)));
804 
805  lrel->remoteid = DatumGetObjectId(slot_getattr(slot, 1, &isnull));
806  Assert(!isnull);
807  lrel->replident = DatumGetChar(slot_getattr(slot, 2, &isnull));
808  Assert(!isnull);
809  lrel->relkind = DatumGetChar(slot_getattr(slot, 3, &isnull));
810  Assert(!isnull);
811 
814 
815 
816  /*
817  * Get column lists for each relation.
818  *
819  * We need to do this before fetching info about column names and types,
820  * so that we can skip columns that should not be replicated.
821  */
823  {
824  WalRcvExecResult *pubres;
825  TupleTableSlot *tslot;
826  Oid attrsRow[] = {INT2VECTOROID};
827  StringInfoData pub_names;
828 
829  initStringInfo(&pub_names);
830  foreach(lc, MySubscription->publications)
831  {
832  if (foreach_current_index(lc) > 0)
833  appendStringInfoString(&pub_names, ", ");
835  }
836 
837  /*
838  * Fetch info about column lists for the relation (from all the
839  * publications).
840  */
841  resetStringInfo(&cmd);
842  appendStringInfo(&cmd,
843  "SELECT DISTINCT"
844  " (CASE WHEN (array_length(gpt.attrs, 1) = c.relnatts)"
845  " THEN NULL ELSE gpt.attrs END)"
846  " FROM pg_publication p,"
847  " LATERAL pg_get_publication_tables(p.pubname) gpt,"
848  " pg_class c"
849  " WHERE gpt.relid = %u AND c.oid = gpt.relid"
850  " AND p.pubname IN ( %s )",
851  lrel->remoteid,
852  pub_names.data);
853 
855  lengthof(attrsRow), attrsRow);
856 
857  if (pubres->status != WALRCV_OK_TUPLES)
858  ereport(ERROR,
859  (errcode(ERRCODE_CONNECTION_FAILURE),
860  errmsg("could not fetch column list info for table \"%s.%s\" from publisher: %s",
861  nspname, relname, pubres->err)));
862 
863  /*
864  * We don't support the case where the column list is different for
865  * the same table when combining publications. See comments atop
866  * fetch_table_list. So there should be only one row returned.
867  * Although we already checked this when creating the subscription, we
868  * still need to check here in case the column list was changed after
869  * creating the subscription and before the sync worker is started.
870  */
871  if (tuplestore_tuple_count(pubres->tuplestore) > 1)
872  ereport(ERROR,
873  errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
874  errmsg("cannot use different column lists for table \"%s.%s\" in different publications",
875  nspname, relname));
876 
877  /*
878  * Get the column list and build a single bitmap with the attnums.
879  *
880  * If we find a NULL value, it means all the columns should be
881  * replicated.
882  */
884  if (tuplestore_gettupleslot(pubres->tuplestore, true, false, tslot))
885  {
886  Datum cfval = slot_getattr(tslot, 1, &isnull);
887 
888  if (!isnull)
889  {
890  ArrayType *arr;
891  int nelems;
892  int16 *elems;
893 
894  arr = DatumGetArrayTypeP(cfval);
895  nelems = ARR_DIMS(arr)[0];
896  elems = (int16 *) ARR_DATA_PTR(arr);
897 
898  for (natt = 0; natt < nelems; natt++)
899  included_cols = bms_add_member(included_cols, elems[natt]);
900  }
901 
902  ExecClearTuple(tslot);
903  }
905 
906  walrcv_clear_result(pubres);
907 
908  pfree(pub_names.data);
909  }
910 
911  /*
912  * Now fetch column names and types.
913  */
914  resetStringInfo(&cmd);
915  appendStringInfo(&cmd,
916  "SELECT a.attnum,"
917  " a.attname,"
918  " a.atttypid,"
919  " a.attnum = ANY(i.indkey)"
920  " FROM pg_catalog.pg_attribute a"
921  " LEFT JOIN pg_catalog.pg_index i"
922  " ON (i.indexrelid = pg_get_replica_identity_index(%u))"
923  " WHERE a.attnum > 0::pg_catalog.int2"
924  " AND NOT a.attisdropped %s"
925  " AND a.attrelid = %u"
926  " ORDER BY a.attnum",
927  lrel->remoteid,
929  "AND a.attgenerated = ''" : ""),
930  lrel->remoteid);
932  lengthof(attrRow), attrRow);
933 
934  if (res->status != WALRCV_OK_TUPLES)
935  ereport(ERROR,
936  (errcode(ERRCODE_CONNECTION_FAILURE),
937  errmsg("could not fetch table info for table \"%s.%s\" from publisher: %s",
938  nspname, relname, res->err)));
939 
940  /* We don't know the number of rows coming, so allocate enough space. */
941  lrel->attnames = palloc0(MaxTupleAttributeNumber * sizeof(char *));
942  lrel->atttyps = palloc0(MaxTupleAttributeNumber * sizeof(Oid));
943  lrel->attkeys = NULL;
944 
945  /*
946  * Store the columns as a list of names. Ignore those that are not
947  * present in the column list, if there is one.
948  */
949  natt = 0;
950  slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple);
951  while (tuplestore_gettupleslot(res->tuplestore, true, false, slot))
952  {
953  char *rel_colname;
955 
956  attnum = DatumGetInt16(slot_getattr(slot, 1, &isnull));
957  Assert(!isnull);
958 
959  /* If the column is not in the column list, skip it. */
960  if (included_cols != NULL && !bms_is_member(attnum, included_cols))
961  {
962  ExecClearTuple(slot);
963  continue;
964  }
965 
966  rel_colname = TextDatumGetCString(slot_getattr(slot, 2, &isnull));
967  Assert(!isnull);
968 
969  lrel->attnames[natt] = rel_colname;
970  lrel->atttyps[natt] = DatumGetObjectId(slot_getattr(slot, 3, &isnull));
971  Assert(!isnull);
972 
973  if (DatumGetBool(slot_getattr(slot, 4, &isnull)))
974  lrel->attkeys = bms_add_member(lrel->attkeys, natt);
975 
976  /* Should never happen. */
977  if (++natt >= MaxTupleAttributeNumber)
978  elog(ERROR, "too many columns in remote table \"%s.%s\"",
979  nspname, relname);
980 
981  ExecClearTuple(slot);
982  }
984 
985  lrel->natts = natt;
986 
988 
989  /*
990  * Get relation's row filter expressions. DISTINCT avoids the same
991  * expression of a table in multiple publications from being included
992  * multiple times in the final expression.
993  *
994  * We need to copy the row even if it matches just one of the
995  * publications, so we later combine all the quals with OR.
996  *
997  * For initial synchronization, row filtering can be ignored in following
998  * cases:
999  *
1000  * 1) one of the subscribed publications for the table hasn't specified
1001  * any row filter
1002  *
1003  * 2) one of the subscribed publications has puballtables set to true
1004  *
1005  * 3) one of the subscribed publications is declared as TABLES IN SCHEMA
1006  * that includes this relation
1007  */
1009  {
1010  StringInfoData pub_names;
1011 
1012  /* Build the pubname list. */
1013  initStringInfo(&pub_names);
1014  foreach(lc, MySubscription->publications)
1015  {
1016  char *pubname = strVal(lfirst(lc));
1017 
1018  if (foreach_current_index(lc) > 0)
1019  appendStringInfoString(&pub_names, ", ");
1020 
1021  appendStringInfoString(&pub_names, quote_literal_cstr(pubname));
1022  }
1023 
1024  /* Check for row filters. */
1025  resetStringInfo(&cmd);
1026  appendStringInfo(&cmd,
1027  "SELECT DISTINCT pg_get_expr(gpt.qual, gpt.relid)"
1028  " FROM pg_publication p,"
1029  " LATERAL pg_get_publication_tables(p.pubname) gpt"
1030  " WHERE gpt.relid = %u"
1031  " AND p.pubname IN ( %s )",
1032  lrel->remoteid,
1033  pub_names.data);
1034 
1035  res = walrcv_exec(LogRepWorkerWalRcvConn, cmd.data, 1, qualRow);
1036 
1037  if (res->status != WALRCV_OK_TUPLES)
1038  ereport(ERROR,
1039  (errmsg("could not fetch table WHERE clause info for table \"%s.%s\" from publisher: %s",
1040  nspname, relname, res->err)));
1041 
1042  /*
1043  * Multiple row filter expressions for the same table will be combined
1044  * by COPY using OR. If any of the filter expressions for this table
1045  * are null, it means the whole table will be copied. In this case it
1046  * is not necessary to construct a unified row filter expression at
1047  * all.
1048  */
1049  slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple);
1050  while (tuplestore_gettupleslot(res->tuplestore, true, false, slot))
1051  {
1052  Datum rf = slot_getattr(slot, 1, &isnull);
1053 
1054  if (!isnull)
1055  *qual = lappend(*qual, makeString(TextDatumGetCString(rf)));
1056  else
1057  {
1058  /* Ignore filters and cleanup as necessary. */
1059  if (*qual)
1060  {
1061  list_free_deep(*qual);
1062  *qual = NIL;
1063  }
1064  break;
1065  }
1066 
1067  ExecClearTuple(slot);
1068  }
1070 
1072  }
1073 
1074  pfree(cmd.data);
1075 }
#define ARR_DATA_PTR(a)
Definition: array.h:315
#define DatumGetArrayTypeP(X)
Definition: array.h:254
#define ARR_DIMS(a)
Definition: array.h:287
int16 AttrNumber
Definition: attnum.h:21
Subscription * MySubscription
Definition: worker.c:314
bool bms_is_member(int x, const Bitmapset *a)
Definition: bitmapset.c:444
Bitmapset * bms_add_member(Bitmapset *a, int x)
Definition: bitmapset.c:755
#define TextDatumGetCString(d)
Definition: builtins.h:95
signed short int16
Definition: c.h:477
#define lengthof(array)
Definition: c.h:772
void ExecDropSingleTupleTableSlot(TupleTableSlot *slot)
Definition: execTuples.c:1254
const TupleTableSlotOps TTSOpsMinimalTuple
Definition: execTuples.c:85
TupleTableSlot * MakeSingleTupleTableSlot(TupleDesc tupdesc, const TupleTableSlotOps *tts_ops)
Definition: execTuples.c:1238
#define MaxTupleAttributeNumber
Definition: htup_details.h:34
List * lappend(List *list, void *datum)
Definition: list.c:338
void * palloc0(Size size)
Definition: mcxt.c:1241
int16 attnum
Definition: pg_attribute.h:83
NameData relname
Definition: pg_class.h:38
#define foreach_current_index(cell)
Definition: pg_list.h:403
static bool DatumGetBool(Datum X)
Definition: postgres.h:90
uintptr_t Datum
Definition: postgres.h:64
static Oid DatumGetObjectId(Datum X)
Definition: postgres.h:242
static char DatumGetChar(Datum X)
Definition: postgres.h:112
static int16 DatumGetInt16(Datum X)
Definition: postgres.h:162
unsigned int Oid
Definition: postgres_ext.h:31
char * quote_literal_cstr(const char *rawstr)
Definition: quote.c:103
void resetStringInfo(StringInfo str)
Definition: stringinfo.c:75
Bitmapset * attkeys
Definition: logicalproto.h:115
Tuplestorestate * tuplestore
Definition: walreceiver.h:223
TupleDesc tupledesc
Definition: walreceiver.h:224
WalRcvExecStatus status
Definition: walreceiver.h:220
bool tuplestore_gettupleslot(Tuplestorestate *state, bool forward, bool copy, TupleTableSlot *slot)
Definition: tuplestore.c:1078
int64 tuplestore_tuple_count(Tuplestorestate *state)
Definition: tuplestore.c:546
static TupleTableSlot * ExecClearTuple(TupleTableSlot *slot)
Definition: tuptable.h:433
static Datum slot_getattr(TupleTableSlot *slot, int attnum, bool *isnull)
Definition: tuptable.h:389
String * makeString(char *str)
Definition: value.c:63
@ WALRCV_OK_TUPLES
Definition: walreceiver.h:207
#define walrcv_server_version(conn)
Definition: walreceiver.h:418

References appendStringInfo(), appendStringInfoString(), ARR_DATA_PTR, ARR_DIMS, Assert(), LogicalRepRelation::attkeys, LogicalRepRelation::attnames, attnum, LogicalRepRelation::atttyps, bms_add_member(), bms_is_member(), StringInfoData::data, DatumGetArrayTypeP, DatumGetBool(), DatumGetChar(), DatumGetInt16(), DatumGetObjectId(), elog(), ereport, WalRcvExecResult::err, errcode(), errmsg(), ERROR, ExecClearTuple(), ExecDropSingleTupleTableSlot(), foreach_current_index, initStringInfo(), lappend(), lengthof, lfirst, list_free_deep(), LogRepWorkerWalRcvConn, MakeSingleTupleTableSlot(), makeString(), MaxTupleAttributeNumber, MySubscription, LogicalRepRelation::natts, NIL, LogicalRepRelation::nspname, palloc0(), pfree(), Subscription::publications, quote_literal_cstr(), LogicalRepRelation::relkind, relname, LogicalRepRelation::relname, LogicalRepRelation::remoteid, LogicalRepRelation::replident, res, resetStringInfo(), slot_getattr(), WalRcvExecResult::status, strVal, TextDatumGetCString, TTSOpsMinimalTuple, WalRcvExecResult::tupledesc, WalRcvExecResult::tuplestore, tuplestore_gettupleslot(), tuplestore_tuple_count(), walrcv_clear_result(), walrcv_exec, WALRCV_OK_TUPLES, and walrcv_server_version.

Referenced by copy_table().

◆ FetchTableStates()

static bool FetchTableStates ( bool started_tx)
static

Definition at line 1497 of file tablesync.c.

1498 {
1499  static bool has_subrels = false;
1500 
1501  *started_tx = false;
1502 
1503  if (!table_states_valid)
1504  {
1505  MemoryContext oldctx;
1506  List *rstates;
1507  ListCell *lc;
1508  SubscriptionRelState *rstate;
1509 
1510  /* Clean the old lists. */
1513 
1514  if (!IsTransactionState())
1515  {
1517  *started_tx = true;
1518  }
1519 
1520  /* Fetch all non-ready tables. */
1521  rstates = GetSubscriptionRelations(MySubscription->oid, true);
1522 
1523  /* Allocate the tracking info in a permanent memory context. */
1525  foreach(lc, rstates)
1526  {
1527  rstate = palloc(sizeof(SubscriptionRelState));
1528  memcpy(rstate, lfirst(lc), sizeof(SubscriptionRelState));
1530  }
1531  MemoryContextSwitchTo(oldctx);
1532 
1533  /*
1534  * Does the subscription have tables?
1535  *
1536  * If there were not-READY relations found then we know it does. But
1537  * if table_state_not_ready was empty we still need to check again to
1538  * see if there are 0 tables.
1539  */
1540  has_subrels = (table_states_not_ready != NIL) ||
1542 
1543  table_states_valid = true;
1544  }
1545 
1546  return has_subrels;
1547 }
MemoryContext CacheMemoryContext
Definition: mcxt.c:144
void * palloc(Size size)
Definition: mcxt.c:1210
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
Definition: palloc.h:138
List * GetSubscriptionRelations(Oid subid, bool not_ready)
bool HasSubscriptionRelations(Oid subid)
static bool table_states_valid
Definition: tablesync.c:123
bool IsTransactionState(void)
Definition: xact.c:378
void StartTransactionCommand(void)
Definition: xact.c:2944

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.

272 {
273  table_states_valid = false;
274 }

References table_states_valid.

Referenced by ApplyWorkerMain(), and ParallelApplyWorkerMain().

◆ LogicalRepSyncTableStart()

char* LogicalRepSyncTableStart ( XLogRecPtr origin_startpos)

Definition at line 1229 of file tablesync.c.

1230 {
1231  char *slotname;
1232  char *err;
1233  char relstate;
1234  XLogRecPtr relstate_lsn;
1235  Relation rel;
1236  AclResult aclresult;
1238  char originname[NAMEDATALEN];
1239  RepOriginId originid;
1240 
1241  /* Check the state of the table synchronization. */
1245  &relstate_lsn);
1247 
1249  MyLogicalRepWorker->relstate = relstate;
1250  MyLogicalRepWorker->relstate_lsn = relstate_lsn;
1252 
1253  /*
1254  * If synchronization is already done or no longer necessary, exit now
1255  * that we've updated shared memory state.
1256  */
1257  switch (relstate)
1258  {
1259  case SUBREL_STATE_SYNCDONE:
1260  case SUBREL_STATE_READY:
1261  case SUBREL_STATE_UNKNOWN:
1262  finish_sync_worker(); /* doesn't return */
1263  }
1264 
1265  /* Calculate the name of the tablesync slot. */
1266  slotname = (char *) palloc(NAMEDATALEN);
1269  slotname,
1270  NAMEDATALEN);
1271 
1272  /*
1273  * Here we use the slot name instead of the subscription name as the
1274  * application_name, so that it is different from the leader apply worker,
1275  * so that synchronous replication can distinguish them.
1276  */
1278  walrcv_connect(MySubscription->conninfo, true, slotname, &err);
1279  if (LogRepWorkerWalRcvConn == NULL)
1280  ereport(ERROR,
1281  (errcode(ERRCODE_CONNECTION_FAILURE),
1282  errmsg("could not connect to the publisher: %s", err)));
1283 
1284  Assert(MyLogicalRepWorker->relstate == SUBREL_STATE_INIT ||
1285  MyLogicalRepWorker->relstate == SUBREL_STATE_DATASYNC ||
1286  MyLogicalRepWorker->relstate == SUBREL_STATE_FINISHEDCOPY);
1287 
1288  /* Assign the origin tracking record name. */
1291  originname,
1292  sizeof(originname));
1293 
1294  if (MyLogicalRepWorker->relstate == SUBREL_STATE_DATASYNC)
1295  {
1296  /*
1297  * We have previously errored out before finishing the copy so the
1298  * replication slot might exist. We want to remove the slot if it
1299  * already exists and proceed.
1300  *
1301  * XXX We could also instead try to drop the slot, last time we failed
1302  * but for that, we might need to clean up the copy state as it might
1303  * be in the middle of fetching the rows. Also, if there is a network
1304  * breakdown then it wouldn't have succeeded so trying it next time
1305  * seems like a better bet.
1306  */
1308  }
1309  else if (MyLogicalRepWorker->relstate == SUBREL_STATE_FINISHEDCOPY)
1310  {
1311  /*
1312  * The COPY phase was previously done, but tablesync then crashed
1313  * before it was able to finish normally.
1314  */
1316 
1317  /*
1318  * The origin tracking name must already exist. It was created first
1319  * time this tablesync was launched.
1320  */
1321  originid = replorigin_by_name(originname, false);
1322  replorigin_session_setup(originid, 0);
1323  replorigin_session_origin = originid;
1324  *origin_startpos = replorigin_session_get_progress(false);
1325 
1327 
1328  goto copy_table_done;
1329  }
1330 
1332  MyLogicalRepWorker->relstate = SUBREL_STATE_DATASYNC;
1335 
1336  /* Update the state and make it visible to others. */
1343  pgstat_report_stat(true);
1344 
1346 
1347  /*
1348  * Use a standard write lock here. It might be better to disallow access
1349  * to the table while it's being synchronized. But we don't want to block
1350  * the main apply process from working and it has to open the relation in
1351  * RowExclusiveLock when remapping remote relation id to local one.
1352  */
1354 
1355  /*
1356  * Check that our table sync worker has permission to insert into the
1357  * target table.
1358  */
1359  aclresult = pg_class_aclcheck(RelationGetRelid(rel), GetUserId(),
1360  ACL_INSERT);
1361  if (aclresult != ACLCHECK_OK)
1362  aclcheck_error(aclresult,
1363  get_relkind_objtype(rel->rd_rel->relkind),
1365 
1366  /*
1367  * COPY FROM does not honor RLS policies. That is not a problem for
1368  * subscriptions owned by roles with BYPASSRLS privilege (or superuser,
1369  * who has it implicitly), but other roles should not be able to
1370  * circumvent RLS. Disallow logical replication into RLS enabled
1371  * relations for such roles.
1372  */
1374  ereport(ERROR,
1375  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1376  errmsg("user \"%s\" cannot replicate into relation with row-level security enabled: \"%s\"",
1377  GetUserNameFromId(GetUserId(), true),
1378  RelationGetRelationName(rel))));
1379 
1380  /*
1381  * Start a transaction in the remote node in REPEATABLE READ mode. This
1382  * ensures that both the replication slot we create (see below) and the
1383  * COPY are consistent with each other.
1384  */
1386  "BEGIN READ ONLY ISOLATION LEVEL REPEATABLE READ",
1387  0, NULL);
1388  if (res->status != WALRCV_OK_COMMAND)
1389  ereport(ERROR,
1390  (errcode(ERRCODE_CONNECTION_FAILURE),
1391  errmsg("table copy could not start transaction on publisher: %s",
1392  res->err)));
1394 
1395  /*
1396  * Create a new permanent logical decoding slot. This slot will be used
1397  * for the catchup phase after COPY is done, so tell it to use the
1398  * snapshot to make the final data consistent.
1399  */
1401  slotname, false /* permanent */ , false /* two_phase */ ,
1402  CRS_USE_SNAPSHOT, origin_startpos);
1403 
1404  /*
1405  * Setup replication origin tracking. The purpose of doing this before the
1406  * copy is to avoid doing the copy again due to any error in setting up
1407  * origin tracking.
1408  */
1409  originid = replorigin_by_name(originname, true);
1410  if (!OidIsValid(originid))
1411  {
1412  /*
1413  * Origin tracking does not exist, so create it now.
1414  *
1415  * Then advance to the LSN got from walrcv_create_slot. This is WAL
1416  * logged for the purpose of recovery. Locks are to prevent the
1417  * replication origin from vanishing while advancing.
1418  */
1419  originid = replorigin_create(originname);
1420 
1421  LockRelationOid(ReplicationOriginRelationId, RowExclusiveLock);
1422  replorigin_advance(originid, *origin_startpos, InvalidXLogRecPtr,
1423  true /* go backward */ , true /* WAL log */ );
1424  UnlockRelationOid(ReplicationOriginRelationId, RowExclusiveLock);
1425 
1426  replorigin_session_setup(originid, 0);
1427  replorigin_session_origin = originid;
1428  }
1429  else
1430  {
1431  ereport(ERROR,
1433  errmsg("replication origin \"%s\" already exists",
1434  originname)));
1435  }
1436 
1437  /* Now do the initial data copy */
1439  copy_table(rel);
1441 
1442  res = walrcv_exec(LogRepWorkerWalRcvConn, "COMMIT", 0, NULL);
1443  if (res->status != WALRCV_OK_COMMAND)
1444  ereport(ERROR,
1445  (errcode(ERRCODE_CONNECTION_FAILURE),
1446  errmsg("table copy could not finish transaction on publisher: %s",
1447  res->err)));
1449 
1450  table_close(rel, NoLock);
1451 
1452  /* Make the copy visible. */
1454 
1455  /*
1456  * Update the persisted state to indicate the COPY phase is done; make it
1457  * visible to others.
1458  */
1461  SUBREL_STATE_FINISHEDCOPY,
1463 
1465 
1466 copy_table_done:
1467 
1468  elog(DEBUG1,
1469  "LogicalRepSyncTableStart: '%s' origin_startpos lsn %X/%X",
1470  originname, LSN_FORMAT_ARGS(*origin_startpos));
1471 
1472  /*
1473  * We are done with the initial data synchronization, update the state.
1474  */
1476  MyLogicalRepWorker->relstate = SUBREL_STATE_SYNCWAIT;
1477  MyLogicalRepWorker->relstate_lsn = *origin_startpos;
1479 
1480  /*
1481  * Finally, wait until the leader apply worker tells us to catch up and
1482  * then return to let LogicalRepApplyLoop do it.
1483  */
1484  wait_for_worker_state_change(SUBREL_STATE_CATCHUP);
1485  return slotname;
1486 }
AclResult
Definition: acl.h:182
@ ACLCHECK_OK
Definition: acl.h:183
void aclcheck_error(AclResult aclerr, ObjectType objtype, const char *objectname)
Definition: aclchk.c:2679
AclResult pg_class_aclcheck(Oid table_oid, Oid roleid, AclMode mode)
Definition: aclchk.c:3931
void ReplicationOriginNameForLogicalRep(Oid suboid, Oid relid, char *originname, Size szoriginname)
Definition: worker.c:457
#define OidIsValid(objectId)
Definition: c.h:759
#define DEBUG1
Definition: elog.h:30
void err(int eval, const char *fmt,...)
Definition: err.c:43
LogicalRepWorker * MyLogicalRepWorker
Definition: launcher.c:61
void UnlockRelationOid(Oid relid, LOCKMODE lockmode)
Definition: lmgr.c:228
void LockRelationOid(Oid relid, LOCKMODE lockmode)
Definition: lmgr.c:109
#define RowExclusiveLock
Definition: lockdefs.h:38
char * GetUserNameFromId(Oid roleid, bool noerr)
Definition: miscinit.c:984
Oid GetUserId(void)
Definition: miscinit.c:510
ObjectType get_relkind_objtype(char relkind)
RepOriginId replorigin_by_name(const char *roname, bool missing_ok)
Definition: origin.c:221
RepOriginId replorigin_create(const char *roname)
Definition: origin.c:252
RepOriginId replorigin_session_origin
Definition: origin.c:156
void replorigin_advance(RepOriginId node, XLogRecPtr remote_commit, XLogRecPtr local_commit, bool go_backward, bool wal_log)
Definition: origin.c:888
void replorigin_session_setup(RepOriginId node, int acquired_by)
Definition: origin.c:1095
XLogRecPtr replorigin_session_get_progress(bool flush)
Definition: origin.c:1234
#define ACL_INSERT
Definition: parsenodes.h:83
#define NAMEDATALEN
char GetSubscriptionRelState(Oid subid, Oid relid, XLogRecPtr *sublsn)
void UpdateSubscriptionRelState(Oid subid, Oid relid, char state, XLogRecPtr sublsn)
#define InvalidOid
Definition: postgres_ext.h:36
#define RelationGetRelid(relation)
Definition: rel.h:503
int check_enable_rls(Oid relid, Oid checkAsUser, bool noError)
Definition: rls.c:52
@ RLS_ENABLED
Definition: rls.h:45
Snapshot GetTransactionSnapshot(void)
Definition: snapmgr.c:251
void PushActiveSnapshot(Snapshot snapshot)
Definition: snapmgr.c:683
void PopActiveSnapshot(void)
Definition: snapmgr.c:778
#define SpinLockRelease(lock)
Definition: spin.h:64
#define SpinLockAcquire(lock)
Definition: spin.h:62
#define ERRCODE_DUPLICATE_OBJECT
Definition: streamutil.c:32
XLogRecPtr relstate_lsn
Form_pg_class rd_rel
Definition: rel.h:110
void ReplicationSlotDropAtPubNode(WalReceiverConn *wrconn, char *slotname, bool missing_ok)
void table_close(Relation relation, LOCKMODE lockmode)
Definition: table.c:126
Relation table_open(Oid relationId, LOCKMODE lockmode)
Definition: table.c:40
static bool wait_for_worker_state_change(char expected_state)
Definition: tablesync.c:222
void ReplicationSlotNameForTablesync(Oid suboid, Oid relid, char *syncslotname, Size szslot)
Definition: tablesync.c:1213
static void copy_table(Relation rel)
Definition: tablesync.c:1083
#define walrcv_create_slot(conn, slotname, temporary, two_phase, snapshot_action, lsn)
Definition: walreceiver.h:430
#define walrcv_connect(conninfo, logical, appname, err)
Definition: walreceiver.h:408
@ WALRCV_OK_COMMAND
Definition: walreceiver.h:205
@ CRS_USE_SNAPSHOT
Definition: walsender.h:24
void CommandCounterIncrement(void)
Definition: xact.c:1078
#define LSN_FORMAT_ARGS(lsn)
Definition: xlogdefs.h:43
uint16 RepOriginId
Definition: xlogdefs.h:65
uint64 XLogRecPtr
Definition: xlogdefs.h:21
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28

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.

667 {
668  List *attnamelist = NIL;
669  int i;
670 
671  for (i = 0; i < rel->remoterel.natts; i++)
672  {
673  attnamelist = lappend(attnamelist,
674  makeString(rel->remoterel.attnames[i]));
675  }
676 
677 
678  return attnamelist;
679 }
LogicalRepRelation remoterel

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.

135 {
136  /*
137  * Commit any outstanding transaction. This is the usual case, unless
138  * there was nothing to do for the table.
139  */
140  if (IsTransactionState())
141  {
143  pgstat_report_stat(true);
144  }
145 
146  /* And flush all writes. */
148 
150  ereport(LOG,
151  (errmsg("logical replication table synchronization worker for subscription \"%s\", table \"%s\" has finished",
155 
156  /* Find the leader apply worker and signal it. */
158 
159  /* Stop gracefully */
160  proc_exit(0);
161 }
#define LOG
Definition: elog.h:31
void proc_exit(int code)
Definition: ipc.c:104
void logicalrep_worker_wakeup(Oid subid, Oid relid)
Definition: launcher.c:643
char * get_rel_name(Oid relid)
Definition: lsyscache.c:1910
XLogRecPtr GetXLogWriteRecPtr(void)
Definition: xlog.c:8873
void XLogFlush(XLogRecPtr record)
Definition: xlog.c:2514

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.

647 {
648  /*
649  * Skip for parallel apply workers because they only operate on tables
650  * that are in a READY state. See pa_can_start() and
651  * should_apply_changes_for_rel().
652  */
654  return;
655 
656  if (am_tablesync_worker())
657  process_syncing_tables_for_sync(current_lsn);
658  else
660 }
static void process_syncing_tables_for_apply(XLogRecPtr current_lsn)
Definition: tablesync.c:408
static void process_syncing_tables_for_sync(XLogRecPtr current_lsn)
Definition: tablesync.c:285
static bool am_parallel_apply_worker(void)
static bool am_tablesync_worker(void)

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.

409 {
410  struct tablesync_start_time_mapping
411  {
412  Oid relid;
413  TimestampTz last_start_time;
414  };
415  static HTAB *last_start_times = NULL;
416  ListCell *lc;
417  bool started_tx = false;
418  bool should_exit = false;
419 
421 
422  /* We need up-to-date sync state info for subscription tables here. */
423  FetchTableStates(&started_tx);
424 
425  /*
426  * Prepare a hash table for tracking last start times of workers, to avoid
427  * immediate restarts. We don't need it if there are no tables that need
428  * syncing.
429  */
431  {
432  HASHCTL ctl;
433 
434  ctl.keysize = sizeof(Oid);
435  ctl.entrysize = sizeof(struct tablesync_start_time_mapping);
436  last_start_times = hash_create("Logical replication table sync worker start times",
437  256, &ctl, HASH_ELEM | HASH_BLOBS);
438  }
439 
440  /*
441  * Clean up the hash table when we're done with all tables (just to
442  * release the bit of memory).
443  */
445  {
447  last_start_times = NULL;
448  }
449 
450  /*
451  * Process all tables that are being synchronized.
452  */
453  foreach(lc, table_states_not_ready)
454  {
456 
457  if (rstate->state == SUBREL_STATE_SYNCDONE)
458  {
459  /*
460  * Apply has caught up to the position where the table sync has
461  * finished. Mark the table as ready so that the apply will just
462  * continue to replicate it normally.
463  */
464  if (current_lsn >= rstate->lsn)
465  {
466  char originname[NAMEDATALEN];
467 
468  rstate->state = SUBREL_STATE_READY;
469  rstate->lsn = current_lsn;
470  if (!started_tx)
471  {
473  started_tx = true;
474  }
475 
476  /*
477  * Remove the tablesync origin tracking if exists.
478  *
479  * There is a chance that the user is concurrently performing
480  * refresh for the subscription where we remove the table
481  * state and its origin or the tablesync worker would have
482  * already removed this origin. We can't rely on tablesync
483  * worker to remove the origin tracking as if there is any
484  * error while dropping we won't restart it to drop the
485  * origin. So passing missing_ok = true.
486  */
488  rstate->relid,
489  originname,
490  sizeof(originname));
491  replorigin_drop_by_name(originname, true, false);
492 
493  /*
494  * Update the state to READY only after the origin cleanup.
495  */
497  rstate->relid, rstate->state,
498  rstate->lsn);
499  }
500  }
501  else
502  {
503  LogicalRepWorker *syncworker;
504 
505  /*
506  * Look for a sync worker for this relation.
507  */
508  LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
509 
511  rstate->relid, false);
512 
513  if (syncworker)
514  {
515  /* Found one, update our copy of its state */
516  SpinLockAcquire(&syncworker->relmutex);
517  rstate->state = syncworker->relstate;
518  rstate->lsn = syncworker->relstate_lsn;
519  if (rstate->state == SUBREL_STATE_SYNCWAIT)
520  {
521  /*
522  * Sync worker is waiting for apply. Tell sync worker it
523  * can catchup now.
524  */
525  syncworker->relstate = SUBREL_STATE_CATCHUP;
526  syncworker->relstate_lsn =
527  Max(syncworker->relstate_lsn, current_lsn);
528  }
529  SpinLockRelease(&syncworker->relmutex);
530 
531  /* If we told worker to catch up, wait for it. */
532  if (rstate->state == SUBREL_STATE_SYNCWAIT)
533  {
534  /* Signal the sync worker, as it may be waiting for us. */
535  if (syncworker->proc)
536  logicalrep_worker_wakeup_ptr(syncworker);
537 
538  /* Now safe to release the LWLock */
539  LWLockRelease(LogicalRepWorkerLock);
540 
541  /*
542  * Enter busy loop and wait for synchronization worker to
543  * reach expected state (or die trying).
544  */
545  if (!started_tx)
546  {
548  started_tx = true;
549  }
550 
552  SUBREL_STATE_SYNCDONE);
553  }
554  else
555  LWLockRelease(LogicalRepWorkerLock);
556  }
557  else
558  {
559  /*
560  * If there is no sync worker for this table yet, count
561  * running sync workers for this subscription, while we have
562  * the lock.
563  */
564  int nsyncworkers =
566 
567  /* Now safe to release the LWLock */
568  LWLockRelease(LogicalRepWorkerLock);
569 
570  /*
571  * If there are free sync worker slot(s), start a new sync
572  * worker for the table.
573  */
574  if (nsyncworkers < max_sync_workers_per_subscription)
575  {
577  struct tablesync_start_time_mapping *hentry;
578  bool found;
579 
580  hentry = hash_search(last_start_times, &rstate->relid,
581  HASH_ENTER, &found);
582 
583  if (!found ||
584  TimestampDifferenceExceeds(hentry->last_start_time, now,
586  {
591  rstate->relid,
593  hentry->last_start_time = now;
594  }
595  }
596  }
597  }
598  }
599 
600  if (started_tx)
601  {
602  /*
603  * Even when the two_phase mode is requested by the user, it remains
604  * as 'pending' until all tablesyncs have reached READY state.
605  *
606  * When this happens, we restart the apply worker and (if the
607  * conditions are still ok) then the two_phase tri-state will become
608  * 'enabled' at that time.
609  *
610  * Note: If the subscription has no tables then leave the state as
611  * PENDING, which allows ALTER SUBSCRIPTION ... REFRESH PUBLICATION to
612  * work.
613  */
615  {
616  CommandCounterIncrement(); /* make updates visible */
617  if (AllTablesyncsReady())
618  {
619  ereport(LOG,
620  (errmsg("logical replication apply worker for subscription \"%s\" will restart so that two_phase can be enabled",
621  MySubscription->name)));
622  should_exit = true;
623  }
624  }
625 
627  pgstat_report_stat(true);
628  }
629 
630  if (should_exit)
631  {
632  /*
633  * Reset the last-start time for this worker so that the launcher will
634  * restart it without waiting for wal_retrieve_retry_interval.
635  */
637 
638  proc_exit(0);
639  }
640 }
bool TimestampDifferenceExceeds(TimestampTz start_time, TimestampTz stop_time, int msec)
Definition: timestamp.c:1727
TimestampTz GetCurrentTimestamp(void)
Definition: timestamp.c:1582
Datum now(PG_FUNCTION_ARGS)
Definition: timestamp.c:1546
#define Max(x, y)
Definition: c.h:982
int64 TimestampTz
Definition: timestamp.h:39
#define DSM_HANDLE_INVALID
Definition: dsm_impl.h:58
void hash_destroy(HTAB *hashp)
Definition: dynahash.c:863
void * hash_search(HTAB *hashp, const void *keyPtr, HASHACTION action, bool *foundPtr)
Definition: dynahash.c:953
HTAB * hash_create(const char *tabname, long nelem, const HASHCTL *info, int flags)
Definition: dynahash.c:350
@ HASH_ENTER
Definition: hsearch.h:114
#define HASH_ELEM
Definition: hsearch.h:95
#define HASH_BLOBS
Definition: hsearch.h:97
LogicalRepWorker * logicalrep_worker_find(Oid subid, Oid relid, bool only_running)
Definition: launcher.c:249
void logicalrep_worker_wakeup_ptr(LogicalRepWorker *worker)
Definition: launcher.c:663
bool logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname, Oid userid, Oid relid, dsm_handle subworker_dsm)
Definition: launcher.c:306
static dshash_table * last_start_times
Definition: launcher.c:95
int max_sync_workers_per_subscription
Definition: launcher.c:58
int logicalrep_sync_worker_count(Oid subid)
Definition: launcher.c:811
void ApplyLauncherForgetWorkerStartTime(Oid subid)
Definition: launcher.c:1031
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1195
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1803
@ LW_SHARED
Definition: lwlock.h:116
void replorigin_drop_by_name(const char *name, bool missing_ok, bool nowait)
Definition: origin.c:411
#define LOGICALREP_TWOPHASE_STATE_PENDING
Size keysize
Definition: hsearch.h:75
Size entrysize
Definition: hsearch.h:76
Definition: dynahash.c:220
bool AllTablesyncsReady(void)
Definition: tablesync.c:1558
static bool wait_for_relation_state_change(Oid relid, char expected_state)
Definition: tablesync.c:174
int wal_retrieve_retry_interval
Definition: xlog.c:137

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.

286 {
288 
289  if (MyLogicalRepWorker->relstate == SUBREL_STATE_CATCHUP &&
290  current_lsn >= MyLogicalRepWorker->relstate_lsn)
291  {
292  TimeLineID tli;
293  char syncslotname[NAMEDATALEN] = {0};
294  char originname[NAMEDATALEN] = {0};
295 
296  MyLogicalRepWorker->relstate = SUBREL_STATE_SYNCDONE;
297  MyLogicalRepWorker->relstate_lsn = current_lsn;
298 
300 
301  /*
302  * UpdateSubscriptionRelState must be called within a transaction.
303  */
304  if (!IsTransactionState())
306 
311 
312  /*
313  * End streaming so that LogRepWorkerWalRcvConn can be used to drop
314  * the slot.
315  */
317 
318  /*
319  * Cleanup the tablesync slot.
320  *
321  * This has to be done after updating the state because otherwise if
322  * there is an error while doing the database operations we won't be
323  * able to rollback dropped slot.
324  */
327  syncslotname,
328  sizeof(syncslotname));
329 
330  /*
331  * It is important to give an error if we are unable to drop the slot,
332  * otherwise, it won't be dropped till the corresponding subscription
333  * is dropped. So passing missing_ok = false.
334  */
336 
338  pgstat_report_stat(false);
339 
340  /*
341  * Start a new transaction to clean up the tablesync origin tracking.
342  * This transaction will be ended within the finish_sync_worker().
343  * Now, even, if we fail to remove this here, the apply worker will
344  * ensure to clean it up afterward.
345  *
346  * We need to do this after the table state is set to SYNCDONE.
347  * Otherwise, if an error occurs while performing the database
348  * operation, the worker will be restarted and the in-memory state of
349  * replication progress (remote_lsn) won't be rolled-back which would
350  * have been cleared before restart. So, the restarted worker will use
351  * invalid replication progress state resulting in replay of
352  * transactions that have already been applied.
353  */
355 
358  originname,
359  sizeof(originname));
360 
361  /*
362  * Resetting the origin session removes the ownership of the slot.
363  * This is needed to allow the origin to be dropped.
364  */
369 
370  /*
371  * Drop the tablesync's origin tracking if exists.
372  *
373  * There is a chance that the user is concurrently performing refresh
374  * for the subscription where we remove the table state and its origin
375  * or the apply worker would have removed this origin. So passing
376  * missing_ok = true.
377  */
378  replorigin_drop_by_name(originname, true, false);
379 
380  finish_sync_worker();
381  }
382  else
384 }
TimestampTz replorigin_session_origin_timestamp
Definition: origin.c:158
void replorigin_session_reset(void)
Definition: origin.c:1187
XLogRecPtr replorigin_session_origin_lsn
Definition: origin.c:157
#define InvalidRepOriginId
Definition: origin.h:33
#define walrcv_endstreaming(conn, next_tli)
Definition: walreceiver.h:424
uint32 TimeLineID
Definition: xlogdefs.h:59

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.

1215 {
1216  snprintf(syncslotname, szslot, "pg_%u_sync_%u_" UINT64_FORMAT, suboid,
1217  relid, GetSystemIdentifier());
1218 }
#define UINT64_FORMAT
Definition: c.h:533
#define snprintf
Definition: port.h:238
uint64 GetSystemIdentifier(void)
Definition: xlog.c:4177

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.

1584 {
1585  Relation rel;
1586  HeapTuple tup;
1587  bool nulls[Natts_pg_subscription];
1588  bool replaces[Natts_pg_subscription];
1589  Datum values[Natts_pg_subscription];
1590 
1592  new_state == LOGICALREP_TWOPHASE_STATE_PENDING ||
1593  new_state == LOGICALREP_TWOPHASE_STATE_ENABLED);
1594 
1595  rel = table_open(SubscriptionRelationId, RowExclusiveLock);
1597  if (!HeapTupleIsValid(tup))
1598  elog(ERROR,
1599  "cache lookup failed for subscription oid %u",
1600  suboid);
1601 
1602  /* Form a new tuple. */
1603  memset(values, 0, sizeof(values));
1604  memset(nulls, false, sizeof(nulls));
1605  memset(replaces, false, sizeof(replaces));
1606 
1607  /* And update/set two_phase state */
1608  values[Anum_pg_subscription_subtwophasestate - 1] = CharGetDatum(new_state);
1609  replaces[Anum_pg_subscription_subtwophasestate - 1] = true;
1610 
1611  tup = heap_modify_tuple(tup, RelationGetDescr(rel),
1612  values, nulls, replaces);
1613  CatalogTupleUpdate(rel, &tup->t_self, tup);
1614 
1615  heap_freetuple(tup);
1617 }
static Datum values[MAXATTR]
Definition: bootstrap.c:156
HeapTuple heap_modify_tuple(HeapTuple tuple, TupleDesc tupleDesc, Datum *replValues, bool *replIsnull, bool *doReplace)
Definition: heaptuple.c:1113
void heap_freetuple(HeapTuple htup)
Definition: heaptuple.c:1338
#define HeapTupleIsValid(tuple)
Definition: htup.h:78
void CatalogTupleUpdate(Relation heapRel, ItemPointer otid, HeapTuple tup)
Definition: indexing.c:313
#define LOGICALREP_TWOPHASE_STATE_DISABLED
#define LOGICALREP_TWOPHASE_STATE_ENABLED
static Datum ObjectIdGetDatum(Oid X)
Definition: postgres.h:252
static Datum CharGetDatum(char X)
Definition: postgres.h:122
#define RelationGetDescr(relation)
Definition: rel.h:529
ItemPointerData t_self
Definition: htup.h:65
#define SearchSysCacheCopy1(cacheId, key1)
Definition: syscache.h:179
@ SUBSCRIPTIONOID
Definition: syscache.h:99

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.

175 {
176  char state;
177 
178  for (;;)
179  {
180  LogicalRepWorker *worker;
181  XLogRecPtr statelsn;
182 
184 
187  relid, &statelsn);
188 
189  if (state == SUBREL_STATE_UNKNOWN)
190  break;
191 
192  if (state == expected_state)
193  return true;
194 
195  /* Check if the sync worker is still running and bail if not. */
196  LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
198  false);
199  LWLockRelease(LogicalRepWorkerLock);
200  if (!worker)
201  break;
202 
203  (void) WaitLatch(MyLatch,
206 
208  }
209 
210  return false;
211 }
int WaitLatch(Latch *latch, int wakeEvents, long timeout, uint32 wait_event_info)
Definition: latch.c:492
void InvalidateCatalogSnapshot(void)
Definition: snapmgr.c:457
Definition: regguts.h:318
@ WAIT_EVENT_LOGICAL_SYNC_STATE_CHANGE
Definition: wait_event.h:112

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.

223 {
224  int rc;
225 
226  for (;;)
227  {
228  LogicalRepWorker *worker;
229 
231 
232  /*
233  * Done if already in correct state. (We assume this fetch is atomic
234  * enough to not give a misleading answer if we do it with no lock.)
235  */
236  if (MyLogicalRepWorker->relstate == expected_state)
237  return true;
238 
239  /*
240  * Bail out if the apply worker has died, else signal it we're
241  * waiting.
242  */
243  LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
245  InvalidOid, false);
246  if (worker && worker->proc)
248  LWLockRelease(LogicalRepWorkerLock);
249  if (!worker)
250  break;
251 
252  /*
253  * Wait. We expect to get a latch signal back from the apply worker,
254  * but use a timeout in case it dies without sending one.
255  */
256  rc = WaitLatch(MyLatch,
259 
260  if (rc & WL_LATCH_SET)
262  }
263 
264  return false;
265 }

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

◆ table_states_not_ready

List* table_states_not_ready = NIL
static

◆ 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().