PostgreSQL Source Code  git master
worker.c File Reference
#include "postgres.h"
#include <sys/stat.h>
#include <unistd.h>
#include "access/genam.h"
#include "access/table.h"
#include "access/tableam.h"
#include "access/twophase.h"
#include "access/xact.h"
#include "access/xlog_internal.h"
#include "catalog/catalog.h"
#include "catalog/indexing.h"
#include "catalog/namespace.h"
#include "catalog/partition.h"
#include "catalog/pg_inherits.h"
#include "catalog/pg_subscription.h"
#include "catalog/pg_subscription_rel.h"
#include "catalog/pg_tablespace.h"
#include "commands/tablecmds.h"
#include "commands/tablespace.h"
#include "commands/trigger.h"
#include "executor/executor.h"
#include "executor/execPartition.h"
#include "executor/nodeModifyTable.h"
#include "funcapi.h"
#include "libpq/pqformat.h"
#include "libpq/pqsignal.h"
#include "mb/pg_wchar.h"
#include "miscadmin.h"
#include "nodes/makefuncs.h"
#include "optimizer/optimizer.h"
#include "parser/parse_relation.h"
#include "pgstat.h"
#include "postmaster/bgworker.h"
#include "postmaster/interrupt.h"
#include "postmaster/postmaster.h"
#include "postmaster/walwriter.h"
#include "replication/decode.h"
#include "replication/logical.h"
#include "replication/logicallauncher.h"
#include "replication/logicalproto.h"
#include "replication/logicalrelation.h"
#include "replication/logicalworker.h"
#include "replication/origin.h"
#include "replication/reorderbuffer.h"
#include "replication/snapbuild.h"
#include "replication/walreceiver.h"
#include "replication/worker_internal.h"
#include "rewrite/rewriteHandler.h"
#include "storage/buffile.h"
#include "storage/bufmgr.h"
#include "storage/fd.h"
#include "storage/ipc.h"
#include "storage/lmgr.h"
#include "storage/proc.h"
#include "storage/procarray.h"
#include "tcop/tcopprot.h"
#include "utils/acl.h"
#include "utils/builtins.h"
#include "utils/catcache.h"
#include "utils/dynahash.h"
#include "utils/datum.h"
#include "utils/fmgroids.h"
#include "utils/guc.h"
#include "utils/inval.h"
#include "utils/lsyscache.h"
#include "utils/memutils.h"
#include "utils/pg_lsn.h"
#include "utils/rel.h"
#include "utils/rls.h"
#include "utils/syscache.h"
#include "utils/timeout.h"
#include "utils/usercontext.h"
Include dependency graph for worker.c:

Go to the source code of this file.

Data Structures

struct  FlushPosition
 
struct  ApplyExecutionData
 
struct  ApplyErrorCallbackArg
 
struct  SubXactInfo
 
struct  ApplySubXactData
 

Macros

#define NAPTIME_PER_CYCLE   1000 /* max sleep time between cycles (1s) */
 
#define is_skipping_changes()   (unlikely(!XLogRecPtrIsInvalid(skip_xact_finish_lsn)))
 

Typedefs

typedef struct FlushPosition FlushPosition
 
typedef struct ApplyExecutionData ApplyExecutionData
 
typedef struct ApplyErrorCallbackArg ApplyErrorCallbackArg
 
typedef struct SubXactInfo SubXactInfo
 
typedef struct ApplySubXactData ApplySubXactData
 

Enumerations

enum  TransApplyAction {
  TRANS_LEADER_APPLY , TRANS_LEADER_SERIALIZE , TRANS_LEADER_SEND_TO_PARALLEL , TRANS_LEADER_PARTIAL_SERIALIZE ,
  TRANS_PARALLEL_APPLY
}
 

Functions

static void subxact_filename (char *path, Oid subid, TransactionId xid)
 
static void changes_filename (char *path, Oid subid, TransactionId xid)
 
static void subxact_info_write (Oid subid, TransactionId xid)
 
static void subxact_info_read (Oid subid, TransactionId xid)
 
static void subxact_info_add (TransactionId xid)
 
static void cleanup_subxact_info (void)
 
static void stream_open_file (Oid subid, TransactionId xid, bool first_segment)
 
static void stream_write_change (char action, StringInfo s)
 
static void stream_open_and_write_change (TransactionId xid, char action, StringInfo s)
 
static void stream_close_file (void)
 
static void send_feedback (XLogRecPtr recvpos, bool force, bool requestReply)
 
static void apply_handle_commit_internal (LogicalRepCommitData *commit_data)
 
static void apply_handle_insert_internal (ApplyExecutionData *edata, ResultRelInfo *relinfo, TupleTableSlot *remoteslot)
 
static void apply_handle_update_internal (ApplyExecutionData *edata, ResultRelInfo *relinfo, TupleTableSlot *remoteslot, LogicalRepTupleData *newtup, Oid localindexoid)
 
static void apply_handle_delete_internal (ApplyExecutionData *edata, ResultRelInfo *relinfo, TupleTableSlot *remoteslot, Oid localindexoid)
 
static bool FindReplTupleInLocalRel (ApplyExecutionData *edata, Relation localrel, LogicalRepRelation *remoterel, Oid localidxoid, TupleTableSlot *remoteslot, TupleTableSlot **localslot)
 
static void apply_handle_tuple_routing (ApplyExecutionData *edata, TupleTableSlot *remoteslot, LogicalRepTupleData *newtup, CmdType operation)
 
static void TwoPhaseTransactionGid (Oid subid, TransactionId xid, char *gid, int szgid)
 
static void maybe_start_skipping_changes (XLogRecPtr finish_lsn)
 
static void stop_skipping_changes (void)
 
static void clear_subscription_skip_lsn (XLogRecPtr finish_lsn)
 
static void set_apply_error_context_xact (TransactionId xid, XLogRecPtr lsn)
 
static void reset_apply_error_context_info (void)
 
static TransApplyAction get_transaction_apply_action (TransactionId xid, ParallelApplyWorkerInfo **winfo)
 
void ReplicationOriginNameForLogicalRep (Oid suboid, Oid relid, char *originname, Size szoriginname)
 
static bool should_apply_changes_for_rel (LogicalRepRelMapEntry *rel)
 
static void begin_replication_step (void)
 
static void end_replication_step (void)
 
static bool handle_streamed_transaction (LogicalRepMsgType action, StringInfo s)
 
static ApplyExecutionDatacreate_edata_for_relation (LogicalRepRelMapEntry *rel)
 
static void finish_edata (ApplyExecutionData *edata)
 
static void slot_fill_defaults (LogicalRepRelMapEntry *rel, EState *estate, TupleTableSlot *slot)
 
static void slot_store_data (TupleTableSlot *slot, LogicalRepRelMapEntry *rel, LogicalRepTupleData *tupleData)
 
static void slot_modify_data (TupleTableSlot *slot, TupleTableSlot *srcslot, LogicalRepRelMapEntry *rel, LogicalRepTupleData *tupleData)
 
static void apply_handle_begin (StringInfo s)
 
static void apply_handle_commit (StringInfo s)
 
static void apply_handle_begin_prepare (StringInfo s)
 
static void apply_handle_prepare_internal (LogicalRepPreparedTxnData *prepare_data)
 
static void apply_handle_prepare (StringInfo s)
 
static void apply_handle_commit_prepared (StringInfo s)
 
static void apply_handle_rollback_prepared (StringInfo s)
 
static void apply_handle_stream_prepare (StringInfo s)
 
static void apply_handle_origin (StringInfo s)
 
void stream_start_internal (TransactionId xid, bool first_segment)
 
static void apply_handle_stream_start (StringInfo s)
 
void stream_stop_internal (TransactionId xid)
 
static void apply_handle_stream_stop (StringInfo s)
 
static void stream_abort_internal (TransactionId xid, TransactionId subxid)
 
static void apply_handle_stream_abort (StringInfo s)
 
static void ensure_last_message (FileSet *stream_fileset, TransactionId xid, int fileno, off_t offset)
 
void apply_spooled_messages (FileSet *stream_fileset, TransactionId xid, XLogRecPtr lsn)
 
static void apply_handle_stream_commit (StringInfo s)
 
static void apply_handle_relation (StringInfo s)
 
static void apply_handle_type (StringInfo s)
 
static void TargetPrivilegesCheck (Relation rel, AclMode mode)
 
static void apply_handle_insert (StringInfo s)
 
static void check_relation_updatable (LogicalRepRelMapEntry *rel)
 
static void apply_handle_update (StringInfo s)
 
static void apply_handle_delete (StringInfo s)
 
static void apply_handle_truncate (StringInfo s)
 
void apply_dispatch (StringInfo s)
 
static void get_flush_position (XLogRecPtr *write, XLogRecPtr *flush, bool *have_pending_txes)
 
void store_flush_position (XLogRecPtr remote_lsn, XLogRecPtr local_lsn)
 
static void UpdateWorkerStats (XLogRecPtr last_lsn, TimestampTz send_time, bool reply)
 
static void LogicalRepApplyLoop (XLogRecPtr last_received)
 
static void apply_worker_exit (void)
 
void maybe_reread_subscription (void)
 
static void subscription_change_cb (Datum arg, int cacheid, uint32 hashvalue)
 
void stream_cleanup_files (Oid subid, TransactionId xid)
 
void set_stream_options (WalRcvStreamOptions *options, char *slotname, XLogRecPtr *origin_startpos)
 
void start_apply (XLogRecPtr origin_startpos)
 
static void run_apply_worker ()
 
void InitializeLogRepWorker (void)
 
void SetupApplyOrSyncWorker (int worker_slot)
 
void ApplyWorkerMain (Datum main_arg)
 
void DisableSubscriptionAndExit (void)
 
bool IsLogicalWorker (void)
 
bool IsLogicalParallelApplyWorker (void)
 
void apply_error_callback (void *arg)
 
void LogicalRepWorkersWakeupAtCommit (Oid subid)
 
void AtEOXact_LogicalRepWorkers (bool isCommit)
 
void set_apply_error_context_origin (char *originname)
 

Variables

static dlist_head lsn_mapping = DLIST_STATIC_INIT(lsn_mapping)
 
ApplyErrorCallbackArg apply_error_callback_arg
 
ErrorContextCallbackapply_error_context_stack = NULL
 
MemoryContext ApplyMessageContext = NULL
 
MemoryContext ApplyContext = NULL
 
static MemoryContext LogicalStreamingContext = NULL
 
WalReceiverConnLogRepWorkerWalRcvConn = NULL
 
SubscriptionMySubscription = NULL
 
static bool MySubscriptionValid = false
 
static Liston_commit_wakeup_workers_subids = NIL
 
bool in_remote_transaction = false
 
static XLogRecPtr remote_final_lsn = InvalidXLogRecPtr
 
static bool in_streamed_transaction = false
 
static TransactionId stream_xid = InvalidTransactionId
 
static uint32 parallel_stream_nchanges = 0
 
bool InitializingApplyWorker = false
 
static XLogRecPtr skip_xact_finish_lsn = InvalidXLogRecPtr
 
static BufFilestream_fd = NULL
 
static ApplySubXactData subxact_data = {0, 0, InvalidTransactionId, NULL}
 

Macro Definition Documentation

◆ is_skipping_changes

#define is_skipping_changes ( )    (unlikely(!XLogRecPtrIsInvalid(skip_xact_finish_lsn)))

Definition at line 354 of file worker.c.

◆ NAPTIME_PER_CYCLE

#define NAPTIME_PER_CYCLE   1000 /* max sleep time between cycles (1s) */

Definition at line 213 of file worker.c.

Typedef Documentation

◆ ApplyErrorCallbackArg

◆ ApplyExecutionData

◆ ApplySubXactData

◆ FlushPosition

typedef struct FlushPosition FlushPosition

◆ SubXactInfo

typedef struct SubXactInfo SubXactInfo

Enumeration Type Documentation

◆ TransApplyAction

Enumerator
TRANS_LEADER_APPLY 
TRANS_LEADER_SERIALIZE 
TRANS_LEADER_SEND_TO_PARALLEL 
TRANS_LEADER_PARTIAL_SERIALIZE 
TRANS_PARALLEL_APPLY 

Definition at line 283 of file worker.c.

284 {
285  /* The action for non-streaming transactions. */
287 
288  /* Actions for streaming transactions. */
TransApplyAction
Definition: worker.c:284
@ TRANS_LEADER_SERIALIZE
Definition: worker.c:289
@ TRANS_PARALLEL_APPLY
Definition: worker.c:292
@ TRANS_LEADER_SEND_TO_PARALLEL
Definition: worker.c:290
@ TRANS_LEADER_APPLY
Definition: worker.c:286
@ TRANS_LEADER_PARTIAL_SERIALIZE
Definition: worker.c:291

Function Documentation

◆ apply_dispatch()

void apply_dispatch ( StringInfo  s)

Definition at line 3292 of file worker.c.

3293 {
3295  LogicalRepMsgType saved_command;
3296 
3297  /*
3298  * Set the current command being applied. Since this function can be
3299  * called recursively when applying spooled changes, save the current
3300  * command.
3301  */
3302  saved_command = apply_error_callback_arg.command;
3304 
3305  switch (action)
3306  {
3307  case LOGICAL_REP_MSG_BEGIN:
3308  apply_handle_begin(s);
3309  break;
3310 
3313  break;
3314 
3317  break;
3318 
3321  break;
3322 
3325  break;
3326 
3329  break;
3330 
3333  break;
3334 
3335  case LOGICAL_REP_MSG_TYPE:
3336  apply_handle_type(s);
3337  break;
3338 
3341  break;
3342 
3344 
3345  /*
3346  * Logical replication does not use generic logical messages yet.
3347  * Although, it could be used by other applications that use this
3348  * output plugin.
3349  */
3350  break;
3351 
3354  break;
3355 
3358  break;
3359 
3362  break;
3363 
3366  break;
3367 
3370  break;
3371 
3374  break;
3375 
3378  break;
3379 
3382  break;
3383 
3386  break;
3387 
3388  default:
3389  ereport(ERROR,
3390  (errcode(ERRCODE_PROTOCOL_VIOLATION),
3391  errmsg("invalid logical replication message type \"??? (%d)\"", action)));
3392  }
3393 
3394  /* Reset the current command */
3395  apply_error_callback_arg.command = saved_command;
3396 }
static void apply_handle_stream_prepare(StringInfo s)
Definition: worker.c:1289
static void apply_handle_type(StringInfo s)
Definition: worker.c:2344
static void apply_handle_truncate(StringInfo s)
Definition: worker.c:3164
static void apply_handle_update(StringInfo s)
Definition: worker.c:2540
static void apply_handle_stream_commit(StringInfo s)
Definition: worker.c:2151
static void apply_handle_commit_prepared(StringInfo s)
Definition: worker.c:1187
ApplyErrorCallbackArg apply_error_callback_arg
Definition: worker.c:296
static void apply_handle_delete(StringInfo s)
Definition: worker.c:2724
static void apply_handle_begin(StringInfo s)
Definition: worker.c:1009
static void apply_handle_commit(StringInfo s)
Definition: worker.c:1034
static void apply_handle_stream_abort(StringInfo s)
Definition: worker.c:1830
static void apply_handle_relation(StringInfo s)
Definition: worker.c:2321
static void apply_handle_prepare(StringInfo s)
Definition: worker.c:1126
static void apply_handle_rollback_prepared(StringInfo s)
Definition: worker.c:1236
static void apply_handle_stream_stop(StringInfo s)
Definition: worker.c:1644
static void apply_handle_origin(StringInfo s)
Definition: worker.c:1426
static void apply_handle_begin_prepare(StringInfo s)
Definition: worker.c:1060
static void apply_handle_stream_start(StringInfo s)
Definition: worker.c:1485
static void apply_handle_insert(StringInfo s)
Definition: worker.c:2391
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
LogicalRepMsgType
Definition: logicalproto.h:58
@ LOGICAL_REP_MSG_INSERT
Definition: logicalproto.h:62
@ LOGICAL_REP_MSG_TRUNCATE
Definition: logicalproto.h:65
@ LOGICAL_REP_MSG_STREAM_STOP
Definition: logicalproto.h:74
@ LOGICAL_REP_MSG_BEGIN
Definition: logicalproto.h:59
@ LOGICAL_REP_MSG_STREAM_PREPARE
Definition: logicalproto.h:77
@ LOGICAL_REP_MSG_STREAM_ABORT
Definition: logicalproto.h:76
@ LOGICAL_REP_MSG_BEGIN_PREPARE
Definition: logicalproto.h:69
@ LOGICAL_REP_MSG_STREAM_START
Definition: logicalproto.h:73
@ LOGICAL_REP_MSG_COMMIT
Definition: logicalproto.h:60
@ LOGICAL_REP_MSG_PREPARE
Definition: logicalproto.h:70
@ LOGICAL_REP_MSG_RELATION
Definition: logicalproto.h:66
@ LOGICAL_REP_MSG_MESSAGE
Definition: logicalproto.h:68
@ LOGICAL_REP_MSG_ROLLBACK_PREPARED
Definition: logicalproto.h:72
@ LOGICAL_REP_MSG_COMMIT_PREPARED
Definition: logicalproto.h:71
@ LOGICAL_REP_MSG_TYPE
Definition: logicalproto.h:67
@ LOGICAL_REP_MSG_DELETE
Definition: logicalproto.h:64
@ LOGICAL_REP_MSG_STREAM_COMMIT
Definition: logicalproto.h:75
@ LOGICAL_REP_MSG_ORIGIN
Definition: logicalproto.h:61
@ LOGICAL_REP_MSG_UPDATE
Definition: logicalproto.h:63
int pq_getmsgbyte(StringInfo msg)
Definition: pqformat.c:402
LogicalRepMsgType command
Definition: worker.c:239

References generate_unaccent_rules::action, apply_error_callback_arg, apply_handle_begin(), apply_handle_begin_prepare(), apply_handle_commit(), apply_handle_commit_prepared(), apply_handle_delete(), apply_handle_insert(), apply_handle_origin(), apply_handle_prepare(), apply_handle_relation(), apply_handle_rollback_prepared(), apply_handle_stream_abort(), apply_handle_stream_commit(), apply_handle_stream_prepare(), apply_handle_stream_start(), apply_handle_stream_stop(), apply_handle_truncate(), apply_handle_type(), apply_handle_update(), ApplyErrorCallbackArg::command, ereport, errcode(), errmsg(), ERROR, LOGICAL_REP_MSG_BEGIN, LOGICAL_REP_MSG_BEGIN_PREPARE, LOGICAL_REP_MSG_COMMIT, LOGICAL_REP_MSG_COMMIT_PREPARED, LOGICAL_REP_MSG_DELETE, LOGICAL_REP_MSG_INSERT, LOGICAL_REP_MSG_MESSAGE, LOGICAL_REP_MSG_ORIGIN, LOGICAL_REP_MSG_PREPARE, LOGICAL_REP_MSG_RELATION, LOGICAL_REP_MSG_ROLLBACK_PREPARED, LOGICAL_REP_MSG_STREAM_ABORT, LOGICAL_REP_MSG_STREAM_COMMIT, LOGICAL_REP_MSG_STREAM_PREPARE, LOGICAL_REP_MSG_STREAM_START, LOGICAL_REP_MSG_STREAM_STOP, LOGICAL_REP_MSG_TRUNCATE, LOGICAL_REP_MSG_TYPE, LOGICAL_REP_MSG_UPDATE, and pq_getmsgbyte().

Referenced by apply_spooled_messages(), LogicalParallelApplyLoop(), and LogicalRepApplyLoop().

◆ apply_error_callback()

void apply_error_callback ( void *  arg)

Definition at line 4893 of file worker.c.

4894 {
4896 
4898  return;
4899 
4900  Assert(errarg->origin_name);
4901 
4902  if (errarg->rel == NULL)
4903  {
4904  if (!TransactionIdIsValid(errarg->remote_xid))
4905  errcontext("processing remote data for replication origin \"%s\" during message type \"%s\"",
4906  errarg->origin_name,
4907  logicalrep_message_type(errarg->command));
4908  else if (XLogRecPtrIsInvalid(errarg->finish_lsn))
4909  errcontext("processing remote data for replication origin \"%s\" during message type \"%s\" in transaction %u",
4910  errarg->origin_name,
4912  errarg->remote_xid);
4913  else
4914  errcontext("processing remote data for replication origin \"%s\" during message type \"%s\" in transaction %u, finished at %X/%X",
4915  errarg->origin_name,
4917  errarg->remote_xid,
4918  LSN_FORMAT_ARGS(errarg->finish_lsn));
4919  }
4920  else
4921  {
4922  if (errarg->remote_attnum < 0)
4923  {
4924  if (XLogRecPtrIsInvalid(errarg->finish_lsn))
4925  errcontext("processing remote data for replication origin \"%s\" during message type \"%s\" for replication target relation \"%s.%s\" in transaction %u",
4926  errarg->origin_name,
4928  errarg->rel->remoterel.nspname,
4929  errarg->rel->remoterel.relname,
4930  errarg->remote_xid);
4931  else
4932  errcontext("processing remote data for replication origin \"%s\" during message type \"%s\" for replication target relation \"%s.%s\" in transaction %u, finished at %X/%X",
4933  errarg->origin_name,
4935  errarg->rel->remoterel.nspname,
4936  errarg->rel->remoterel.relname,
4937  errarg->remote_xid,
4938  LSN_FORMAT_ARGS(errarg->finish_lsn));
4939  }
4940  else
4941  {
4942  if (XLogRecPtrIsInvalid(errarg->finish_lsn))
4943  errcontext("processing remote data for replication origin \"%s\" during message type \"%s\" for replication target relation \"%s.%s\" column \"%s\" in transaction %u",
4944  errarg->origin_name,
4946  errarg->rel->remoterel.nspname,
4947  errarg->rel->remoterel.relname,
4948  errarg->rel->remoterel.attnames[errarg->remote_attnum],
4949  errarg->remote_xid);
4950  else
4951  errcontext("processing remote data for replication origin \"%s\" during message type \"%s\" for replication target relation \"%s.%s\" column \"%s\" in transaction %u, finished at %X/%X",
4952  errarg->origin_name,
4954  errarg->rel->remoterel.nspname,
4955  errarg->rel->remoterel.relname,
4956  errarg->rel->remoterel.attnames[errarg->remote_attnum],
4957  errarg->remote_xid,
4958  LSN_FORMAT_ARGS(errarg->finish_lsn));
4959  }
4960  }
4961 }
#define errcontext
Definition: elog.h:196
Assert(fmt[strlen(fmt) - 1] !='\n')
const char * logicalrep_message_type(LogicalRepMsgType action)
Definition: proto.c:1217
TransactionId remote_xid
Definition: worker.c:244
XLogRecPtr finish_lsn
Definition: worker.c:245
LogicalRepRelMapEntry * rel
Definition: worker.c:240
LogicalRepRelation remoterel
#define TransactionIdIsValid(xid)
Definition: transam.h:41
#define LSN_FORMAT_ARGS(lsn)
Definition: xlogdefs.h:43
#define XLogRecPtrIsInvalid(r)
Definition: xlogdefs.h:29

References apply_error_callback_arg, Assert(), LogicalRepRelation::attnames, ApplyErrorCallbackArg::command, errcontext, ApplyErrorCallbackArg::finish_lsn, logicalrep_message_type(), LSN_FORMAT_ARGS, LogicalRepRelation::nspname, ApplyErrorCallbackArg::origin_name, ApplyErrorCallbackArg::rel, LogicalRepRelation::relname, ApplyErrorCallbackArg::remote_attnum, ApplyErrorCallbackArg::remote_xid, LogicalRepRelMapEntry::remoterel, TransactionIdIsValid, and XLogRecPtrIsInvalid.

Referenced by LogicalParallelApplyLoop(), and LogicalRepApplyLoop().

◆ apply_handle_begin()

static void apply_handle_begin ( StringInfo  s)
static

Definition at line 1009 of file worker.c.

1010 {
1011  LogicalRepBeginData begin_data;
1012 
1013  /* There must not be an active streaming transaction. */
1015 
1016  logicalrep_read_begin(s, &begin_data);
1017  set_apply_error_context_xact(begin_data.xid, begin_data.final_lsn);
1018 
1019  remote_final_lsn = begin_data.final_lsn;
1020 
1022 
1023  in_remote_transaction = true;
1024 
1026 }
bool in_remote_transaction
Definition: worker.c:321
static void set_apply_error_context_xact(TransactionId xid, XLogRecPtr lsn)
Definition: worker.c:4965
static XLogRecPtr remote_final_lsn
Definition: worker.c:322
static void maybe_start_skipping_changes(XLogRecPtr finish_lsn)
Definition: worker.c:4763
static TransactionId stream_xid
Definition: worker.c:327
void pgstat_report_activity(BackendState state, const char *cmd_str)
@ STATE_RUNNING
void logicalrep_read_begin(StringInfo in, LogicalRepBeginData *begin_data)
Definition: proto.c:74
XLogRecPtr final_lsn
Definition: logicalproto.h:129
TransactionId xid
Definition: logicalproto.h:131

References Assert(), LogicalRepBeginData::final_lsn, in_remote_transaction, logicalrep_read_begin(), maybe_start_skipping_changes(), pgstat_report_activity(), remote_final_lsn, set_apply_error_context_xact(), STATE_RUNNING, stream_xid, TransactionIdIsValid, and LogicalRepBeginData::xid.

Referenced by apply_dispatch().

◆ apply_handle_begin_prepare()

static void apply_handle_begin_prepare ( StringInfo  s)
static

Definition at line 1060 of file worker.c.

1061 {
1062  LogicalRepPreparedTxnData begin_data;
1063 
1064  /* Tablesync should never receive prepare. */
1065  if (am_tablesync_worker())
1066  ereport(ERROR,
1067  (errcode(ERRCODE_PROTOCOL_VIOLATION),
1068  errmsg_internal("tablesync worker received a BEGIN PREPARE message")));
1069 
1070  /* There must not be an active streaming transaction. */
1072 
1073  logicalrep_read_begin_prepare(s, &begin_data);
1074  set_apply_error_context_xact(begin_data.xid, begin_data.prepare_lsn);
1075 
1076  remote_final_lsn = begin_data.prepare_lsn;
1077 
1079 
1080  in_remote_transaction = true;
1081 
1083 }
int errmsg_internal(const char *fmt,...)
Definition: elog.c:1156
void logicalrep_read_begin_prepare(StringInfo in, LogicalRepPreparedTxnData *begin_data)
Definition: proto.c:145
static bool am_tablesync_worker(void)

References am_tablesync_worker(), Assert(), ereport, errcode(), errmsg_internal(), ERROR, in_remote_transaction, logicalrep_read_begin_prepare(), maybe_start_skipping_changes(), pgstat_report_activity(), LogicalRepPreparedTxnData::prepare_lsn, remote_final_lsn, set_apply_error_context_xact(), STATE_RUNNING, stream_xid, TransactionIdIsValid, and LogicalRepPreparedTxnData::xid.

Referenced by apply_dispatch().

◆ apply_handle_commit()

static void apply_handle_commit ( StringInfo  s)
static

Definition at line 1034 of file worker.c.

1035 {
1036  LogicalRepCommitData commit_data;
1037 
1038  logicalrep_read_commit(s, &commit_data);
1039 
1040  if (commit_data.commit_lsn != remote_final_lsn)
1041  ereport(ERROR,
1042  (errcode(ERRCODE_PROTOCOL_VIOLATION),
1043  errmsg_internal("incorrect commit LSN %X/%X in commit message (expected %X/%X)",
1044  LSN_FORMAT_ARGS(commit_data.commit_lsn),
1046 
1047  apply_handle_commit_internal(&commit_data);
1048 
1049  /* Process any tables that are being synchronized in parallel. */
1050  process_syncing_tables(commit_data.end_lsn);
1051 
1054 }
static void apply_handle_commit_internal(LogicalRepCommitData *commit_data)
Definition: worker.c:2261
static void reset_apply_error_context_info(void)
Definition: worker.c:4973
@ STATE_IDLE
void logicalrep_read_commit(StringInfo in, LogicalRepCommitData *commit_data)
Definition: proto.c:109
void process_syncing_tables(XLogRecPtr current_lsn)
Definition: tablesync.c:650

References apply_handle_commit_internal(), LogicalRepCommitData::commit_lsn, LogicalRepCommitData::end_lsn, ereport, errcode(), errmsg_internal(), ERROR, logicalrep_read_commit(), LSN_FORMAT_ARGS, pgstat_report_activity(), process_syncing_tables(), remote_final_lsn, reset_apply_error_context_info(), and STATE_IDLE.

Referenced by apply_dispatch().

◆ apply_handle_commit_internal()

static void apply_handle_commit_internal ( LogicalRepCommitData commit_data)
static

Definition at line 2261 of file worker.c.

2262 {
2263  if (is_skipping_changes())
2264  {
2266 
2267  /*
2268  * Start a new transaction to clear the subskiplsn, if not started
2269  * yet.
2270  */
2271  if (!IsTransactionState())
2273  }
2274 
2275  if (IsTransactionState())
2276  {
2277  /*
2278  * The transaction is either non-empty or skipped, so we clear the
2279  * subskiplsn.
2280  */
2282 
2283  /*
2284  * Update origin state so we can restart streaming from correct
2285  * position in case of crash.
2286  */
2287  replorigin_session_origin_lsn = commit_data->end_lsn;
2289 
2291 
2292  if (IsTransactionBlock())
2293  {
2294  EndTransactionBlock(false);
2296  }
2297 
2298  pgstat_report_stat(false);
2299 
2301  }
2302  else
2303  {
2304  /* Process any invalidation messages that might have accumulated. */
2307  }
2308 
2309  in_remote_transaction = false;
2310 }
static void stop_skipping_changes(void)
Definition: worker.c:4790
#define is_skipping_changes()
Definition: worker.c:354
static void clear_subscription_skip_lsn(XLogRecPtr finish_lsn)
Definition: worker.c:4812
void store_flush_position(XLogRecPtr remote_lsn, XLogRecPtr local_lsn)
Definition: worker.c:3456
void maybe_reread_subscription(void)
Definition: worker.c:3885
void AcceptInvalidationMessages(void)
Definition: inval.c:746
TimestampTz replorigin_session_origin_timestamp
Definition: origin.c:158
XLogRecPtr replorigin_session_origin_lsn
Definition: origin.c:157
long pgstat_report_stat(bool force)
Definition: pgstat.c:582
TimestampTz committime
Definition: logicalproto.h:138
bool IsTransactionState(void)
Definition: xact.c:378
void StartTransactionCommand(void)
Definition: xact.c:2937
bool IsTransactionBlock(void)
Definition: xact.c:4816
void CommitTransactionCommand(void)
Definition: xact.c:3034
bool EndTransactionBlock(bool chain)
Definition: xact.c:3890
XLogRecPtr XactLastCommitEnd
Definition: xlog.c:258

References AcceptInvalidationMessages(), clear_subscription_skip_lsn(), LogicalRepCommitData::commit_lsn, LogicalRepCommitData::committime, CommitTransactionCommand(), LogicalRepCommitData::end_lsn, EndTransactionBlock(), in_remote_transaction, is_skipping_changes, IsTransactionBlock(), IsTransactionState(), maybe_reread_subscription(), pgstat_report_stat(), replorigin_session_origin_lsn, replorigin_session_origin_timestamp, StartTransactionCommand(), stop_skipping_changes(), store_flush_position(), and XactLastCommitEnd.

Referenced by apply_handle_commit(), and apply_handle_stream_commit().

◆ apply_handle_commit_prepared()

static void apply_handle_commit_prepared ( StringInfo  s)
static

Definition at line 1187 of file worker.c.

1188 {
1189  LogicalRepCommitPreparedTxnData prepare_data;
1190  char gid[GIDSIZE];
1191 
1192  logicalrep_read_commit_prepared(s, &prepare_data);
1193  set_apply_error_context_xact(prepare_data.xid, prepare_data.commit_lsn);
1194 
1195  /* Compute GID for two_phase transactions. */
1197  gid, sizeof(gid));
1198 
1199  /* There is no transaction when COMMIT PREPARED is called */
1201 
1202  /*
1203  * Update origin state so we can restart streaming from correct position
1204  * in case of crash.
1205  */
1206  replorigin_session_origin_lsn = prepare_data.end_lsn;
1208 
1209  FinishPreparedTransaction(gid, true);
1212  pgstat_report_stat(false);
1213 
1215  in_remote_transaction = false;
1216 
1217  /* Process any tables that are being synchronized in parallel. */
1218  process_syncing_tables(prepare_data.end_lsn);
1219 
1220  clear_subscription_skip_lsn(prepare_data.end_lsn);
1221 
1224 }
static void begin_replication_step(void)
Definition: worker.c:526
static void end_replication_step(void)
Definition: worker.c:549
static void TwoPhaseTransactionGid(Oid subid, TransactionId xid, char *gid, int szgid)
Definition: worker.c:4410
Subscription * MySubscription
Definition: worker.c:316
void logicalrep_read_commit_prepared(StringInfo in, LogicalRepCommitPreparedTxnData *prepare_data)
Definition: proto.c:278
void FinishPreparedTransaction(const char *gid, bool isCommit)
Definition: twophase.c:1481
#define GIDSIZE
Definition: xact.h:31

References begin_replication_step(), clear_subscription_skip_lsn(), LogicalRepCommitPreparedTxnData::commit_lsn, LogicalRepCommitPreparedTxnData::commit_time, CommitTransactionCommand(), LogicalRepCommitPreparedTxnData::end_lsn, end_replication_step(), FinishPreparedTransaction(), GIDSIZE, in_remote_transaction, logicalrep_read_commit_prepared(), MySubscription, Subscription::oid, pgstat_report_activity(), pgstat_report_stat(), process_syncing_tables(), replorigin_session_origin_lsn, replorigin_session_origin_timestamp, reset_apply_error_context_info(), set_apply_error_context_xact(), STATE_IDLE, store_flush_position(), TwoPhaseTransactionGid(), XactLastCommitEnd, and LogicalRepCommitPreparedTxnData::xid.

Referenced by apply_dispatch().

◆ apply_handle_delete()

static void apply_handle_delete ( StringInfo  s)
static

Definition at line 2724 of file worker.c.

2725 {
2726  LogicalRepRelMapEntry *rel;
2727  LogicalRepTupleData oldtup;
2728  LogicalRepRelId relid;
2729  UserContext ucxt;
2730  ApplyExecutionData *edata;
2731  EState *estate;
2732  TupleTableSlot *remoteslot;
2733  MemoryContext oldctx;
2734  bool run_as_owner;
2735 
2736  /*
2737  * Quick return if we are skipping data modification changes or handling
2738  * streamed transactions.
2739  */
2740  if (is_skipping_changes() ||
2742  return;
2743 
2745 
2746  relid = logicalrep_read_delete(s, &oldtup);
2747  rel = logicalrep_rel_open(relid, RowExclusiveLock);
2748  if (!should_apply_changes_for_rel(rel))
2749  {
2750  /*
2751  * The relation can't become interesting in the middle of the
2752  * transaction so it's safe to unlock it.
2753  */
2756  return;
2757  }
2758 
2759  /* Set relation for error callback */
2761 
2762  /* Check if we can do the delete. */
2764 
2765  /*
2766  * Make sure that any user-supplied code runs as the table owner, unless
2767  * the user has opted out of that behavior.
2768  */
2769  run_as_owner = MySubscription->runasowner;
2770  if (!run_as_owner)
2771  SwitchToUntrustedUser(rel->localrel->rd_rel->relowner, &ucxt);
2772 
2773  /* Initialize the executor state. */
2774  edata = create_edata_for_relation(rel);
2775  estate = edata->estate;
2776  remoteslot = ExecInitExtraTupleSlot(estate,
2777  RelationGetDescr(rel->localrel),
2778  &TTSOpsVirtual);
2779 
2780  /* Build the search tuple. */
2782  slot_store_data(remoteslot, rel, &oldtup);
2783  MemoryContextSwitchTo(oldctx);
2784 
2785  /* For a partitioned table, apply delete to correct partition. */
2786  if (rel->localrel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
2788  remoteslot, NULL, CMD_DELETE);
2789  else
2791  remoteslot, rel->localindexoid);
2792 
2793  finish_edata(edata);
2794 
2795  /* Reset relation for error callback */
2797 
2798  if (!run_as_owner)
2799  RestoreUserContext(&ucxt);
2800 
2802 
2804 }
static void check_relation_updatable(LogicalRepRelMapEntry *rel)
Definition: worker.c:2499
static ApplyExecutionData * create_edata_for_relation(LogicalRepRelMapEntry *rel)
Definition: worker.c:670
static bool should_apply_changes_for_rel(LogicalRepRelMapEntry *rel)
Definition: worker.c:486
static bool handle_streamed_transaction(LogicalRepMsgType action, StringInfo s)
Definition: worker.c:577
static void apply_handle_tuple_routing(ApplyExecutionData *edata, TupleTableSlot *remoteslot, LogicalRepTupleData *newtup, CmdType operation)
Definition: worker.c:2913
static void slot_store_data(TupleTableSlot *slot, LogicalRepRelMapEntry *rel, LogicalRepTupleData *tupleData)
Definition: worker.c:815
static void finish_edata(ApplyExecutionData *edata)
Definition: worker.c:727
static void apply_handle_delete_internal(ApplyExecutionData *edata, ResultRelInfo *relinfo, TupleTableSlot *remoteslot, Oid localindexoid)
Definition: worker.c:2812
const TupleTableSlotOps TTSOpsVirtual
Definition: execTuples.c:83
TupleTableSlot * ExecInitExtraTupleSlot(EState *estate, TupleDesc tupledesc, const TupleTableSlotOps *tts_ops)
Definition: execTuples.c:1832
#define GetPerTupleMemoryContext(estate)
Definition: executor.h:554
#define NoLock
Definition: lockdefs.h:34
#define RowExclusiveLock
Definition: lockdefs.h:38
uint32 LogicalRepRelId
Definition: logicalproto.h:101
@ CMD_DELETE
Definition: nodes.h:279
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
Definition: palloc.h:138
LogicalRepRelId logicalrep_read_delete(StringInfo in, LogicalRepTupleData *oldtup)
Definition: proto.c:564
#define RelationGetDescr(relation)
Definition: rel.h:530
LogicalRepRelMapEntry * logicalrep_rel_open(LogicalRepRelId remoteid, LOCKMODE lockmode)
Definition: relation.c:328
void logicalrep_rel_close(LogicalRepRelMapEntry *rel, LOCKMODE lockmode)
Definition: relation.c:474
ResultRelInfo * targetRelInfo
Definition: worker.c:229
EState * estate
Definition: worker.c:226
Form_pg_class rd_rel
Definition: rel.h:111
void SwitchToUntrustedUser(Oid userid, UserContext *context)
Definition: usercontext.c:33
void RestoreUserContext(UserContext *context)
Definition: usercontext.c:87

References apply_error_callback_arg, apply_handle_delete_internal(), apply_handle_tuple_routing(), begin_replication_step(), check_relation_updatable(), CMD_DELETE, create_edata_for_relation(), end_replication_step(), ApplyExecutionData::estate, ExecInitExtraTupleSlot(), finish_edata(), GetPerTupleMemoryContext, handle_streamed_transaction(), is_skipping_changes, LogicalRepRelMapEntry::localindexoid, LogicalRepRelMapEntry::localrel, LOGICAL_REP_MSG_DELETE, logicalrep_read_delete(), logicalrep_rel_close(), logicalrep_rel_open(), MemoryContextSwitchTo(), MySubscription, NoLock, RelationData::rd_rel, ApplyErrorCallbackArg::rel, RelationGetDescr, RestoreUserContext(), RowExclusiveLock, Subscription::runasowner, should_apply_changes_for_rel(), slot_store_data(), SwitchToUntrustedUser(), ApplyExecutionData::targetRelInfo, and TTSOpsVirtual.

Referenced by apply_dispatch().

◆ apply_handle_delete_internal()

static void apply_handle_delete_internal ( ApplyExecutionData edata,
ResultRelInfo relinfo,
TupleTableSlot remoteslot,
Oid  localindexoid 
)
static

Definition at line 2812 of file worker.c.

2816 {
2817  EState *estate = edata->estate;
2818  Relation localrel = relinfo->ri_RelationDesc;
2819  LogicalRepRelation *remoterel = &edata->targetRel->remoterel;
2820  EPQState epqstate;
2821  TupleTableSlot *localslot;
2822  bool found;
2823 
2824  EvalPlanQualInit(&epqstate, estate, NULL, NIL, -1, NIL);
2825  ExecOpenIndices(relinfo, false);
2826 
2827  found = FindReplTupleInLocalRel(edata, localrel, remoterel, localindexoid,
2828  remoteslot, &localslot);
2829 
2830  /* If found delete it. */
2831  if (found)
2832  {
2833  EvalPlanQualSetSlot(&epqstate, localslot);
2834 
2835  /* Do the actual delete. */
2837  ExecSimpleRelationDelete(relinfo, estate, &epqstate, localslot);
2838  }
2839  else
2840  {
2841  /*
2842  * The tuple to be deleted could not be found. Do nothing except for
2843  * emitting a log message.
2844  *
2845  * XXX should this be promoted to ereport(LOG) perhaps?
2846  */
2847  elog(DEBUG1,
2848  "logical replication did not find row to be deleted "
2849  "in replication target relation \"%s\"",
2850  RelationGetRelationName(localrel));
2851  }
2852 
2853  /* Cleanup. */
2854  ExecCloseIndices(relinfo);
2855  EvalPlanQualEnd(&epqstate);
2856 }
static bool FindReplTupleInLocalRel(ApplyExecutionData *edata, Relation localrel, LogicalRepRelation *remoterel, Oid localidxoid, TupleTableSlot *remoteslot, TupleTableSlot **localslot)
Definition: worker.c:2866
static void TargetPrivilegesCheck(Relation rel, AclMode mode)
Definition: worker.c:2359
#define DEBUG1
Definition: elog.h:30
void ExecCloseIndices(ResultRelInfo *resultRelInfo)
Definition: execIndexing.c:231
void ExecOpenIndices(ResultRelInfo *resultRelInfo, bool speculative)
Definition: execIndexing.c:156
void EvalPlanQualInit(EPQState *epqstate, EState *parentestate, Plan *subplan, List *auxrowmarks, int epqParam, List *resultRelations)
Definition: execMain.c:2563
void EvalPlanQualEnd(EPQState *epqstate)
Definition: execMain.c:3006
void ExecSimpleRelationDelete(ResultRelInfo *resultRelInfo, EState *estate, EPQState *epqstate, TupleTableSlot *searchslot)
#define EvalPlanQualSetSlot(epqstate, slot)
Definition: executor.h:243
#define ACL_DELETE
Definition: parsenodes.h:86
#define NIL
Definition: pg_list.h:68
#define RelationGetRelationName(relation)
Definition: rel.h:538
LogicalRepRelMapEntry * targetRel
Definition: worker.c:228
Relation ri_RelationDesc
Definition: execnodes.h:450

References ACL_DELETE, DEBUG1, elog(), ApplyExecutionData::estate, EvalPlanQualEnd(), EvalPlanQualInit(), EvalPlanQualSetSlot, ExecCloseIndices(), ExecOpenIndices(), ExecSimpleRelationDelete(), FindReplTupleInLocalRel(), NIL, RelationGetRelationName, LogicalRepRelMapEntry::remoterel, ResultRelInfo::ri_RelationDesc, TargetPrivilegesCheck(), and ApplyExecutionData::targetRel.

Referenced by apply_handle_delete(), and apply_handle_tuple_routing().

◆ apply_handle_insert()

static void apply_handle_insert ( StringInfo  s)
static

Definition at line 2391 of file worker.c.

2392 {
2393  LogicalRepRelMapEntry *rel;
2394  LogicalRepTupleData newtup;
2395  LogicalRepRelId relid;
2396  UserContext ucxt;
2397  ApplyExecutionData *edata;
2398  EState *estate;
2399  TupleTableSlot *remoteslot;
2400  MemoryContext oldctx;
2401  bool run_as_owner;
2402 
2403  /*
2404  * Quick return if we are skipping data modification changes or handling
2405  * streamed transactions.
2406  */
2407  if (is_skipping_changes() ||
2409  return;
2410 
2412 
2413  relid = logicalrep_read_insert(s, &newtup);
2414  rel = logicalrep_rel_open(relid, RowExclusiveLock);
2415  if (!should_apply_changes_for_rel(rel))
2416  {
2417  /*
2418  * The relation can't become interesting in the middle of the
2419  * transaction so it's safe to unlock it.
2420  */
2423  return;
2424  }
2425 
2426  /*
2427  * Make sure that any user-supplied code runs as the table owner, unless
2428  * the user has opted out of that behavior.
2429  */
2430  run_as_owner = MySubscription->runasowner;
2431  if (!run_as_owner)
2432  SwitchToUntrustedUser(rel->localrel->rd_rel->relowner, &ucxt);
2433 
2434  /* Set relation for error callback */
2436 
2437  /* Initialize the executor state. */
2438  edata = create_edata_for_relation(rel);
2439  estate = edata->estate;
2440  remoteslot = ExecInitExtraTupleSlot(estate,
2441  RelationGetDescr(rel->localrel),
2442  &TTSOpsVirtual);
2443 
2444  /* Process and store remote tuple in the slot */
2446  slot_store_data(remoteslot, rel, &newtup);
2447  slot_fill_defaults(rel, estate, remoteslot);
2448  MemoryContextSwitchTo(oldctx);
2449 
2450  /* For a partitioned table, insert the tuple into a partition. */
2451  if (rel->localrel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
2453  remoteslot, NULL, CMD_INSERT);
2454  else
2456  remoteslot);
2457 
2458  finish_edata(edata);
2459 
2460  /* Reset relation for error callback */
2462 
2463  if (!run_as_owner)
2464  RestoreUserContext(&ucxt);
2465 
2467 
2469 }
static void apply_handle_insert_internal(ApplyExecutionData *edata, ResultRelInfo *relinfo, TupleTableSlot *remoteslot)
Definition: worker.c:2477
static void slot_fill_defaults(LogicalRepRelMapEntry *rel, EState *estate, TupleTableSlot *slot)
Definition: worker.c:758
@ CMD_INSERT
Definition: nodes.h:278
LogicalRepRelId logicalrep_read_insert(StringInfo in, LogicalRepTupleData *newtup)
Definition: proto.c:436

References apply_error_callback_arg, apply_handle_insert_internal(), apply_handle_tuple_routing(), begin_replication_step(), CMD_INSERT, create_edata_for_relation(), end_replication_step(), ApplyExecutionData::estate, ExecInitExtraTupleSlot(), finish_edata(), GetPerTupleMemoryContext, handle_streamed_transaction(), is_skipping_changes, LogicalRepRelMapEntry::localrel, LOGICAL_REP_MSG_INSERT, logicalrep_read_insert(), logicalrep_rel_close(), logicalrep_rel_open(), MemoryContextSwitchTo(), MySubscription, NoLock, RelationData::rd_rel, ApplyErrorCallbackArg::rel, RelationGetDescr, RestoreUserContext(), RowExclusiveLock, Subscription::runasowner, should_apply_changes_for_rel(), slot_fill_defaults(), slot_store_data(), SwitchToUntrustedUser(), ApplyExecutionData::targetRelInfo, and TTSOpsVirtual.

Referenced by apply_dispatch().

◆ apply_handle_insert_internal()

static void apply_handle_insert_internal ( ApplyExecutionData edata,
ResultRelInfo relinfo,
TupleTableSlot remoteslot 
)
static

Definition at line 2477 of file worker.c.

2480 {
2481  EState *estate = edata->estate;
2482 
2483  /* We must open indexes here. */
2484  ExecOpenIndices(relinfo, false);
2485 
2486  /* Do the insert. */
2488  ExecSimpleRelationInsert(relinfo, estate, remoteslot);
2489 
2490  /* Cleanup. */
2491  ExecCloseIndices(relinfo);
2492 }
void ExecSimpleRelationInsert(ResultRelInfo *resultRelInfo, EState *estate, TupleTableSlot *slot)
#define ACL_INSERT
Definition: parsenodes.h:83

References ACL_INSERT, ApplyExecutionData::estate, ExecCloseIndices(), ExecOpenIndices(), ExecSimpleRelationInsert(), ResultRelInfo::ri_RelationDesc, and TargetPrivilegesCheck().

Referenced by apply_handle_insert(), and apply_handle_tuple_routing().

◆ apply_handle_origin()

static void apply_handle_origin ( StringInfo  s)
static

Definition at line 1426 of file worker.c.

1427 {
1428  /*
1429  * ORIGIN message can only come inside streaming transaction or inside
1430  * remote transaction and before any actual writes.
1431  */
1432  if (!in_streamed_transaction &&
1435  ereport(ERROR,
1436  (errcode(ERRCODE_PROTOCOL_VIOLATION),
1437  errmsg_internal("ORIGIN message sent out of order")));
1438 }
static bool in_streamed_transaction
Definition: worker.c:325

References am_tablesync_worker(), ereport, errcode(), errmsg_internal(), ERROR, in_remote_transaction, in_streamed_transaction, and IsTransactionState().

Referenced by apply_dispatch().

◆ apply_handle_prepare()

static void apply_handle_prepare ( StringInfo  s)
static

Definition at line 1126 of file worker.c.

1127 {
1128  LogicalRepPreparedTxnData prepare_data;
1129 
1130  logicalrep_read_prepare(s, &prepare_data);
1131 
1132  if (prepare_data.prepare_lsn != remote_final_lsn)
1133  ereport(ERROR,
1134  (errcode(ERRCODE_PROTOCOL_VIOLATION),
1135  errmsg_internal("incorrect prepare LSN %X/%X in prepare message (expected %X/%X)",
1136  LSN_FORMAT_ARGS(prepare_data.prepare_lsn),
1138 
1139  /*
1140  * Unlike commit, here, we always prepare the transaction even though no
1141  * change has happened in this transaction or all changes are skipped. It
1142  * is done this way because at commit prepared time, we won't know whether
1143  * we have skipped preparing a transaction because of those reasons.
1144  *
1145  * XXX, We can optimize such that at commit prepared time, we first check
1146  * whether we have prepared the transaction or not but that doesn't seem
1147  * worthwhile because such cases shouldn't be common.
1148  */
1150 
1151  apply_handle_prepare_internal(&prepare_data);
1152 
1155  pgstat_report_stat(false);
1156 
1158 
1159  in_remote_transaction = false;
1160 
1161  /* Process any tables that are being synchronized in parallel. */
1162  process_syncing_tables(prepare_data.end_lsn);
1163 
1164  /*
1165  * Since we have already prepared the transaction, in a case where the
1166  * server crashes before clearing the subskiplsn, it will be left but the
1167  * transaction won't be resent. But that's okay because it's a rare case
1168  * and the subskiplsn will be cleared when finishing the next transaction.
1169  */
1172 
1175 }
static void apply_handle_prepare_internal(LogicalRepPreparedTxnData *prepare_data)
Definition: worker.c:1089
void logicalrep_read_prepare(StringInfo in, LogicalRepPreparedTxnData *prepare_data)
Definition: proto.c:239

References apply_handle_prepare_internal(), begin_replication_step(), clear_subscription_skip_lsn(), CommitTransactionCommand(), LogicalRepPreparedTxnData::end_lsn, end_replication_step(), ereport, errcode(), errmsg_internal(), ERROR, in_remote_transaction, logicalrep_read_prepare(), LSN_FORMAT_ARGS, pgstat_report_activity(), pgstat_report_stat(), LogicalRepPreparedTxnData::prepare_lsn, process_syncing_tables(), remote_final_lsn, reset_apply_error_context_info(), STATE_IDLE, stop_skipping_changes(), store_flush_position(), and XactLastCommitEnd.

Referenced by apply_dispatch().

◆ apply_handle_prepare_internal()

static void apply_handle_prepare_internal ( LogicalRepPreparedTxnData prepare_data)
static

Definition at line 1089 of file worker.c.

1090 {
1091  char gid[GIDSIZE];
1092 
1093  /*
1094  * Compute unique GID for two_phase transactions. We don't use GID of
1095  * prepared transaction sent by server as that can lead to deadlock when
1096  * we have multiple subscriptions from same node point to publications on
1097  * the same node. See comments atop worker.c
1098  */
1099  TwoPhaseTransactionGid(MySubscription->oid, prepare_data->xid,
1100  gid, sizeof(gid));
1101 
1102  /*
1103  * BeginTransactionBlock is necessary to balance the EndTransactionBlock
1104  * called within the PrepareTransactionBlock below.
1105  */
1106  if (!IsTransactionBlock())
1107  {
1109  CommitTransactionCommand(); /* Completes the preceding Begin command. */
1110  }
1111 
1112  /*
1113  * Update origin state so we can restart streaming from correct position
1114  * in case of crash.
1115  */
1116  replorigin_session_origin_lsn = prepare_data->end_lsn;
1118 
1120 }
bool PrepareTransactionBlock(const char *gid)
Definition: xact.c:3838
void BeginTransactionBlock(void)
Definition: xact.c:3770

References BeginTransactionBlock(), CommitTransactionCommand(), LogicalRepPreparedTxnData::end_lsn, GIDSIZE, IsTransactionBlock(), MySubscription, Subscription::oid, LogicalRepPreparedTxnData::prepare_time, PrepareTransactionBlock(), replorigin_session_origin_lsn, replorigin_session_origin_timestamp, TwoPhaseTransactionGid(), and LogicalRepPreparedTxnData::xid.

Referenced by apply_handle_prepare(), and apply_handle_stream_prepare().

◆ apply_handle_relation()

static void apply_handle_relation ( StringInfo  s)
static

Definition at line 2321 of file worker.c.

2322 {
2323  LogicalRepRelation *rel;
2324 
2326  return;
2327 
2328  rel = logicalrep_read_rel(s);
2330 
2331  /* Also reset all entries in the partition map that refer to remoterel. */
2333 }
LogicalRepRelation * logicalrep_read_rel(StringInfo in)
Definition: proto.c:700
void logicalrep_partmap_reset_relmap(LogicalRepRelation *remoterel)
Definition: relation.c:541
void logicalrep_relmap_update(LogicalRepRelation *remoterel)
Definition: relation.c:165

References handle_streamed_transaction(), LOGICAL_REP_MSG_RELATION, logicalrep_partmap_reset_relmap(), logicalrep_read_rel(), and logicalrep_relmap_update().

Referenced by apply_dispatch().

◆ apply_handle_rollback_prepared()

static void apply_handle_rollback_prepared ( StringInfo  s)
static

Definition at line 1236 of file worker.c.

1237 {
1238  LogicalRepRollbackPreparedTxnData rollback_data;
1239  char gid[GIDSIZE];
1240 
1241  logicalrep_read_rollback_prepared(s, &rollback_data);
1242  set_apply_error_context_xact(rollback_data.xid, rollback_data.rollback_end_lsn);
1243 
1244  /* Compute GID for two_phase transactions. */
1245  TwoPhaseTransactionGid(MySubscription->oid, rollback_data.xid,
1246  gid, sizeof(gid));
1247 
1248  /*
1249  * It is possible that we haven't received prepare because it occurred
1250  * before walsender reached a consistent point or the two_phase was still
1251  * not enabled by that time, so in such cases, we need to skip rollback
1252  * prepared.
1253  */
1254  if (LookupGXact(gid, rollback_data.prepare_end_lsn,
1255  rollback_data.prepare_time))
1256  {
1257  /*
1258  * Update origin state so we can restart streaming from correct
1259  * position in case of crash.
1260  */
1263 
1264  /* There is no transaction when ABORT/ROLLBACK PREPARED is called */
1266  FinishPreparedTransaction(gid, false);
1269 
1271  }
1272 
1273  pgstat_report_stat(false);
1274 
1276  in_remote_transaction = false;
1277 
1278  /* Process any tables that are being synchronized in parallel. */
1280 
1283 }
void logicalrep_read_rollback_prepared(StringInfo in, LogicalRepRollbackPreparedTxnData *rollback_data)
Definition: proto.c:336
bool LookupGXact(const char *gid, XLogRecPtr prepare_end_lsn, TimestampTz origin_prepare_timestamp)
Definition: twophase.c:2610

References begin_replication_step(), clear_subscription_skip_lsn(), CommitTransactionCommand(), end_replication_step(), FinishPreparedTransaction(), GIDSIZE, in_remote_transaction, logicalrep_read_rollback_prepared(), LookupGXact(), MySubscription, Subscription::oid, pgstat_report_activity(), pgstat_report_stat(), LogicalRepRollbackPreparedTxnData::prepare_end_lsn, LogicalRepRollbackPreparedTxnData::prepare_time, process_syncing_tables(), replorigin_session_origin_lsn, replorigin_session_origin_timestamp, reset_apply_error_context_info(), LogicalRepRollbackPreparedTxnData::rollback_end_lsn, LogicalRepRollbackPreparedTxnData::rollback_time, set_apply_error_context_xact(), STATE_IDLE, store_flush_position(), TwoPhaseTransactionGid(), XactLastCommitEnd, and LogicalRepRollbackPreparedTxnData::xid.

Referenced by apply_dispatch().

◆ apply_handle_stream_abort()

static void apply_handle_stream_abort ( StringInfo  s)
static

Definition at line 1830 of file worker.c.

1831 {
1832  TransactionId xid;
1833  TransactionId subxid;
1834  LogicalRepStreamAbortData abort_data;
1835  ParallelApplyWorkerInfo *winfo;
1836  TransApplyAction apply_action;
1837 
1838  /* Save the message before it is consumed. */
1839  StringInfoData original_msg = *s;
1840  bool toplevel_xact;
1841 
1843  ereport(ERROR,
1844  (errcode(ERRCODE_PROTOCOL_VIOLATION),
1845  errmsg_internal("STREAM ABORT message without STREAM STOP")));
1846 
1847  /* We receive abort information only when we can apply in parallel. */
1848  logicalrep_read_stream_abort(s, &abort_data,
1850 
1851  xid = abort_data.xid;
1852  subxid = abort_data.subxid;
1853  toplevel_xact = (xid == subxid);
1854 
1855  set_apply_error_context_xact(subxid, abort_data.abort_lsn);
1856 
1857  apply_action = get_transaction_apply_action(xid, &winfo);
1858 
1859  switch (apply_action)
1860  {
1861  case TRANS_LEADER_APPLY:
1862 
1863  /*
1864  * We are in the leader apply worker and the transaction has been
1865  * serialized to file.
1866  */
1867  stream_abort_internal(xid, subxid);
1868 
1869  elog(DEBUG1, "finished processing the STREAM ABORT command");
1870  break;
1871 
1873  Assert(winfo);
1874 
1875  /*
1876  * For the case of aborting the subtransaction, we increment the
1877  * number of streaming blocks and take the lock again before
1878  * sending the STREAM_ABORT to ensure that the parallel apply
1879  * worker will wait on the lock for the next set of changes after
1880  * processing the STREAM_ABORT message if it is not already
1881  * waiting for STREAM_STOP message.
1882  *
1883  * It is important to perform this locking before sending the
1884  * STREAM_ABORT message so that the leader can hold the lock first
1885  * and the parallel apply worker will wait for the leader to
1886  * release the lock. This is the same as what we do in
1887  * apply_handle_stream_stop. See Locking Considerations atop
1888  * applyparallelworker.c.
1889  */
1890  if (!toplevel_xact)
1891  {
1895  }
1896 
1897  if (pa_send_data(winfo, s->len, s->data))
1898  {
1899  /*
1900  * Unlike STREAM_COMMIT and STREAM_PREPARE, we don't need to
1901  * wait here for the parallel apply worker to finish as that
1902  * is not required to maintain the commit order and won't have
1903  * the risk of failures due to transaction dependencies and
1904  * deadlocks. However, it is possible that before the parallel
1905  * worker finishes and we clear the worker info, the xid
1906  * wraparound happens on the upstream and a new transaction
1907  * with the same xid can appear and that can lead to duplicate
1908  * entries in ParallelApplyTxnHash. Yet another problem could
1909  * be that we may have serialized the changes in partial
1910  * serialize mode and the file containing xact changes may
1911  * already exist, and after xid wraparound trying to create
1912  * the file for the same xid can lead to an error. To avoid
1913  * these problems, we decide to wait for the aborts to finish.
1914  *
1915  * Note, it is okay to not update the flush location position
1916  * for aborts as in worst case that means such a transaction
1917  * won't be sent again after restart.
1918  */
1919  if (toplevel_xact)
1921 
1922  break;
1923  }
1924 
1925  /*
1926  * Switch to serialize mode when we are not able to send the
1927  * change to parallel apply worker.
1928  */
1929  pa_switch_to_partial_serialize(winfo, true);
1930 
1931  /* fall through */
1933  Assert(winfo);
1934 
1935  /*
1936  * Parallel apply worker might have applied some changes, so write
1937  * the STREAM_ABORT message so that it can rollback the
1938  * subtransaction if needed.
1939  */
1941  &original_msg);
1942 
1943  if (toplevel_xact)
1944  {
1947  }
1948  break;
1949 
1950  case TRANS_PARALLEL_APPLY:
1951 
1952  /*
1953  * If the parallel apply worker is applying spooled messages then
1954  * close the file before aborting.
1955  */
1956  if (toplevel_xact && stream_fd)
1958 
1959  pa_stream_abort(&abort_data);
1960 
1961  /*
1962  * We need to wait after processing rollback to savepoint for the
1963  * next set of changes.
1964  *
1965  * We have a race condition here due to which we can start waiting
1966  * here when there are more chunk of streams in the queue. See
1967  * apply_handle_stream_stop.
1968  */
1969  if (!toplevel_xact)
1971 
1972  elog(DEBUG1, "finished processing the STREAM ABORT command");
1973  break;
1974 
1975  default:
1976  elog(ERROR, "unexpected apply action: %d", (int) apply_action);
1977  break;
1978  }
1979 
1981 }
void pa_unlock_stream(TransactionId xid, LOCKMODE lockmode)
void pa_stream_abort(LogicalRepStreamAbortData *abort_data)
void pa_lock_stream(TransactionId xid, LOCKMODE lockmode)
void pa_set_fileset_state(ParallelApplyWorkerShared *wshared, PartialFileSetState fileset_state)
void pa_switch_to_partial_serialize(ParallelApplyWorkerInfo *winfo, bool stream_locked)
void pa_xact_finish(ParallelApplyWorkerInfo *winfo, XLogRecPtr remote_lsn)
bool pa_send_data(ParallelApplyWorkerInfo *winfo, Size nbytes, const void *data)
void pa_decr_and_wait_stream_block(void)
static uint32 pg_atomic_add_fetch_u32(volatile pg_atomic_uint32 *ptr, int32 add_)
Definition: atomics.h:381
static TransApplyAction get_transaction_apply_action(TransactionId xid, ParallelApplyWorkerInfo **winfo)
Definition: worker.c:5050
static void stream_open_and_write_change(TransactionId xid, char action, StringInfo s)
Definition: worker.c:4327
static BufFile * stream_fd
Definition: worker.c:357
static void stream_abort_internal(TransactionId xid, TransactionId subxid)
Definition: worker.c:1747
static void stream_close_file(void)
Definition: worker.c:4279
uint32 TransactionId
Definition: c.h:641
LogicalRepWorker * MyLogicalRepWorker
Definition: launcher.c:61
#define AccessExclusiveLock
Definition: lockdefs.h:43
void logicalrep_read_stream_abort(StringInfo in, LogicalRepStreamAbortData *abort_data, bool read_abort_info)
Definition: proto.c:1192
ParallelApplyWorkerShared * shared
pg_atomic_uint32 pending_stream_count
@ FS_SERIALIZE_DONE
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28

References LogicalRepStreamAbortData::abort_lsn, AccessExclusiveLock, Assert(), StringInfoData::data, DEBUG1, elog(), ereport, errcode(), errmsg_internal(), ERROR, FS_SERIALIZE_DONE, get_transaction_apply_action(), in_streamed_transaction, InvalidXLogRecPtr, StringInfoData::len, LOGICAL_REP_MSG_STREAM_ABORT, logicalrep_read_stream_abort(), MyLogicalRepWorker, pa_decr_and_wait_stream_block(), pa_lock_stream(), pa_send_data(), pa_set_fileset_state(), pa_stream_abort(), pa_switch_to_partial_serialize(), pa_unlock_stream(), pa_xact_finish(), LogicalRepWorker::parallel_apply, ParallelApplyWorkerShared::pending_stream_count, pg_atomic_add_fetch_u32(), reset_apply_error_context_info(), set_apply_error_context_xact(), ParallelApplyWorkerInfo::shared, stream_abort_internal(), stream_close_file(), stream_fd, stream_open_and_write_change(), LogicalRepStreamAbortData::subxid, TRANS_LEADER_APPLY, TRANS_LEADER_PARTIAL_SERIALIZE, TRANS_LEADER_SEND_TO_PARALLEL, TRANS_PARALLEL_APPLY, and LogicalRepStreamAbortData::xid.

Referenced by apply_dispatch().

◆ apply_handle_stream_commit()

static void apply_handle_stream_commit ( StringInfo  s)
static

Definition at line 2151 of file worker.c.

2152 {
2153  TransactionId xid;
2154  LogicalRepCommitData commit_data;
2155  ParallelApplyWorkerInfo *winfo;
2156  TransApplyAction apply_action;
2157 
2158  /* Save the message before it is consumed. */
2159  StringInfoData original_msg = *s;
2160 
2162  ereport(ERROR,
2163  (errcode(ERRCODE_PROTOCOL_VIOLATION),
2164  errmsg_internal("STREAM COMMIT message without STREAM STOP")));
2165 
2166  xid = logicalrep_read_stream_commit(s, &commit_data);
2167  set_apply_error_context_xact(xid, commit_data.commit_lsn);
2168 
2169  apply_action = get_transaction_apply_action(xid, &winfo);
2170 
2171  switch (apply_action)
2172  {
2173  case TRANS_LEADER_APPLY:
2174 
2175  /*
2176  * The transaction has been serialized to file, so replay all the
2177  * spooled operations.
2178  */
2180  commit_data.commit_lsn);
2181 
2182  apply_handle_commit_internal(&commit_data);
2183 
2184  /* Unlink the files with serialized changes and subxact info. */
2186 
2187  elog(DEBUG1, "finished processing the STREAM COMMIT command");
2188  break;
2189 
2191  Assert(winfo);
2192 
2193  if (pa_send_data(winfo, s->len, s->data))
2194  {
2195  /* Finish processing the streaming transaction. */
2196  pa_xact_finish(winfo, commit_data.end_lsn);
2197  break;
2198  }
2199 
2200  /*
2201  * Switch to serialize mode when we are not able to send the
2202  * change to parallel apply worker.
2203  */
2204  pa_switch_to_partial_serialize(winfo, true);
2205 
2206  /* fall through */
2208  Assert(winfo);
2209 
2211  &original_msg);
2212 
2214 
2215  /* Finish processing the streaming transaction. */
2216  pa_xact_finish(winfo, commit_data.end_lsn);
2217  break;
2218 
2219  case TRANS_PARALLEL_APPLY:
2220 
2221  /*
2222  * If the parallel apply worker is applying spooled messages then
2223  * close the file before committing.
2224  */
2225  if (stream_fd)
2227 
2228  apply_handle_commit_internal(&commit_data);
2229 
2231 
2232  /*
2233  * It is important to set the transaction state as finished before
2234  * releasing the lock. See pa_wait_for_xact_finish.
2235  */
2238 
2240 
2241  elog(DEBUG1, "finished processing the STREAM COMMIT command");
2242  break;
2243 
2244  default:
2245  elog(ERROR, "unexpected apply action: %d", (int) apply_action);
2246  break;
2247  }
2248 
2249  /* Process any tables that are being synchronized in parallel. */
2250  process_syncing_tables(commit_data.end_lsn);
2251 
2253 
2255 }
void pa_set_xact_state(ParallelApplyWorkerShared *wshared, ParallelTransState xact_state)
void pa_reset_subtrans(void)
ParallelApplyWorkerShared * MyParallelShared
void pa_unlock_transaction(TransactionId xid, LOCKMODE lockmode)
void stream_cleanup_files(Oid subid, TransactionId xid)
Definition: worker.c:4210
void apply_spooled_messages(FileSet *stream_fileset, TransactionId xid, XLogRecPtr lsn)
Definition: worker.c:2019
TransactionId logicalrep_read_stream_commit(StringInfo in, LogicalRepCommitData *commit_data)
Definition: proto.c:1137
FileSet * stream_fileset
@ PARALLEL_TRANS_FINISHED

References AccessExclusiveLock, apply_handle_commit_internal(), apply_spooled_messages(), Assert(), LogicalRepCommitData::commit_lsn, StringInfoData::data, DEBUG1, elog(), LogicalRepCommitData::end_lsn, ereport, errcode(), errmsg_internal(), ERROR, FS_SERIALIZE_DONE, get_transaction_apply_action(), in_streamed_transaction, ParallelApplyWorkerShared::last_commit_end, StringInfoData::len, LOGICAL_REP_MSG_STREAM_COMMIT, logicalrep_read_stream_commit(), MyLogicalRepWorker, MyParallelShared, pa_reset_subtrans(), pa_send_data(), pa_set_fileset_state(), pa_set_xact_state(), pa_switch_to_partial_serialize(), pa_unlock_transaction(), pa_xact_finish(), PARALLEL_TRANS_FINISHED, pgstat_report_activity(), process_syncing_tables(), reset_apply_error_context_info(), set_apply_error_context_xact(), ParallelApplyWorkerInfo::shared, STATE_IDLE, stream_cleanup_files(), stream_close_file(), stream_fd, LogicalRepWorker::stream_fileset, stream_open_and_write_change(), LogicalRepWorker::subid, TRANS_LEADER_APPLY, TRANS_LEADER_PARTIAL_SERIALIZE, TRANS_LEADER_SEND_TO_PARALLEL, TRANS_PARALLEL_APPLY, and XactLastCommitEnd.

Referenced by apply_dispatch().

◆ apply_handle_stream_prepare()

static void apply_handle_stream_prepare ( StringInfo  s)
static

Definition at line 1289 of file worker.c.

1290 {
1291  LogicalRepPreparedTxnData prepare_data;
1292  ParallelApplyWorkerInfo *winfo;
1293  TransApplyAction apply_action;
1294 
1295  /* Save the message before it is consumed. */
1296  StringInfoData original_msg = *s;
1297 
1299  ereport(ERROR,
1300  (errcode(ERRCODE_PROTOCOL_VIOLATION),
1301  errmsg_internal("STREAM PREPARE message without STREAM STOP")));
1302 
1303  /* Tablesync should never receive prepare. */
1304  if (am_tablesync_worker())
1305  ereport(ERROR,
1306  (errcode(ERRCODE_PROTOCOL_VIOLATION),
1307  errmsg_internal("tablesync worker received a STREAM PREPARE message")));
1308 
1309  logicalrep_read_stream_prepare(s, &prepare_data);
1310  set_apply_error_context_xact(prepare_data.xid, prepare_data.prepare_lsn);
1311 
1312  apply_action = get_transaction_apply_action(prepare_data.xid, &winfo);
1313 
1314  switch (apply_action)
1315  {
1316  case TRANS_LEADER_APPLY:
1317 
1318  /*
1319  * The transaction has been serialized to file, so replay all the
1320  * spooled operations.
1321  */
1323  prepare_data.xid, prepare_data.prepare_lsn);
1324 
1325  /* Mark the transaction as prepared. */
1326  apply_handle_prepare_internal(&prepare_data);
1327 
1329 
1331 
1332  in_remote_transaction = false;
1333 
1334  /* Unlink the files with serialized changes and subxact info. */
1336 
1337  elog(DEBUG1, "finished processing the STREAM PREPARE command");
1338  break;
1339 
1341  Assert(winfo);
1342 
1343  if (pa_send_data(winfo, s->len, s->data))
1344  {
1345  /* Finish processing the streaming transaction. */
1346  pa_xact_finish(winfo, prepare_data.end_lsn);
1347  break;
1348  }
1349 
1350  /*
1351  * Switch to serialize mode when we are not able to send the
1352  * change to parallel apply worker.
1353  */
1354  pa_switch_to_partial_serialize(winfo, true);
1355 
1356  /* fall through */
1358  Assert(winfo);
1359 
1360  stream_open_and_write_change(prepare_data.xid,
1362  &original_msg);
1363 
1365 
1366  /* Finish processing the streaming transaction. */
1367  pa_xact_finish(winfo, prepare_data.end_lsn);
1368  break;
1369 
1370  case TRANS_PARALLEL_APPLY:
1371 
1372  /*
1373  * If the parallel apply worker is applying spooled messages then
1374  * close the file before preparing.
1375  */
1376  if (stream_fd)
1378 
1380 
1381  /* Mark the transaction as prepared. */
1382  apply_handle_prepare_internal(&prepare_data);
1383 
1385 
1387 
1389 
1392 
1394 
1395  elog(DEBUG1, "finished processing the STREAM PREPARE command");
1396  break;
1397 
1398  default:
1399  elog(ERROR, "unexpected apply action: %d", (int) apply_action);
1400  break;
1401  }
1402 
1403  pgstat_report_stat(false);
1404 
1405  /* Process any tables that are being synchronized in parallel. */
1406  process_syncing_tables(prepare_data.end_lsn);
1407 
1408  /*
1409  * Similar to prepare case, the subskiplsn could be left in a case of
1410  * server crash but it's okay. See the comments in apply_handle_prepare().
1411  */
1414 
1416 
1418 }
void logicalrep_read_stream_prepare(StringInfo in, LogicalRepPreparedTxnData *prepare_data)
Definition: proto.c:376

References AccessExclusiveLock, am_tablesync_worker(), apply_handle_prepare_internal(), apply_spooled_messages(), Assert(), begin_replication_step(), clear_subscription_skip_lsn(), CommitTransactionCommand(), StringInfoData::data, DEBUG1, elog(), LogicalRepPreparedTxnData::end_lsn, end_replication_step(), ereport, errcode(), errmsg_internal(), ERROR, FS_SERIALIZE_DONE, get_transaction_apply_action(), in_remote_transaction, in_streamed_transaction, ParallelApplyWorkerShared::last_commit_end, StringInfoData::len, LOGICAL_REP_MSG_STREAM_PREPARE, logicalrep_read_stream_prepare(), MyLogicalRepWorker, MyParallelShared, pa_reset_subtrans(), pa_send_data(), pa_set_fileset_state(), pa_set_xact_state(), pa_switch_to_partial_serialize(), pa_unlock_transaction(), pa_xact_finish(), PARALLEL_TRANS_FINISHED, pgstat_report_activity(), pgstat_report_stat(), LogicalRepPreparedTxnData::prepare_lsn, process_syncing_tables(), reset_apply_error_context_info(), set_apply_error_context_xact(), ParallelApplyWorkerInfo::shared, STATE_IDLE, stop_skipping_changes(), store_flush_position(), stream_cleanup_files(), stream_close_file(), stream_fd, LogicalRepWorker::stream_fileset, stream_open_and_write_change(), LogicalRepWorker::subid, TRANS_LEADER_APPLY, TRANS_LEADER_PARTIAL_SERIALIZE, TRANS_LEADER_SEND_TO_PARALLEL, TRANS_PARALLEL_APPLY, XactLastCommitEnd, LogicalRepPreparedTxnData::xid, and ParallelApplyWorkerShared::xid.

Referenced by apply_dispatch().

◆ apply_handle_stream_start()

static void apply_handle_stream_start ( StringInfo  s)
static

Definition at line 1485 of file worker.c.

1486 {
1487  bool first_segment;
1488  ParallelApplyWorkerInfo *winfo;
1489  TransApplyAction apply_action;
1490 
1491  /* Save the message before it is consumed. */
1492  StringInfoData original_msg = *s;
1493 
1495  ereport(ERROR,
1496  (errcode(ERRCODE_PROTOCOL_VIOLATION),
1497  errmsg_internal("duplicate STREAM START message")));
1498 
1499  /* There must not be an active streaming transaction. */
1501 
1502  /* notify handle methods we're processing a remote transaction */
1503  in_streamed_transaction = true;
1504 
1505  /* extract XID of the top-level transaction */
1506  stream_xid = logicalrep_read_stream_start(s, &first_segment);
1507 
1509  ereport(ERROR,
1510  (errcode(ERRCODE_PROTOCOL_VIOLATION),
1511  errmsg_internal("invalid transaction ID in streamed replication transaction")));
1512 
1514 
1515  /* Try to allocate a worker for the streaming transaction. */
1516  if (first_segment)
1518 
1519  apply_action = get_transaction_apply_action(stream_xid, &winfo);
1520 
1521  switch (apply_action)
1522  {
1524 
1525  /*
1526  * Function stream_start_internal starts a transaction. This
1527  * transaction will be committed on the stream stop unless it is a
1528  * tablesync worker in which case it will be committed after
1529  * processing all the messages. We need this transaction for
1530  * handling the BufFile, used for serializing the streaming data
1531  * and subxact info.
1532  */
1533  stream_start_internal(stream_xid, first_segment);
1534  break;
1535 
1537  Assert(winfo);
1538 
1539  /*
1540  * Once we start serializing the changes, the parallel apply
1541  * worker will wait for the leader to release the stream lock
1542  * until the end of the transaction. So, we don't need to release
1543  * the lock or increment the stream count in that case.
1544  */
1545  if (pa_send_data(winfo, s->len, s->data))
1546  {
1547  /*
1548  * Unlock the shared object lock so that the parallel apply
1549  * worker can continue to receive changes.
1550  */
1551  if (!first_segment)
1553 
1554  /*
1555  * Increment the number of streaming blocks waiting to be
1556  * processed by parallel apply worker.
1557  */
1559 
1560  /* Cache the parallel apply worker for this transaction. */
1562  break;
1563  }
1564 
1565  /*
1566  * Switch to serialize mode when we are not able to send the
1567  * change to parallel apply worker.
1568  */
1569  pa_switch_to_partial_serialize(winfo, !first_segment);
1570 
1571  /* fall through */
1573  Assert(winfo);
1574 
1575  /*
1576  * Open the spool file unless it was already opened when switching
1577  * to serialize mode. The transaction started in
1578  * stream_start_internal will be committed on the stream stop.
1579  */
1580  if (apply_action != TRANS_LEADER_SEND_TO_PARALLEL)
1581  stream_start_internal(stream_xid, first_segment);
1582 
1584 
1585  /* Cache the parallel apply worker for this transaction. */
1587  break;
1588 
1589  case TRANS_PARALLEL_APPLY:
1590  if (first_segment)
1591  {
1592  /* Hold the lock until the end of the transaction. */
1595 
1596  /*
1597  * Signal the leader apply worker, as it may be waiting for
1598  * us.
1599  */
1601  }
1602 
1604  break;
1605 
1606  default:
1607  elog(ERROR, "unexpected apply action: %d", (int) apply_action);
1608  break;
1609  }
1610 
1612 }
void pa_lock_transaction(TransactionId xid, LOCKMODE lockmode)
void pa_allocate_worker(TransactionId xid)
void pa_set_stream_apply_worker(ParallelApplyWorkerInfo *winfo)
static uint32 parallel_stream_nchanges
Definition: worker.c:333
static void stream_write_change(char action, StringInfo s)
Definition: worker.c:4297
void stream_start_internal(TransactionId xid, bool first_segment)
Definition: worker.c:1447
void logicalrep_worker_wakeup(Oid subid, Oid relid)
Definition: launcher.c:682
#define InvalidOid
Definition: postgres_ext.h:36
TransactionId logicalrep_read_stream_start(StringInfo in, bool *first_segment)
Definition: proto.c:1087
@ PARALLEL_TRANS_STARTED

References AccessExclusiveLock, Assert(), StringInfoData::data, elog(), ereport, errcode(), errmsg_internal(), ERROR, get_transaction_apply_action(), in_streamed_transaction, InvalidOid, InvalidXLogRecPtr, StringInfoData::len, LOGICAL_REP_MSG_STREAM_START, logicalrep_read_stream_start(), logicalrep_worker_wakeup(), MyLogicalRepWorker, MyParallelShared, pa_allocate_worker(), pa_lock_transaction(), pa_send_data(), pa_set_stream_apply_worker(), pa_set_xact_state(), pa_switch_to_partial_serialize(), pa_unlock_stream(), parallel_stream_nchanges, PARALLEL_TRANS_STARTED, ParallelApplyWorkerShared::pending_stream_count, pg_atomic_add_fetch_u32(), pgstat_report_activity(), set_apply_error_context_xact(), ParallelApplyWorkerInfo::shared, STATE_RUNNING, stream_start_internal(), stream_write_change(), stream_xid, LogicalRepWorker::subid, TRANS_LEADER_PARTIAL_SERIALIZE, TRANS_LEADER_SEND_TO_PARALLEL, TRANS_LEADER_SERIALIZE, TRANS_PARALLEL_APPLY, TransactionIdIsValid, and ParallelApplyWorkerShared::xid.

Referenced by apply_dispatch().

◆ apply_handle_stream_stop()

static void apply_handle_stream_stop ( StringInfo  s)
static

Definition at line 1644 of file worker.c.

1645 {
1646  ParallelApplyWorkerInfo *winfo;
1647  TransApplyAction apply_action;
1648 
1650  ereport(ERROR,
1651  (errcode(ERRCODE_PROTOCOL_VIOLATION),
1652  errmsg_internal("STREAM STOP message without STREAM START")));
1653 
1654  apply_action = get_transaction_apply_action(stream_xid, &winfo);
1655 
1656  switch (apply_action)
1657  {
1660  break;
1661 
1663  Assert(winfo);
1664 
1665  /*
1666  * Lock before sending the STREAM_STOP message so that the leader
1667  * can hold the lock first and the parallel apply worker will wait
1668  * for leader to release the lock. See Locking Considerations atop
1669  * applyparallelworker.c.
1670  */
1672 
1673  if (pa_send_data(winfo, s->len, s->data))
1674  {
1676  break;
1677  }
1678 
1679  /*
1680  * Switch to serialize mode when we are not able to send the
1681  * change to parallel apply worker.
1682  */
1683  pa_switch_to_partial_serialize(winfo, true);
1684 
1685  /* fall through */
1690  break;
1691 
1692  case TRANS_PARALLEL_APPLY:
1693  elog(DEBUG1, "applied %u changes in the streaming chunk",
1695 
1696  /*
1697  * By the time parallel apply worker is processing the changes in
1698  * the current streaming block, the leader apply worker may have
1699  * sent multiple streaming blocks. This can lead to parallel apply
1700  * worker start waiting even when there are more chunk of streams
1701  * in the queue. So, try to lock only if there is no message left
1702  * in the queue. See Locking Considerations atop
1703  * applyparallelworker.c.
1704  *
1705  * Note that here we have a race condition where we can start
1706  * waiting even when there are pending streaming chunks. This can
1707  * happen if the leader sends another streaming block and acquires
1708  * the stream lock again after the parallel apply worker checks
1709  * that there is no pending streaming block and before it actually
1710  * starts waiting on a lock. We can handle this case by not
1711  * allowing the leader to increment the stream block count during
1712  * the time parallel apply worker acquires the lock but it is not
1713  * clear whether that is worth the complexity.
1714  *
1715  * Now, if this missed chunk contains rollback to savepoint, then
1716  * there is a risk of deadlock which probably shouldn't happen
1717  * after restart.
1718  */
1720  break;
1721 
1722  default:
1723  elog(ERROR, "unexpected apply action: %d", (int) apply_action);
1724  break;
1725  }
1726 
1727  in_streamed_transaction = false;
1729 
1730  /*
1731  * The parallel apply worker could be in a transaction in which case we
1732  * need to report the state as STATE_IDLEINTRANSACTION.
1733  */
1736  else
1738 
1740 }
void stream_stop_internal(TransactionId xid)
Definition: worker.c:1621
@ STATE_IDLEINTRANSACTION
#define InvalidTransactionId
Definition: transam.h:31
bool IsTransactionOrTransactionBlock(void)
Definition: xact.c:4834

References AccessExclusiveLock, Assert(), StringInfoData::data, DEBUG1, elog(), ereport, errcode(), errmsg_internal(), ERROR, get_transaction_apply_action(), in_streamed_transaction, InvalidTransactionId, IsTransactionOrTransactionBlock(), StringInfoData::len, LOGICAL_REP_MSG_STREAM_STOP, pa_decr_and_wait_stream_block(), pa_lock_stream(), pa_send_data(), pa_set_stream_apply_worker(), pa_switch_to_partial_serialize(), parallel_stream_nchanges, pgstat_report_activity(), reset_apply_error_context_info(), ParallelApplyWorkerInfo::shared, STATE_IDLE, STATE_IDLEINTRANSACTION, stream_stop_internal(), stream_write_change(), stream_xid, TRANS_LEADER_PARTIAL_SERIALIZE, TRANS_LEADER_SEND_TO_PARALLEL, TRANS_LEADER_SERIALIZE, TRANS_PARALLEL_APPLY, and ParallelApplyWorkerShared::xid.

Referenced by apply_dispatch().

◆ apply_handle_truncate()

static void apply_handle_truncate ( StringInfo  s)
static

Definition at line 3164 of file worker.c.

3165 {
3166  bool cascade = false;
3167  bool restart_seqs = false;
3168  List *remote_relids = NIL;
3169  List *remote_rels = NIL;
3170  List *rels = NIL;
3171  List *part_rels = NIL;
3172  List *relids = NIL;
3173  List *relids_logged = NIL;
3174  ListCell *lc;
3175  LOCKMODE lockmode = AccessExclusiveLock;
3176 
3177  /*
3178  * Quick return if we are skipping data modification changes or handling
3179  * streamed transactions.
3180  */
3181  if (is_skipping_changes() ||
3183  return;
3184 
3186 
3187  remote_relids = logicalrep_read_truncate(s, &cascade, &restart_seqs);
3188 
3189  foreach(lc, remote_relids)
3190  {
3191  LogicalRepRelId relid = lfirst_oid(lc);
3192  LogicalRepRelMapEntry *rel;
3193 
3194  rel = logicalrep_rel_open(relid, lockmode);
3195  if (!should_apply_changes_for_rel(rel))
3196  {
3197  /*
3198  * The relation can't become interesting in the middle of the
3199  * transaction so it's safe to unlock it.
3200  */
3201  logicalrep_rel_close(rel, lockmode);
3202  continue;
3203  }
3204 
3205  remote_rels = lappend(remote_rels, rel);
3207  rels = lappend(rels, rel->localrel);
3208  relids = lappend_oid(relids, rel->localreloid);
3210  relids_logged = lappend_oid(relids_logged, rel->localreloid);
3211 
3212  /*
3213  * Truncate partitions if we got a message to truncate a partitioned
3214  * table.
3215  */
3216  if (rel->localrel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
3217  {
3218  ListCell *child;
3219  List *children = find_all_inheritors(rel->localreloid,
3220  lockmode,
3221  NULL);
3222 
3223  foreach(child, children)
3224  {
3225  Oid childrelid = lfirst_oid(child);
3226  Relation childrel;
3227 
3228  if (list_member_oid(relids, childrelid))
3229  continue;
3230 
3231  /* find_all_inheritors already got lock */
3232  childrel = table_open(childrelid, NoLock);
3233 
3234  /*
3235  * Ignore temp tables of other backends. See similar code in
3236  * ExecuteTruncate().
3237  */
3238  if (RELATION_IS_OTHER_TEMP(childrel))
3239  {
3240  table_close(childrel, lockmode);
3241  continue;
3242  }
3243 
3245  rels = lappend(rels, childrel);
3246  part_rels = lappend(part_rels, childrel);
3247  relids = lappend_oid(relids, childrelid);
3248  /* Log this relation only if needed for logical decoding */
3249  if (RelationIsLogicallyLogged(childrel))
3250  relids_logged = lappend_oid(relids_logged, childrelid);
3251  }
3252  }
3253  }
3254 
3255  /*
3256  * Even if we used CASCADE on the upstream primary we explicitly default
3257  * to replaying changes without further cascading. This might be later
3258  * changeable with a user specified option.
3259  *
3260  * MySubscription->runasowner tells us whether we want to execute
3261  * replication actions as the subscription owner; the last argument to
3262  * TruncateGuts tells it whether we want to switch to the table owner.
3263  * Those are exactly opposite conditions.
3264  */
3265  ExecuteTruncateGuts(rels,
3266  relids,
3267  relids_logged,
3268  DROP_RESTRICT,
3269  restart_seqs,
3271  foreach(lc, remote_rels)
3272  {
3273  LogicalRepRelMapEntry *rel = lfirst(lc);
3274 
3276  }
3277  foreach(lc, part_rels)
3278  {
3279  Relation rel = lfirst(lc);
3280 
3281  table_close(rel, NoLock);
3282  }
3283 
3285 }
List * lappend(List *list, void *datum)
Definition: list.c:338
List * lappend_oid(List *list, Oid datum)
Definition: list.c:374
bool list_member_oid(const List *list, Oid datum)
Definition: list.c:721
int LOCKMODE
Definition: lockdefs.h:26
@ DROP_RESTRICT
Definition: parsenodes.h:2193
#define ACL_TRUNCATE
Definition: parsenodes.h:87
List * find_all_inheritors(Oid parentrelId, LOCKMODE lockmode, List **numparents)
Definition: pg_inherits.c:256
#define lfirst(lc)
Definition: pg_list.h:172
#define lfirst_oid(lc)
Definition: pg_list.h:174
unsigned int Oid
Definition: postgres_ext.h:31
List * logicalrep_read_truncate(StringInfo in, bool *cascade, bool *restart_seqs)
Definition: proto.c:618
#define RelationIsLogicallyLogged(relation)
Definition: rel.h:702
#define RELATION_IS_OTHER_TEMP(relation)
Definition: rel.h:659
Definition: pg_list.h:54
void table_close(Relation relation, LOCKMODE lockmode)
Definition: table.c:126
Relation table_open(Oid relationId, LOCKMODE lockmode)
Definition: table.c:40
void ExecuteTruncateGuts(List *explicit_rels, List *relids, List *relids_logged, DropBehavior behavior, bool restart_seqs, bool run_as_table_owner)
Definition: tablecmds.c:1809

References AccessExclusiveLock, ACL_TRUNCATE, begin_replication_step(), DROP_RESTRICT, end_replication_step(), ExecuteTruncateGuts(), find_all_inheritors(), handle_streamed_transaction(), is_skipping_changes, lappend(), lappend_oid(), lfirst, lfirst_oid, list_member_oid(), LogicalRepRelMapEntry::localrel, LogicalRepRelMapEntry::localreloid, LOGICAL_REP_MSG_TRUNCATE, logicalrep_read_truncate(), logicalrep_rel_close(), logicalrep_rel_open(), MySubscription, NIL, NoLock, RelationData::rd_rel, RELATION_IS_OTHER_TEMP, RelationIsLogicallyLogged, Subscription::runasowner, should_apply_changes_for_rel(), table_close(), table_open(), and TargetPrivilegesCheck().

Referenced by apply_dispatch().

◆ apply_handle_tuple_routing()

static void apply_handle_tuple_routing ( ApplyExecutionData edata,
TupleTableSlot remoteslot,
LogicalRepTupleData newtup,
CmdType  operation 
)
static

Definition at line 2913 of file worker.c.

2917 {
2918  EState *estate = edata->estate;
2919  LogicalRepRelMapEntry *relmapentry = edata->targetRel;
2920  ResultRelInfo *relinfo = edata->targetRelInfo;
2921  Relation parentrel = relinfo->ri_RelationDesc;
2922  ModifyTableState *mtstate;
2923  PartitionTupleRouting *proute;
2924  ResultRelInfo *partrelinfo;
2925  Relation partrel;
2926  TupleTableSlot *remoteslot_part;
2927  TupleConversionMap *map;
2928  MemoryContext oldctx;
2929  LogicalRepRelMapEntry *part_entry = NULL;
2930  AttrMap *attrmap = NULL;
2931 
2932  /* ModifyTableState is needed for ExecFindPartition(). */
2933  edata->mtstate = mtstate = makeNode(ModifyTableState);
2934  mtstate->ps.plan = NULL;
2935  mtstate->ps.state = estate;
2936  mtstate->operation = operation;
2937  mtstate->resultRelInfo = relinfo;
2938 
2939  /* ... as is PartitionTupleRouting. */
2940  edata->proute = proute = ExecSetupPartitionTupleRouting(estate, parentrel);
2941 
2942  /*
2943  * Find the partition to which the "search tuple" belongs.
2944  */
2945  Assert(remoteslot != NULL);
2947  partrelinfo = ExecFindPartition(mtstate, relinfo, proute,
2948  remoteslot, estate);
2949  Assert(partrelinfo != NULL);
2950  partrel = partrelinfo->ri_RelationDesc;
2951 
2952  /*
2953  * Check for supported relkind. We need this since partitions might be of
2954  * unsupported relkinds; and the set of partitions can change, so checking
2955  * at CREATE/ALTER SUBSCRIPTION would be insufficient.
2956  */
2957  CheckSubscriptionRelkind(partrel->rd_rel->relkind,
2959  RelationGetRelationName(partrel));
2960 
2961  /*
2962  * To perform any of the operations below, the tuple must match the
2963  * partition's rowtype. Convert if needed or just copy, using a dedicated
2964  * slot to store the tuple in any case.
2965  */
2966  remoteslot_part = partrelinfo->ri_PartitionTupleSlot;
2967  if (remoteslot_part == NULL)
2968  remoteslot_part = table_slot_create(partrel, &estate->es_tupleTable);
2969  map = ExecGetRootToChildMap(partrelinfo, estate);
2970  if (map != NULL)
2971  {
2972  attrmap = map->attrMap;
2973  remoteslot_part = execute_attr_map_slot(attrmap, remoteslot,
2974  remoteslot_part);
2975  }
2976  else
2977  {
2978  remoteslot_part = ExecCopySlot(remoteslot_part, remoteslot);
2979  slot_getallattrs(remoteslot_part);
2980  }
2981  MemoryContextSwitchTo(oldctx);
2982 
2983  /* Check if we can do the update or delete on the leaf partition. */
2984  if (operation == CMD_UPDATE || operation == CMD_DELETE)
2985  {
2986  part_entry = logicalrep_partition_open(relmapentry, partrel,
2987  attrmap);
2988  check_relation_updatable(part_entry);
2989  }
2990 
2991  switch (operation)
2992  {
2993  case CMD_INSERT:
2994  apply_handle_insert_internal(edata, partrelinfo,
2995  remoteslot_part);
2996  break;
2997 
2998  case CMD_DELETE:
2999  apply_handle_delete_internal(edata, partrelinfo,
3000  remoteslot_part,
3001  part_entry->localindexoid);
3002  break;
3003 
3004  case CMD_UPDATE:
3005 
3006  /*
3007  * For UPDATE, depending on whether or not the updated tuple
3008  * satisfies the partition's constraint, perform a simple UPDATE
3009  * of the partition or move the updated tuple into a different
3010  * suitable partition.
3011  */
3012  {
3013  TupleTableSlot *localslot;
3014  ResultRelInfo *partrelinfo_new;
3015  Relation partrel_new;
3016  bool found;
3017 
3018  /* Get the matching local tuple from the partition. */
3019  found = FindReplTupleInLocalRel(edata, partrel,
3020  &part_entry->remoterel,
3021  part_entry->localindexoid,
3022  remoteslot_part, &localslot);
3023  if (!found)
3024  {
3025  /*
3026  * The tuple to be updated could not be found. Do nothing
3027  * except for emitting a log message.
3028  *
3029  * XXX should this be promoted to ereport(LOG) perhaps?
3030  */
3031  elog(DEBUG1,
3032  "logical replication did not find row to be updated "
3033  "in replication target relation's partition \"%s\"",
3034  RelationGetRelationName(partrel));
3035  return;
3036  }
3037 
3038  /*
3039  * Apply the update to the local tuple, putting the result in
3040  * remoteslot_part.
3041  */
3043  slot_modify_data(remoteslot_part, localslot, part_entry,
3044  newtup);
3045  MemoryContextSwitchTo(oldctx);
3046 
3047  /*
3048  * Does the updated tuple still satisfy the current
3049  * partition's constraint?
3050  */
3051  if (!partrel->rd_rel->relispartition ||
3052  ExecPartitionCheck(partrelinfo, remoteslot_part, estate,
3053  false))
3054  {
3055  /*
3056  * Yes, so simply UPDATE the partition. We don't call
3057  * apply_handle_update_internal() here, which would
3058  * normally do the following work, to avoid repeating some
3059  * work already done above to find the local tuple in the
3060  * partition.
3061  */
3062  EPQState epqstate;
3063 
3064  EvalPlanQualInit(&epqstate, estate, NULL, NIL, -1, NIL);
3065  ExecOpenIndices(partrelinfo, false);
3066 
3067  EvalPlanQualSetSlot(&epqstate, remoteslot_part);
3069  ACL_UPDATE);
3070  ExecSimpleRelationUpdate(partrelinfo, estate, &epqstate,
3071  localslot, remoteslot_part);
3072  ExecCloseIndices(partrelinfo);
3073  EvalPlanQualEnd(&epqstate);
3074  }
3075  else
3076  {
3077  /* Move the tuple into the new partition. */
3078 
3079  /*
3080  * New partition will be found using tuple routing, which
3081  * can only occur via the parent table. We might need to
3082  * convert the tuple to the parent's rowtype. Note that
3083  * this is the tuple found in the partition, not the
3084  * original search tuple received by this function.
3085  */
3086  if (map)
3087  {
3088  TupleConversionMap *PartitionToRootMap =
3090  RelationGetDescr(parentrel));
3091 
3092  remoteslot =
3093  execute_attr_map_slot(PartitionToRootMap->attrMap,
3094  remoteslot_part, remoteslot);
3095  }
3096  else
3097  {
3098  remoteslot = ExecCopySlot(remoteslot, remoteslot_part);
3099  slot_getallattrs(remoteslot);
3100  }
3101 
3102  /* Find the new partition. */
3104  partrelinfo_new = ExecFindPartition(mtstate, relinfo,
3105  proute, remoteslot,
3106  estate);
3107  MemoryContextSwitchTo(oldctx);
3108  Assert(partrelinfo_new != partrelinfo);
3109  partrel_new = partrelinfo_new->ri_RelationDesc;
3110 
3111  /* Check that new partition also has supported relkind. */
3112  CheckSubscriptionRelkind(partrel_new->rd_rel->relkind,
3114  RelationGetRelationName(partrel_new));
3115 
3116  /* DELETE old tuple found in the old partition. */
3117  apply_handle_delete_internal(edata, partrelinfo,
3118  localslot,
3119  part_entry->localindexoid);
3120 
3121  /* INSERT new tuple into the new partition. */
3122 
3123  /*
3124  * Convert the replacement tuple to match the destination
3125  * partition rowtype.
3126  */
3128  remoteslot_part = partrelinfo_new->ri_PartitionTupleSlot;
3129  if (remoteslot_part == NULL)
3130  remoteslot_part = table_slot_create(partrel_new,
3131  &estate->es_tupleTable);
3132  map = ExecGetRootToChildMap(partrelinfo_new, estate);
3133  if (map != NULL)
3134  {
3135  remoteslot_part = execute_attr_map_slot(map->attrMap,
3136  remoteslot,
3137  remoteslot_part);
3138  }
3139  else
3140  {
3141  remoteslot_part = ExecCopySlot(remoteslot_part,
3142  remoteslot);
3143  slot_getallattrs(remoteslot);
3144  }
3145  MemoryContextSwitchTo(oldctx);
3146  apply_handle_insert_internal(edata, partrelinfo_new,
3147  remoteslot_part);
3148  }
3149  }
3150  break;
3151 
3152  default:
3153  elog(ERROR, "unrecognized CmdType: %d", (int) operation);
3154  break;
3155  }
3156 }
static void slot_modify_data(TupleTableSlot *slot, TupleTableSlot *srcslot, LogicalRepRelMapEntry *rel, LogicalRepTupleData *tupleData)
Definition: worker.c:916
bool ExecPartitionCheck(ResultRelInfo *resultRelInfo, TupleTableSlot *slot, EState *estate, bool emitError)
Definition: execMain.c:1816
ResultRelInfo * ExecFindPartition(ModifyTableState *mtstate, ResultRelInfo *rootResultRelInfo, PartitionTupleRouting *proute, TupleTableSlot *slot, EState *estate)
PartitionTupleRouting * ExecSetupPartitionTupleRouting(EState *estate, Relation rel)
void CheckSubscriptionRelkind(char relkind, const char *nspname, const char *relname)
void ExecSimpleRelationUpdate(ResultRelInfo *resultRelInfo, EState *estate, EPQState *epqstate, TupleTableSlot *searchslot, TupleTableSlot *slot)
TupleConversionMap * ExecGetRootToChildMap(ResultRelInfo *resultRelInfo, EState *estate)
Definition: execUtils.c:1237
char * get_namespace_name(Oid nspid)
Definition: lsyscache.c:3348
@ CMD_UPDATE
Definition: nodes.h:277
#define makeNode(_type_)
Definition: nodes.h:176
#define ACL_UPDATE
Definition: parsenodes.h:85
#define RelationGetNamespace(relation)
Definition: rel.h:545
LogicalRepRelMapEntry * logicalrep_partition_open(LogicalRepRelMapEntry *root, Relation partrel, AttrMap *map)
Definition: relation.c:603
PartitionTupleRouting * proute
Definition: worker.c:233
ModifyTableState * mtstate
Definition: worker.c:232
Definition: attmap.h:35
List * es_tupleTable
Definition: execnodes.h:661
CmdType operation
Definition: execnodes.h:1276
ResultRelInfo * resultRelInfo
Definition: execnodes.h:1280
PlanState ps
Definition: execnodes.h:1275
Plan * plan
Definition: execnodes.h:1037
EState * state
Definition: execnodes.h:1039
TupleTableSlot * ri_PartitionTupleSlot
Definition: execnodes.h:575
AttrMap * attrMap
Definition: tupconvert.h:28
TupleTableSlot * table_slot_create(Relation relation, List **reglist)
Definition: tableam.c:91
TupleTableSlot * execute_attr_map_slot(AttrMap *attrMap, TupleTableSlot *in_slot, TupleTableSlot *out_slot)
Definition: tupconvert.c:192
TupleConversionMap * convert_tuples_by_name(TupleDesc indesc, TupleDesc outdesc)
Definition: tupconvert.c:102
static TupleTableSlot * ExecCopySlot(TupleTableSlot *dstslot, TupleTableSlot *srcslot)
Definition: tuptable.h:482
static void slot_getallattrs(TupleTableSlot *slot)
Definition: tuptable.h:361

References ACL_UPDATE, apply_handle_delete_internal(), apply_handle_insert_internal(), Assert(), TupleConversionMap::attrMap, check_relation_updatable(), CheckSubscriptionRelkind(), CMD_DELETE, CMD_INSERT, CMD_UPDATE, convert_tuples_by_name(), DEBUG1, elog(), ERROR, EState::es_tupleTable, ApplyExecutionData::estate, EvalPlanQualEnd(), EvalPlanQualInit(), EvalPlanQualSetSlot, ExecCloseIndices(), ExecCopySlot(), ExecFindPartition(), ExecGetRootToChildMap(), ExecOpenIndices(), ExecPartitionCheck(), ExecSetupPartitionTupleRouting(), ExecSimpleRelationUpdate(), execute_attr_map_slot(), FindReplTupleInLocalRel(), get_namespace_name(), GetPerTupleMemoryContext, LogicalRepRelMapEntry::localindexoid, logicalrep_partition_open(), makeNode, MemoryContextSwitchTo(), ApplyExecutionData::mtstate, NIL, ModifyTableState::operation, PlanState::plan, ApplyExecutionData::proute, ModifyTableState::ps, RelationData::rd_rel, RelationGetDescr, RelationGetNamespace, RelationGetRelationName, LogicalRepRelMapEntry::remoterel, ModifyTableState::resultRelInfo, ResultRelInfo::ri_PartitionTupleSlot, ResultRelInfo::ri_RelationDesc, slot_getallattrs(), slot_modify_data(), PlanState::state, table_slot_create(), TargetPrivilegesCheck(), ApplyExecutionData::targetRel, and ApplyExecutionData::targetRelInfo.

Referenced by apply_handle_delete(), apply_handle_insert(), and apply_handle_update().

◆ apply_handle_type()

static void apply_handle_type ( StringInfo  s)
static

Definition at line 2344 of file worker.c.

2345 {
2346  LogicalRepTyp typ;
2347 
2349  return;
2350 
2351  logicalrep_read_typ(s, &typ);
2352 }
void logicalrep_read_typ(StringInfo in, LogicalRepTyp *ltyp)
Definition: proto.c:756

References handle_streamed_transaction(), LOGICAL_REP_MSG_TYPE, and logicalrep_read_typ().

Referenced by apply_dispatch().

◆ apply_handle_update()

static void apply_handle_update ( StringInfo  s)
static

Definition at line 2540 of file worker.c.

2541 {
2542  LogicalRepRelMapEntry *rel;
2543  LogicalRepRelId relid;
2544  UserContext ucxt;
2545  ApplyExecutionData *edata;
2546  EState *estate;
2547  LogicalRepTupleData oldtup;
2548  LogicalRepTupleData newtup;
2549  bool has_oldtup;
2550  TupleTableSlot *remoteslot;
2551  RTEPermissionInfo *target_perminfo;
2552  MemoryContext oldctx;
2553  bool run_as_owner;
2554 
2555  /*
2556  * Quick return if we are skipping data modification changes or handling
2557  * streamed transactions.
2558  */
2559  if (is_skipping_changes() ||
2561  return;
2562 
2564 
2565  relid = logicalrep_read_update(s, &has_oldtup, &oldtup,
2566  &newtup);
2567  rel = logicalrep_rel_open(relid, RowExclusiveLock);
2568  if (!should_apply_changes_for_rel(rel))
2569  {
2570  /*
2571  * The relation can't become interesting in the middle of the
2572  * transaction so it's safe to unlock it.
2573  */
2576  return;
2577  }
2578 
2579  /* Set relation for error callback */
2581 
2582  /* Check if we can do the update. */
2584 
2585  /*
2586  * Make sure that any user-supplied code runs as the table owner, unless
2587  * the user has opted out of that behavior.
2588  */
2589  run_as_owner = MySubscription->runasowner;
2590  if (!run_as_owner)
2591  SwitchToUntrustedUser(rel->localrel->rd_rel->relowner, &ucxt);
2592 
2593  /* Initialize the executor state. */
2594  edata = create_edata_for_relation(rel);
2595  estate = edata->estate;
2596  remoteslot = ExecInitExtraTupleSlot(estate,
2597  RelationGetDescr(rel->localrel),
2598  &TTSOpsVirtual);
2599 
2600  /*
2601  * Populate updatedCols so that per-column triggers can fire, and so
2602  * executor can correctly pass down indexUnchanged hint. This could
2603  * include more columns than were actually changed on the publisher
2604  * because the logical replication protocol doesn't contain that
2605  * information. But it would for example exclude columns that only exist
2606  * on the subscriber, since we are not touching those.
2607  */
2608  target_perminfo = list_nth(estate->es_rteperminfos, 0);
2609  for (int i = 0; i < remoteslot->tts_tupleDescriptor->natts; i++)
2610  {
2612  int remoteattnum = rel->attrmap->attnums[i];
2613 
2614  if (!att->attisdropped && remoteattnum >= 0)
2615  {
2616  Assert(remoteattnum < newtup.ncols);
2617  if (newtup.colstatus[remoteattnum] != LOGICALREP_COLUMN_UNCHANGED)
2618  target_perminfo->updatedCols =
2619  bms_add_member(target_perminfo->updatedCols,
2621  }
2622  }
2623 
2624  /* Build the search tuple. */
2626  slot_store_data(remoteslot, rel,
2627  has_oldtup ? &oldtup : &newtup);
2628  MemoryContextSwitchTo(oldctx);
2629 
2630  /* For a partitioned table, apply update to correct partition. */
2631  if (rel->localrel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
2633  remoteslot, &newtup, CMD_UPDATE);
2634  else
2636  remoteslot, &newtup, rel->localindexoid);
2637 
2638  finish_edata(edata);
2639 
2640  /* Reset relation for error callback */
2642 
2643  if (!run_as_owner)
2644  RestoreUserContext(&ucxt);
2645 
2647 
2649 }
static void apply_handle_update_internal(ApplyExecutionData *edata, ResultRelInfo *relinfo, TupleTableSlot *remoteslot, LogicalRepTupleData *newtup, Oid localindexoid)
Definition: worker.c:2657
Bitmapset * bms_add_member(Bitmapset *a, int x)
Definition: bitmapset.c:753
int i
Definition: isn.c:73
#define LOGICALREP_COLUMN_UNCHANGED
Definition: logicalproto.h:97
FormData_pg_attribute * Form_pg_attribute
Definition: pg_attribute.h:209
static void * list_nth(const List *list, int n)
Definition: pg_list.h:299
LogicalRepRelId logicalrep_read_update(StringInfo in, bool *has_oldtuple, LogicalRepTupleData *oldtup, LogicalRepTupleData *newtup)
Definition: proto.c:492
AttrNumber * attnums
Definition: attmap.h:36
List * es_rteperminfos
Definition: execnodes.h:624
Bitmapset * updatedCols
Definition: parsenodes.h:1249
TupleDesc tts_tupleDescriptor
Definition: tuptable.h:123
#define FirstLowInvalidHeapAttributeNumber
Definition: sysattr.h:27
#define TupleDescAttr(tupdesc, i)
Definition: tupdesc.h:92

References apply_error_callback_arg, apply_handle_tuple_routing(), apply_handle_update_internal(), Assert(), AttrMap::attnums, LogicalRepRelMapEntry::attrmap, begin_replication_step(), bms_add_member(), check_relation_updatable(), CMD_UPDATE, LogicalRepTupleData::colstatus, create_edata_for_relation(), end_replication_step(), EState::es_rteperminfos, ApplyExecutionData::estate, ExecInitExtraTupleSlot(), finish_edata(), FirstLowInvalidHeapAttributeNumber, GetPerTupleMemoryContext, handle_streamed_transaction(), i, is_skipping_changes, list_nth(), LogicalRepRelMapEntry::localindexoid, LogicalRepRelMapEntry::localrel, LOGICAL_REP_MSG_UPDATE, LOGICALREP_COLUMN_UNCHANGED, logicalrep_read_update(), logicalrep_rel_close(), logicalrep_rel_open(), MemoryContextSwitchTo(), MySubscription, TupleDescData::natts, LogicalRepTupleData::ncols, NoLock, RelationData::rd_rel, ApplyErrorCallbackArg::rel, RelationGetDescr, RestoreUserContext(), RowExclusiveLock, Subscription::runasowner, should_apply_changes_for_rel(), slot_store_data(), SwitchToUntrustedUser(), ApplyExecutionData::targetRelInfo, TupleTableSlot::tts_tupleDescriptor, TTSOpsVirtual, TupleDescAttr, and RTEPermissionInfo::updatedCols.

Referenced by apply_dispatch().

◆ apply_handle_update_internal()

static void apply_handle_update_internal ( ApplyExecutionData edata,
ResultRelInfo relinfo,
TupleTableSlot remoteslot,
LogicalRepTupleData newtup,
Oid  localindexoid 
)
static

Definition at line 2657 of file worker.c.

2662 {
2663  EState *estate = edata->estate;
2664  LogicalRepRelMapEntry *relmapentry = edata->targetRel;
2665  Relation localrel = relinfo->ri_RelationDesc;
2666  EPQState epqstate;
2667  TupleTableSlot *localslot;
2668  bool found;
2669  MemoryContext oldctx;
2670 
2671  EvalPlanQualInit(&epqstate, estate, NULL, NIL, -1, NIL);
2672  ExecOpenIndices(relinfo, false);
2673 
2674  found = FindReplTupleInLocalRel(edata, localrel,
2675  &relmapentry->remoterel,
2676  localindexoid,
2677  remoteslot, &localslot);
2678  ExecClearTuple(remoteslot);
2679 
2680  /*
2681  * Tuple found.
2682  *
2683  * Note this will fail if there are other conflicting unique indexes.
2684  */
2685  if (found)
2686  {
2687  /* Process and store remote tuple in the slot */
2689  slot_modify_data(remoteslot, localslot, relmapentry, newtup);
2690  MemoryContextSwitchTo(oldctx);
2691 
2692  EvalPlanQualSetSlot(&epqstate, remoteslot);
2693 
2694  /* Do the actual update. */
2696  ExecSimpleRelationUpdate(relinfo, estate, &epqstate, localslot,
2697  remoteslot);
2698  }
2699  else
2700  {
2701  /*
2702  * The tuple to be updated could not be found. Do nothing except for
2703  * emitting a log message.
2704  *
2705  * XXX should this be promoted to ereport(LOG) perhaps?
2706  */
2707  elog(DEBUG1,
2708  "logical replication did not find row to be updated "
2709  "in replication target relation \"%s\"",
2710  RelationGetRelationName(localrel));
2711  }
2712 
2713  /* Cleanup. */
2714  ExecCloseIndices(relinfo);
2715  EvalPlanQualEnd(&epqstate);
2716 }
static TupleTableSlot * ExecClearTuple(TupleTableSlot *slot)
Definition: tuptable.h:432

References ACL_UPDATE, DEBUG1, elog(), ApplyExecutionData::estate, EvalPlanQualEnd(), EvalPlanQualInit(), EvalPlanQualSetSlot, ExecClearTuple(), ExecCloseIndices(), ExecOpenIndices(), ExecSimpleRelationUpdate(), FindReplTupleInLocalRel(), GetPerTupleMemoryContext, MemoryContextSwitchTo(), NIL, RelationGetRelationName, LogicalRepRelMapEntry::remoterel, ResultRelInfo::ri_RelationDesc, slot_modify_data(), TargetPrivilegesCheck(), and ApplyExecutionData::targetRel.

Referenced by apply_handle_update().

◆ apply_spooled_messages()

void apply_spooled_messages ( FileSet stream_fileset,
TransactionId  xid,
XLogRecPtr  lsn 
)

Definition at line 2019 of file worker.c.

2021 {
2023  int nchanges;
2024  char path[MAXPGPATH];
2025  char *buffer = NULL;
2026  MemoryContext oldcxt;
2027  ResourceOwner oldowner;
2028  int fileno;
2029  off_t offset;
2030 
2031  if (!am_parallel_apply_worker())
2033 
2034  /* Make sure we have an open transaction */
2036 
2037  /*
2038  * Allocate file handle and memory required to process all the messages in
2039  * TopTransactionContext to avoid them getting reset after each message is
2040  * processed.
2041  */
2043 
2044  /* Open the spool file for the committed/prepared transaction */
2046  elog(DEBUG1, "replaying changes from file \"%s\"", path);
2047 
2048  /*
2049  * Make sure the file is owned by the toplevel transaction so that the
2050  * file will not be accidentally closed when aborting a subtransaction.
2051  */
2052  oldowner = CurrentResourceOwner;
2054 
2055  stream_fd = BufFileOpenFileSet(stream_fileset, path, O_RDONLY, false);
2056 
2057  CurrentResourceOwner = oldowner;
2058 
2059  buffer = palloc(BLCKSZ);
2060  initStringInfo(&s2);
2061 
2062  MemoryContextSwitchTo(oldcxt);
2063 
2064  remote_final_lsn = lsn;
2065 
2066  /*
2067  * Make sure the handle apply_dispatch methods are aware we're in a remote
2068  * transaction.
2069  */
2070  in_remote_transaction = true;
2072 
2074 
2075  /*
2076  * Read the entries one by one and pass them through the same logic as in
2077  * apply_dispatch.
2078  */
2079  nchanges = 0;
2080  while (true)
2081  {
2082  size_t nbytes;
2083  int len;
2084 
2086 
2087  /* read length of the on-disk record */
2088  nbytes = BufFileReadMaybeEOF(stream_fd, &len, sizeof(len), true);
2089 
2090  /* have we reached end of the file? */
2091  if (nbytes == 0)
2092  break;
2093 
2094  /* do we have a correct length? */
2095  if (len <= 0)
2096  elog(ERROR, "incorrect length %d in streaming transaction's changes file \"%s\"",
2097  len, path);
2098 
2099  /* make sure we have sufficiently large buffer */
2100  buffer = repalloc(buffer, len);
2101 
2102  /* and finally read the data into the buffer */
2103  BufFileReadExact(stream_fd, buffer, len);
2104 
2105  BufFileTell(stream_fd, &fileno, &offset);
2106 
2107  /* copy the buffer to the stringinfo and call apply_dispatch */
2108  resetStringInfo(&s2);
2109  appendBinaryStringInfo(&s2, buffer, len);
2110 
2111  /* Ensure we are reading the data into our memory context. */
2113 
2114  apply_dispatch(&s2);
2115 
2117 
2118  MemoryContextSwitchTo(oldcxt);
2119 
2120  nchanges++;
2121 
2122  /*
2123  * It is possible the file has been closed because we have processed
2124  * the transaction end message like stream_commit in which case that
2125  * must be the last message.
2126  */
2127  if (!stream_fd)
2128  {
2129  ensure_last_message(stream_fileset, xid, fileno, offset);
2130  break;
2131  }
2132 
2133  if (nchanges % 1000 == 0)
2134  elog(DEBUG1, "replayed %d changes from file \"%s\"",
2135  nchanges, path);
2136  }
2137 
2138  if (stream_fd)
2140 
2141  elog(DEBUG1, "replayed %d (all) changes from file \"%s\"",
2142  nchanges, path);
2143 
2144  return;
2145 }
MemoryContext ApplyMessageContext
Definition: worker.c:308
static void changes_filename(char *path, Oid subid, TransactionId xid)
Definition: worker.c:4196
void apply_dispatch(StringInfo s)
Definition: worker.c:3292
static void ensure_last_message(FileSet *stream_fileset, TransactionId xid, int fileno, off_t offset)
Definition: worker.c:1987
void BufFileReadExact(BufFile *file, void *ptr, size_t size)
Definition: buffile.c:654
BufFile * BufFileOpenFileSet(FileSet *fileset, const char *name, int mode, bool missing_ok)
Definition: buffile.c:291
void BufFileTell(BufFile *file, int *fileno, off_t *offset)
Definition: buffile.c:833
size_t BufFileReadMaybeEOF(BufFile *file, void *ptr, size_t size, bool eofOK)
Definition: buffile.c:664
void MemoryContextReset(MemoryContext context)
Definition: mcxt.c:330
MemoryContext TopTransactionContext
Definition: mcxt.c:146
void * repalloc(void *pointer, Size size)
Definition: mcxt.c:1476
void * palloc(Size size)
Definition: mcxt.c:1226
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:121
#define MAXPGPATH
const void size_t len
char * s2
ResourceOwner TopTransactionResourceOwner
Definition: resowner.c:149
ResourceOwner CurrentResourceOwner
Definition: resowner.c:147
void resetStringInfo(StringInfo str)
Definition: stringinfo.c:75
void appendBinaryStringInfo(StringInfo str, const void *data, int datalen)
Definition: stringinfo.c:227
void initStringInfo(StringInfo str)
Definition: stringinfo.c:59
static bool am_parallel_apply_worker(void)

References am_parallel_apply_worker(), appendBinaryStringInfo(), apply_dispatch(), ApplyMessageContext, begin_replication_step(), BufFileOpenFileSet(), BufFileReadExact(), BufFileReadMaybeEOF(), BufFileTell(), changes_filename(), CHECK_FOR_INTERRUPTS, CurrentResourceOwner, DEBUG1, elog(), end_replication_step(), ensure_last_message(), ERROR, in_remote_transaction, initStringInfo(), len, MAXPGPATH, maybe_start_skipping_changes(), MemoryContextReset(), MemoryContextSwitchTo(), MyLogicalRepWorker, palloc(), pgstat_report_activity(), remote_final_lsn, repalloc(), resetStringInfo(), s2, STATE_RUNNING, stream_close_file(), stream_fd, LogicalRepWorker::subid, TopTransactionContext, and TopTransactionResourceOwner.

Referenced by apply_handle_stream_commit(), apply_handle_stream_prepare(), and pa_process_spooled_messages_if_required().

◆ apply_worker_exit()

static void apply_worker_exit ( void  )
static

Definition at line 3854 of file worker.c.

3855 {
3857  {
3858  /*
3859  * Don't stop the parallel apply worker as the leader will detect the
3860  * subscription parameter change and restart logical replication later
3861  * anyway. This also prevents the leader from reporting errors when
3862  * trying to communicate with a stopped parallel apply worker, which
3863  * would accidentally disable subscriptions if disable_on_error was
3864  * set.
3865  */
3866  return;
3867  }
3868 
3869  /*
3870  * Reset the last-start time for this apply worker so that the launcher
3871  * will restart it without waiting for wal_retrieve_retry_interval if the
3872  * subscription is still active, and so that we won't leak that hash table
3873  * entry if it isn't.
3874  */
3875  if (am_leader_apply_worker())
3877 
3878  proc_exit(0);
3879 }
void proc_exit(int code)
Definition: ipc.c:104
void ApplyLauncherForgetWorkerStartTime(Oid subid)
Definition: launcher.c:1074
static bool am_leader_apply_worker(void)

References am_leader_apply_worker(), am_parallel_apply_worker(), ApplyLauncherForgetWorkerStartTime(), MyLogicalRepWorker, proc_exit(), and LogicalRepWorker::subid.

Referenced by InitializeLogRepWorker(), and maybe_reread_subscription().

◆ ApplyWorkerMain()

void ApplyWorkerMain ( Datum  main_arg)

Definition at line 4685 of file worker.c.

4686 {
4687  int worker_slot = DatumGetInt32(main_arg);
4688 
4689  InitializingApplyWorker = true;
4690 
4691  SetupApplyOrSyncWorker(worker_slot);
4692 
4693  InitializingApplyWorker = false;
4694 
4695  run_apply_worker();
4696 
4697  proc_exit(0);
4698 }
bool InitializingApplyWorker
Definition: worker.c:336
static void run_apply_worker()
Definition: worker.c:4462
void SetupApplyOrSyncWorker(int worker_slot)
Definition: worker.c:4644
static int32 DatumGetInt32(Datum X)
Definition: postgres.h:202

References DatumGetInt32(), InitializingApplyWorker, proc_exit(), run_apply_worker(), and SetupApplyOrSyncWorker().

◆ AtEOXact_LogicalRepWorkers()

void AtEOXact_LogicalRepWorkers ( bool  isCommit)

Definition at line 5003 of file worker.c.

5004 {
5005  if (isCommit && on_commit_wakeup_workers_subids != NIL)
5006  {
5007  ListCell *lc;
5008 
5009  LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
5010  foreach(lc, on_commit_wakeup_workers_subids)
5011  {
5012  Oid subid = lfirst_oid(lc);
5013  List *workers;
5014  ListCell *lc2;
5015 
5016  workers = logicalrep_workers_find(subid, true);
5017  foreach(lc2, workers)
5018  {
5019  LogicalRepWorker *worker = (LogicalRepWorker *) lfirst(lc2);
5020 
5022  }
5023  }
5024  LWLockRelease(LogicalRepWorkerLock);
5025  }
5026 
5027  /* The List storage will be reclaimed automatically in xact cleanup. */
5029 }
static List * on_commit_wakeup_workers_subids
Definition: worker.c:319
void logicalrep_worker_wakeup_ptr(LogicalRepWorker *worker)
Definition: launcher.c:702
List * logicalrep_workers_find(Oid subid, bool only_running)
Definition: launcher.c:281
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1195
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1808
@ LW_SHARED
Definition: lwlock.h:117

References lfirst, lfirst_oid, logicalrep_worker_wakeup_ptr(), logicalrep_workers_find(), LW_SHARED, LWLockAcquire(), LWLockRelease(), NIL, and on_commit_wakeup_workers_subids.

Referenced by AbortTransaction(), CommitTransaction(), and PrepareTransaction().

◆ begin_replication_step()

◆ changes_filename()

static void changes_filename ( char *  path,
Oid  subid,
TransactionId  xid 
)
inlinestatic

Definition at line 4196 of file worker.c.

4197 {
4198  snprintf(path, MAXPGPATH, "%u-%u.changes", subid, xid);
4199 }
#define snprintf
Definition: port.h:238

References MAXPGPATH, and snprintf.

Referenced by apply_spooled_messages(), ensure_last_message(), stream_abort_internal(), stream_cleanup_files(), and stream_open_file().

◆ check_relation_updatable()

static void check_relation_updatable ( LogicalRepRelMapEntry rel)
static

Definition at line 2499 of file worker.c.

2500 {
2501  /*
2502  * For partitioned tables, we only need to care if the target partition is
2503  * updatable (aka has PK or RI defined for it).
2504  */
2505  if (rel->localrel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
2506  return;
2507 
2508  /* Updatable, no error. */
2509  if (rel->updatable)
2510  return;
2511 
2512  /*
2513  * We are in error mode so it's fine this is somewhat slow. It's better to
2514  * give user correct error.
2515  */
2517  {
2518  ereport(ERROR,
2519  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
2520  errmsg("publisher did not send replica identity column "
2521  "expected by the logical replication target relation \"%s.%s\"",
2522  rel->remoterel.nspname, rel->remoterel.relname)));
2523  }
2524 
2525  ereport(ERROR,
2526  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
2527  errmsg("logical replication target relation \"%s.%s\" has "
2528  "neither REPLICA IDENTITY index nor PRIMARY "
2529  "KEY and published relation does not have "
2530  "REPLICA IDENTITY FULL",
2531  rel->remoterel.nspname, rel->remoterel.relname)));
2532 }
#define OidIsValid(objectId)
Definition: c.h:764
Oid GetRelationIdentityOrPK(Relation rel)
Definition: relation.c:855

References ereport, errcode(), errmsg(), ERROR, GetRelationIdentityOrPK(), LogicalRepRelMapEntry::localrel, LogicalRepRelation::nspname, OidIsValid, RelationData::rd_rel, LogicalRepRelation::relname, LogicalRepRelMapEntry::remoterel, and LogicalRepRelMapEntry::updatable.

Referenced by apply_handle_delete(), apply_handle_tuple_routing(), and apply_handle_update().

◆ cleanup_subxact_info()

static void cleanup_subxact_info ( void  )
inlinestatic

Definition at line 4393 of file worker.c.

4394 {
4395  if (subxact_data.subxacts)
4397 
4398  subxact_data.subxacts = NULL;
4400  subxact_data.nsubxacts = 0;
4402 }
static ApplySubXactData subxact_data
Definition: worker.c:375
void pfree(void *pointer)
Definition: mcxt.c:1456
uint32 nsubxacts
Definition: worker.c:369
uint32 nsubxacts_max
Definition: worker.c:370
SubXactInfo * subxacts
Definition: worker.c:372
TransactionId subxact_last
Definition: worker.c:371

References InvalidTransactionId, ApplySubXactData::nsubxacts, ApplySubXactData::nsubxacts_max, pfree(), subxact_data, ApplySubXactData::subxact_last, and ApplySubXactData::subxacts.

Referenced by stream_abort_internal(), and subxact_info_write().

◆ clear_subscription_skip_lsn()

static void clear_subscription_skip_lsn ( XLogRecPtr  finish_lsn)
static

Definition at line 4812 of file worker.c.

4813 {
4814  Relation rel;
4815  Form_pg_subscription subform;
4816  HeapTuple tup;
4817  XLogRecPtr myskiplsn = MySubscription->skiplsn;
4818  bool started_tx = false;
4819 
4821  return;
4822 
4823  if (!IsTransactionState())
4824  {
4826  started_tx = true;
4827  }
4828 
4829  /*
4830  * Protect subskiplsn of pg_subscription from being concurrently updated
4831  * while clearing it.
4832  */
4833  LockSharedObject(SubscriptionRelationId, MySubscription->oid, 0,
4834  AccessShareLock);
4835 
4836  rel = table_open(SubscriptionRelationId, RowExclusiveLock);
4837 
4838  /* Fetch the existing tuple. */
4841 
4842  if (!HeapTupleIsValid(tup))
4843  elog(ERROR, "subscription \"%s\" does not exist", MySubscription->name);
4844 
4845  subform = (Form_pg_subscription) GETSTRUCT(tup);
4846 
4847  /*
4848  * Clear the subskiplsn. If the user has already changed subskiplsn before
4849  * clearing it we don't update the catalog and the replication origin
4850  * state won't get advanced. So in the worst case, if the server crashes
4851  * before sending an acknowledgment of the flush position the transaction
4852  * will be sent again and the user needs to set subskiplsn again. We can
4853  * reduce the possibility by logging a replication origin WAL record to
4854  * advance the origin LSN instead but there is no way to advance the
4855  * origin timestamp and it doesn't seem to be worth doing anything about
4856  * it since it's a very rare case.
4857  */
4858  if (subform->subskiplsn == myskiplsn)
4859  {
4860  bool nulls[Natts_pg_subscription];
4861  bool replaces[Natts_pg_subscription];
4862  Datum values[Natts_pg_subscription];
4863 
4864  memset(values, 0, sizeof(values));
4865  memset(nulls, false, sizeof(nulls));
4866  memset(replaces, false, sizeof(replaces));
4867 
4868  /* reset subskiplsn */
4869  values[Anum_pg_subscription_subskiplsn - 1] = LSNGetDatum(InvalidXLogRecPtr);
4870  replaces[Anum_pg_subscription_subskiplsn - 1] = true;
4871 
4872  tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
4873  replaces);
4874  CatalogTupleUpdate(rel, &tup->t_self, tup);
4875 
4876  if (myskiplsn != finish_lsn)
4877  ereport(WARNING,
4878  errmsg("skip-LSN of subscription \"%s\" cleared", MySubscription->name),
4879  errdetail("Remote transaction's finish WAL location (LSN) %X/%X did not match skip-LSN %X/%X.",
4880  LSN_FORMAT_ARGS(finish_lsn),
4881  LSN_FORMAT_ARGS(myskiplsn)));
4882  }
4883 
4884  heap_freetuple(tup);
4885  table_close(rel, NoLock);
4886 
4887  if (started_tx)
4889 }
static Datum values[MAXATTR]
Definition: bootstrap.c:156
#define likely(x)
Definition: c.h:299
int errdetail(const char *fmt,...)
Definition: elog.c:1202
#define WARNING
Definition: elog.h:36
HeapTuple heap_modify_tuple(HeapTuple tuple, TupleDesc tupleDesc, Datum *replValues, bool *replIsnull, bool *doReplace)
Definition: heaptuple.c:1201
void heap_freetuple(HeapTuple htup)
Definition: heaptuple.c:1426
#define HeapTupleIsValid(tuple)
Definition: htup.h:78
#define GETSTRUCT(TUP)
Definition: htup_details.h:653
void CatalogTupleUpdate(Relation heapRel, ItemPointer otid, HeapTuple tup)
Definition: indexing.c:313
void LockSharedObject(Oid classid, Oid objid, uint16 objsubid, LOCKMODE lockmode)
Definition: lmgr.c:1046
#define AccessShareLock
Definition: lockdefs.h:36
static Datum LSNGetDatum(XLogRecPtr X)
Definition: pg_lsn.h:28
FormData_pg_subscription * Form_pg_subscription
uintptr_t Datum
Definition: postgres.h:64
static Datum ObjectIdGetDatum(Oid X)
Definition: postgres.h:252
ItemPointerData t_self
Definition: htup.h:65
XLogRecPtr skiplsn
#define SearchSysCacheCopy1(cacheId, key1)
Definition: syscache.h:182
@ SUBSCRIPTIONOID
Definition: syscache.h:99
uint64 XLogRecPtr
Definition: xlogdefs.h:21

References AccessShareLock, am_parallel_apply_worker(), CatalogTupleUpdate(), CommitTransactionCommand(), elog(), ereport, errdetail(), errmsg(), ERROR, GETSTRUCT, heap_freetuple(), heap_modify_tuple(), HeapTupleIsValid, InvalidXLogRecPtr, IsTransactionState(), likely, LockSharedObject(), LSN_FORMAT_ARGS, LSNGetDatum(), MySubscription, Subscription::name, NoLock, ObjectIdGetDatum(), Subscription::oid, RelationGetDescr, RowExclusiveLock, SearchSysCacheCopy1, Subscription::skiplsn, StartTransactionCommand(), SUBSCRIPTIONOID, HeapTupleData::t_self, table_close(), table_open(), values, WARNING, and XLogRecPtrIsInvalid.

Referenced by apply_handle_commit_internal(), apply_handle_commit_prepared(), apply_handle_prepare(), apply_handle_rollback_prepared(), and apply_handle_stream_prepare().

◆ create_edata_for_relation()

static ApplyExecutionData* create_edata_for_relation ( LogicalRepRelMapEntry rel)
static

Definition at line 670 of file worker.c.

671 {
672  ApplyExecutionData *edata;
673  EState *estate;
674  RangeTblEntry *rte;
675  List *perminfos = NIL;
676  ResultRelInfo *resultRelInfo;
677 
678  edata = (ApplyExecutionData *) palloc0(sizeof(ApplyExecutionData));
679  edata->targetRel = rel;
680 
681  edata->estate = estate = CreateExecutorState();
682 
683  rte = makeNode(RangeTblEntry);
684  rte->rtekind = RTE_RELATION;
685  rte->relid = RelationGetRelid(rel->localrel);
686  rte->relkind = rel->localrel->rd_rel->relkind;
688 
689  addRTEPermissionInfo(&perminfos, rte);
690 
691  ExecInitRangeTable(estate, list_make1(rte), perminfos);
692 
693  edata->targetRelInfo = resultRelInfo = makeNode(ResultRelInfo);
694 
695  /*
696  * Use Relation opened by logicalrep_rel_open() instead of opening it
697  * again.
698  */
699  InitResultRelInfo(resultRelInfo, rel->localrel, 1, NULL, 0);
700 
701  /*
702  * We put the ResultRelInfo in the es_opened_result_relations list, even
703  * though we don't populate the es_result_relations array. That's a bit
704  * bogus, but it's enough to make ExecGetTriggerResultRel() find them.
705  *
706  * ExecOpenIndices() is not called here either, each execution path doing
707  * an apply operation being responsible for that.
708  */
710  lappend(estate->es_opened_result_relations, resultRelInfo);
711 
712  estate->es_output_cid = GetCurrentCommandId(true);
713 
714  /* Prepare to catch AFTER triggers. */
716 
717  /* other fields of edata remain NULL for now */
718 
719  return edata;
720 }
void InitResultRelInfo(ResultRelInfo *resultRelInfo, Relation resultRelationDesc, Index resultRelationIndex, ResultRelInfo *partition_root_rri, int instrument_options)
Definition: execMain.c:1225
void ExecInitRangeTable(EState *estate, List *rangeTable, List *permInfos)
Definition: execUtils.c:733
EState * CreateExecutorState(void)
Definition: execUtils.c:93
void * palloc0(Size size)
Definition: mcxt.c:1257
RTEPermissionInfo * addRTEPermissionInfo(List **rteperminfos, RangeTblEntry *rte)
@ RTE_RELATION
Definition: parsenodes.h:1013
#define list_make1(x1)
Definition: pg_list.h:212
#define RelationGetRelid(relation)
Definition: rel.h:504
List * es_opened_result_relations
Definition: execnodes.h:637
CommandId es_output_cid
Definition: execnodes.h:631
RTEKind rtekind
Definition: parsenodes.h:1032
void AfterTriggerBeginQuery(void)
Definition: trigger.c:5006
CommandId GetCurrentCommandId(bool used)
Definition: xact.c:818

References AccessShareLock, addRTEPermissionInfo(), AfterTriggerBeginQuery(), CreateExecutorState(), EState::es_opened_result_relations, EState::es_output_cid, ApplyExecutionData::estate, ExecInitRangeTable(), GetCurrentCommandId(), InitResultRelInfo(), lappend(), list_make1, LogicalRepRelMapEntry::localrel, makeNode, NIL, palloc0(), RelationData::rd_rel, RelationGetRelid, RangeTblEntry::relid, RangeTblEntry::relkind, RangeTblEntry::rellockmode, RTE_RELATION, RangeTblEntry::rtekind, ApplyExecutionData::targetRel, and ApplyExecutionData::targetRelInfo.

Referenced by apply_handle_delete(), apply_handle_insert(), and apply_handle_update().

◆ DisableSubscriptionAndExit()

void DisableSubscriptionAndExit ( void  )

Definition at line 4705 of file worker.c.

4706 {
4707  /*
4708  * Emit the error message, and recover from the error state to an idle
4709  * state
4710  */
4711  HOLD_INTERRUPTS();
4712 
4713  EmitErrorReport();
4715  FlushErrorState();
4716 
4718 
4719  /* Report the worker failed during either table synchronization or apply */
4721  !am_tablesync_worker());
4722 
4723  /* Disable the subscription */
4727 
4728  /* Ensure we remove no-longer-useful entry for worker's start time */
4729  if (am_leader_apply_worker())
4731 
4732  /* Notify the subscription has been disabled and exit */
4733  ereport(LOG,
4734  errmsg("subscription \"%s\" has been disabled because of an error",
4735  MySubscription->name));
4736 
4737  proc_exit(0);
4738 }
void EmitErrorReport(void)
Definition: elog.c:1669
void FlushErrorState(void)
Definition: elog.c:1825
#define LOG
Definition: elog.h:31
#define RESUME_INTERRUPTS()
Definition: miscadmin.h:134
#define HOLD_INTERRUPTS()
Definition: miscadmin.h:132
void DisableSubscription(Oid subid)
void pgstat_report_subscription_error(Oid subid, bool is_apply_error)
void AbortOutOfAnyTransaction(void)
Definition: xact.c:4712

References AbortOutOfAnyTransaction(), am_leader_apply_worker(), am_tablesync_worker(), ApplyLauncherForgetWorkerStartTime(), CommitTransactionCommand(), DisableSubscription(), EmitErrorReport(), ereport, errmsg(), FlushErrorState(), HOLD_INTERRUPTS, LOG, MyLogicalRepWorker, MySubscription, Subscription::name, Subscription::oid, pgstat_report_subscription_error(), proc_exit(), RESUME_INTERRUPTS, StartTransactionCommand(), and LogicalRepWorker::subid.

Referenced by start_apply(), and start_table_sync().

◆ end_replication_step()

◆ ensure_last_message()

static void ensure_last_message ( FileSet stream_fileset,
TransactionId  xid,
int  fileno,
off_t  offset 
)
static

Definition at line 1987 of file worker.c.

1989 {
1990  char path[MAXPGPATH];
1991  BufFile *fd;
1992  int last_fileno;
1993  off_t last_offset;
1994 
1996 
1998 
2000 
2001  fd = BufFileOpenFileSet(stream_fileset, path, O_RDONLY, false);
2002 
2003  BufFileSeek(fd, 0, 0, SEEK_END);
2004  BufFileTell(fd, &last_fileno, &last_offset);
2005 
2006  BufFileClose(fd);
2007 
2009 
2010  if (last_fileno != fileno || last_offset != offset)
2011  elog(ERROR, "unexpected message left in streaming transaction's changes file \"%s\"",
2012  path);
2013 }
int BufFileSeek(BufFile *file, int fileno, off_t offset, int whence)
Definition: buffile.c:740
void BufFileClose(BufFile *file)
Definition: buffile.c:412
static int fd(const char *x, int i)
Definition: preproc-init.c:105

References Assert(), begin_replication_step(), BufFileClose(), BufFileOpenFileSet(), BufFileSeek(), BufFileTell(), changes_filename(), elog(), end_replication_step(), ERROR, fd(), IsTransactionState(), MAXPGPATH, MyLogicalRepWorker, and LogicalRepWorker::subid.

Referenced by apply_spooled_messages().

◆ FindReplTupleInLocalRel()

static bool FindReplTupleInLocalRel ( ApplyExecutionData edata,
Relation  localrel,
LogicalRepRelation remoterel,
Oid  localidxoid,
TupleTableSlot remoteslot,
TupleTableSlot **  localslot 
)
static

Definition at line 2866 of file worker.c.

2871 {
2872  EState *estate = edata->estate;
2873  bool found;
2874 
2875  /*
2876  * Regardless of the top-level operation, we're performing a read here, so
2877  * check for SELECT privileges.
2878  */
2879  TargetPrivilegesCheck(localrel, ACL_SELECT);
2880 
2881  *localslot = table_slot_create(localrel, &estate->es_tupleTable);
2882 
2883  Assert(OidIsValid(localidxoid) ||
2884  (remoterel->replident == REPLICA_IDENTITY_FULL));
2885 
2886  if (OidIsValid(localidxoid))
2887  {
2888 #ifdef USE_ASSERT_CHECKING
2889  Relation idxrel = index_open(localidxoid, AccessShareLock);
2890 
2891  /* Index must be PK, RI, or usable for REPLICA IDENTITY FULL tables */
2892  Assert(GetRelationIdentityOrPK(idxrel) == localidxoid ||
2894  edata->targetRel->attrmap));
2895  index_close(idxrel, AccessShareLock);
2896 #endif
2897 
2898  found = RelationFindReplTupleByIndex(localrel, localidxoid,
2900  remoteslot, *localslot);
2901  }
2902  else
2903  found = RelationFindReplTupleSeq(localrel, LockTupleExclusive,
2904  remoteslot, *localslot);
2905 
2906  return found;
2907 }
bool RelationFindReplTupleSeq(Relation rel, LockTupleMode lockmode, TupleTableSlot *searchslot, TupleTableSlot *outslot)
bool RelationFindReplTupleByIndex(Relation rel, Oid idxoid, LockTupleMode lockmode, TupleTableSlot *searchslot, TupleTableSlot *outslot)
IndexInfo * BuildIndexInfo(Relation index)
Definition: index.c:2426
void index_close(Relation relation, LOCKMODE lockmode)
Definition: indexam.c:158
Relation index_open(Oid relationId, LOCKMODE lockmode)
Definition: indexam.c:132
@ LockTupleExclusive
Definition: lockoptions.h:58
#define ACL_SELECT
Definition: parsenodes.h:84
bool IsIndexUsableForReplicaIdentityFull(IndexInfo *indexInfo, AttrMap *attrmap)
Definition: relation.c:808

References AccessShareLock, ACL_SELECT, Assert(), LogicalRepRelMapEntry::attrmap, BuildIndexInfo(), EState::es_tupleTable, ApplyExecutionData::estate, GetRelationIdentityOrPK(), index_close(), index_open(), IsIndexUsableForReplicaIdentityFull(), LockTupleExclusive, OidIsValid, RelationFindReplTupleByIndex(), RelationFindReplTupleSeq(), LogicalRepRelation::replident, table_slot_create(), TargetPrivilegesCheck(), and ApplyExecutionData::targetRel.

Referenced by apply_handle_delete_internal(), apply_handle_tuple_routing(), and apply_handle_update_internal().

◆ finish_edata()

static void finish_edata ( ApplyExecutionData edata)
static

Definition at line 727 of file worker.c.

728 {
729  EState *estate = edata->estate;
730 
731  /* Handle any queued AFTER triggers. */
732  AfterTriggerEndQuery(estate);
733 
734  /* Shut down tuple routing, if any was done. */
735  if (edata->proute)
736  ExecCleanupTupleRouting(edata->mtstate, edata->proute);
737 
738  /*
739  * Cleanup. It might seem that we should call ExecCloseResultRelations()
740  * here, but we intentionally don't. It would close the rel we added to
741  * es_opened_result_relations above, which is wrong because we took no
742  * corresponding refcount. We rely on ExecCleanupTupleRouting() to close
743  * any other relations opened during execution.
744  */
745  ExecResetTupleTable(estate->es_tupleTable, false);
746  FreeExecutorState(estate);
747  pfree(edata);
748 }
void ExecCleanupTupleRouting(ModifyTableState *mtstate, PartitionTupleRouting *proute)
void ExecResetTupleTable(List *tupleTable, bool shouldFree)
Definition: execTuples.c:1192
void FreeExecutorState(EState *estate)
Definition: execUtils.c:194
void AfterTriggerEndQuery(EState *estate)
Definition: trigger.c:5026

References AfterTriggerEndQuery(), EState::es_tupleTable, ApplyExecutionData::estate, ExecCleanupTupleRouting(), ExecResetTupleTable(), FreeExecutorState(), ApplyExecutionData::mtstate, pfree(), and ApplyExecutionData::proute.

Referenced by apply_handle_delete(), apply_handle_insert(), and apply_handle_update().

◆ get_flush_position()

static void get_flush_position ( XLogRecPtr write,
XLogRecPtr flush,
bool have_pending_txes 
)
static

Definition at line 3412 of file worker.c.

3414 {
3415  dlist_mutable_iter iter;
3416  XLogRecPtr local_flush = GetFlushRecPtr(NULL);
3417 
3419  *flush = InvalidXLogRecPtr;
3420 
3422  {
3423  FlushPosition *pos =
3424  dlist_container(FlushPosition, node, iter.cur);
3425 
3426  *write = pos->remote_end;
3427 
3428  if (pos->local_end <= local_flush)
3429  {
3430  *flush = pos->remote_end;
3431  dlist_delete(iter.cur);
3432  pfree(pos);
3433  }
3434  else
3435  {
3436  /*
3437  * Don't want to uselessly iterate over the rest of the list which
3438  * could potentially be long. Instead get the last element and
3439  * grab the write position from there.
3440  */
3441  pos = dlist_tail_element(FlushPosition, node,
3442  &lsn_mapping);
3443  *write = pos->remote_end;
3444  *have_pending_txes = true;
3445  return;
3446  }
3447  }
3448 
3449  *have_pending_txes = !dlist_is_empty(&lsn_mapping);
3450 }
static dlist_head lsn_mapping
Definition: worker.c:222
static void dlist_delete(dlist_node *node)
Definition: ilist.h:405
#define dlist_tail_element(type, membername, lhead)
Definition: ilist.h:612
#define dlist_foreach_modify(iter, lhead)
Definition: ilist.h:640
static bool dlist_is_empty(const dlist_head *head)
Definition: ilist.h:336
#define dlist_container(type, membername, ptr)
Definition: ilist.h:593
#define write(a, b, c)
Definition: win32.h:14
XLogRecPtr remote_end
Definition: worker.c:219
XLogRecPtr local_end
Definition: worker.c:218
dlist_node * cur
Definition: ilist.h:200
XLogRecPtr GetFlushRecPtr(TimeLineID *insertTLI)
Definition: xlog.c:6113

References dlist_mutable_iter::cur, dlist_container, dlist_delete(), dlist_foreach_modify, dlist_is_empty(), dlist_tail_element, GetFlushRecPtr(), InvalidXLogRecPtr, FlushPosition::local_end, lsn_mapping, pfree(), FlushPosition::remote_end, and write.

Referenced by send_feedback().

◆ get_transaction_apply_action()

static TransApplyAction get_transaction_apply_action ( TransactionId  xid,
ParallelApplyWorkerInfo **  winfo 
)
static

Definition at line 5050 of file worker.c.

5051 {
5052  *winfo = NULL;
5053 
5055  {
5056  return TRANS_PARALLEL_APPLY;
5057  }
5058 
5059  /*
5060  * If we are processing this transaction using a parallel apply worker
5061  * then either we send the changes to the parallel worker or if the worker
5062  * is busy then serialize the changes to the file which will later be
5063  * processed by the parallel worker.
5064  */
5065  *winfo = pa_find_worker(xid);
5066 
5067  if (*winfo && (*winfo)->serialize_changes)
5068  {
5070  }
5071  else if (*winfo)
5072  {
5074  }
5075 
5076  /*
5077  * If there is no parallel worker involved to process this transaction
5078  * then we either directly apply the change or serialize it to a file
5079  * which will later be applied when the transaction finish message is
5080  * processed.
5081  */
5082  else if (in_streamed_transaction)
5083  {
5084  return TRANS_LEADER_SERIALIZE;
5085  }
5086  else
5087  {
5088  return TRANS_LEADER_APPLY;
5089  }
5090 }
ParallelApplyWorkerInfo * pa_find_worker(TransactionId xid)

References am_parallel_apply_worker(), in_streamed_transaction, pa_find_worker(), ParallelApplyWorkerInfo::serialize_changes, TRANS_LEADER_APPLY, TRANS_LEADER_PARTIAL_SERIALIZE, TRANS_LEADER_SEND_TO_PARALLEL, TRANS_LEADER_SERIALIZE, and TRANS_PARALLEL_APPLY.

Referenced by apply_handle_stream_abort(), apply_handle_stream_commit(), apply_handle_stream_prepare(), apply_handle_stream_start(), apply_handle_stream_stop(), and handle_streamed_transaction().

◆ handle_streamed_transaction()

static bool handle_streamed_transaction ( LogicalRepMsgType  action,
StringInfo  s 
)
static

Definition at line 577 of file worker.c.

578 {
579  TransactionId current_xid;
581  TransApplyAction apply_action;
582  StringInfoData original_msg;
583 
584  apply_action = get_transaction_apply_action(stream_xid, &winfo);
585 
586  /* not in streaming mode */
587  if (apply_action == TRANS_LEADER_APPLY)
588  return false;
589 
591 
592  /*
593  * The parallel apply worker needs the xid in this message to decide
594  * whether to define a savepoint, so save the original message that has
595  * not moved the cursor after the xid. We will serialize this message to a
596  * file in PARTIAL_SERIALIZE mode.
597  */
598  original_msg = *s;
599 
600  /*
601  * We should have received XID of the subxact as the first part of the
602  * message, so extract it.
603  */
604  current_xid = pq_getmsgint(s, 4);
605 
606  if (!TransactionIdIsValid(current_xid))
607  ereport(ERROR,
608  (errcode(ERRCODE_PROTOCOL_VIOLATION),
609  errmsg_internal("invalid transaction ID in streamed replication transaction")));
610 
611  switch (apply_action)
612  {
614  Assert(stream_fd);
615 
616  /* Add the new subxact to the array (unless already there). */
617  subxact_info_add(current_xid);
618 
619  /* Write the change to the current file */
621  return true;
622 
624  Assert(winfo);
625 
626  /*
627  * XXX The publisher side doesn't always send relation/type update
628  * messages after the streaming transaction, so also update the
629  * relation/type in leader apply worker. See function
630  * cleanup_rel_sync_cache.
631  */
632  if (pa_send_data(winfo, s->len, s->data))
633  return (action != LOGICAL_REP_MSG_RELATION &&
635 
636  /*
637  * Switch to serialize mode when we are not able to send the
638  * change to parallel apply worker.
639  */
640  pa_switch_to_partial_serialize(winfo, false);
641 
642  /* fall through */
644  stream_write_change(action, &original_msg);
645 
646  /* Same reason as TRANS_LEADER_SEND_TO_PARALLEL case. */
647  return (action != LOGICAL_REP_MSG_RELATION &&
649 
652 
653  /* Define a savepoint for a subxact if needed. */
654  pa_start_subtrans(current_xid, stream_xid);
655  return false;
656 
657  default:
658  elog(ERROR, "unexpected apply action: %d", (int) apply_action);
659  return false; /* silence compiler warning */
660  }
661 }
void pa_start_subtrans(TransactionId current_xid, TransactionId top_xid)
static void subxact_info_add(TransactionId xid)
Definition: worker.c:4111
unsigned int pq_getmsgint(StringInfo msg, int b)
Definition: pqformat.c:418

References generate_unaccent_rules::action, Assert(), StringInfoData::data, elog(), ereport, errcode(), errmsg_internal(), ERROR, get_transaction_apply_action(), StringInfoData::len, LOGICAL_REP_MSG_RELATION, LOGICAL_REP_MSG_TYPE, pa_send_data(), pa_start_subtrans(), pa_switch_to_partial_serialize(), parallel_stream_nchanges, pq_getmsgint(), stream_fd, stream_write_change(), stream_xid, subxact_info_add(), TRANS_LEADER_APPLY, TRANS_LEADER_PARTIAL_SERIALIZE, TRANS_LEADER_SEND_TO_PARALLEL, TRANS_LEADER_SERIALIZE, TRANS_PARALLEL_APPLY, and TransactionIdIsValid.

Referenced by apply_handle_delete(), apply_handle_insert(), apply_handle_relation(), apply_handle_truncate(), apply_handle_type(), and apply_handle_update().

◆ InitializeLogRepWorker()

void InitializeLogRepWorker ( void  )

Definition at line 4568 of file worker.c.

4569 {
4570  MemoryContext oldctx;
4571 
4572  /* Run as replica session replication role. */
4573  SetConfigOption("session_replication_role", "replica",
4575 
4576  /* Connect to our database. */
4579  0);
4580 
4581  /*
4582  * Set always-secure search path, so malicious users can't redirect user
4583  * code (e.g. pg_index.indexprs).
4584  */
4585  SetConfigOption("search_path", "", PGC_SUSET, PGC_S_OVERRIDE);
4586 
4587  /* Load the subscription into persistent memory context. */
4589  "ApplyContext",
4593 
4595  if (!MySubscription)
4596  {
4597  ereport(LOG,
4598  (errmsg("logical replication worker for subscription %u will not start because the subscription was removed during startup",
4600 
4601  /* Ensure we remove no-longer-useful entry for worker's start time */
4602  if (am_leader_apply_worker())
4604 
4605  proc_exit(0);
4606  }
4607 
4608  MySubscriptionValid = true;
4609  MemoryContextSwitchTo(oldctx);
4610 
4611  if (!MySubscription->enabled)
4612  {
4613  ereport(LOG,
4614  (errmsg("logical replication worker for subscription \"%s\" will not start because the subscription was disabled during startup",
4615  MySubscription->name)));
4616 
4618  }
4619 
4620  /* Setup synchronous commit according to the user's wishes */
4621  SetConfigOption("synchronous_commit", MySubscription->synccommit,
4623 
4624  /* Keep us informed about subscription changes. */
4627  (Datum) 0);
4628 
4629  if (am_tablesync_worker())
4630  ereport(LOG,
4631  (errmsg("logical replication table synchronization worker for subscription \"%s\", table \"%s\" has started",
4634  else
4635  ereport(LOG,
4636  (errmsg("logical replication apply worker for subscription \"%s\" has started",
4637  MySubscription->name)));
4638 
4640 }
static void subscription_change_cb(Datum arg, int cacheid, uint32 hashvalue)
Definition: worker.c:3996
static void apply_worker_exit(void)
Definition: worker.c:3854
MemoryContext ApplyContext
Definition: worker.c:309
static bool MySubscriptionValid
Definition: worker.c:317
void SetConfigOption(const char *name, const char *value, GucContext context, GucSource source)
Definition: guc.c:4176
@ PGC_S_OVERRIDE
Definition: guc.h:119
@ PGC_SUSET
Definition: guc.h:74
@ PGC_BACKEND
Definition: guc.h:73
void CacheRegisterSyscacheCallback(int cacheid, SyscacheCallbackFunction func, Datum arg)
Definition: inval.c:1519
char * get_rel_name(Oid relid)
Definition: lsyscache.c:1932
MemoryContext TopMemoryContext
Definition: mcxt.c:141
#define AllocSetContextCreate
Definition: memutils.h:129
#define ALLOCSET_DEFAULT_SIZES
Definition: memutils.h:153
Subscription * GetSubscription(Oid subid, bool missing_ok)
void BackgroundWorkerInitializeConnectionByOid(Oid dboid, Oid useroid, uint32 flags)
Definition: postmaster.c:5605

References ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate, am_leader_apply_worker(), am_tablesync_worker(), apply_worker_exit(), ApplyContext, ApplyLauncherForgetWorkerStartTime(), BackgroundWorkerInitializeConnectionByOid(), CacheRegisterSyscacheCallback(), CommitTransactionCommand(), LogicalRepWorker::dbid, Subscription::enabled, ereport, errmsg(), get_rel_name(), GetSubscription(), LOG, MemoryContextSwitchTo(), MyLogicalRepWorker, MySubscription, MySubscriptionValid, Subscription::name, PGC_BACKEND, PGC_S_OVERRIDE, PGC_SUSET, proc_exit(), LogicalRepWorker::relid, SetConfigOption(), StartTransactionCommand(), LogicalRepWorker::subid, subscription_change_cb(), SUBSCRIPTIONOID, Subscription::synccommit, TopMemoryContext, and LogicalRepWorker::userid.

Referenced by ParallelApplyWorkerMain(), and SetupApplyOrSyncWorker().

◆ IsLogicalParallelApplyWorker()

bool IsLogicalParallelApplyWorker ( void  )

Definition at line 4753 of file worker.c.

4754 {
4756 }
bool IsLogicalWorker(void)
Definition: worker.c:4744

References am_parallel_apply_worker(), and IsLogicalWorker().

Referenced by mq_putmessage().

◆ IsLogicalWorker()

bool IsLogicalWorker ( void  )

Definition at line 4744 of file worker.c.

4745 {
4746  return MyLogicalRepWorker != NULL;
4747 }

References MyLogicalRepWorker.

Referenced by IsLogicalParallelApplyWorker(), and ProcessInterrupts().

◆ LogicalRepApplyLoop()

static void LogicalRepApplyLoop ( XLogRecPtr  last_received)
static

Definition at line 3498 of file worker.c.

3499 {
3500  TimestampTz last_recv_timestamp = GetCurrentTimestamp();
3501  bool ping_sent = false;
3502  TimeLineID tli;
3503  ErrorContextCallback errcallback;
3504 
3505  /*
3506  * Init the ApplyMessageContext which we clean up after each replication
3507  * protocol message.
3508  */
3510  "ApplyMessageContext",
3512 
3513  /*
3514  * This memory context is used for per-stream data when the streaming mode
3515  * is enabled. This context is reset on each stream stop.
3516  */
3518  "LogicalStreamingContext",
3520 
3521  /* mark as idle, before starting to loop */
3523 
3524  /*
3525  * Push apply error context callback. Fields will be filled while applying
3526  * a change.
3527  */
3528  errcallback.callback = apply_error_callback;
3529  errcallback.previous = error_context_stack;
3530  error_context_stack = &errcallback;
3532 
3533  /* This outer loop iterates once per wait. */
3534  for (;;)
3535  {
3537  int rc;
3538  int len;
3539  char *buf = NULL;
3540  bool endofstream = false;
3541  long wait_time;
3542 
3544 
3546 
3548 
3549  if (len != 0)
3550  {
3551  /* Loop to process all available data (without blocking). */
3552  for (;;)
3553  {
3555 
3556  if (len == 0)
3557  {
3558  break;
3559  }
3560  else if (len < 0)
3561  {
3562  ereport(LOG,
3563  (errmsg("data stream from publisher has ended")));
3564  endofstream = true;
3565  break;
3566  }
3567  else
3568  {
3569  int c;
3570  StringInfoData s;
3571 
3572  if (ConfigReloadPending)
3573  {
3574  ConfigReloadPending = false;
3576  }
3577 
3578  /* Reset timeout. */
3579  last_recv_timestamp = GetCurrentTimestamp();
3580  ping_sent = false;
3581 
3582  /* Ensure we are reading the data into our memory context. */
3584 
3585  s.data = buf;
3586  s.len = len;
3587  s.cursor = 0;
3588  s.maxlen = -1;
3589 
3590  c = pq_getmsgbyte(&s);
3591 
3592  if (c == 'w')
3593  {
3594  XLogRecPtr start_lsn;
3595  XLogRecPtr end_lsn;
3596  TimestampTz send_time;
3597 
3598  start_lsn = pq_getmsgint64(&s);
3599  end_lsn = pq_getmsgint64(&s);
3600  send_time = pq_getmsgint64(&s);
3601 
3602  if (last_received < start_lsn)
3603  last_received = start_lsn;
3604 
3605  if (last_received < end_lsn)
3606  last_received = end_lsn;
3607 
3608  UpdateWorkerStats(last_received, send_time, false);
3609 
3610  apply_dispatch(&s);
3611  }
3612  else if (c == 'k')
3613  {
3614  XLogRecPtr end_lsn;
3616  bool reply_requested;
3617 
3618  end_lsn = pq_getmsgint64(&s);
3619  timestamp = pq_getmsgint64(&s);
3620  reply_requested = pq_getmsgbyte(&s);
3621 
3622  if (last_received < end_lsn)
3623  last_received = end_lsn;
3624 
3625  send_feedback(last_received, reply_requested, false);
3626  UpdateWorkerStats(last_received, timestamp, true);
3627  }
3628  /* other message types are purposefully ignored */
3629 
3631  }
3632 
3634  }
3635  }
3636 
3637  /* confirm all writes so far */
3638  send_feedback(last_received, false, false);
3639 
3641  {
3642  /*
3643  * If we didn't get any transactions for a while there might be
3644  * unconsumed invalidation messages in the queue, consume them
3645  * now.
3646  */
3649 
3650  /* Process any table synchronization changes. */
3651  process_syncing_tables(last_received);
3652  }
3653 
3654  /* Cleanup the memory. */
3657 
3658  /* Check if we need to exit the streaming loop. */
3659  if (endofstream)
3660  break;
3661 
3662  /*
3663  * Wait for more data or latch. If we have unflushed transactions,
3664  * wake up after WalWriterDelay to see if they've been flushed yet (in
3665  * which case we should send a feedback message). Otherwise, there's
3666  * no particular urgency about waking up unless we get data or a
3667  * signal.
3668  */
3669  if (!dlist_is_empty(&lsn_mapping))
3670  wait_time = WalWriterDelay;
3671  else
3672  wait_time = NAPTIME_PER_CYCLE;
3673 
3677  fd, wait_time,
3678  WAIT_EVENT_LOGICAL_APPLY_MAIN);
3679 
3680  if (rc & WL_LATCH_SET)
3681  {
3684  }
3685 
3686  if (ConfigReloadPending)
3687  {
3688  ConfigReloadPending = false;
3690  }
3691 
3692  if (rc & WL_TIMEOUT)
3693  {
3694  /*
3695  * We didn't receive anything new. If we haven't heard anything
3696  * from the server for more than wal_receiver_timeout / 2, ping
3697  * the server. Also, if it's been longer than
3698  * wal_receiver_status_interval since the last update we sent,
3699  * send a status update to the primary anyway, to report any
3700  * progress in applying WAL.
3701  */
3702  bool requestReply = false;
3703 
3704  /*
3705  * Check if time since last receive from primary has reached the
3706  * configured limit.
3707  */
3708  if (wal_receiver_timeout > 0)
3709  {
3711  TimestampTz timeout;
3712 
3713  timeout =
3714  TimestampTzPlusMilliseconds(last_recv_timestamp,
3716 
3717  if (now >= timeout)
3718  ereport(ERROR,
3719  (errcode(ERRCODE_CONNECTION_FAILURE),
3720  errmsg("terminating logical replication worker due to timeout")));
3721 
3722  /* Check to see if it's time for a ping. */
3723  if (!ping_sent)
3724  {
3725  timeout = TimestampTzPlusMilliseconds(last_recv_timestamp,
3726  (wal_receiver_timeout / 2));
3727  if (now >= timeout)
3728  {
3729  requestReply = true;
3730  ping_sent = true;
3731  }
3732  }
3733  }
3734 
3735  send_feedback(last_received, requestReply, requestReply);
3736 
3737  /*
3738  * Force reporting to ensure long idle periods don't lead to
3739  * arbitrarily delayed stats. Stats can only be reported outside
3740  * of (implicit or explicit) transactions. That shouldn't lead to
3741  * stats being delayed for long, because transactions are either
3742  * sent as a whole on commit or streamed. Streamed transactions
3743  * are spilled to disk and applied on commit.
3744  */
3745  if (!IsTransactionState())
3746  pgstat_report_stat(true);
3747  }
3748  }
3749 
3750  /* Pop the error context stack */
3751  error_context_stack = errcallback.previous;
3753 
3754  /* All done */
3756 }
static void UpdateWorkerStats(XLogRecPtr last_lsn, TimestampTz send_time, bool reply)
Definition: worker.c:3482
#define NAPTIME_PER_CYCLE
Definition: worker.c:213
ErrorContextCallback * apply_error_context_stack
Definition: worker.c:306
static void send_feedback(XLogRecPtr recvpos, bool force, bool requestReply)
Definition: worker.c:3765
WalReceiverConn * LogRepWorkerWalRcvConn
Definition: worker.c:314
void apply_error_callback(void *arg)
Definition: worker.c:4893
static MemoryContext LogicalStreamingContext
Definition: worker.c:312
TimestampTz GetCurrentTimestamp(void)
Definition: timestamp.c:1583
Datum now(PG_FUNCTION_ARGS)
Definition: timestamp.c:1547
int64 TimestampTz
Definition: timestamp.h:39
ErrorContextCallback * error_context_stack
Definition: elog.c:95
struct Latch * MyLatch
Definition: globals.c:58
@ PGC_SIGHUP
Definition: guc.h:71
void ProcessConfigFile(GucContext context)
volatile sig_atomic_t ConfigReloadPending
Definition: interrupt.c:27
int WaitLatchOrSocket(Latch *latch, int wakeEvents, pgsocket sock, long timeout, uint32 wait_event_info)
Definition: latch.c:538
void ResetLatch(Latch *latch)
Definition: latch.c:697
#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 MemoryContextResetAndDeleteChildren(ctx)
Definition: memutils.h:70
static char * buf
Definition: pg_test_fsync.c:67
int64 timestamp
int pgsocket
Definition: port.h:29
#define PGINVALID_SOCKET
Definition: port.h:31
int64 pq_getmsgint64(StringInfo msg)
Definition: pqformat.c:456
char * c
struct ErrorContextCallback * previous
Definition: elog.h:295
void(* callback)(void *arg)
Definition: elog.h:296
#define TimestampTzPlusMilliseconds(tz, ms)
Definition: timestamp.h:85
int wal_receiver_timeout
Definition: walreceiver.c:91
#define walrcv_endstreaming(conn, next_tli)
Definition: walreceiver.h:426
#define walrcv_receive(conn, buffer, wait_fd)
Definition: walreceiver.h:428
int WalWriterDelay
Definition: walwriter.c:70
uint32 TimeLineID
Definition: xlogdefs.h:59

References AcceptInvalidationMessages(), ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate, apply_dispatch(), apply_error_callback(), apply_error_context_stack, ApplyContext, ApplyMessageContext, buf, ErrorContextCallback::callback, CHECK_FOR_INTERRUPTS, ConfigReloadPending, StringInfoData::cursor, StringInfoData::data, dlist_is_empty(), ereport, errcode(), errmsg(), ERROR, error_context_stack, fd(), GetCurrentTimestamp(), in_remote_transaction, in_streamed_transaction, IsTransactionState(), StringInfoData::len, len, LOG, LogicalStreamingContext, LogRepWorkerWalRcvConn, lsn_mapping, StringInfoData::maxlen, maybe_reread_subscription(), MemoryContextReset(), MemoryContextResetAndDeleteChildren, MemoryContextSwitchTo(), MyLatch, NAPTIME_PER_CYCLE, now(), PGC_SIGHUP, PGINVALID_SOCKET, pgstat_report_activity(), pgstat_report_stat(), pq_getmsgbyte(), pq_getmsgint64(), ErrorContextCallback::previous, process_syncing_tables(), ProcessConfigFile(), ResetLatch(), send_feedback(), STATE_IDLE, TimestampTzPlusMilliseconds, TopMemoryContext, UpdateWorkerStats(), WaitLatchOrSocket(), wal_receiver_timeout, walrcv_endstreaming, walrcv_receive, WalWriterDelay, WL_EXIT_ON_PM_DEATH, WL_LATCH_SET, WL_SOCKET_READABLE, and WL_TIMEOUT.

Referenced by start_apply().

◆ LogicalRepWorkersWakeupAtCommit()

void LogicalRepWorkersWakeupAtCommit ( Oid  subid)

◆ maybe_reread_subscription()

void maybe_reread_subscription ( void  )

Definition at line 3885 of file worker.c.

3886 {
3887  MemoryContext oldctx;
3889  bool started_tx = false;
3890 
3891  /* When cache state is valid there is nothing to do here. */
3892  if (MySubscriptionValid)
3893  return;
3894 
3895  /* This function might be called inside or outside of transaction. */
3896  if (!IsTransactionState())
3897  {
3899  started_tx = true;
3900  }
3901 
3902  /* Ensure allocations in permanent context. */
3904 
3906 
3907  /*
3908  * Exit if the subscription was removed. This normally should not happen
3909  * as the worker gets killed during DROP SUBSCRIPTION.
3910  */
3911  if (!newsub)
3912  {
3913  ereport(LOG,
3914  (errmsg("logical replication worker for subscription \"%s\" will stop because the subscription was removed",
3915  MySubscription->name)));
3916 
3917  /* Ensure we remove no-longer-useful entry for worker's start time */
3918  if (am_leader_apply_worker())
3920 
3921  proc_exit(0);
3922  }
3923 
3924  /* Exit if the subscription was disabled. */
3925  if (!newsub->enabled)
3926  {
3927  ereport(LOG,
3928  (errmsg("logical replication worker for subscription \"%s\" will stop because the subscription was disabled",
3929  MySubscription->name)));
3930 
3932  }
3933 
3934  /* !slotname should never happen when enabled is true. */
3935  Assert(newsub->slotname);
3936 
3937  /* two-phase should not be altered */
3938  Assert(newsub->twophasestate == MySubscription->twophasestate);
3939 
3940  /*
3941  * Exit if any parameter that affects the remote connection was changed.
3942  * The launcher will start a new worker but note that the parallel apply
3943  * worker won't restart if the streaming option's value is changed from
3944  * 'parallel' to any other value or the server decides not to stream the
3945  * in-progress transaction.
3946  */
3947  if (strcmp(newsub->conninfo, MySubscription->conninfo) != 0 ||
3948  strcmp(newsub->name, MySubscription->name) != 0 ||
3949  strcmp(newsub->slotname, MySubscription->slotname) != 0 ||
3950  newsub->binary != MySubscription->binary ||
3951  newsub->stream != MySubscription->stream ||
3952  newsub->passwordrequired != MySubscription->passwordrequired ||
3953  strcmp(newsub->origin, MySubscription->origin) != 0 ||
3954  newsub->owner != MySubscription->owner ||
3955  !equal(newsub->publications, MySubscription->publications))
3956  {
3958  ereport(LOG,
3959  (errmsg("logical replication parallel apply worker for subscription \"%s\" will stop because of a parameter change",
3960  MySubscription->name)));
3961  else
3962  ereport(LOG,
3963  (errmsg("logical replication worker for subscription \"%s\" will restart because of a parameter change",
3964  MySubscription->name)));
3965 
3967  }
3968 
3969  /* Check for other changes that should never happen too. */
3970  if (newsub->dbid != MySubscription->dbid)
3971  {
3972  elog(ERROR, "subscription %u changed unexpectedly",
3974  }
3975 
3976  /* Clean old subscription info and switch to new one. */
3979 
3980  MemoryContextSwitchTo(oldctx);
3981 
3982  /* Change synchronous commit according to the user's wishes */
3983  SetConfigOption("synchronous_commit", MySubscription->synccommit,
3985 
3986  if (started_tx)
3988 
3989  MySubscriptionValid = true;
3990 }
bool equal(const void *a, const void *b)
Definition: equalfuncs.c:223
void FreeSubscription(Subscription *sub)
static color newsub(struct colormap *cm, color co)
Definition: regc_color.c:389

References am_leader_apply_worker(), am_parallel_apply_worker(), apply_worker_exit(), ApplyContext, ApplyLauncherForgetWorkerStartTime(), Assert(), Subscription::binary, CommitTransactionCommand(), Subscription::conninfo, Subscription::dbid, elog(), equal(), ereport, errmsg(), ERROR, FreeSubscription(), GetSubscription(), IsTransactionState(), LOG, MemoryContextSwitchTo(), MyLogicalRepWorker, MySubscription, MySubscriptionValid, Subscription::name, newsub(), Subscription::origin, Subscription::owner, Subscription::passwordrequired, PGC_BACKEND, PGC_S_OVERRIDE, proc_exit(), Subscription::publications, SetConfigOption(), Subscription::slotname, StartTransactionCommand(), Subscription::stream, LogicalRepWorker::subid, Subscription::synccommit, and Subscription::twophasestate.

Referenced by apply_handle_commit_internal(), begin_replication_step(), LogicalRepApplyLoop(), and pa_can_start().

◆ maybe_start_skipping_changes()

static void maybe_start_skipping_changes ( XLogRecPtr  finish_lsn)
static

Definition at line 4763 of file worker.c.

4764 {
4768 
4769  /*
4770  * Quick return if it's not requested to skip this transaction. This
4771  * function is called for every remote transaction and we assume that
4772  * skipping the transaction is not used often.
4773  */
4775  MySubscription->skiplsn != finish_lsn))
4776  return;
4777 
4778  /* Start skipping all changes of this transaction */
4779  skip_xact_finish_lsn = finish_lsn;
4780 
4781  ereport(LOG,
4782  errmsg("logical replication starts skipping transaction at LSN %X/%X",
4784 }
static XLogRecPtr skip_xact_finish_lsn
Definition: worker.c:353

References Assert(), ereport, errmsg(), in_remote_transaction, in_streamed_transaction, is_skipping_changes, likely, LOG, LSN_FORMAT_ARGS, MySubscription, skip_xact_finish_lsn, Subscription::skiplsn, and XLogRecPtrIsInvalid.

Referenced by apply_handle_begin(), apply_handle_begin_prepare(), and apply_spooled_messages().

◆ ReplicationOriginNameForLogicalRep()

void ReplicationOriginNameForLogicalRep ( Oid  suboid,
Oid  relid,
char *  originname,
Size  szoriginname 
)

Definition at line 446 of file worker.c.

448 {
449  if (OidIsValid(relid))
450  {
451  /* Replication origin name for tablesync workers. */
452  snprintf(originname, szoriginname, "pg_%u_%u", suboid, relid);
453  }
454  else
455  {
456  /* Replication origin name for non-tablesync workers. */
457  snprintf(originname, szoriginname, "pg_%u", suboid);
458  }
459 }

References OidIsValid, and snprintf.

Referenced by AlterSubscription(), AlterSubscription_refresh(), CreateSubscription(), DropSubscription(), LogicalRepSyncTableStart(), ParallelApplyWorkerMain(), process_syncing_tables_for_apply(), process_syncing_tables_for_sync(), run_apply_worker(), and run_tablesync_worker().

◆ reset_apply_error_context_info()

◆ run_apply_worker()

static void run_apply_worker ( )
static

Definition at line 4462 of file worker.c.

4463 {
4464  char originname[NAMEDATALEN];
4465  XLogRecPtr origin_startpos = InvalidXLogRecPtr;
4466  char *slotname = NULL;
4468  RepOriginId originid;
4469  TimeLineID startpointTLI;
4470  char *