worker.c File Reference

#include "postgres.h"
#include <sys/stat.h>
#include <unistd.h>
#include "access/genam.h"
#include "access/commit_ts.h"
#include "access/table.h"
#include "access/tableam.h"
#include "access/tupconvert.h"
#include "access/twophase.h"
#include "access/xact.h"
#include "catalog/indexing.h"
#include "catalog/pg_inherits.h"
#include "catalog/pg_subscription.h"
#include "catalog/pg_subscription_rel.h"
#include "commands/subscriptioncmds.h"
#include "commands/tablecmds.h"
#include "commands/trigger.h"
#include "executor/executor.h"
#include "executor/execPartition.h"
#include "libpq/pqformat.h"
#include "miscadmin.h"
#include "optimizer/optimizer.h"
#include "parser/parse_relation.h"
#include "pgstat.h"
#include "port/pg_bitutils.h"
#include "postmaster/bgworker.h"
#include "postmaster/interrupt.h"
#include "postmaster/walwriter.h"
#include "replication/conflict.h"
#include "replication/logicallauncher.h"
#include "replication/logicalproto.h"
#include "replication/logicalrelation.h"
#include "replication/logicalworker.h"
#include "replication/origin.h"
#include "replication/slot.h"
#include "replication/walreceiver.h"
#include "replication/worker_internal.h"
#include "rewrite/rewriteHandler.h"
#include "storage/buffile.h"
#include "storage/ipc.h"
#include "storage/latch.h"
#include "storage/lmgr.h"
#include "storage/procarray.h"
#include "tcop/tcopprot.h"
#include "utils/acl.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/snapmgr.h"
#include "utils/syscache.h"
#include "utils/usercontext.h"
#include "utils/wait_event.h"

Include dependency graph for worker.c:

Go to the source code of this file.

Data Structures
struct	FlushPosition
struct	ApplyExecutionData
struct	ApplyErrorCallbackArg
struct	RetainDeadTuplesData
struct	SubXactInfo
struct	ApplySubXactData

Macros
#define	NAPTIME_PER_CYCLE   1000 /* max sleep time between cycles (1s) */
#define	MIN_XID_ADVANCE_INTERVAL   100
#define	MAX_XID_ADVANCE_INTERVAL   180000
#define	is_skipping_changes()   (unlikely(XLogRecPtrIsValid(skip_xact_finish_lsn)))

Typedefs
typedef struct FlushPosition	FlushPosition
typedef struct ApplyExecutionData	ApplyExecutionData
typedef struct ApplyErrorCallbackArg	ApplyErrorCallbackArg
typedef struct RetainDeadTuplesData	RetainDeadTuplesData
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 }
enum	RetainDeadTuplesPhase {   RDT_GET_CANDIDATE_XID , RDT_REQUEST_PUBLISHER_STATUS , RDT_WAIT_FOR_PUBLISHER_STATUS , RDT_WAIT_FOR_LOCAL_FLUSH ,   RDT_STOP_CONFLICT_INFO_RETENTION , RDT_RESUME_CONFLICT_INFO_RETENTION }

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	maybe_advance_nonremovable_xid (RetainDeadTuplesData *rdt_data, bool status_received)
static bool	can_advance_nonremovable_xid (RetainDeadTuplesData *rdt_data)
static void	process_rdt_phase_transition (RetainDeadTuplesData *rdt_data, bool status_received)
static void	get_candidate_xid (RetainDeadTuplesData *rdt_data)
static void	request_publisher_status (RetainDeadTuplesData *rdt_data)
static void	wait_for_publisher_status (RetainDeadTuplesData *rdt_data, bool status_received)
static void	wait_for_local_flush (RetainDeadTuplesData *rdt_data)
static bool	should_stop_conflict_info_retention (RetainDeadTuplesData *rdt_data)
static void	stop_conflict_info_retention (RetainDeadTuplesData *rdt_data)
static void	resume_conflict_info_retention (RetainDeadTuplesData *rdt_data)
static bool	update_retention_status (bool active)
static void	reset_retention_data_fields (RetainDeadTuplesData *rdt_data)
static void	adjust_xid_advance_interval (RetainDeadTuplesData *rdt_data, bool new_xid_found)
static void	apply_worker_exit (void)
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 bool	FindDeletedTupleInLocalRel (Relation localrel, Oid localidxoid, TupleTableSlot *remoteslot, TransactionId *delete_xid, ReplOriginId *delete_origin, TimestampTz *delete_time)
static void	apply_handle_tuple_routing (ApplyExecutionData *edata, TupleTableSlot *remoteslot, LogicalRepTupleData *newtup, CmdType operation)
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)
static void	set_wal_receiver_timeout (void)
static void	on_exit_clear_xact_state (int code, Datum arg)
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 ApplyExecutionData *	create_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, pgoff_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 bool	IsIndexUsableForFindingDeletedTuple (Oid localindexoid, TransactionId conflict_detection_xmin)
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)
void	maybe_reread_subscription (void)
static void	subscription_change_cb (Datum arg, SysCacheIdentifier 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)
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)
static ApplyErrorCallbackArg	apply_error_callback_arg
ErrorContextCallback *	apply_error_context_stack = NULL
MemoryContext	ApplyMessageContext = NULL
MemoryContext	ApplyContext = NULL
static MemoryContext	LogicalStreamingContext = NULL
WalReceiverConn *	LogRepWorkerWalRcvConn = NULL
Subscription *	MySubscription = NULL
static bool	MySubscriptionValid = false
static List *	on_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 BufFile *	stream_fd = NULL
static XLogRecPtr	last_flushpos = InvalidXLogRecPtr
static ApplySubXactData	subxact_data = {0, 0, InvalidTransactionId, NULL}

Macro Definition Documentation

is_skipping_changes

#define is_skipping_changes

(

)

(unlikely(XLogRecPtrIsValid(skip_xact_finish_lsn)))

Definition at line 522 of file worker.c.

MAX_XID_ADVANCE_INTERVAL

#define MAX_XID_ADVANCE_INTERVAL   180000

Definition at line 461 of file worker.c.

MIN_XID_ADVANCE_INTERVAL

#define MIN_XID_ADVANCE_INTERVAL   100

Definition at line 460 of file worker.c.

NAPTIME_PER_CYCLE

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

Definition at line 304 of file worker.c.

Typedef Documentation

ApplyErrorCallbackArg

typedef struct ApplyErrorCallbackArg ApplyErrorCallbackArg

ApplyExecutionData

typedef struct ApplyExecutionData ApplyExecutionData

ApplySubXactData

typedef struct ApplySubXactData ApplySubXactData

FlushPosition

typedef struct FlushPosition FlushPosition

RetainDeadTuplesData

typedef struct RetainDeadTuplesData RetainDeadTuplesData

SubXactInfo

typedef struct SubXactInfo SubXactInfo

Enumeration Type Documentation

RetainDeadTuplesPhase

enum RetainDeadTuplesPhase

Enumerator
RDT_GET_CANDIDATE_XID
RDT_REQUEST_PUBLISHER_STATUS
RDT_WAIT_FOR_PUBLISHER_STATUS
RDT_WAIT_FOR_LOCAL_FLUSH
RDT_STOP_CONFLICT_INFO_RETENTION
RDT_RESUME_CONFLICT_INFO_RETENTION

Definition at line 392 of file worker.c.

{
    RDT_GET_CANDIDATE_XID,
    RDT_REQUEST_PUBLISHER_STATUS,
    RDT_WAIT_FOR_PUBLISHER_STATUS,
    RDT_WAIT_FOR_LOCAL_FLUSH,
    RDT_STOP_CONFLICT_INFO_RETENTION,
    RDT_RESUME_CONFLICT_INFO_RETENTION,
} RetainDeadTuplesPhase;

TransApplyAction

enum TransApplyAction

Enumerator
TRANS_LEADER_APPLY
TRANS_LEADER_SERIALIZE
TRANS_LEADER_SEND_TO_PARALLEL
TRANS_LEADER_PARTIAL_SERIALIZE
TRANS_PARALLEL_APPLY

Definition at line 374 of file worker.c.

{
    /* The action for non-streaming transactions. */
    TRANS_LEADER_APPLY,
 
    /* Actions for streaming transactions. */
    TRANS_LEADER_SERIALIZE,
    TRANS_LEADER_SEND_TO_PARALLEL,
    TRANS_LEADER_PARTIAL_SERIALIZE,
    TRANS_PARALLEL_APPLY,
} TransApplyAction;

Function Documentation

adjust_xid_advance_interval()

static void adjust_xid_advance_interval ( RetainDeadTuplesData *  rdt_data, bool  new_xid_found  )

static

Definition at line 4977 of file worker.c.

{
    if (rdt_data->xid_advance_interval && !new_xid_found)
    {
        int         max_interval = wal_receiver_status_interval
            ? wal_receiver_status_interval * 1000
            : MAX_XID_ADVANCE_INTERVAL;
 
        /*
         * No new transaction ID has been assigned since the last check, so
         * double the interval, but not beyond the maximum allowable value.
         */
        rdt_data->xid_advance_interval = Min(rdt_data->xid_advance_interval * 2,
                                             max_interval);
    }
    else if (rdt_data->xid_advance_interval &&
             !MySubscription->retentionactive)
    {
        /*
         * Retention has been stopped, so double the interval-capped at a
         * maximum of 3 minutes. The wal_receiver_status_interval is
         * intentionally not used as an upper bound, since the likelihood of
         * retention resuming is lower than that of general activity resuming.
         */
        rdt_data->xid_advance_interval = Min(rdt_data->xid_advance_interval * 2,
                                             MAX_XID_ADVANCE_INTERVAL);
    }
    else
    {
        /*
         * A new transaction ID was found or the interval is not yet
         * initialized, so set the interval to the minimum value.
         */
        rdt_data->xid_advance_interval = MIN_XID_ADVANCE_INTERVAL;
    }
 
    /*
     * Ensure the wait time remains within the maximum retention time limit
     * when retention is active.  Skip this cap when maxretention is zero,
     * which means unlimited retention (no timeout).
     */
    if (MySubscription->retentionactive && MySubscription->maxretention > 0)
        rdt_data->xid_advance_interval = Min(rdt_data->xid_advance_interval,
                                             MySubscription->maxretention);
}

References fb(), MAX_XID_ADVANCE_INTERVAL, Subscription::maxretention, Min, MIN_XID_ADVANCE_INTERVAL, MySubscription, Subscription::retentionactive, and wal_receiver_status_interval.

Referenced by get_candidate_xid().

apply_dispatch()

void apply_dispatch

(

StringInfo

s

)

Definition at line 3797 of file worker.c.

{
    LogicalRepMsgType action = pq_getmsgbyte(s);
    LogicalRepMsgType saved_command;
 
    /*
     * Set the current command being applied. Since this function can be
     * called recursively when applying spooled changes, save the current
     * command.
     */
    saved_command = apply_error_callback_arg.command;
    apply_error_callback_arg.command = action;
 
    switch (action)
    {
        case LOGICAL_REP_MSG_BEGIN:
            apply_handle_begin(s);
            break;
 
        case LOGICAL_REP_MSG_COMMIT:
            apply_handle_commit(s);
            break;
 
        case LOGICAL_REP_MSG_INSERT:
            apply_handle_insert(s);
            break;
 
        case LOGICAL_REP_MSG_UPDATE:
            apply_handle_update(s);
            break;
 
        case LOGICAL_REP_MSG_DELETE:
            apply_handle_delete(s);
            break;
 
        case LOGICAL_REP_MSG_TRUNCATE:
            apply_handle_truncate(s);
            break;
 
        case LOGICAL_REP_MSG_RELATION:
            apply_handle_relation(s);
            break;
 
        case LOGICAL_REP_MSG_TYPE:
            apply_handle_type(s);
            break;
 
        case LOGICAL_REP_MSG_ORIGIN:
            apply_handle_origin(s);
            break;
 
        case LOGICAL_REP_MSG_MESSAGE:
 
            /*
             * Logical replication does not use generic logical messages yet.
             * Although, it could be used by other applications that use this
             * output plugin.
             */
            break;
 
        case LOGICAL_REP_MSG_STREAM_START:
            apply_handle_stream_start(s);
            break;
 
        case LOGICAL_REP_MSG_STREAM_STOP:
            apply_handle_stream_stop(s);
            break;
 
        case LOGICAL_REP_MSG_STREAM_ABORT:
            apply_handle_stream_abort(s);
            break;
 
        case LOGICAL_REP_MSG_STREAM_COMMIT:
            apply_handle_stream_commit(s);
            break;
 
        case LOGICAL_REP_MSG_BEGIN_PREPARE:
            apply_handle_begin_prepare(s);
            break;
 
        case LOGICAL_REP_MSG_PREPARE:
            apply_handle_prepare(s);
            break;
 
        case LOGICAL_REP_MSG_COMMIT_PREPARED:
            apply_handle_commit_prepared(s);
            break;
 
        case LOGICAL_REP_MSG_ROLLBACK_PREPARED:
            apply_handle_rollback_prepared(s);
            break;
 
        case LOGICAL_REP_MSG_STREAM_PREPARE:
            apply_handle_stream_prepare(s);
            break;
 
        default:
            ereport(ERROR,
                    (errcode(ERRCODE_PROTOCOL_VIOLATION),
                     errmsg("invalid logical replication message type \"??? (%d)\"", action)));
    }
 
    /* Reset the current command */
    apply_error_callback_arg.command = saved_command;
}

References 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(), ERRCODE_PROTOCOL_VIOLATION, errmsg, ERROR, fb(), 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 6238 of file worker.c.

{
    ApplyErrorCallbackArg *errarg = &apply_error_callback_arg;
 
    if (apply_error_callback_arg.command == 0)
        return;
 
    Assert(errarg->origin_name);
 
    if (errarg->rel == NULL)
    {
        if (!TransactionIdIsValid(errarg->remote_xid))
            errcontext("processing remote data for replication origin \"%s\" during message type \"%s\"",
                       errarg->origin_name,
                       logicalrep_message_type(errarg->command));
        else if (!XLogRecPtrIsValid(errarg->finish_lsn))
            errcontext("processing remote data for replication origin \"%s\" during message type \"%s\" in transaction %u",
                       errarg->origin_name,
                       logicalrep_message_type(errarg->command),
                       errarg->remote_xid);
        else
            errcontext("processing remote data for replication origin \"%s\" during message type \"%s\" in transaction %u, finished at %X/%08X",
                       errarg->origin_name,
                       logicalrep_message_type(errarg->command),
                       errarg->remote_xid,
                       LSN_FORMAT_ARGS(errarg->finish_lsn));
    }
    else
    {
        if (errarg->remote_attnum < 0)
        {
            if (!XLogRecPtrIsValid(errarg->finish_lsn))
                errcontext("processing remote data for replication origin \"%s\" during message type \"%s\" for replication target relation \"%s.%s\" in transaction %u",
                           errarg->origin_name,
                           logicalrep_message_type(errarg->command),
                           errarg->rel->remoterel.nspname,
                           errarg->rel->remoterel.relname,
                           errarg->remote_xid);
            else
                errcontext("processing remote data for replication origin \"%s\" during message type \"%s\" for replication target relation \"%s.%s\" in transaction %u, finished at %X/%08X",
                           errarg->origin_name,
                           logicalrep_message_type(errarg->command),
                           errarg->rel->remoterel.nspname,
                           errarg->rel->remoterel.relname,
                           errarg->remote_xid,
                           LSN_FORMAT_ARGS(errarg->finish_lsn));
        }
        else
        {
            if (!XLogRecPtrIsValid(errarg->finish_lsn))
                errcontext("processing remote data for replication origin \"%s\" during message type \"%s\" for replication target relation \"%s.%s\" column \"%s\" in transaction %u",
                           errarg->origin_name,
                           logicalrep_message_type(errarg->command),
                           errarg->rel->remoterel.nspname,
                           errarg->rel->remoterel.relname,
                           errarg->rel->remoterel.attnames[errarg->remote_attnum],
                           errarg->remote_xid);
            else
                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/%08X",
                           errarg->origin_name,
                           logicalrep_message_type(errarg->command),
                           errarg->rel->remoterel.nspname,
                           errarg->rel->remoterel.relname,
                           errarg->rel->remoterel.attnames[errarg->remote_attnum],
                           errarg->remote_xid,
                           LSN_FORMAT_ARGS(errarg->finish_lsn));
        }
    }
}

References apply_error_callback_arg, Assert, ApplyErrorCallbackArg::command, errcontext, fb(), logicalrep_message_type(), LSN_FORMAT_ARGS, TransactionIdIsValid, and XLogRecPtrIsValid.

Referenced by LogicalParallelApplyLoop(), and LogicalRepApplyLoop().

apply_handle_begin()

static void apply_handle_begin ( StringInfo  s )

static

Definition at line 1228 of file worker.c.

{
    LogicalRepBeginData begin_data;
 
    /* There must not be an active streaming transaction. */
    Assert(!TransactionIdIsValid(stream_xid));
 
    logicalrep_read_begin(s, &begin_data);
    set_apply_error_context_xact(begin_data.xid, begin_data.final_lsn);
 
    remote_final_lsn = begin_data.final_lsn;
 
    maybe_start_skipping_changes(begin_data.final_lsn);
 
    in_remote_transaction = true;
 
    pgstat_report_activity(STATE_RUNNING, NULL);
}

References Assert, fb(), 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, and TransactionIdIsValid.

Referenced by apply_dispatch().

apply_handle_begin_prepare()

static void apply_handle_begin_prepare ( StringInfo  s )

static

Definition at line 1282 of file worker.c.

{
    LogicalRepPreparedTxnData begin_data;
 
    /* Tablesync should never receive prepare. */
    if (am_tablesync_worker())
        ereport(ERROR,
                (errcode(ERRCODE_PROTOCOL_VIOLATION),
                 errmsg_internal("tablesync worker received a BEGIN PREPARE message")));
 
    /* There must not be an active streaming transaction. */
    Assert(!TransactionIdIsValid(stream_xid));
 
    logicalrep_read_begin_prepare(s, &begin_data);
    set_apply_error_context_xact(begin_data.xid, begin_data.prepare_lsn);
 
    remote_final_lsn = begin_data.prepare_lsn;
 
    maybe_start_skipping_changes(begin_data.prepare_lsn);
 
    in_remote_transaction = true;
 
    pgstat_report_activity(STATE_RUNNING, NULL);
}

References am_tablesync_worker(), Assert, ereport, errcode(), ERRCODE_PROTOCOL_VIOLATION, errmsg_internal(), ERROR, fb(), in_remote_transaction, logicalrep_read_begin_prepare(), maybe_start_skipping_changes(), pgstat_report_activity(), remote_final_lsn, set_apply_error_context_xact(), STATE_RUNNING, stream_xid, and TransactionIdIsValid.

Referenced by apply_dispatch().

apply_handle_commit()

static void apply_handle_commit ( StringInfo  s )

static

Definition at line 1253 of file worker.c.

{
    LogicalRepCommitData commit_data;
 
    logicalrep_read_commit(s, &commit_data);
 
    if (commit_data.commit_lsn != remote_final_lsn)
        ereport(ERROR,
                (errcode(ERRCODE_PROTOCOL_VIOLATION),
                 errmsg_internal("incorrect commit LSN %X/%08X in commit message (expected %X/%08X)",
                                 LSN_FORMAT_ARGS(commit_data.commit_lsn),
                                 LSN_FORMAT_ARGS(remote_final_lsn))));
 
    apply_handle_commit_internal(&commit_data);
 
    /*
     * Process any tables that are being synchronized in parallel, as well as
     * any newly added tables or sequences.
     */
    ProcessSyncingRelations(commit_data.end_lsn);
 
    pgstat_report_activity(STATE_IDLE, NULL);
    reset_apply_error_context_info();
}

References apply_handle_commit_internal(), ereport, errcode(), ERRCODE_PROTOCOL_VIOLATION, errmsg_internal(), ERROR, fb(), logicalrep_read_commit(), LSN_FORMAT_ARGS, pgstat_report_activity(), ProcessSyncingRelations(), 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 2520 of file worker.c.

{
    if (is_skipping_changes())
    {
        stop_skipping_changes();
 
        /*
         * Start a new transaction to clear the subskiplsn, if not started
         * yet.
         */
        if (!IsTransactionState())
            StartTransactionCommand();
    }
 
    if (IsTransactionState())
    {
        /*
         * The transaction is either non-empty or skipped, so we clear the
         * subskiplsn.
         */
        clear_subscription_skip_lsn(commit_data->commit_lsn);
 
        /*
         * Update origin state so we can restart streaming from correct
         * position in case of crash.
         */
        replorigin_xact_state.origin_lsn = commit_data->end_lsn;
        replorigin_xact_state.origin_timestamp = commit_data->committime;
 
        CommitTransactionCommand();
 
        if (IsTransactionBlock())
        {
            EndTransactionBlock(false);
            CommitTransactionCommand();
        }
 
        pgstat_report_stat(false);
 
        store_flush_position(commit_data->end_lsn, XactLastCommitEnd);
    }
    else
    {
        /* Process any invalidation messages that might have accumulated. */
        AcceptInvalidationMessages();
        maybe_reread_subscription();
    }
 
    in_remote_transaction = false;
}

References AcceptInvalidationMessages(), clear_subscription_skip_lsn(), CommitTransactionCommand(), EndTransactionBlock(), fb(), in_remote_transaction, is_skipping_changes, IsTransactionBlock(), IsTransactionState(), maybe_reread_subscription(), ReplOriginXactState::origin_lsn, ReplOriginXactState::origin_timestamp, pgstat_report_stat(), replorigin_xact_state, 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 1422 of file worker.c.

{
    LogicalRepCommitPreparedTxnData prepare_data;
    char        gid[GIDSIZE];
 
    logicalrep_read_commit_prepared(s, &prepare_data);
    set_apply_error_context_xact(prepare_data.xid, prepare_data.commit_lsn);
 
    /* Compute GID for two_phase transactions. */
    TwoPhaseTransactionGid(MySubscription->oid, prepare_data.xid,
                           gid, sizeof(gid));
 
    /* There is no transaction when COMMIT PREPARED is called */
    begin_replication_step();
 
    /*
     * Update origin state so we can restart streaming from correct position
     * in case of crash.
     */
    replorigin_xact_state.origin_lsn = prepare_data.end_lsn;
    replorigin_xact_state.origin_timestamp = prepare_data.commit_time;
 
    FinishPreparedTransaction(gid, true);
    end_replication_step();
    CommitTransactionCommand();
    pgstat_report_stat(false);
 
    store_flush_position(prepare_data.end_lsn, XactLastCommitEnd);
    in_remote_transaction = false;
 
    /*
     * Process any tables that are being synchronized in parallel, as well as
     * any newly added tables or sequences.
     */
    ProcessSyncingRelations(prepare_data.end_lsn);
 
    clear_subscription_skip_lsn(prepare_data.end_lsn);
 
    pgstat_report_activity(STATE_IDLE, NULL);
    reset_apply_error_context_info();
}

References begin_replication_step(), clear_subscription_skip_lsn(), CommitTransactionCommand(), end_replication_step(), fb(), FinishPreparedTransaction(), GIDSIZE, in_remote_transaction, logicalrep_read_commit_prepared(), MySubscription, Subscription::oid, ReplOriginXactState::origin_lsn, ReplOriginXactState::origin_timestamp, pgstat_report_activity(), pgstat_report_stat(), ProcessSyncingRelations(), replorigin_xact_state, reset_apply_error_context_info(), set_apply_error_context_xact(), STATE_IDLE, store_flush_position(), TwoPhaseTransactionGid(), and XactLastCommitEnd.

Referenced by apply_dispatch().

apply_handle_delete()

static void apply_handle_delete ( StringInfo  s )

static

Definition at line 3034 of file worker.c.

{
    LogicalRepRelMapEntry *rel;
    LogicalRepTupleData oldtup;
    LogicalRepRelId relid;
    UserContext ucxt;
    ApplyExecutionData *edata;
    EState     *estate;
    TupleTableSlot *remoteslot;
    MemoryContext oldctx;
    bool        run_as_owner;
 
    /*
     * Quick return if we are skipping data modification changes or handling
     * streamed transactions.
     */
    if (is_skipping_changes() ||
        handle_streamed_transaction(LOGICAL_REP_MSG_DELETE, s))
        return;
 
    begin_replication_step();
 
    relid = logicalrep_read_delete(s, &oldtup);
    rel = logicalrep_rel_open(relid, RowExclusiveLock);
    if (!should_apply_changes_for_rel(rel))
    {
        /*
         * The relation can't become interesting in the middle of the
         * transaction so it's safe to unlock it.
         */
        logicalrep_rel_close(rel, RowExclusiveLock);
        end_replication_step();
        return;
    }
 
    /* Set relation for error callback */
    apply_error_callback_arg.rel = rel;
 
    /* Check if we can do the delete. */
    check_relation_updatable(rel);
 
    /*
     * Make sure that any user-supplied code runs as the table owner, unless
     * the user has opted out of that behavior.
     */
    run_as_owner = MySubscription->runasowner;
    if (!run_as_owner)
        SwitchToUntrustedUser(rel->localrel->rd_rel->relowner, &ucxt);
 
    /* Initialize the executor state. */
    edata = create_edata_for_relation(rel);
    estate = edata->estate;
    remoteslot = ExecInitExtraTupleSlot(estate,
                                        RelationGetDescr(rel->localrel),
                                        &TTSOpsVirtual);
 
    /* Build the search tuple. */
    oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
    slot_store_data(remoteslot, rel, &oldtup);
    MemoryContextSwitchTo(oldctx);
 
    /* For a partitioned table, apply delete to correct partition. */
    if (rel->localrel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
        apply_handle_tuple_routing(edata,
                                   remoteslot, NULL, CMD_DELETE);
    else
    {
        ResultRelInfo *relinfo = edata->targetRelInfo;
 
        ExecOpenIndices(relinfo, false);
        apply_handle_delete_internal(edata, relinfo,
                                     remoteslot, rel->localindexoid);
        ExecCloseIndices(relinfo);
    }
 
    finish_edata(edata);
 
    /* Reset relation for error callback */
    apply_error_callback_arg.rel = NULL;
 
    if (!run_as_owner)
        RestoreUserContext(&ucxt);
 
    logicalrep_rel_close(rel, NoLock);
 
    end_replication_step();
}

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(), ExecCloseIndices(), ExecInitExtraTupleSlot(), ExecOpenIndices(), fb(), 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(), 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 3128 of file worker.c.

{
    EState     *estate = edata->estate;
    Relation    localrel = relinfo->ri_RelationDesc;
    LogicalRepRelation *remoterel = &edata->targetRel->remoterel;
    EPQState    epqstate;
    TupleTableSlot *localslot;
    ConflictTupleInfo conflicttuple = {0};
    bool        found;
 
    EvalPlanQualInit(&epqstate, estate, NULL, NIL, -1, NIL);
 
    /* Caller should have opened indexes already. */
    Assert(relinfo->ri_IndexRelationDescs != NULL ||
           !localrel->rd_rel->relhasindex ||
           RelationGetIndexList(localrel) == NIL);
 
    found = FindReplTupleInLocalRel(edata, localrel, remoterel, localindexoid,
                                    remoteslot, &localslot);
 
    /* If found delete it. */
    if (found)
    {
        /*
         * Report the conflict if the tuple was modified by a different
         * origin.
         */
        if (GetTupleTransactionInfo(localslot, &conflicttuple.xmin,
                                    &conflicttuple.origin, &conflicttuple.ts) &&
            conflicttuple.origin != replorigin_xact_state.origin)
        {
            conflicttuple.slot = localslot;
            ReportApplyConflict(estate, relinfo, LOG, CT_DELETE_ORIGIN_DIFFERS,
                                remoteslot, NULL,
                                list_make1(&conflicttuple));
        }
 
        EvalPlanQualSetSlot(&epqstate, localslot);
 
        /* Do the actual delete. */
        TargetPrivilegesCheck(relinfo->ri_RelationDesc, ACL_DELETE);
        ExecSimpleRelationDelete(relinfo, estate, &epqstate, localslot);
    }
    else
    {
        /*
         * The tuple to be deleted could not be found.  Do nothing except for
         * emitting a log message.
         */
        ReportApplyConflict(estate, relinfo, LOG, CT_DELETE_MISSING,
                            remoteslot, NULL, list_make1(&conflicttuple));
    }
 
    /* Cleanup. */
    EvalPlanQualEnd(&epqstate);
}

References ACL_DELETE, Assert, CT_DELETE_MISSING, CT_DELETE_ORIGIN_DIFFERS, EvalPlanQualEnd(), EvalPlanQualInit(), EvalPlanQualSetSlot, ExecSimpleRelationDelete(), fb(), FindReplTupleInLocalRel(), GetTupleTransactionInfo(), list_make1, LOG, NIL, ReplOriginXactState::origin, RelationData::rd_rel, RelationGetIndexList(), replorigin_xact_state, ReportApplyConflict(), and TargetPrivilegesCheck().

Referenced by apply_handle_delete(), and apply_handle_tuple_routing().

apply_handle_insert()

static void apply_handle_insert ( StringInfo  s )

static

Definition at line 2650 of file worker.c.

{
    LogicalRepRelMapEntry *rel;
    LogicalRepTupleData newtup;
    LogicalRepRelId relid;
    UserContext ucxt;
    ApplyExecutionData *edata;
    EState     *estate;
    TupleTableSlot *remoteslot;
    MemoryContext oldctx;
    bool        run_as_owner;
 
    /*
     * Quick return if we are skipping data modification changes or handling
     * streamed transactions.
     */
    if (is_skipping_changes() ||
        handle_streamed_transaction(LOGICAL_REP_MSG_INSERT, s))
        return;
 
    begin_replication_step();
 
    relid = logicalrep_read_insert(s, &newtup);
    rel = logicalrep_rel_open(relid, RowExclusiveLock);
    if (!should_apply_changes_for_rel(rel))
    {
        /*
         * The relation can't become interesting in the middle of the
         * transaction so it's safe to unlock it.
         */
        logicalrep_rel_close(rel, RowExclusiveLock);
        end_replication_step();
        return;
    }
 
    /*
     * Make sure that any user-supplied code runs as the table owner, unless
     * the user has opted out of that behavior.
     */
    run_as_owner = MySubscription->runasowner;
    if (!run_as_owner)
        SwitchToUntrustedUser(rel->localrel->rd_rel->relowner, &ucxt);
 
    /* Set relation for error callback */
    apply_error_callback_arg.rel = rel;
 
    /* Initialize the executor state. */
    edata = create_edata_for_relation(rel);
    estate = edata->estate;
    remoteslot = ExecInitExtraTupleSlot(estate,
                                        RelationGetDescr(rel->localrel),
                                        &TTSOpsVirtual);
 
    /* Process and store remote tuple in the slot */
    oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
    slot_store_data(remoteslot, rel, &newtup);
    slot_fill_defaults(rel, estate, remoteslot);
    MemoryContextSwitchTo(oldctx);
 
    /* For a partitioned table, insert the tuple into a partition. */
    if (rel->localrel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
        apply_handle_tuple_routing(edata,
                                   remoteslot, NULL, CMD_INSERT);
    else
    {
        ResultRelInfo *relinfo = edata->targetRelInfo;
 
        ExecOpenIndices(relinfo, false);
        apply_handle_insert_internal(edata, relinfo, remoteslot);
        ExecCloseIndices(relinfo);
    }
 
    finish_edata(edata);
 
    /* Reset relation for error callback */
    apply_error_callback_arg.rel = NULL;
 
    if (!run_as_owner)
        RestoreUserContext(&ucxt);
 
    logicalrep_rel_close(rel, NoLock);
 
    end_replication_step();
}

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(), ExecCloseIndices(), ExecInitExtraTupleSlot(), ExecOpenIndices(), fb(), 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(), 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 2741 of file worker.c.

{
    EState     *estate = edata->estate;
 
    /* Caller should have opened indexes already. */
    Assert(relinfo->ri_IndexRelationDescs != NULL ||
           !relinfo->ri_RelationDesc->rd_rel->relhasindex ||
           RelationGetIndexList(relinfo->ri_RelationDesc) == NIL);
 
    /* Caller will not have done this bit. */
    Assert(relinfo->ri_onConflictArbiterIndexes == NIL);
    InitConflictIndexes(relinfo);
 
    /* Do the insert. */
    TargetPrivilegesCheck(relinfo->ri_RelationDesc, ACL_INSERT);
    ExecSimpleRelationInsert(relinfo, estate, remoteslot);
}

References ACL_INSERT, Assert, ExecSimpleRelationInsert(), fb(), InitConflictIndexes(), NIL, RelationGetIndexList(), 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 1683 of file worker.c.

{
    /*
     * ORIGIN message can only come inside streaming transaction or inside
     * remote transaction and before any actual writes.
     */
    if (!in_streamed_transaction &&
        (!in_remote_transaction ||
         (IsTransactionState() && !am_tablesync_worker())))
        ereport(ERROR,
                (errcode(ERRCODE_PROTOCOL_VIOLATION),
                 errmsg_internal("ORIGIN message sent out of order")));
}

References am_tablesync_worker(), ereport, errcode(), ERRCODE_PROTOCOL_VIOLATION, 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 1348 of file worker.c.

{
    LogicalRepPreparedTxnData prepare_data;
 
    logicalrep_read_prepare(s, &prepare_data);
 
    if (prepare_data.prepare_lsn != remote_final_lsn)
        ereport(ERROR,
                (errcode(ERRCODE_PROTOCOL_VIOLATION),
                 errmsg_internal("incorrect prepare LSN %X/%08X in prepare message (expected %X/%08X)",
                                 LSN_FORMAT_ARGS(prepare_data.prepare_lsn),
                                 LSN_FORMAT_ARGS(remote_final_lsn))));
 
    /*
     * Unlike commit, here, we always prepare the transaction even though no
     * change has happened in this transaction or all changes are skipped. It
     * is done this way because at commit prepared time, we won't know whether
     * we have skipped preparing a transaction because of those reasons.
     *
     * XXX, We can optimize such that at commit prepared time, we first check
     * whether we have prepared the transaction or not but that doesn't seem
     * worthwhile because such cases shouldn't be common.
     */
    begin_replication_step();
 
    apply_handle_prepare_internal(&prepare_data);
 
    end_replication_step();
    CommitTransactionCommand();
    pgstat_report_stat(false);
 
    /*
     * It is okay not to set the local_end LSN for the prepare because we
     * always flush the prepare record. So, we can send the acknowledgment of
     * the remote_end LSN as soon as prepare is finished.
     *
     * XXX For the sake of consistency with commit, we could have set it with
     * the LSN of prepare but as of now we don't track that value similar to
     * XactLastCommitEnd, and adding it for this purpose doesn't seems worth
     * it.
     */
    store_flush_position(prepare_data.end_lsn, InvalidXLogRecPtr);
 
    in_remote_transaction = false;
 
    /*
     * Process any tables that are being synchronized in parallel, as well as
     * any newly added tables or sequences.
     */
    ProcessSyncingRelations(prepare_data.end_lsn);
 
    /*
     * Since we have already prepared the transaction, in a case where the
     * server crashes before clearing the subskiplsn, it will be left but the
     * transaction won't be resent. But that's okay because it's a rare case
     * and the subskiplsn will be cleared when finishing the next transaction.
     */
    stop_skipping_changes();
    clear_subscription_skip_lsn(prepare_data.prepare_lsn);
 
    pgstat_report_activity(STATE_IDLE, NULL);
    reset_apply_error_context_info();
}

References apply_handle_prepare_internal(), begin_replication_step(), clear_subscription_skip_lsn(), CommitTransactionCommand(), end_replication_step(), ereport, errcode(), ERRCODE_PROTOCOL_VIOLATION, errmsg_internal(), ERROR, fb(), in_remote_transaction, InvalidXLogRecPtr, logicalrep_read_prepare(), LSN_FORMAT_ARGS, pgstat_report_activity(), pgstat_report_stat(), ProcessSyncingRelations(), remote_final_lsn, reset_apply_error_context_info(), STATE_IDLE, stop_skipping_changes(), and store_flush_position().

Referenced by apply_dispatch().

apply_handle_prepare_internal()

static void apply_handle_prepare_internal ( LogicalRepPreparedTxnData *  prepare_data )

static

Definition at line 1311 of file worker.c.

{
    char        gid[GIDSIZE];
 
    /*
     * Compute unique GID for two_phase transactions. We don't use GID of
     * prepared transaction sent by server as that can lead to deadlock when
     * we have multiple subscriptions from same node point to publications on
     * the same node. See comments atop worker.c
     */
    TwoPhaseTransactionGid(MySubscription->oid, prepare_data->xid,
                           gid, sizeof(gid));
 
    /*
     * BeginTransactionBlock is necessary to balance the EndTransactionBlock
     * called within the PrepareTransactionBlock below.
     */
    if (!IsTransactionBlock())
    {
        BeginTransactionBlock();
        CommitTransactionCommand(); /* Completes the preceding Begin command. */
    }
 
    /*
     * Update origin state so we can restart streaming from correct position
     * in case of crash.
     */
    replorigin_xact_state.origin_lsn = prepare_data->end_lsn;
    replorigin_xact_state.origin_timestamp = prepare_data->prepare_time;
 
    PrepareTransactionBlock(gid);
}

References BeginTransactionBlock(), CommitTransactionCommand(), fb(), GIDSIZE, IsTransactionBlock(), MySubscription, Subscription::oid, ReplOriginXactState::origin_lsn, ReplOriginXactState::origin_timestamp, PrepareTransactionBlock(), replorigin_xact_state, and TwoPhaseTransactionGid().

Referenced by apply_handle_prepare(), and apply_handle_stream_prepare().

apply_handle_relation()

static void apply_handle_relation ( StringInfo  s )

static

Definition at line 2580 of file worker.c.

{
    LogicalRepRelation *rel;
 
    if (handle_streamed_transaction(LOGICAL_REP_MSG_RELATION, s))
        return;
 
    rel = logicalrep_read_rel(s);
    logicalrep_relmap_update(rel);
 
    /* Also reset all entries in the partition map that refer to remoterel. */
    logicalrep_partmap_reset_relmap(rel);
}

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 1474 of file worker.c.

{
    LogicalRepRollbackPreparedTxnData rollback_data;
    char        gid[GIDSIZE];
 
    logicalrep_read_rollback_prepared(s, &rollback_data);
    set_apply_error_context_xact(rollback_data.xid, rollback_data.rollback_end_lsn);
 
    /* Compute GID for two_phase transactions. */
    TwoPhaseTransactionGid(MySubscription->oid, rollback_data.xid,
                           gid, sizeof(gid));
 
    /*
     * It is possible that we haven't received prepare because it occurred
     * before walsender reached a consistent point or the two_phase was still
     * not enabled by that time, so in such cases, we need to skip rollback
     * prepared.
     */
    if (LookupGXact(gid, rollback_data.prepare_end_lsn,
                    rollback_data.prepare_time))
    {
        /*
         * Update origin state so we can restart streaming from correct
         * position in case of crash.
         */
        replorigin_xact_state.origin_lsn = rollback_data.rollback_end_lsn;
        replorigin_xact_state.origin_timestamp = rollback_data.rollback_time;
 
        /* There is no transaction when ABORT/ROLLBACK PREPARED is called */
        begin_replication_step();
        FinishPreparedTransaction(gid, false);
        end_replication_step();
        CommitTransactionCommand();
 
        clear_subscription_skip_lsn(rollback_data.rollback_end_lsn);
    }
 
    pgstat_report_stat(false);
 
    /*
     * It is okay not to set the local_end LSN for the rollback of prepared
     * transaction because we always flush the WAL record for it. See
     * apply_handle_prepare.
     */
    store_flush_position(rollback_data.rollback_end_lsn, InvalidXLogRecPtr);
    in_remote_transaction = false;
 
    /*
     * Process any tables that are being synchronized in parallel, as well as
     * any newly added tables or sequences.
     */
    ProcessSyncingRelations(rollback_data.rollback_end_lsn);
 
    pgstat_report_activity(STATE_IDLE, NULL);
    reset_apply_error_context_info();
}

References begin_replication_step(), clear_subscription_skip_lsn(), CommitTransactionCommand(), end_replication_step(), fb(), FinishPreparedTransaction(), GIDSIZE, in_remote_transaction, InvalidXLogRecPtr, logicalrep_read_rollback_prepared(), LookupGXact(), MySubscription, Subscription::oid, ReplOriginXactState::origin_lsn, ReplOriginXactState::origin_timestamp, pgstat_report_activity(), pgstat_report_stat(), ProcessSyncingRelations(), replorigin_xact_state, reset_apply_error_context_info(), set_apply_error_context_xact(), STATE_IDLE, store_flush_position(), and TwoPhaseTransactionGid().

Referenced by apply_dispatch().

apply_handle_stream_abort()

static void apply_handle_stream_abort ( StringInfo  s )

static

Definition at line 2088 of file worker.c.

{
    TransactionId xid;
    TransactionId subxid;
    LogicalRepStreamAbortData abort_data;
    ParallelApplyWorkerInfo *winfo;
    TransApplyAction apply_action;
 
    /* Save the message before it is consumed. */
    StringInfoData original_msg = *s;
    bool        toplevel_xact;
 
    if (in_streamed_transaction)
        ereport(ERROR,
                (errcode(ERRCODE_PROTOCOL_VIOLATION),
                 errmsg_internal("STREAM ABORT message without STREAM STOP")));
 
    /* We receive abort information only when we can apply in parallel. */
    logicalrep_read_stream_abort(s, &abort_data,
                                 MyLogicalRepWorker->parallel_apply);
 
    xid = abort_data.xid;
    subxid = abort_data.subxid;
    toplevel_xact = (xid == subxid);
 
    set_apply_error_context_xact(subxid, abort_data.abort_lsn);
 
    apply_action = get_transaction_apply_action(xid, &winfo);
 
    switch (apply_action)
    {
        case TRANS_LEADER_APPLY:
 
            /*
             * We are in the leader apply worker and the transaction has been
             * serialized to file.
             */
            stream_abort_internal(xid, subxid);
 
            elog(DEBUG1, "finished processing the STREAM ABORT command");
            break;
 
        case TRANS_LEADER_SEND_TO_PARALLEL:
            Assert(winfo);
 
            /*
             * For the case of aborting the subtransaction, we increment the
             * number of streaming blocks and take the lock again before
             * sending the STREAM_ABORT to ensure that the parallel apply
             * worker will wait on the lock for the next set of changes after
             * processing the STREAM_ABORT message if it is not already
             * waiting for STREAM_STOP message.
             *
             * It is important to perform this locking before sending the
             * STREAM_ABORT message so that the leader can hold the lock first
             * and the parallel apply worker will wait for the leader to
             * release the lock. This is the same as what we do in
             * apply_handle_stream_stop. See Locking Considerations atop
             * applyparallelworker.c.
             */
            if (!toplevel_xact)
            {
                pa_unlock_stream(xid, AccessExclusiveLock);
                pg_atomic_add_fetch_u32(&winfo->shared->pending_stream_count, 1);
                pa_lock_stream(xid, AccessExclusiveLock);
            }
 
            if (pa_send_data(winfo, s->len, s->data))
            {
                /*
                 * Unlike STREAM_COMMIT and STREAM_PREPARE, we don't need to
                 * wait here for the parallel apply worker to finish as that
                 * is not required to maintain the commit order and won't have
                 * the risk of failures due to transaction dependencies and
                 * deadlocks. However, it is possible that before the parallel
                 * worker finishes and we clear the worker info, the xid
                 * wraparound happens on the upstream and a new transaction
                 * with the same xid can appear and that can lead to duplicate
                 * entries in ParallelApplyTxnHash. Yet another problem could
                 * be that we may have serialized the changes in partial
                 * serialize mode and the file containing xact changes may
                 * already exist, and after xid wraparound trying to create
                 * the file for the same xid can lead to an error. To avoid
                 * these problems, we decide to wait for the aborts to finish.
                 *
                 * Note, it is okay to not update the flush location position
                 * for aborts as in worst case that means such a transaction
                 * won't be sent again after restart.
                 */
                if (toplevel_xact)
                    pa_xact_finish(winfo, InvalidXLogRecPtr);
 
                break;
            }
 
            /*
             * Switch to serialize mode when we are not able to send the
             * change to parallel apply worker.
             */
            pa_switch_to_partial_serialize(winfo, true);
 
            pg_fallthrough;
        case TRANS_LEADER_PARTIAL_SERIALIZE:
            Assert(winfo);
 
            /*
             * Parallel apply worker might have applied some changes, so write
             * the STREAM_ABORT message so that it can rollback the
             * subtransaction if needed.
             */
            stream_open_and_write_change(xid, LOGICAL_REP_MSG_STREAM_ABORT,
                                         &original_msg);
 
            if (toplevel_xact)
            {
                pa_set_fileset_state(winfo->shared, FS_SERIALIZE_DONE);
                pa_xact_finish(winfo, InvalidXLogRecPtr);
            }
            break;
 
        case TRANS_PARALLEL_APPLY:
 
            /*
             * If the parallel apply worker is applying spooled messages then
             * close the file before aborting.
             */
            if (toplevel_xact && stream_fd)
                stream_close_file();
 
            pa_stream_abort(&abort_data);
 
            /*
             * We need to wait after processing rollback to savepoint for the
             * next set of changes.
             *
             * We have a race condition here due to which we can start waiting
             * here when there are more chunk of streams in the queue. See
             * apply_handle_stream_stop.
             */
            if (!toplevel_xact)
                pa_decr_and_wait_stream_block();
 
            elog(DEBUG1, "finished processing the STREAM ABORT command");
            break;
 
        default:
            elog(ERROR, "unexpected apply action: %d", (int) apply_action);
            break;
    }
 
    reset_apply_error_context_info();
}

References AccessExclusiveLock, Assert, StringInfoData::data, DEBUG1, elog, ereport, errcode(), ERRCODE_PROTOCOL_VIOLATION, errmsg_internal(), ERROR, fb(), 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(), pg_fallthrough, 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(), TRANS_LEADER_APPLY, TRANS_LEADER_PARTIAL_SERIALIZE, TRANS_LEADER_SEND_TO_PARALLEL, and TRANS_PARALLEL_APPLY.

Referenced by apply_dispatch().

apply_handle_stream_commit()

static void apply_handle_stream_commit ( StringInfo  s )

static

Definition at line 2407 of file worker.c.

{
    TransactionId xid;
    LogicalRepCommitData commit_data;
    ParallelApplyWorkerInfo *winfo;
    TransApplyAction apply_action;
 
    /* Save the message before it is consumed. */
    StringInfoData original_msg = *s;
 
    if (in_streamed_transaction)
        ereport(ERROR,
                (errcode(ERRCODE_PROTOCOL_VIOLATION),
                 errmsg_internal("STREAM COMMIT message without STREAM STOP")));
 
    xid = logicalrep_read_stream_commit(s, &commit_data);
    set_apply_error_context_xact(xid, commit_data.commit_lsn);
 
    apply_action = get_transaction_apply_action(xid, &winfo);
 
    switch (apply_action)
    {
        case TRANS_LEADER_APPLY:
 
            /*
             * The transaction has been serialized to file, so replay all the
             * spooled operations.
             */
            apply_spooled_messages(MyLogicalRepWorker->stream_fileset, xid,
                                   commit_data.commit_lsn);
 
            apply_handle_commit_internal(&commit_data);
 
            /* Unlink the files with serialized changes and subxact info. */
            stream_cleanup_files(MyLogicalRepWorker->subid, xid);
 
            elog(DEBUG1, "finished processing the STREAM COMMIT command");
            break;
 
        case TRANS_LEADER_SEND_TO_PARALLEL:
            Assert(winfo);
 
            if (pa_send_data(winfo, s->len, s->data))
            {
                /* Finish processing the streaming transaction. */
                pa_xact_finish(winfo, commit_data.end_lsn);
                break;
            }
 
            /*
             * Switch to serialize mode when we are not able to send the
             * change to parallel apply worker.
             */
            pa_switch_to_partial_serialize(winfo, true);
 
            pg_fallthrough;
        case TRANS_LEADER_PARTIAL_SERIALIZE:
            Assert(winfo);
 
            stream_open_and_write_change(xid, LOGICAL_REP_MSG_STREAM_COMMIT,
                                         &original_msg);
 
            pa_set_fileset_state(winfo->shared, FS_SERIALIZE_DONE);
 
            /* Finish processing the streaming transaction. */
            pa_xact_finish(winfo, commit_data.end_lsn);
            break;
 
        case TRANS_PARALLEL_APPLY:
 
            /*
             * If the parallel apply worker is applying spooled messages then
             * close the file before committing.
             */
            if (stream_fd)
                stream_close_file();
 
            apply_handle_commit_internal(&commit_data);
 
            MyParallelShared->last_commit_end = XactLastCommitEnd;
 
            /*
             * It is important to set the transaction state as finished before
             * releasing the lock. See pa_wait_for_xact_finish.
             */
            pa_set_xact_state(MyParallelShared, PARALLEL_TRANS_FINISHED);
            pa_unlock_transaction(xid, AccessExclusiveLock);
 
            pa_reset_subtrans();
 
            elog(DEBUG1, "finished processing the STREAM COMMIT command");
            break;
 
        default:
            elog(ERROR, "unexpected apply action: %d", (int) apply_action);
            break;
    }
 
    /*
     * Process any tables that are being synchronized in parallel, as well as
     * any newly added tables or sequences.
     */
    ProcessSyncingRelations(commit_data.end_lsn);
 
    pgstat_report_activity(STATE_IDLE, NULL);
 
    reset_apply_error_context_info();
}

References AccessExclusiveLock, apply_handle_commit_internal(), apply_spooled_messages(), Assert, StringInfoData::data, DEBUG1, elog, ereport, errcode(), ERRCODE_PROTOCOL_VIOLATION, errmsg_internal(), ERROR, fb(), 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, pg_fallthrough, pgstat_report_activity(), ProcessSyncingRelations(), 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 1535 of file worker.c.

{
    LogicalRepPreparedTxnData prepare_data;
    ParallelApplyWorkerInfo *winfo;
    TransApplyAction apply_action;
 
    /* Save the message before it is consumed. */
    StringInfoData original_msg = *s;
 
    if (in_streamed_transaction)
        ereport(ERROR,
                (errcode(ERRCODE_PROTOCOL_VIOLATION),
                 errmsg_internal("STREAM PREPARE message without STREAM STOP")));
 
    /* Tablesync should never receive prepare. */
    if (am_tablesync_worker())
        ereport(ERROR,
                (errcode(ERRCODE_PROTOCOL_VIOLATION),
                 errmsg_internal("tablesync worker received a STREAM PREPARE message")));
 
    logicalrep_read_stream_prepare(s, &prepare_data);
    set_apply_error_context_xact(prepare_data.xid, prepare_data.prepare_lsn);
 
    apply_action = get_transaction_apply_action(prepare_data.xid, &winfo);
 
    switch (apply_action)
    {
        case TRANS_LEADER_APPLY:
 
            /*
             * The transaction has been serialized to file, so replay all the
             * spooled operations.
             */
            apply_spooled_messages(MyLogicalRepWorker->stream_fileset,
                                   prepare_data.xid, prepare_data.prepare_lsn);
 
            /* Mark the transaction as prepared. */
            apply_handle_prepare_internal(&prepare_data);
 
            CommitTransactionCommand();
 
            /*
             * It is okay not to set the local_end LSN for the prepare because
             * we always flush the prepare record. See apply_handle_prepare.
             */
            store_flush_position(prepare_data.end_lsn, InvalidXLogRecPtr);
 
            in_remote_transaction = false;
 
            /* Unlink the files with serialized changes and subxact info. */
            stream_cleanup_files(MyLogicalRepWorker->subid, prepare_data.xid);
 
            elog(DEBUG1, "finished processing the STREAM PREPARE command");
            break;
 
        case TRANS_LEADER_SEND_TO_PARALLEL:
            Assert(winfo);
 
            if (pa_send_data(winfo, s->len, s->data))
            {
                /* Finish processing the streaming transaction. */
                pa_xact_finish(winfo, prepare_data.end_lsn);
                break;
            }
 
            /*
             * Switch to serialize mode when we are not able to send the
             * change to parallel apply worker.
             */
            pa_switch_to_partial_serialize(winfo, true);
 
            pg_fallthrough;
        case TRANS_LEADER_PARTIAL_SERIALIZE:
            Assert(winfo);
 
            stream_open_and_write_change(prepare_data.xid,
                                         LOGICAL_REP_MSG_STREAM_PREPARE,
                                         &original_msg);
 
            pa_set_fileset_state(winfo->shared, FS_SERIALIZE_DONE);
 
            /* Finish processing the streaming transaction. */
            pa_xact_finish(winfo, prepare_data.end_lsn);
            break;
 
        case TRANS_PARALLEL_APPLY:
 
            /*
             * If the parallel apply worker is applying spooled messages then
             * close the file before preparing.
             */
            if (stream_fd)
                stream_close_file();
 
            begin_replication_step();
 
            /* Mark the transaction as prepared. */
            apply_handle_prepare_internal(&prepare_data);
 
            end_replication_step();
 
            CommitTransactionCommand();
 
            /*
             * It is okay not to set the local_end LSN for the prepare because
             * we always flush the prepare record. See apply_handle_prepare.
             */
            MyParallelShared->last_commit_end = InvalidXLogRecPtr;
 
            pa_set_xact_state(MyParallelShared, PARALLEL_TRANS_FINISHED);
            pa_unlock_transaction(MyParallelShared->xid, AccessExclusiveLock);
 
            pa_reset_subtrans();
 
            elog(DEBUG1, "finished processing the STREAM PREPARE command");
            break;
 
        default:
            elog(ERROR, "unexpected apply action: %d", (int) apply_action);
            break;
    }
 
    pgstat_report_stat(false);
 
    /*
     * Process any tables that are being synchronized in parallel, as well as
     * any newly added tables or sequences.
     */
    ProcessSyncingRelations(prepare_data.end_lsn);
 
    /*
     * Similar to prepare case, the subskiplsn could be left in a case of
     * server crash but it's okay. See the comments in apply_handle_prepare().
     */
    stop_skipping_changes();
    clear_subscription_skip_lsn(prepare_data.prepare_lsn);
 
    pgstat_report_activity(STATE_IDLE, NULL);
 
    reset_apply_error_context_info();
}

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, end_replication_step(), ereport, errcode(), ERRCODE_PROTOCOL_VIOLATION, errmsg_internal(), ERROR, fb(), FS_SERIALIZE_DONE, get_transaction_apply_action(), in_remote_transaction, in_streamed_transaction, InvalidXLogRecPtr, 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, pg_fallthrough, pgstat_report_activity(), pgstat_report_stat(), ProcessSyncingRelations(), 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, and ParallelApplyWorkerShared::xid.

Referenced by apply_dispatch().

apply_handle_stream_start()

static void apply_handle_stream_start ( StringInfo  s )

static

Definition at line 1742 of file worker.c.

{
    bool        first_segment;
    ParallelApplyWorkerInfo *winfo;
    TransApplyAction apply_action;
 
    /* Save the message before it is consumed. */
    StringInfoData original_msg = *s;
 
    if (in_streamed_transaction)
        ereport(ERROR,
                (errcode(ERRCODE_PROTOCOL_VIOLATION),
                 errmsg_internal("duplicate STREAM START message")));
 
    /* There must not be an active streaming transaction. */
    Assert(!TransactionIdIsValid(stream_xid));
 
    /* notify handle methods we're processing a remote transaction */
    in_streamed_transaction = true;
 
    /* extract XID of the top-level transaction */
    stream_xid = logicalrep_read_stream_start(s, &first_segment);
 
    if (!TransactionIdIsValid(stream_xid))
        ereport(ERROR,
                (errcode(ERRCODE_PROTOCOL_VIOLATION),
                 errmsg_internal("invalid transaction ID in streamed replication transaction")));
 
    set_apply_error_context_xact(stream_xid, InvalidXLogRecPtr);
 
    /* Try to allocate a worker for the streaming transaction. */
    if (first_segment)
        pa_allocate_worker(stream_xid);
 
    apply_action = get_transaction_apply_action(stream_xid, &winfo);
 
    switch (apply_action)
    {
        case TRANS_LEADER_SERIALIZE:
 
            /*
             * Function stream_start_internal starts a transaction. This
             * transaction will be committed on the stream stop unless it is a
             * tablesync worker in which case it will be committed after
             * processing all the messages. We need this transaction for
             * handling the BufFile, used for serializing the streaming data
             * and subxact info.
             */
            stream_start_internal(stream_xid, first_segment);
            break;
 
        case TRANS_LEADER_SEND_TO_PARALLEL:
            Assert(winfo);
 
            /*
             * Once we start serializing the changes, the parallel apply
             * worker will wait for the leader to release the stream lock
             * until the end of the transaction. So, we don't need to release
             * the lock or increment the stream count in that case.
             */
            if (pa_send_data(winfo, s->len, s->data))
            {
                /*
                 * Unlock the shared object lock so that the parallel apply
                 * worker can continue to receive changes.
                 */
                if (!first_segment)
                    pa_unlock_stream(winfo->shared->xid, AccessExclusiveLock);
 
                /*
                 * Increment the number of streaming blocks waiting to be
                 * processed by parallel apply worker.
                 */
                pg_atomic_add_fetch_u32(&winfo->shared->pending_stream_count, 1);
 
                /* Cache the parallel apply worker for this transaction. */
                pa_set_stream_apply_worker(winfo);
                break;
            }
 
            /*
             * Switch to serialize mode when we are not able to send the
             * change to parallel apply worker.
             */
            pa_switch_to_partial_serialize(winfo, !first_segment);
 
            pg_fallthrough;
        case TRANS_LEADER_PARTIAL_SERIALIZE:
            Assert(winfo);
 
            /*
             * Open the spool file unless it was already opened when switching
             * to serialize mode. The transaction started in
             * stream_start_internal will be committed on the stream stop.
             */
            if (apply_action != TRANS_LEADER_SEND_TO_PARALLEL)
                stream_start_internal(stream_xid, first_segment);
 
            stream_write_change(LOGICAL_REP_MSG_STREAM_START, &original_msg);
 
            /* Cache the parallel apply worker for this transaction. */
            pa_set_stream_apply_worker(winfo);
            break;
 
        case TRANS_PARALLEL_APPLY:
            if (first_segment)
            {
                /* Hold the lock until the end of the transaction. */
                pa_lock_transaction(MyParallelShared->xid, AccessExclusiveLock);
                pa_set_xact_state(MyParallelShared, PARALLEL_TRANS_STARTED);
 
                /*
                 * Signal the leader apply worker, as it may be waiting for
                 * us.
                 */
                logicalrep_worker_wakeup(WORKERTYPE_APPLY,
                                         MyLogicalRepWorker->subid, InvalidOid);
            }
 
            parallel_stream_nchanges = 0;
            break;
 
        default:
            elog(ERROR, "unexpected apply action: %d", (int) apply_action);
            break;
    }
 
    pgstat_report_activity(STATE_RUNNING, NULL);
}

References AccessExclusiveLock, Assert, StringInfoData::data, elog, ereport, errcode(), ERRCODE_PROTOCOL_VIOLATION, errmsg_internal(), ERROR, fb(), 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(), pg_fallthrough, 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, WORKERTYPE_APPLY, and ParallelApplyWorkerShared::xid.

Referenced by apply_dispatch().

apply_handle_stream_stop()

static void apply_handle_stream_stop ( StringInfo  s )

static

Definition at line 1902 of file worker.c.

{
    ParallelApplyWorkerInfo *winfo;
    TransApplyAction apply_action;
 
    if (!in_streamed_transaction)
        ereport(ERROR,
                (errcode(ERRCODE_PROTOCOL_VIOLATION),
                 errmsg_internal("STREAM STOP message without STREAM START")));
 
    apply_action = get_transaction_apply_action(stream_xid, &winfo);
 
    switch (apply_action)
    {
        case TRANS_LEADER_SERIALIZE:
            stream_stop_internal(stream_xid);
            break;
 
        case TRANS_LEADER_SEND_TO_PARALLEL:
            Assert(winfo);
 
            /*
             * Lock before sending the STREAM_STOP message so that the leader
             * can hold the lock first and the parallel apply worker will wait
             * for leader to release the lock. See Locking Considerations atop
             * applyparallelworker.c.
             */
            pa_lock_stream(winfo->shared->xid, AccessExclusiveLock);
 
            if (pa_send_data(winfo, s->len, s->data))
            {
                pa_set_stream_apply_worker(NULL);
                break;
            }
 
            /*
             * Switch to serialize mode when we are not able to send the
             * change to parallel apply worker.
             */
            pa_switch_to_partial_serialize(winfo, true);
 
            pg_fallthrough;
        case TRANS_LEADER_PARTIAL_SERIALIZE:
            stream_write_change(LOGICAL_REP_MSG_STREAM_STOP, s);
            stream_stop_internal(stream_xid);
            pa_set_stream_apply_worker(NULL);
            break;
 
        case TRANS_PARALLEL_APPLY:
            elog(DEBUG1, "applied %u changes in the streaming chunk",
                 parallel_stream_nchanges);
 
            /*
             * By the time parallel apply worker is processing the changes in
             * the current streaming block, the leader apply worker may have
             * sent multiple streaming blocks. This can lead to parallel apply
             * worker start waiting even when there are more chunk of streams
             * in the queue. So, try to lock only if there is no message left
             * in the queue. See Locking Considerations atop
             * applyparallelworker.c.
             *
             * Note that here we have a race condition where we can start
             * waiting even when there are pending streaming chunks. This can
             * happen if the leader sends another streaming block and acquires
             * the stream lock again after the parallel apply worker checks
             * that there is no pending streaming block and before it actually
             * starts waiting on a lock. We can handle this case by not
             * allowing the leader to increment the stream block count during
             * the time parallel apply worker acquires the lock but it is not
             * clear whether that is worth the complexity.
             *
             * Now, if this missed chunk contains rollback to savepoint, then
             * there is a risk of deadlock which probably shouldn't happen
             * after restart.
             */
            pa_decr_and_wait_stream_block();
            break;
 
        default:
            elog(ERROR, "unexpected apply action: %d", (int) apply_action);
            break;
    }
 
    in_streamed_transaction = false;
    stream_xid = InvalidTransactionId;
 
    /*
     * The parallel apply worker could be in a transaction in which case we
     * need to report the state as STATE_IDLEINTRANSACTION.
     */
    if (IsTransactionOrTransactionBlock())
        pgstat_report_activity(STATE_IDLEINTRANSACTION, NULL);
    else
        pgstat_report_activity(STATE_IDLE, NULL);
 
    reset_apply_error_context_info();
}

References AccessExclusiveLock, Assert, StringInfoData::data, DEBUG1, elog, ereport, errcode(), ERRCODE_PROTOCOL_VIOLATION, errmsg_internal(), ERROR, fb(), 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, pg_fallthrough, 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 3669 of file worker.c.

{
    bool        cascade = false;
    bool        restart_seqs = false;
    List       *remote_relids = NIL;
    List       *remote_rels = NIL;
    List       *rels = NIL;
    List       *part_rels = NIL;
    List       *relids = NIL;
    List       *relids_logged = NIL;
    ListCell   *lc;
    LOCKMODE    lockmode = AccessExclusiveLock;
 
    /*
     * Quick return if we are skipping data modification changes or handling
     * streamed transactions.
     */
    if (is_skipping_changes() ||
        handle_streamed_transaction(LOGICAL_REP_MSG_TRUNCATE, s))
        return;
 
    begin_replication_step();
 
    remote_relids = logicalrep_read_truncate(s, &cascade, &restart_seqs);
 
    foreach(lc, remote_relids)
    {
        LogicalRepRelId relid = lfirst_oid(lc);
        LogicalRepRelMapEntry *rel;
 
        rel = logicalrep_rel_open(relid, lockmode);
        if (!should_apply_changes_for_rel(rel))
        {
            /*
             * The relation can't become interesting in the middle of the
             * transaction so it's safe to unlock it.
             */
            logicalrep_rel_close(rel, lockmode);
            continue;
        }
 
        remote_rels = lappend(remote_rels, rel);
        TargetPrivilegesCheck(rel->localrel, ACL_TRUNCATE);
        rels = lappend(rels, rel->localrel);
        relids = lappend_oid(relids, rel->localreloid);
        if (RelationIsLogicallyLogged(rel->localrel))
            relids_logged = lappend_oid(relids_logged, rel->localreloid);
 
        /*
         * Truncate partitions if we got a message to truncate a partitioned
         * table.
         */
        if (rel->localrel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
        {
            ListCell   *child;
            List       *children = find_all_inheritors(rel->localreloid,
                                                       lockmode,
                                                       NULL);
 
            foreach(child, children)
            {
                Oid         childrelid = lfirst_oid(child);
                Relation    childrel;
 
                if (list_member_oid(relids, childrelid))
                    continue;
 
                /* find_all_inheritors already got lock */
                childrel = table_open(childrelid, NoLock);
 
                /*
                 * Ignore temp tables of other backends.  See similar code in
                 * ExecuteTruncate().
                 */
                if (RELATION_IS_OTHER_TEMP(childrel))
                {
                    table_close(childrel, lockmode);
                    continue;
                }
 
                TargetPrivilegesCheck(childrel, ACL_TRUNCATE);
                rels = lappend(rels, childrel);
                part_rels = lappend(part_rels, childrel);
                relids = lappend_oid(relids, childrelid);
                /* Log this relation only if needed for logical decoding */
                if (RelationIsLogicallyLogged(childrel))
                    relids_logged = lappend_oid(relids_logged, childrelid);
            }
        }
    }
 
    /*
     * Even if we used CASCADE on the upstream primary we explicitly default
     * to replaying changes without further cascading. This might be later
     * changeable with a user specified option.
     *
     * MySubscription->runasowner tells us whether we want to execute
     * replication actions as the subscription owner; the last argument to
     * TruncateGuts tells it whether we want to switch to the table owner.
     * Those are exactly opposite conditions.
     */
    ExecuteTruncateGuts(rels,
                        relids,
                        relids_logged,
                        DROP_RESTRICT,
                        restart_seqs,
                        !MySubscription->runasowner);
    foreach(lc, remote_rels)
    {
        LogicalRepRelMapEntry *rel = lfirst(lc);
 
        logicalrep_rel_close(rel, NoLock);
    }
    foreach(lc, part_rels)
    {
        Relation    rel = lfirst(lc);
 
        table_close(rel, NoLock);
    }
 
    end_replication_step();
}

References AccessExclusiveLock, ACL_TRUNCATE, begin_replication_step(), DROP_RESTRICT, end_replication_step(), ExecuteTruncateGuts(), fb(), 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 3373 of file worker.c.

{
    EState     *estate = edata->estate;
    LogicalRepRelMapEntry *relmapentry = edata->targetRel;
    ResultRelInfo *relinfo = edata->targetRelInfo;
    Relation    parentrel = relinfo->ri_RelationDesc;
    ModifyTableState *mtstate;
    PartitionTupleRouting *proute;
    ResultRelInfo *partrelinfo;
    Relation    partrel;
    TupleTableSlot *remoteslot_part;
    TupleConversionMap *map;
    MemoryContext oldctx;
    LogicalRepRelMapEntry *part_entry = NULL;
    AttrMap    *attrmap = NULL;
 
    /* ModifyTableState is needed for ExecFindPartition(). */
    edata->mtstate = mtstate = makeNode(ModifyTableState);
    mtstate->ps.plan = NULL;
    mtstate->ps.state = estate;
    mtstate->operation = operation;
    mtstate->resultRelInfo = relinfo;
 
    /* ... as is PartitionTupleRouting. */
    edata->proute = proute = ExecSetupPartitionTupleRouting(estate, parentrel);
 
    /*
     * Find the partition to which the "search tuple" belongs.
     */
    Assert(remoteslot != NULL);
    oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
    partrelinfo = ExecFindPartition(mtstate, relinfo, proute,
                                    remoteslot, estate);
    Assert(partrelinfo != NULL);
    partrel = partrelinfo->ri_RelationDesc;
 
    /*
     * Check for supported relkind.  We need this since partitions might be of
     * unsupported relkinds; and the set of partitions can change, so checking
     * at CREATE/ALTER SUBSCRIPTION would be insufficient.
     */
    CheckSubscriptionRelkind(partrel->rd_rel->relkind,
                             relmapentry->remoterel.relkind,
                             get_namespace_name(RelationGetNamespace(partrel)),
                             RelationGetRelationName(partrel));
 
    /*
     * To perform any of the operations below, the tuple must match the
     * partition's rowtype. Convert if needed or just copy, using a dedicated
     * slot to store the tuple in any case.
     */
    remoteslot_part = partrelinfo->ri_PartitionTupleSlot;
    if (remoteslot_part == NULL)
        remoteslot_part = table_slot_create(partrel, &estate->es_tupleTable);
    map = ExecGetRootToChildMap(partrelinfo, estate);
    if (map != NULL)
    {
        attrmap = map->attrMap;
        remoteslot_part = execute_attr_map_slot(attrmap, remoteslot,
                                                remoteslot_part);
    }
    else
    {
        remoteslot_part = ExecCopySlot(remoteslot_part, remoteslot);
        slot_getallattrs(remoteslot_part);
    }
    MemoryContextSwitchTo(oldctx);
 
    /* Check if we can do the update or delete on the leaf partition. */
    if (operation == CMD_UPDATE || operation == CMD_DELETE)
    {
        part_entry = logicalrep_partition_open(relmapentry, partrel,
                                               attrmap);
        check_relation_updatable(part_entry);
    }
 
    switch (operation)
    {
        case CMD_INSERT:
            apply_handle_insert_internal(edata, partrelinfo,
                                         remoteslot_part);
            break;
 
        case CMD_DELETE:
            apply_handle_delete_internal(edata, partrelinfo,
                                         remoteslot_part,
                                         part_entry->localindexoid);
            break;
 
        case CMD_UPDATE:
 
            /*
             * For UPDATE, depending on whether or not the updated tuple
             * satisfies the partition's constraint, perform a simple UPDATE
             * of the partition or move the updated tuple into a different
             * suitable partition.
             */
            {
                TupleTableSlot *localslot;
                ResultRelInfo *partrelinfo_new;
                Relation    partrel_new;
                bool        found;
                EPQState    epqstate;
                ConflictTupleInfo conflicttuple = {0};
 
                /* Get the matching local tuple from the partition. */
                found = FindReplTupleInLocalRel(edata, partrel,
                                                &part_entry->remoterel,
                                                part_entry->localindexoid,
                                                remoteslot_part, &localslot);
                if (!found)
                {
                    ConflictType type;
                    TupleTableSlot *newslot = localslot;
 
                    /*
                     * Detecting whether the tuple was recently deleted or
                     * never existed is crucial to avoid misleading the user
                     * during conflict handling.
                     */
                    if (FindDeletedTupleInLocalRel(partrel,
                                                   part_entry->localindexoid,
                                                   remoteslot_part,
                                                   &conflicttuple.xmin,
                                                   &conflicttuple.origin,
                                                   &conflicttuple.ts) &&
                        conflicttuple.origin != replorigin_xact_state.origin)
                        type = CT_UPDATE_DELETED;
                    else
                        type = CT_UPDATE_MISSING;
 
                    /* Store the new tuple for conflict reporting */
                    slot_store_data(newslot, part_entry, newtup);
 
                    /*
                     * The tuple to be updated could not be found or was
                     * deleted.  Do nothing except for emitting a log message.
                     */
                    ReportApplyConflict(estate, partrelinfo, LOG,
                                        type, remoteslot_part, newslot,
                                        list_make1(&conflicttuple));
 
                    return;
                }
 
                /*
                 * Report the conflict if the tuple was modified by a
                 * different origin.
                 */
                if (GetTupleTransactionInfo(localslot, &conflicttuple.xmin,
                                            &conflicttuple.origin,
                                            &conflicttuple.ts) &&
                    conflicttuple.origin != replorigin_xact_state.origin)
                {
                    TupleTableSlot *newslot;
 
                    /* Store the new tuple for conflict reporting */
                    newslot = table_slot_create(partrel, &estate->es_tupleTable);
                    slot_store_data(newslot, part_entry, newtup);
 
                    conflicttuple.slot = localslot;
 
                    ReportApplyConflict(estate, partrelinfo, LOG, CT_UPDATE_ORIGIN_DIFFERS,
                                        remoteslot_part, newslot,
                                        list_make1(&conflicttuple));
                }
 
                /*
                 * Apply the update to the local tuple, putting the result in
                 * remoteslot_part.
                 */
                oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
                slot_modify_data(remoteslot_part, localslot, part_entry,
                                 newtup);
                MemoryContextSwitchTo(oldctx);
 
                EvalPlanQualInit(&epqstate, estate, NULL, NIL, -1, NIL);
 
                /*
                 * Does the updated tuple still satisfy the current
                 * partition's constraint?
                 */
                if (!partrel->rd_rel->relispartition ||
                    ExecPartitionCheck(partrelinfo, remoteslot_part, estate,
                                       false))
                {
                    /*
                     * Yes, so simply UPDATE the partition.  We don't call
                     * apply_handle_update_internal() here, which would
                     * normally do the following work, to avoid repeating some
                     * work already done above to find the local tuple in the
                     * partition.
                     */
                    InitConflictIndexes(partrelinfo);
 
                    EvalPlanQualSetSlot(&epqstate, remoteslot_part);
                    TargetPrivilegesCheck(partrelinfo->ri_RelationDesc,
                                          ACL_UPDATE);
                    ExecSimpleRelationUpdate(partrelinfo, estate, &epqstate,
                                             localslot, remoteslot_part);
                }
                else
                {
                    /* Move the tuple into the new partition. */
 
                    /*
                     * New partition will be found using tuple routing, which
                     * can only occur via the parent table.  We might need to
                     * convert the tuple to the parent's rowtype.  Note that
                     * this is the tuple found in the partition, not the
                     * original search tuple received by this function.
                     */
                    if (map)
                    {
                        TupleConversionMap *PartitionToRootMap =
                            convert_tuples_by_name(RelationGetDescr(partrel),
                                                   RelationGetDescr(parentrel));
 
                        remoteslot =
                            execute_attr_map_slot(PartitionToRootMap->attrMap,
                                                  remoteslot_part, remoteslot);
                    }
                    else
                    {
                        remoteslot = ExecCopySlot(remoteslot, remoteslot_part);
                        slot_getallattrs(remoteslot);
                    }
 
                    /* Find the new partition. */
                    oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
                    partrelinfo_new = ExecFindPartition(mtstate, relinfo,
                                                        proute, remoteslot,
                                                        estate);
                    MemoryContextSwitchTo(oldctx);
                    Assert(partrelinfo_new != partrelinfo);
                    partrel_new = partrelinfo_new->ri_RelationDesc;
 
                    /* Check that new partition also has supported relkind. */
                    CheckSubscriptionRelkind(partrel_new->rd_rel->relkind,
                                             relmapentry->remoterel.relkind,
                                             get_namespace_name(RelationGetNamespace(partrel_new)),
                                             RelationGetRelationName(partrel_new));
 
                    /* DELETE old tuple found in the old partition. */
                    EvalPlanQualSetSlot(&epqstate, localslot);
                    TargetPrivilegesCheck(partrelinfo->ri_RelationDesc, ACL_DELETE);
                    ExecSimpleRelationDelete(partrelinfo, estate, &epqstate, localslot);
 
                    /* INSERT new tuple into the new partition. */
 
                    /*
                     * Convert the replacement tuple to match the destination
                     * partition rowtype.
                     */
                    oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
                    remoteslot_part = partrelinfo_new->ri_PartitionTupleSlot;
                    if (remoteslot_part == NULL)
                        remoteslot_part = table_slot_create(partrel_new,
                                                            &estate->es_tupleTable);
                    map = ExecGetRootToChildMap(partrelinfo_new, estate);
                    if (map != NULL)
                    {
                        remoteslot_part = execute_attr_map_slot(map->attrMap,
                                                                remoteslot,
                                                                remoteslot_part);
                    }
                    else
                    {
                        remoteslot_part = ExecCopySlot(remoteslot_part,
                                                       remoteslot);
                        slot_getallattrs(remoteslot);
                    }
                    MemoryContextSwitchTo(oldctx);
                    apply_handle_insert_internal(edata, partrelinfo_new,
                                                 remoteslot_part);
                }
 
                EvalPlanQualEnd(&epqstate);
            }
            break;
 
        default:
            elog(ERROR, "unrecognized CmdType: %d", (int) operation);
            break;
    }
}

References ACL_DELETE, 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(), CT_UPDATE_DELETED, CT_UPDATE_MISSING, CT_UPDATE_ORIGIN_DIFFERS, elog, ERROR, EState::es_tupleTable, EvalPlanQualEnd(), EvalPlanQualInit(), EvalPlanQualSetSlot, ExecCopySlot(), ExecFindPartition(), ExecGetRootToChildMap(), ExecPartitionCheck(), ExecSetupPartitionTupleRouting(), ExecSimpleRelationDelete(), ExecSimpleRelationUpdate(), execute_attr_map_slot(), fb(), FindDeletedTupleInLocalRel(), FindReplTupleInLocalRel(), get_namespace_name(), GetPerTupleMemoryContext, GetTupleTransactionInfo(), InitConflictIndexes(), list_make1, LOG, logicalrep_partition_open(), makeNode, MemoryContextSwitchTo(), NIL, operation, ModifyTableState::operation, ReplOriginXactState::origin, PlanState::plan, ModifyTableState::ps, RelationData::rd_rel, RelationGetDescr, RelationGetNamespace, RelationGetRelationName, LogicalRepRelation::relkind, LogicalRepRelMapEntry::remoterel, replorigin_xact_state, ReportApplyConflict(), ModifyTableState::resultRelInfo, slot_getallattrs(), slot_modify_data(), slot_store_data(), PlanState::state, table_slot_create(), TargetPrivilegesCheck(), and type.

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 2603 of file worker.c.

{
    LogicalRepTyp typ;
 
    if (handle_streamed_transaction(LOGICAL_REP_MSG_TYPE, s))
        return;
 
    logicalrep_read_typ(s, &typ);
}

References fb(), 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 2807 of file worker.c.

{
    LogicalRepRelMapEntry *rel;
    LogicalRepRelId relid;
    UserContext ucxt;
    ApplyExecutionData *edata;
    EState     *estate;
    LogicalRepTupleData oldtup;
    LogicalRepTupleData newtup;
    bool        has_oldtup;
    TupleTableSlot *remoteslot;
    RTEPermissionInfo *target_perminfo;
    MemoryContext oldctx;
    bool        run_as_owner;
 
    /*
     * Quick return if we are skipping data modification changes or handling
     * streamed transactions.
     */
    if (is_skipping_changes() ||
        handle_streamed_transaction(LOGICAL_REP_MSG_UPDATE, s))
        return;
 
    begin_replication_step();
 
    relid = logicalrep_read_update(s, &has_oldtup, &oldtup,
                                   &newtup);
    rel = logicalrep_rel_open(relid, RowExclusiveLock);
    if (!should_apply_changes_for_rel(rel))
    {
        /*
         * The relation can't become interesting in the middle of the
         * transaction so it's safe to unlock it.
         */
        logicalrep_rel_close(rel, RowExclusiveLock);
        end_replication_step();
        return;
    }
 
    /* Set relation for error callback */
    apply_error_callback_arg.rel = rel;
 
    /* Check if we can do the update. */
    check_relation_updatable(rel);
 
    /*
     * Make sure that any user-supplied code runs as the table owner, unless
     * the user has opted out of that behavior.
     */
    run_as_owner = MySubscription->runasowner;
    if (!run_as_owner)
        SwitchToUntrustedUser(rel->localrel->rd_rel->relowner, &ucxt);
 
    /* Initialize the executor state. */
    edata = create_edata_for_relation(rel);
    estate = edata->estate;
    remoteslot = ExecInitExtraTupleSlot(estate,
                                        RelationGetDescr(rel->localrel),
                                        &TTSOpsVirtual);
 
    /*
     * Populate updatedCols so that per-column triggers can fire, and so
     * executor can correctly pass down indexUnchanged hint.  This could
     * include more columns than were actually changed on the publisher
     * because the logical replication protocol doesn't contain that
     * information.  But it would for example exclude columns that only exist
     * on the subscriber, since we are not touching those.
     */
    target_perminfo = list_nth(estate->es_rteperminfos, 0);
    for (int i = 0; i < remoteslot->tts_tupleDescriptor->natts; i++)
    {
        CompactAttribute *att = TupleDescCompactAttr(remoteslot->tts_tupleDescriptor, i);
        int         remoteattnum = rel->attrmap->attnums[i];
 
        if (!att->attisdropped && remoteattnum >= 0)
        {
            if (remoteattnum >= newtup.ncols)
                ereport(ERROR,
                        (errcode(ERRCODE_PROTOCOL_VIOLATION),
                         errmsg("logical replication column %d not found in tuple: only %d column(s) received",
                                remoteattnum + 1, newtup.ncols)));
 
            if (newtup.colstatus[remoteattnum] != LOGICALREP_COLUMN_UNCHANGED)
                target_perminfo->updatedCols =
                    bms_add_member(target_perminfo->updatedCols,
                                   i + 1 - FirstLowInvalidHeapAttributeNumber);
        }
    }
 
    /* Build the search tuple. */
    oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
    slot_store_data(remoteslot, rel,
                    has_oldtup ? &oldtup : &newtup);
    MemoryContextSwitchTo(oldctx);
 
    /* For a partitioned table, apply update to correct partition. */
    if (rel->localrel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
        apply_handle_tuple_routing(edata,
                                   remoteslot, &newtup, CMD_UPDATE);
    else
        apply_handle_update_internal(edata, edata->targetRelInfo,
                                     remoteslot, &newtup, rel->localindexoid);
 
    finish_edata(edata);
 
    /* Reset relation for error callback */
    apply_error_callback_arg.rel = NULL;
 
    if (!run_as_owner)
        RestoreUserContext(&ucxt);
 
    logicalrep_rel_close(rel, NoLock);
 
    end_replication_step();
}

References apply_error_callback_arg, apply_handle_tuple_routing(), apply_handle_update_internal(), CompactAttribute::attisdropped, AttrMap::attnums, LogicalRepRelMapEntry::attrmap, begin_replication_step(), bms_add_member(), check_relation_updatable(), CMD_UPDATE, create_edata_for_relation(), end_replication_step(), ereport, errcode(), ERRCODE_PROTOCOL_VIOLATION, errmsg, ERROR, EState::es_rteperminfos, ExecInitExtraTupleSlot(), fb(), 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, NoLock, RelationData::rd_rel, ApplyErrorCallbackArg::rel, RelationGetDescr, RestoreUserContext(), RowExclusiveLock, Subscription::runasowner, should_apply_changes_for_rel(), slot_store_data(), SwitchToUntrustedUser(), TTSOpsVirtual, and TupleDescCompactAttr().

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 2929 of file worker.c.

{
    EState     *estate = edata->estate;
    LogicalRepRelMapEntry *relmapentry = edata->targetRel;
    Relation    localrel = relinfo->ri_RelationDesc;
    EPQState    epqstate;
    TupleTableSlot *localslot = NULL;
    ConflictTupleInfo conflicttuple = {0};
    bool        found;
    MemoryContext oldctx;
 
    EvalPlanQualInit(&epqstate, estate, NULL, NIL, -1, NIL);
    ExecOpenIndices(relinfo, false);
 
    found = FindReplTupleInLocalRel(edata, localrel,
                                    &relmapentry->remoterel,
                                    localindexoid,
                                    remoteslot, &localslot);
 
    /*
     * Tuple found.
     *
     * Note this will fail if there are other conflicting unique indexes.
     */
    if (found)
    {
        /*
         * Report the conflict if the tuple was modified by a different
         * origin.
         */
        if (GetTupleTransactionInfo(localslot, &conflicttuple.xmin,
                                    &conflicttuple.origin, &conflicttuple.ts) &&
            conflicttuple.origin != replorigin_xact_state.origin)
        {
            TupleTableSlot *newslot;
 
            /* Store the new tuple for conflict reporting */
            newslot = table_slot_create(localrel, &estate->es_tupleTable);
            slot_store_data(newslot, relmapentry, newtup);
 
            conflicttuple.slot = localslot;
 
            ReportApplyConflict(estate, relinfo, LOG, CT_UPDATE_ORIGIN_DIFFERS,
                                remoteslot, newslot,
                                list_make1(&conflicttuple));
        }
 
        /* Process and store remote tuple in the slot */
        oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
        slot_modify_data(remoteslot, localslot, relmapentry, newtup);
        MemoryContextSwitchTo(oldctx);
 
        EvalPlanQualSetSlot(&epqstate, remoteslot);
 
        InitConflictIndexes(relinfo);
 
        /* Do the actual update. */
        TargetPrivilegesCheck(relinfo->ri_RelationDesc, ACL_UPDATE);
        ExecSimpleRelationUpdate(relinfo, estate, &epqstate, localslot,
                                 remoteslot);
    }
    else
    {
        ConflictType type;
        TupleTableSlot *newslot = localslot;
 
        /*
         * Detecting whether the tuple was recently deleted or never existed
         * is crucial to avoid misleading the user during conflict handling.
         */
        if (FindDeletedTupleInLocalRel(localrel, localindexoid, remoteslot,
                                       &conflicttuple.xmin,
                                       &conflicttuple.origin,
                                       &conflicttuple.ts) &&
            conflicttuple.origin != replorigin_xact_state.origin)
            type = CT_UPDATE_DELETED;
        else
            type = CT_UPDATE_MISSING;
 
        /* Store the new tuple for conflict reporting */
        slot_store_data(newslot, relmapentry, newtup);
 
        /*
         * The tuple to be updated could not be found or was deleted.  Do
         * nothing except for emitting a log message.
         */
        ReportApplyConflict(estate, relinfo, LOG, type, remoteslot, newslot,
                            list_make1(&conflicttuple));
    }
 
    /* Cleanup. */
    ExecCloseIndices(relinfo);
    EvalPlanQualEnd(&epqstate);
}

References ACL_UPDATE, CT_UPDATE_DELETED, CT_UPDATE_MISSING, CT_UPDATE_ORIGIN_DIFFERS, EState::es_tupleTable, EvalPlanQualEnd(), EvalPlanQualInit(), EvalPlanQualSetSlot, ExecCloseIndices(), ExecOpenIndices(), ExecSimpleRelationUpdate(), fb(), FindDeletedTupleInLocalRel(), FindReplTupleInLocalRel(), GetPerTupleMemoryContext, GetTupleTransactionInfo(), InitConflictIndexes(), list_make1, LOG, MemoryContextSwitchTo(), NIL, ReplOriginXactState::origin, LogicalRepRelMapEntry::remoterel, replorigin_xact_state, ReportApplyConflict(), slot_modify_data(), slot_store_data(), table_slot_create(), TargetPrivilegesCheck(), and type.

Referenced by apply_handle_update().

apply_spooled_messages()

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

Definition at line 2277 of file worker.c.

{
    int         nchanges;
    char        path[MAXPGPATH];
    char       *buffer = NULL;
    MemoryContext oldcxt;
    ResourceOwner oldowner;
    int         fileno;
    pgoff_t     offset;
 
    if (!am_parallel_apply_worker())
        maybe_start_skipping_changes(lsn);
 
    /* Make sure we have an open transaction */
    begin_replication_step();
 
    /*
     * Allocate file handle and memory required to process all the messages in
     * TopTransactionContext to avoid them getting reset after each message is
     * processed.
     */
    oldcxt = MemoryContextSwitchTo(TopTransactionContext);
 
    /* Open the spool file for the committed/prepared transaction */
    changes_filename(path, MyLogicalRepWorker->subid, xid);
    elog(DEBUG1, "replaying changes from file \"%s\"", path);
 
    /*
     * Make sure the file is owned by the toplevel transaction so that the
     * file will not be accidentally closed when aborting a subtransaction.
     */
    oldowner = CurrentResourceOwner;
    CurrentResourceOwner = TopTransactionResourceOwner;
 
    stream_fd = BufFileOpenFileSet(stream_fileset, path, O_RDONLY, false);
 
    CurrentResourceOwner = oldowner;
 
    buffer = palloc(BLCKSZ);
 
    MemoryContextSwitchTo(oldcxt);
 
    remote_final_lsn = lsn;
 
    /*
     * Make sure the handle apply_dispatch methods are aware we're in a remote
     * transaction.
     */
    in_remote_transaction = true;
    pgstat_report_activity(STATE_RUNNING, NULL);
 
    end_replication_step();
 
    /*
     * Read the entries one by one and pass them through the same logic as in
     * apply_dispatch.
     */
    nchanges = 0;
    while (true)
    {
        StringInfoData s2;
        size_t      nbytes;
        int         len;
 
        CHECK_FOR_INTERRUPTS();
 
        /* read length of the on-disk record */
        nbytes = BufFileReadMaybeEOF(stream_fd, &len, sizeof(len), true);
 
        /* have we reached end of the file? */
        if (nbytes == 0)
            break;
 
        /* do we have a correct length? */
        if (len <= 0)
            elog(ERROR, "incorrect length %d in streaming transaction's changes file \"%s\"",
                 len, path);
 
        /* make sure we have sufficiently large buffer */
        buffer = repalloc(buffer, len);
 
        /* and finally read the data into the buffer */
        BufFileReadExact(stream_fd, buffer, len);
 
        BufFileTell(stream_fd, &fileno, &offset);
 
        /* init a stringinfo using the buffer and call apply_dispatch */
        initReadOnlyStringInfo(&s2, buffer, len);
 
        /* Ensure we are reading the data into our memory context. */
        oldcxt = MemoryContextSwitchTo(ApplyMessageContext);
 
        apply_dispatch(&s2);
 
        MemoryContextReset(ApplyMessageContext);
 
        MemoryContextSwitchTo(oldcxt);
 
        nchanges++;
 
        /*
         * It is possible the file has been closed because we have processed
         * the transaction end message like stream_commit in which case that
         * must be the last message.
         */
        if (!stream_fd)
        {
            ensure_last_message(stream_fileset, xid, fileno, offset);
            break;
        }
 
        if (nchanges % 1000 == 0)
            elog(DEBUG1, "replayed %d changes from file \"%s\"",
                 nchanges, path);
    }
 
    if (stream_fd)
        stream_close_file();
 
    elog(DEBUG1, "replayed %d (all) changes from file \"%s\"",
         nchanges, path);
 
    return;
}

References am_parallel_apply_worker(), 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, fb(), in_remote_transaction, initReadOnlyStringInfo(), len, MAXPGPATH, maybe_start_skipping_changes(), MemoryContextReset(), MemoryContextSwitchTo(), MyLogicalRepWorker, palloc(), pgstat_report_activity(), remote_final_lsn, repalloc(), 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 5027 of file worker.c.

{
    if (am_parallel_apply_worker())
    {
        /*
         * Don't stop the parallel apply worker as the leader will detect the
         * subscription parameter change and restart logical replication later
         * anyway. This also prevents the leader from reporting errors when
         * trying to communicate with a stopped parallel apply worker, which
         * would accidentally disable subscriptions if disable_on_error was
         * set.
         */
        return;
    }
 
    /*
     * Reset the last-start time for this apply worker so that the launcher
     * will restart it without waiting for wal_retrieve_retry_interval if the
     * subscription is still active, and so that we won't leak that hash table
     * entry if it isn't.
     */
    if (am_leader_apply_worker())
        ApplyLauncherForgetWorkerStartTime(MyLogicalRepWorker->subid);
 
    proc_exit(0);
}

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

Referenced by InitializeLogRepWorker(), maybe_reread_subscription(), and resume_conflict_info_retention().

ApplyWorkerMain()

void ApplyWorkerMain

(

Datum

main_arg

)

Definition at line 6003 of file worker.c.

{
    int         worker_slot = DatumGetInt32(main_arg);
 
    InitializingApplyWorker = true;
 
    SetupApplyOrSyncWorker(worker_slot);
 
    InitializingApplyWorker = false;
 
    run_apply_worker();
 
    proc_exit(0);
}

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

AtEOXact_LogicalRepWorkers()

void AtEOXact_LogicalRepWorkers

(

bool

isCommit

)

Definition at line 6348 of file worker.c.

{
    if (isCommit && on_commit_wakeup_workers_subids != NIL)
    {
        ListCell   *lc;
 
        LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
        foreach(lc, on_commit_wakeup_workers_subids)
        {
            Oid         subid = lfirst_oid(lc);
            List       *workers;
            ListCell   *lc2;
 
            workers = logicalrep_workers_find(subid, true, false);
            foreach(lc2, workers)
            {
                LogicalRepWorker *worker = (LogicalRepWorker *) lfirst(lc2);
 
                logicalrep_worker_wakeup_ptr(worker);
            }
        }
        LWLockRelease(LogicalRepWorkerLock);
    }
 
    /* The List storage will be reclaimed automatically in xact cleanup. */
    on_commit_wakeup_workers_subids = NIL;
}

References fb(), 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()

static void begin_replication_step ( void  )

static

Definition at line 733 of file worker.c.

{
    SetCurrentStatementStartTimestamp();
 
    if (!IsTransactionState())
    {
        StartTransactionCommand();
        maybe_reread_subscription();
    }
 
    PushActiveSnapshot(GetTransactionSnapshot());
 
    MemoryContextSwitchTo(ApplyMessageContext);
}

References ApplyMessageContext, GetTransactionSnapshot(), IsTransactionState(), maybe_reread_subscription(), MemoryContextSwitchTo(), PushActiveSnapshot(), SetCurrentStatementStartTimestamp(), and StartTransactionCommand().

Referenced by apply_handle_commit_prepared(), apply_handle_delete(), apply_handle_insert(), apply_handle_prepare(), apply_handle_rollback_prepared(), apply_handle_stream_prepare(), apply_handle_truncate(), apply_handle_update(), apply_spooled_messages(), ensure_last_message(), stream_abort_internal(), and stream_start_internal().

can_advance_nonremovable_xid()

static bool can_advance_nonremovable_xid ( RetainDeadTuplesData *  rdt_data )

static

Definition at line 4423 of file worker.c.

{
    /*
     * It is sufficient to manage non-removable transaction ID for a
     * subscription by the main apply worker to detect update_deleted reliably
     * even for table sync or parallel apply workers.
     */
    if (!am_leader_apply_worker())
        return false;
 
    /* No need to advance if retaining dead tuples is not required */
    if (!MySubscription->retaindeadtuples)
        return false;
 
    return true;
}

References am_leader_apply_worker(), MySubscription, and Subscription::retaindeadtuples.

Referenced by maybe_advance_nonremovable_xid().

changes_filename()

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

inlinestatic

Definition at line 5426 of file worker.c.

{
    snprintf(path, MAXPGPATH, "%u-%u.changes", subid, xid);
}

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 2766 of file worker.c.

{
    /*
     * For partitioned tables, we only need to care if the target partition is
     * updatable (aka has PK or RI defined for it).
     */
    if (rel->localrel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
        return;
 
    /* Updatable, no error. */
    if (rel->updatable)
        return;
 
    /*
     * We are in error mode so it's fine this is somewhat slow. It's better to
     * give user correct error.
     */
    if (OidIsValid(GetRelationIdentityOrPK(rel->localrel)))
    {
        ereport(ERROR,
                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                 errmsg("publisher did not send replica identity column "
                        "expected by the logical replication target relation \"%s.%s\"",
                        rel->remoterel.nspname, rel->remoterel.relname)));
    }
 
    ereport(ERROR,
            (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
             errmsg("logical replication target relation \"%s.%s\" has "
                    "neither REPLICA IDENTITY index nor PRIMARY "
                    "KEY and published relation does not have "
                    "REPLICA IDENTITY FULL",
                    rel->remoterel.nspname, rel->remoterel.relname)));
}

References ereport, errcode(), errmsg, ERROR, fb(), 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 5623 of file worker.c.

{
    if (subxact_data.subxacts)
        pfree(subxact_data.subxacts);
 
    subxact_data.subxacts = NULL;
    subxact_data.subxact_last = InvalidTransactionId;
    subxact_data.nsubxacts = 0;
    subxact_data.nsubxacts_max = 0;
}

References fb(), 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 6149 of file worker.c.

{
    Relation    rel;
    Form_pg_subscription subform;
    HeapTuple   tup;
    XLogRecPtr  myskiplsn = MySubscription->skiplsn;
    bool        started_tx = false;
 
    if (likely(!XLogRecPtrIsValid(myskiplsn)) || am_parallel_apply_worker())
        return;
 
    if (!IsTransactionState())
    {
        StartTransactionCommand();
        started_tx = true;
    }
 
    /*
     * Updating pg_subscription might involve TOAST table access, so ensure we
     * have a valid snapshot.
     */
    PushActiveSnapshot(GetTransactionSnapshot());
 
    /*
     * Protect subskiplsn of pg_subscription from being concurrently updated
     * while clearing it.
     */
    LockSharedObject(SubscriptionRelationId, MySubscription->oid, 0,
                     AccessShareLock);
 
    rel = table_open(SubscriptionRelationId, RowExclusiveLock);
 
    /* Fetch the existing tuple. */
    tup = SearchSysCacheCopy1(SUBSCRIPTIONOID,
                              ObjectIdGetDatum(MySubscription->oid));
 
    if (!HeapTupleIsValid(tup))
        elog(ERROR, "subscription \"%s\" does not exist", MySubscription->name);
 
    subform = (Form_pg_subscription) GETSTRUCT(tup);
 
    /*
     * Clear the subskiplsn. If the user has already changed subskiplsn before
     * clearing it we don't update the catalog and the replication origin
     * state won't get advanced. So in the worst case, if the server crashes
     * before sending an acknowledgment of the flush position the transaction
     * will be sent again and the user needs to set subskiplsn again. We can
     * reduce the possibility by logging a replication origin WAL record to
     * advance the origin LSN instead but there is no way to advance the
     * origin timestamp and it doesn't seem to be worth doing anything about
     * it since it's a very rare case.
     */
    if (subform->subskiplsn == myskiplsn)
    {
        bool        nulls[Natts_pg_subscription];
        bool        replaces[Natts_pg_subscription];
        Datum       values[Natts_pg_subscription];
 
        memset(values, 0, sizeof(values));
        memset(nulls, false, sizeof(nulls));
        memset(replaces, false, sizeof(replaces));
 
        /* reset subskiplsn */
        values[Anum_pg_subscription_subskiplsn - 1] = LSNGetDatum(InvalidXLogRecPtr);
        replaces[Anum_pg_subscription_subskiplsn - 1] = true;
 
        tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
                                replaces);
        CatalogTupleUpdate(rel, &tup->t_self, tup);
 
        if (myskiplsn != finish_lsn)
            ereport(WARNING,
                    errmsg("skip-LSN of subscription \"%s\" cleared", MySubscription->name),
                    errdetail("Remote transaction's finish WAL location (LSN) %X/%08X did not match skip-LSN %X/%08X.",
                              LSN_FORMAT_ARGS(finish_lsn),
                              LSN_FORMAT_ARGS(myskiplsn)));
    }
 
    heap_freetuple(tup);
    table_close(rel, NoLock);
 
    PopActiveSnapshot();
 
    if (started_tx)
        CommitTransactionCommand();
}

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

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 877 of file worker.c.

{
    ApplyExecutionData *edata;
    EState     *estate;
    RangeTblEntry *rte;
    List       *perminfos = NIL;
    ResultRelInfo *resultRelInfo;
 
    edata = palloc0_object(ApplyExecutionData);
    edata->targetRel = rel;
 
    edata->estate = estate = CreateExecutorState();
 
    rte = makeNode(RangeTblEntry);
    rte->rtekind = RTE_RELATION;
    rte->relid = RelationGetRelid(rel->localrel);
    rte->relkind = rel->localrel->rd_rel->relkind;
    rte->rellockmode = AccessShareLock;
 
    addRTEPermissionInfo(&perminfos, rte);
 
    ExecInitRangeTable(estate, list_make1(rte), perminfos,
                       bms_make_singleton(1));
 
    edata->targetRelInfo = resultRelInfo = makeNode(ResultRelInfo);
 
    /*
     * Use Relation opened by logicalrep_rel_open() instead of opening it
     * again.
     */
    InitResultRelInfo(resultRelInfo, rel->localrel, 1, NULL, 0);
 
    /*
     * We put the ResultRelInfo in the es_opened_result_relations list, even
     * though we don't populate the es_result_relations array.  That's a bit
     * bogus, but it's enough to make ExecGetTriggerResultRel() find them.
     *
     * ExecOpenIndices() is not called here either, each execution path doing
     * an apply operation being responsible for that.
     */
    estate->es_opened_result_relations =
        lappend(estate->es_opened_result_relations, resultRelInfo);
 
    estate->es_output_cid = GetCurrentCommandId(true);
 
    /* Prepare to catch AFTER triggers. */
    AfterTriggerBeginQuery();
 
    /* other fields of edata remain NULL for now */
 
    return edata;
}

References AccessShareLock, addRTEPermissionInfo(), AfterTriggerBeginQuery(), bms_make_singleton(), CreateExecutorState(), EState::es_opened_result_relations, EState::es_output_cid, ExecInitRangeTable(), fb(), GetCurrentCommandId(), InitResultRelInfo(), lappend(), list_make1, LogicalRepRelMapEntry::localrel, makeNode, NIL, palloc0_object, RelationData::rd_rel, RelationGetRelid, and RTE_RELATION.

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

DisableSubscriptionAndExit()

void DisableSubscriptionAndExit

(

void

)

Definition at line 6023 of file worker.c.

{
    /*
     * Emit the error message, and recover from the error state to an idle
     * state
     */
    HOLD_INTERRUPTS();
 
    EmitErrorReport();
    AbortOutOfAnyTransaction();
    FlushErrorState();
 
    RESUME_INTERRUPTS();
 
    /*
     * Report the worker failed during sequence synchronization, table
     * synchronization, or apply.
     */
    pgstat_report_subscription_error(MyLogicalRepWorker->subid);
 
    /* Disable the subscription */
    StartTransactionCommand();
 
    /*
     * Updating pg_subscription might involve TOAST table access, so ensure we
     * have a valid snapshot.
     */
    PushActiveSnapshot(GetTransactionSnapshot());
 
    DisableSubscription(MySubscription->oid);
    PopActiveSnapshot();
    CommitTransactionCommand();
 
    /* Ensure we remove no-longer-useful entry for worker's start time */
    if (am_leader_apply_worker())
        ApplyLauncherForgetWorkerStartTime(MyLogicalRepWorker->subid);
 
    /* Notify the subscription has been disabled and exit */
    ereport(LOG,
            errmsg("subscription \"%s\" has been disabled because of an error",
                   MySubscription->name));
 
    /*
     * Skip the track_commit_timestamp check when disabling the worker due to
     * an error, as verifying commit timestamps is unnecessary in this
     * context.
     */
    CheckSubDeadTupleRetention(false, true, WARNING,
                               MySubscription->retaindeadtuples,
                               MySubscription->retentionactive, false);
 
    proc_exit(0);
}

References AbortOutOfAnyTransaction(), am_leader_apply_worker(), ApplyLauncherForgetWorkerStartTime(), CheckSubDeadTupleRetention(), CommitTransactionCommand(), DisableSubscription(), EmitErrorReport(), ereport, errmsg, FlushErrorState(), GetTransactionSnapshot(), HOLD_INTERRUPTS, LOG, MyLogicalRepWorker, MySubscription, Subscription::name, Subscription::oid, pgstat_report_subscription_error(), PopActiveSnapshot(), proc_exit(), PushActiveSnapshot(), RESUME_INTERRUPTS, Subscription::retaindeadtuples, Subscription::retentionactive, StartTransactionCommand(), LogicalRepWorker::subid, and WARNING.

Referenced by start_apply(), start_sequence_sync(), and start_table_sync().

end_replication_step()

static void end_replication_step ( void  )

static

Definition at line 756 of file worker.c.

{
    PopActiveSnapshot();
 
    CommandCounterIncrement();
}

References CommandCounterIncrement(), and PopActiveSnapshot().

Referenced by apply_handle_commit_prepared(), apply_handle_delete(), apply_handle_insert(), apply_handle_prepare(), apply_handle_rollback_prepared(), apply_handle_stream_prepare(), apply_handle_truncate(), apply_handle_update(), apply_spooled_messages(), ensure_last_message(), stream_abort_internal(), and stream_start_internal().

ensure_last_message()

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

static

Definition at line 2245 of file worker.c.

{
    char        path[MAXPGPATH];
    BufFile    *fd;
    int         last_fileno;
    pgoff_t     last_offset;
 
    Assert(!IsTransactionState());
 
    begin_replication_step();
 
    changes_filename(path, MyLogicalRepWorker->subid, xid);
 
    fd = BufFileOpenFileSet(stream_fileset, path, O_RDONLY, false);
 
    BufFileSeek(fd, 0, 0, SEEK_END);
    BufFileTell(fd, &last_fileno, &last_offset);
 
    BufFileClose(fd);
 
    end_replication_step();
 
    if (last_fileno != fileno || last_offset != offset)
        elog(ERROR, "unexpected message left in streaming transaction's changes file \"%s\"",
             path);
}

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

Referenced by apply_spooled_messages().

FindDeletedTupleInLocalRel()

static bool FindDeletedTupleInLocalRel ( Relation  localrel, Oid  localidxoid, TupleTableSlot *  remoteslot, TransactionId *  delete_xid, ReplOriginId *  delete_origin, TimestampTz *  delete_time  )

static

Definition at line 3291 of file worker.c.

{
    TransactionId oldestxmin;
 
    /*
     * Return false if either dead tuples are not retained or commit timestamp
     * data is not available.
     */
    if (!MySubscription->retaindeadtuples || !track_commit_timestamp)
        return false;
 
    /*
     * For conflict detection, we use the leader worker's
     * oldest_nonremovable_xid value instead of invoking
     * GetOldestNonRemovableTransactionId() or using the conflict detection
     * slot's xmin. The oldest_nonremovable_xid acts as a threshold to
     * identify tuples that were recently deleted. These deleted tuples are no
     * longer visible to concurrent transactions. However, if a remote update
     * matches such a tuple, we log an update_deleted conflict.
     *
     * While GetOldestNonRemovableTransactionId() and slot.xmin may return
     * transaction IDs older than oldest_nonremovable_xid, for our current
     * purpose, it is acceptable to treat tuples deleted by transactions prior
     * to oldest_nonremovable_xid as update_missing conflicts.
     */
    if (am_leader_apply_worker())
    {
        oldestxmin = MyLogicalRepWorker->oldest_nonremovable_xid;
    }
    else
    {
        LogicalRepWorker *leader;
 
        /*
         * Obtain the information from the leader apply worker as only the
         * leader manages oldest_nonremovable_xid (see
         * maybe_advance_nonremovable_xid() for details).
         */
        LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
        leader = logicalrep_worker_find(WORKERTYPE_APPLY,
                                        MyLogicalRepWorker->subid, InvalidOid,
                                        false);
        if (!leader)
        {
            ereport(ERROR,
                    (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                     errmsg("could not detect conflict as the leader apply worker has exited")));
        }
 
        SpinLockAcquire(&leader->relmutex);
        oldestxmin = leader->oldest_nonremovable_xid;
        SpinLockRelease(&leader->relmutex);
        LWLockRelease(LogicalRepWorkerLock);
    }
 
    /*
     * Return false if the leader apply worker has stopped retaining
     * information for detecting conflicts. This implies that update_deleted
     * can no longer be reliably detected.
     */
    if (!TransactionIdIsValid(oldestxmin))
        return false;
 
    if (OidIsValid(localidxoid) &&
        IsIndexUsableForFindingDeletedTuple(localidxoid, oldestxmin))
        return RelationFindDeletedTupleInfoByIndex(localrel, localidxoid,
                                                   remoteslot, oldestxmin,
                                                   delete_xid, delete_origin,
                                                   delete_time);
    else
        return RelationFindDeletedTupleInfoSeq(localrel, remoteslot,
                                               oldestxmin, delete_xid,
                                               delete_origin, delete_time);
}

References am_leader_apply_worker(), ereport, errcode(), errmsg, ERROR, fb(), InvalidOid, IsIndexUsableForFindingDeletedTuple(), logicalrep_worker_find(), LW_SHARED, LWLockAcquire(), LWLockRelease(), MyLogicalRepWorker, MySubscription, OidIsValid, LogicalRepWorker::oldest_nonremovable_xid, RelationFindDeletedTupleInfoByIndex(), RelationFindDeletedTupleInfoSeq(), LogicalRepWorker::relmutex, Subscription::retaindeadtuples, SpinLockAcquire(), SpinLockRelease(), LogicalRepWorker::subid, track_commit_timestamp, TransactionIdIsValid, and WORKERTYPE_APPLY.

Referenced by apply_handle_tuple_routing(), and apply_handle_update_internal().

FindReplTupleInLocalRel()

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

static

Definition at line 3196 of file worker.c.

{
    EState     *estate = edata->estate;
    bool        found;
 
    /*
     * Regardless of the top-level operation, we're performing a read here, so
     * check for SELECT privileges.
     */
    TargetPrivilegesCheck(localrel, ACL_SELECT);
 
    *localslot = table_slot_create(localrel, &estate->es_tupleTable);
 
    Assert(OidIsValid(localidxoid) ||
           (remoterel->replident == REPLICA_IDENTITY_FULL));
 
    if (OidIsValid(localidxoid))
    {
#ifdef USE_ASSERT_CHECKING
        Relation    idxrel = index_open(localidxoid, AccessShareLock);
 
        /* Index must be PK, RI, or usable for REPLICA IDENTITY FULL tables */
        Assert(GetRelationIdentityOrPK(localrel) == localidxoid ||
               (remoterel->replident == REPLICA_IDENTITY_FULL &&
                IsIndexUsableForReplicaIdentityFull(idxrel,
                                                    edata->targetRel->attrmap)));
        index_close(idxrel, AccessShareLock);
#endif
 
        found = RelationFindReplTupleByIndex(localrel, localidxoid,
                                             LockTupleExclusive,
                                             remoteslot, *localslot);
    }
    else
        found = RelationFindReplTupleSeq(localrel, LockTupleExclusive,
                                         remoteslot, *localslot);
 
    return found;
}

References AccessShareLock, ACL_SELECT, Assert, EState::es_tupleTable, fb(), GetRelationIdentityOrPK(), index_close(), index_open(), IsIndexUsableForReplicaIdentityFull(), LockTupleExclusive, OidIsValid, RelationFindReplTupleByIndex(), RelationFindReplTupleSeq(), LogicalRepRelation::replident, table_slot_create(), and TargetPrivilegesCheck().

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 935 of file worker.c.

{
    EState     *estate = edata->estate;
 
    /* Handle any queued AFTER triggers. */
    AfterTriggerEndQuery(estate);
 
    /* Shut down tuple routing, if any was done. */
    if (edata->proute)
        ExecCleanupTupleRouting(edata->mtstate, edata->proute);
 
    /*
     * Cleanup.  It might seem that we should call ExecCloseResultRelations()
     * here, but we intentionally don't.  It would close the rel we added to
     * es_opened_result_relations above, which is wrong because we took no
     * corresponding refcount.  We rely on ExecCleanupTupleRouting() to close
     * any other relations opened during execution.
     */
    ExecResetTupleTable(estate->es_tupleTable, false);
    FreeExecutorState(estate);
    pfree(edata);
}

References AfterTriggerEndQuery(), EState::es_tupleTable, ExecCleanupTupleRouting(), ExecResetTupleTable(), fb(), FreeExecutorState(), and pfree().

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

get_candidate_xid()

static void get_candidate_xid ( RetainDeadTuplesData *  rdt_data )

static

Definition at line 4475 of file worker.c.

{
    TransactionId oldest_running_xid;
    TimestampTz now;
 
    /*
     * Use last_recv_time when applying changes in the loop to avoid
     * unnecessary system time retrieval. If last_recv_time is not available,
     * obtain the current timestamp.
     */
    now = rdt_data->last_recv_time ? rdt_data->last_recv_time : GetCurrentTimestamp();
 
    /*
     * Compute the candidate_xid and request the publisher status at most once
     * per xid_advance_interval. Refer to adjust_xid_advance_interval() for
     * details on how this value is dynamically adjusted. This is to avoid
     * using CPU and network resources without making much progress.
     */
    if (!TimestampDifferenceExceeds(rdt_data->candidate_xid_time, now,
                                    rdt_data->xid_advance_interval))
        return;
 
    /*
     * Immediately update the timer, even if the function returns later
     * without setting candidate_xid due to inactivity on the subscriber. This
     * avoids frequent calls to GetOldestActiveTransactionId.
     */
    rdt_data->candidate_xid_time = now;
 
    /*
     * Consider transactions in the current database, as only dead tuples from
     * this database are required for conflict detection.
     */
    oldest_running_xid = GetOldestActiveTransactionId(false, false);
 
    /*
     * Oldest active transaction ID (oldest_running_xid) can't be behind any
     * of its previously computed value.
     */
    Assert(TransactionIdPrecedesOrEquals(MyLogicalRepWorker->oldest_nonremovable_xid,
                                         oldest_running_xid));
 
    /* Return if the oldest_nonremovable_xid cannot be advanced */
    if (TransactionIdEquals(MyLogicalRepWorker->oldest_nonremovable_xid,
                            oldest_running_xid))
    {
        adjust_xid_advance_interval(rdt_data, false);
        return;
    }
 
    adjust_xid_advance_interval(rdt_data, true);
 
    rdt_data->candidate_xid = oldest_running_xid;
    rdt_data->phase = RDT_REQUEST_PUBLISHER_STATUS;
 
    /* process the next phase */
    process_rdt_phase_transition(rdt_data, false);
}

References adjust_xid_advance_interval(), Assert, fb(), GetCurrentTimestamp(), GetOldestActiveTransactionId(), MyLogicalRepWorker, now(), LogicalRepWorker::oldest_nonremovable_xid, process_rdt_phase_transition(), RDT_REQUEST_PUBLISHER_STATUS, TimestampDifferenceExceeds(), TransactionIdEquals, and TransactionIdPrecedesOrEquals().

Referenced by process_rdt_phase_transition().

get_flush_position()

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

static

Definition at line 3917 of file worker.c.

{
    dlist_mutable_iter iter;
    XLogRecPtr  local_flush = GetFlushRecPtr(NULL);
 
    *write = InvalidXLogRecPtr;
    *flush = InvalidXLogRecPtr;
 
    dlist_foreach_modify(iter, &lsn_mapping)
    {
        FlushPosition *pos =
            dlist_container(FlushPosition, node, iter.cur);
 
        *write = pos->remote_end;
 
        if (pos->local_end <= local_flush)
        {
            *flush = pos->remote_end;
            dlist_delete(iter.cur);
            pfree(pos);
        }
        else
        {
            /*
             * Don't want to uselessly iterate over the rest of the list which
             * could potentially be long. Instead get the last element and
             * grab the write position from there.
             */
            pos = dlist_tail_element(FlushPosition, node,
                                     &lsn_mapping);
            *write = pos->remote_end;
            *have_pending_txes = true;
            return;
        }
    }
 
    *have_pending_txes = !dlist_is_empty(&lsn_mapping);
}

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

Referenced by send_feedback(), and wait_for_local_flush().

get_transaction_apply_action()

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

static

Definition at line 6395 of file worker.c.

{
    *winfo = NULL;
 
    if (am_parallel_apply_worker())
    {
        return TRANS_PARALLEL_APPLY;
    }
 
    /*
     * If we are processing this transaction using a parallel apply worker
     * then either we send the changes to the parallel worker or if the worker
     * is busy then serialize the changes to the file which will later be
     * processed by the parallel worker.
     */
    *winfo = pa_find_worker(xid);
 
    if (*winfo && (*winfo)->serialize_changes)
    {
        return TRANS_LEADER_PARTIAL_SERIALIZE;
    }
    else if (*winfo)
    {
        return TRANS_LEADER_SEND_TO_PARALLEL;
    }
 
    /*
     * If there is no parallel worker involved to process this transaction
     * then we either directly apply the change or serialize it to a file
     * which will later be applied when the transaction finish message is
     * processed.
     */
    else if (in_streamed_transaction)
    {
        return TRANS_LEADER_SERIALIZE;
    }
    else
    {
        return TRANS_LEADER_APPLY;
    }
}

References am_parallel_apply_worker(), fb(), 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 784 of file worker.c.

{
    TransactionId current_xid;
    ParallelApplyWorkerInfo *winfo;
    TransApplyAction apply_action;
    StringInfoData original_msg;
 
    apply_action = get_transaction_apply_action(stream_xid, &winfo);
 
    /* not in streaming mode */
    if (apply_action == TRANS_LEADER_APPLY)
        return false;
 
    Assert(TransactionIdIsValid(stream_xid));
 
    /*
     * The parallel apply worker needs the xid in this message to decide
     * whether to define a savepoint, so save the original message that has
     * not moved the cursor after the xid. We will serialize this message to a
     * file in PARTIAL_SERIALIZE mode.
     */
    original_msg = *s;
 
    /*
     * We should have received XID of the subxact as the first part of the
     * message, so extract it.
     */
    current_xid = pq_getmsgint(s, 4);
 
    if (!TransactionIdIsValid(current_xid))
        ereport(ERROR,
                (errcode(ERRCODE_PROTOCOL_VIOLATION),
                 errmsg_internal("invalid transaction ID in streamed replication transaction")));
 
    switch (apply_action)
    {
        case TRANS_LEADER_SERIALIZE:
            Assert(stream_fd);
 
            /* Add the new subxact to the array (unless already there). */
            subxact_info_add(current_xid);
 
            /* Write the change to the current file */
            stream_write_change(action, s);
            return true;
 
        case TRANS_LEADER_SEND_TO_PARALLEL:
            Assert(winfo);
 
            /*
             * XXX The publisher side doesn't always send relation/type update
             * messages after the streaming transaction, so also update the
             * relation/type in leader apply worker. See function
             * cleanup_rel_sync_cache.
             */
            if (pa_send_data(winfo, s->len, s->data))
                return (action != LOGICAL_REP_MSG_RELATION &&
                        action != LOGICAL_REP_MSG_TYPE);
 
            /*
             * Switch to serialize mode when we are not able to send the
             * change to parallel apply worker.
             */
            pa_switch_to_partial_serialize(winfo, false);
 
            pg_fallthrough;
        case TRANS_LEADER_PARTIAL_SERIALIZE:
            stream_write_change(action, &original_msg);
 
            /* Same reason as TRANS_LEADER_SEND_TO_PARALLEL case. */
            return (action != LOGICAL_REP_MSG_RELATION &&
                    action != LOGICAL_REP_MSG_TYPE);
 
        case TRANS_PARALLEL_APPLY:
            parallel_stream_nchanges += 1;
 
            /* Define a savepoint for a subxact if needed. */
            pa_start_subtrans(current_xid, stream_xid);
            return false;
 
        default:
            elog(ERROR, "unexpected apply action: %d", (int) apply_action);
            return false;       /* silence compiler warning */
    }
}

References Assert, StringInfoData::data, elog, ereport, errcode(), ERRCODE_PROTOCOL_VIOLATION, errmsg_internal(), ERROR, fb(), 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, pg_fallthrough, 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 5795 of file worker.c.

{
    /* Run as replica session replication role. */
    SetConfigOption("session_replication_role", "replica",
                    PGC_SUSET, PGC_S_OVERRIDE);
 
    /* Connect to our database. */
    BackgroundWorkerInitializeConnectionByOid(MyLogicalRepWorker->dbid,
                                              MyLogicalRepWorker->userid,
                                              0);
 
    /*
     * Set always-secure search path, so malicious users can't redirect user
     * code (e.g. pg_index.indexprs).
     */
    SetConfigOption("search_path", "", PGC_SUSET, PGC_S_OVERRIDE);
 
    ApplyContext = AllocSetContextCreate(TopMemoryContext,
                                         "ApplyContext",
                                         ALLOCSET_DEFAULT_SIZES);
 
    StartTransactionCommand();
 
    /*
     * Lock the subscription to prevent it from being concurrently dropped,
     * then re-verify its existence. After the initialization, the worker will
     * be terminated gracefully if the subscription is dropped.
     */
    LockSharedObject(SubscriptionRelationId, MyLogicalRepWorker->subid, 0,
                     AccessShareLock);
 
    MySubscription = GetSubscription(MyLogicalRepWorker->subid, true, true);
 
    if (MySubscription)
    {
        MemoryContextSetParent(MySubscription->cxt, ApplyContext);
    }
    else
    {
        ereport(LOG,
                (errmsg("logical replication worker for subscription %u will not start because the subscription was removed during startup",
                        MyLogicalRepWorker->subid)));
 
        /* Ensure we remove no-longer-useful entry for worker's start time */
        if (am_leader_apply_worker())
            ApplyLauncherForgetWorkerStartTime(MyLogicalRepWorker->subid);
 
        proc_exit(0);
    }
 
    MySubscriptionValid = true;
 
    if (!MySubscription->enabled)
    {
        ereport(LOG,
                (errmsg("logical replication worker for subscription \"%s\" will not start because the subscription was disabled during startup",
                        MySubscription->name)));
 
        apply_worker_exit();
    }
 
    /*
     * Restart the worker if retain_dead_tuples was enabled during startup.
     *
     * At this point, the replication slot used for conflict detection might
     * not exist yet, or could be dropped soon if the launcher perceives
     * retain_dead_tuples as disabled. To avoid unnecessary tracking of
     * oldest_nonremovable_xid when the slot is absent or at risk of being
     * dropped, a restart is initiated.
     *
     * The oldest_nonremovable_xid should be initialized only when the
     * subscription's retention is active before launching the worker. See
     * logicalrep_worker_launch.
     */
    if (am_leader_apply_worker() &&
        MySubscription->retaindeadtuples &&
        MySubscription->retentionactive &&
        !TransactionIdIsValid(MyLogicalRepWorker->oldest_nonremovable_xid))
    {
        ereport(LOG,
                errmsg("logical replication worker for subscription \"%s\" will restart because the option %s was enabled during startup",
                       MySubscription->name, "retain_dead_tuples"));
 
        apply_worker_exit();
    }
 
    /* Setup synchronous commit according to the user's wishes */
    SetConfigOption("synchronous_commit", MySubscription->synccommit,
                    PGC_BACKEND, PGC_S_OVERRIDE);
 
    /* Change wal_receiver_timeout according to the user's wishes */
    set_wal_receiver_timeout();
 
    /*
     * Keep us informed about subscription or role changes. Note that the
     * role's superuser privilege can be revoked.
     */
    CacheRegisterSyscacheCallback(SUBSCRIPTIONOID,
                                  subscription_change_cb,
                                  (Datum) 0);
    /* Changes to foreign servers may affect subscriptions using SERVER. */
    CacheRegisterSyscacheCallback(FOREIGNSERVEROID,
                                  subscription_change_cb,
                                  (Datum) 0);
    /* Changes to user mappings may affect subscriptions using SERVER. */
    CacheRegisterSyscacheCallback(USERMAPPINGOID,
                                  subscription_change_cb,
                                  (Datum) 0);
 
    /*
     * Changes to FDW connection_function may affect subscriptions using
     * SERVER.
     */
    CacheRegisterSyscacheCallback(FOREIGNDATAWRAPPEROID,
                                  subscription_change_cb,
                                  (Datum) 0);
 
    CacheRegisterSyscacheCallback(AUTHOID,
                                  subscription_change_cb,
                                  (Datum) 0);
 
    if (am_tablesync_worker())
        ereport(LOG,
                errmsg("logical replication table synchronization worker for subscription \"%s\", table \"%s\" has started",
                       MySubscription->name,
                       get_rel_name(MyLogicalRepWorker->relid)));
    else if (am_sequencesync_worker())
        ereport(LOG,
                errmsg("logical replication sequence synchronization worker for subscription \"%s\" has started",
                       MySubscription->name));
    else
        ereport(LOG,
                errmsg("logical replication apply worker for subscription \"%s\" has started",
                       MySubscription->name));
 
    CommitTransactionCommand();
 
    /*
     * Register a callback to reset the origin state before aborting any
     * pending transaction during shutdown (see ShutdownPostgres()). This will
     * avoid origin advancement for an incomplete transaction which could
     * otherwise lead to its loss as such a transaction won't be sent by the
     * server again.
     *
     * Note that even a LOG or DEBUG statement placed after setting the origin
     * state may process a shutdown signal before committing the current apply
     * operation. So, it is important to register such a callback here.
     *
     * Register this callback here to ensure that all types of logical
     * replication workers that set up origins and apply remote transactions
     * are protected.
     */
    before_shmem_exit(on_exit_clear_xact_state, (Datum) 0);
}

References AccessShareLock, ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate, am_leader_apply_worker(), am_sequencesync_worker(), am_tablesync_worker(), apply_worker_exit(), ApplyContext, ApplyLauncherForgetWorkerStartTime(), BackgroundWorkerInitializeConnectionByOid(), before_shmem_exit(), CacheRegisterSyscacheCallback(), CommitTransactionCommand(), Subscription::cxt, LogicalRepWorker::dbid, Subscription::enabled, ereport, errmsg, fb(), get_rel_name(), GetSubscription(), LockSharedObject(), LOG, MemoryContextSetParent(), MyLogicalRepWorker, MySubscription, MySubscriptionValid, Subscription::name, LogicalRepWorker::oldest_nonremovable_xid, on_exit_clear_xact_state(), PGC_BACKEND, PGC_S_OVERRIDE, PGC_SUSET, proc_exit(), LogicalRepWorker::relid, Subscription::retaindeadtuples, Subscription::retentionactive, set_wal_receiver_timeout(), SetConfigOption(), StartTransactionCommand(), LogicalRepWorker::subid, subscription_change_cb(), Subscription::synccommit, TopMemoryContext, TransactionIdIsValid, and LogicalRepWorker::userid.

Referenced by ParallelApplyWorkerMain(), and SetupApplyOrSyncWorker().

IsIndexUsableForFindingDeletedTuple()

static bool IsIndexUsableForFindingDeletedTuple ( Oid  localindexoid, TransactionId  conflict_detection_xmin  )

static

Definition at line 3257 of file worker.c.

{
    HeapTuple   index_tuple;
    TransactionId index_xmin;
 
    index_tuple = SearchSysCache1(INDEXRELID, ObjectIdGetDatum(localindexoid));
 
    if (!HeapTupleIsValid(index_tuple)) /* should not happen */
        elog(ERROR, "cache lookup failed for index %u", localindexoid);
 
    /*
     * No need to check for a frozen transaction ID, as
     * TransactionIdPrecedes() manages it internally, treating it as falling
     * behind the conflict_detection_xmin.
     */
    index_xmin = HeapTupleHeaderGetXmin(index_tuple->t_data);
 
    ReleaseSysCache(index_tuple);
 
    return TransactionIdPrecedes(index_xmin, conflict_detection_xmin);
}

References elog, ERROR, fb(), HeapTupleHeaderGetXmin(), HeapTupleIsValid, ObjectIdGetDatum(), ReleaseSysCache(), SearchSysCache1(), and TransactionIdPrecedes().

Referenced by FindDeletedTupleInLocalRel().

IsLogicalParallelApplyWorker()

bool IsLogicalParallelApplyWorker

(

void

)

Definition at line 6090 of file worker.c.

{
    return IsLogicalWorker() && am_parallel_apply_worker();
}

References am_parallel_apply_worker(), and IsLogicalWorker().

Referenced by mq_putmessage().

IsLogicalWorker()

bool IsLogicalWorker

(

void

)

Definition at line 6081 of file worker.c.

{
    return MyLogicalRepWorker != NULL;
}

References fb(), and MyLogicalRepWorker.

Referenced by IsLogicalParallelApplyWorker(), and ProcessInterrupts().

LogicalRepApplyLoop()

static void LogicalRepApplyLoop ( XLogRecPtr  last_received )

static

Definition at line 4003 of file worker.c.

{
    TimestampTz last_recv_timestamp = GetCurrentTimestamp();
    bool        ping_sent = false;
    TimeLineID  tli;
    ErrorContextCallback errcallback;
    RetainDeadTuplesData rdt_data = {0};
 
    /*
     * Init the ApplyMessageContext which we clean up after each replication
     * protocol message.
     */
    ApplyMessageContext = AllocSetContextCreate(ApplyContext,
                                                "ApplyMessageContext",
                                                ALLOCSET_DEFAULT_SIZES);
 
    /*
     * This memory context is used for per-stream data when the streaming mode
     * is enabled. This context is reset on each stream stop.
     */
    LogicalStreamingContext = AllocSetContextCreate(ApplyContext,
                                                    "LogicalStreamingContext",
                                                    ALLOCSET_DEFAULT_SIZES);
 
    /* mark as idle, before starting to loop */
    pgstat_report_activity(STATE_IDLE, NULL);
 
    /*
     * Push apply error context callback. Fields will be filled while applying
     * a change.
     */
    errcallback.callback = apply_error_callback;
    errcallback.previous = error_context_stack;
    error_context_stack = &errcallback;
    apply_error_context_stack = error_context_stack;
 
    /* This outer loop iterates once per wait. */
    for (;;)
    {
        pgsocket    fd = PGINVALID_SOCKET;
        int         rc;
        int         len;
        char       *buf = NULL;
        bool        endofstream = false;
        long        wait_time;
 
        CHECK_FOR_INTERRUPTS();
 
        MemoryContextSwitchTo(ApplyMessageContext);
 
        len = walrcv_receive(LogRepWorkerWalRcvConn, &buf, &fd);
 
        if (len != 0)
        {
            /* Loop to process all available data (without blocking). */
            for (;;)
            {
                CHECK_FOR_INTERRUPTS();
 
                if (len == 0)
                {
                    break;
                }
                else if (len < 0)
                {
                    ereport(LOG,
                            (errmsg("data stream from publisher has ended")));
                    endofstream = true;
                    break;
                }
                else
                {
                    int         c;
                    StringInfoData s;
 
                    if (ConfigReloadPending)
                    {
                        ConfigReloadPending = false;
                        ProcessConfigFile(PGC_SIGHUP);
                    }
 
                    /* Reset timeout. */
                    last_recv_timestamp = GetCurrentTimestamp();
                    ping_sent = false;
 
                    rdt_data.last_recv_time = last_recv_timestamp;
 
                    /* Ensure we are reading the data into our memory context. */
                    MemoryContextSwitchTo(ApplyMessageContext);
 
                    initReadOnlyStringInfo(&s, buf, len);
 
                    c = pq_getmsgbyte(&s);
 
                    if (c == PqReplMsg_WALData)
                    {
                        XLogRecPtr  start_lsn;
                        XLogRecPtr  end_lsn;
                        TimestampTz send_time;
 
                        start_lsn = pq_getmsgint64(&s);
                        end_lsn = pq_getmsgint64(&s);
                        send_time = pq_getmsgint64(&s);
 
                        if (last_received < start_lsn)
                            last_received = start_lsn;
 
                        if (last_received < end_lsn)
                            last_received = end_lsn;
 
                        UpdateWorkerStats(last_received, send_time, false);
 
                        apply_dispatch(&s);
 
                        maybe_advance_nonremovable_xid(&rdt_data, false);
                    }
                    else if (c == PqReplMsg_Keepalive)
                    {
                        XLogRecPtr  end_lsn;
                        TimestampTz timestamp;
                        bool        reply_requested;
 
                        end_lsn = pq_getmsgint64(&s);
                        timestamp = pq_getmsgint64(&s);
                        reply_requested = pq_getmsgbyte(&s);
 
                        if (last_received < end_lsn)
                            last_received = end_lsn;
 
                        send_feedback(last_received, reply_requested, false);
 
                        maybe_advance_nonremovable_xid(&rdt_data, false);
 
                        UpdateWorkerStats(last_received, timestamp, true);
                    }
                    else if (c == PqReplMsg_PrimaryStatusUpdate)
                    {
                        rdt_data.remote_lsn = pq_getmsgint64(&s);
                        rdt_data.remote_oldestxid = FullTransactionIdFromU64((uint64) pq_getmsgint64(&s));
                        rdt_data.remote_nextxid = FullTransactionIdFromU64((uint64) pq_getmsgint64(&s));
                        rdt_data.reply_time = pq_getmsgint64(&s);
 
                        /*
                         * This should never happen, see
                         * ProcessStandbyPSRequestMessage. But if it happens
                         * due to a bug, we don't want to proceed as it can
                         * incorrectly advance oldest_nonremovable_xid.
                         */
                        if (!XLogRecPtrIsValid(rdt_data.remote_lsn))
                            elog(ERROR, "cannot get the latest WAL position from the publisher");
 
                        maybe_advance_nonremovable_xid(&rdt_data, true);
 
                        UpdateWorkerStats(last_received, rdt_data.reply_time, false);
                    }
                    /* other message types are purposefully ignored */
 
                    MemoryContextReset(ApplyMessageContext);
                }
 
                len = walrcv_receive(LogRepWorkerWalRcvConn, &buf, &fd);
            }
        }
 
        /* confirm all writes so far */
        send_feedback(last_received, false, false);
 
        /* Reset the timestamp if no message was received */
        rdt_data.last_recv_time = 0;
 
        maybe_advance_nonremovable_xid(&rdt_data, false);
 
        if (!in_remote_transaction && !in_streamed_transaction)
        {
            /*
             * If we didn't get any transactions for a while there might be
             * unconsumed invalidation messages in the queue, consume them
             * now.
             */
            AcceptInvalidationMessages();
            maybe_reread_subscription();
 
            /*
             * Process any relations that are being synchronized in parallel
             * and any newly added tables or sequences.
             */
            ProcessSyncingRelations(last_received);
        }
 
        /* Cleanup the memory. */
        MemoryContextReset(ApplyMessageContext);
        MemoryContextSwitchTo(TopMemoryContext);
 
        /* Check if we need to exit the streaming loop. */
        if (endofstream)
            break;
 
        /*
         * Wait for more data or latch.  If we have unflushed transactions,
         * wake up after WalWriterDelay to see if they've been flushed yet (in
         * which case we should send a feedback message).  Otherwise, there's
         * no particular urgency about waking up unless we get data or a
         * signal.
         */
        if (!dlist_is_empty(&lsn_mapping))
            wait_time = WalWriterDelay;
        else
            wait_time = NAPTIME_PER_CYCLE;
 
        /*
         * Ensure to wake up when it's possible to advance the non-removable
         * transaction ID, or when the retention duration may have exceeded
         * max_retention_duration.
         */
        if (MySubscription->retentionactive)
        {
            if (rdt_data.phase == RDT_GET_CANDIDATE_XID &&
                rdt_data.xid_advance_interval)
                wait_time = Min(wait_time, rdt_data.xid_advance_interval);
            else if (MySubscription->maxretention > 0)
                wait_time = Min(wait_time, MySubscription->maxretention);
        }
 
        rc = WaitLatchOrSocket(MyLatch,
                               WL_SOCKET_READABLE | WL_LATCH_SET |
                               WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
                               fd, wait_time,
                               WAIT_EVENT_LOGICAL_APPLY_MAIN);
 
        if (rc & WL_LATCH_SET)
        {
            ResetLatch(MyLatch);
            CHECK_FOR_INTERRUPTS();
        }
 
        if (ConfigReloadPending)
        {
            ConfigReloadPending = false;
            ProcessConfigFile(PGC_SIGHUP);
        }
 
        if (rc & WL_TIMEOUT)
        {
            /*
             * We didn't receive anything new. If we haven't heard anything
             * from the server for more than wal_receiver_timeout / 2, ping
             * the server. Also, if it's been longer than
             * wal_receiver_status_interval since the last update we sent,
             * send a status update to the primary anyway, to report any
             * progress in applying WAL.
             */
            bool        requestReply = false;
 
            /*
             * Check if time since last receive from primary has reached the
             * configured limit.
             */
            if (wal_receiver_timeout > 0)
            {
                TimestampTz now = GetCurrentTimestamp();
                TimestampTz timeout;
 
                timeout =
                    TimestampTzPlusMilliseconds(last_recv_timestamp,
                                                wal_receiver_timeout);
 
                if (now >= timeout)
                    ereport(ERROR,
                            (errcode(ERRCODE_CONNECTION_FAILURE),
                             errmsg("terminating logical replication worker due to timeout")));
 
                /* Check to see if it's time for a ping. */
                if (!ping_sent)
                {
                    timeout = TimestampTzPlusMilliseconds(last_recv_timestamp,
                                                          (wal_receiver_timeout / 2));
                    if (now >= timeout)
                    {
                        requestReply = true;
                        ping_sent = true;
                    }
                }
            }
 
            send_feedback(last_received, requestReply, requestReply);
 
            maybe_advance_nonremovable_xid(&rdt_data, false);
 
            /*
             * Force reporting to ensure long idle periods don't lead to
             * arbitrarily delayed stats. Stats can only be reported outside
             * of (implicit or explicit) transactions. That shouldn't lead to
             * stats being delayed for long, because transactions are either
             * sent as a whole on commit or streamed. Streamed transactions
             * are spilled to disk and applied on commit.
             */
            if (!IsTransactionState())
                pgstat_report_stat(true);
        }
    }
 
    /* Pop the error context stack */
    error_context_stack = errcallback.previous;
    apply_error_context_stack = error_context_stack;
 
    /* All done */
    walrcv_endstreaming(LogRepWorkerWalRcvConn, &tli);
}

References AcceptInvalidationMessages(), ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate, apply_dispatch(), apply_error_callback(), apply_error_context_stack, ApplyContext, ApplyMessageContext, buf, ErrorContextCallback::callback, CHECK_FOR_INTERRUPTS, ConfigReloadPending, dlist_is_empty(), elog, ereport, errcode(), errmsg, ERROR, error_context_stack, fb(), fd(), FullTransactionIdFromU64(), GetCurrentTimestamp(), in_remote_transaction, in_streamed_transaction, initReadOnlyStringInfo(), IsTransactionState(), len, LOG, LogicalStreamingContext, LogRepWorkerWalRcvConn, lsn_mapping, Subscription::maxretention, maybe_advance_nonremovable_xid(), maybe_reread_subscription(), MemoryContextReset(), MemoryContextSwitchTo(), Min, MyLatch, MySubscription, NAPTIME_PER_CYCLE, now(), PGC_SIGHUP, PGINVALID_SOCKET, pgstat_report_activity(), pgstat_report_stat(), pq_getmsgbyte(), pq_getmsgint64(), PqReplMsg_Keepalive, PqReplMsg_PrimaryStatusUpdate, PqReplMsg_WALData, ErrorContextCallback::previous, ProcessConfigFile(), ProcessSyncingRelations(), RDT_GET_CANDIDATE_XID, ResetLatch(), Subscription::retentionactive, 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, WL_TIMEOUT, and XLogRecPtrIsValid.

Referenced by start_apply().

LogicalRepWorkersWakeupAtCommit()

void LogicalRepWorkersWakeupAtCommit

(

Oid

subid

)

Definition at line 6334 of file worker.c.

{
    MemoryContext oldcxt;
 
    oldcxt = MemoryContextSwitchTo(TopTransactionContext);
    on_commit_wakeup_workers_subids =
        list_append_unique_oid(on_commit_wakeup_workers_subids, subid);
    MemoryContextSwitchTo(oldcxt);
}

References fb(), list_append_unique_oid(), MemoryContextSwitchTo(), on_commit_wakeup_workers_subids, and TopTransactionContext.

Referenced by AlterObjectRename_internal(), AlterSubscription(), and AlterSubscriptionOwner_internal().

maybe_advance_nonremovable_xid()

static void maybe_advance_nonremovable_xid ( RetainDeadTuplesData *  rdt_data, bool  status_received  )

static

Definition at line 4409 of file worker.c.

{
    if (!can_advance_nonremovable_xid(rdt_data))
        return;
 
    process_rdt_phase_transition(rdt_data, status_received);
}

References can_advance_nonremovable_xid(), fb(), and process_rdt_phase_transition().

Referenced by LogicalRepApplyLoop().

maybe_reread_subscription()

void maybe_reread_subscription

(

void

)

Definition at line 5061 of file worker.c.

{
    Subscription *newsub;
    bool        started_tx = false;
 
    /* When cache state is valid there is nothing to do here. */
    if (MySubscriptionValid)
        return;
 
    /* This function might be called inside or outside of transaction. */
    if (!IsTransactionState())
    {
        StartTransactionCommand();
        started_tx = true;
    }
 
    newsub = GetSubscription(MyLogicalRepWorker->subid, true, true);
 
    if (newsub)
    {
        MemoryContextSetParent(newsub->cxt, ApplyContext);
    }
    else
    {
        /*
         * Exit if the subscription was removed. This normally should not
         * happen as the worker gets killed during DROP SUBSCRIPTION.
         */
        ereport(LOG,
                (errmsg("logical replication worker for subscription \"%s\" will stop because the subscription was removed",
                        MySubscription->name)));
 
        /* Ensure we remove no-longer-useful entry for worker's start time */
        if (am_leader_apply_worker())
            ApplyLauncherForgetWorkerStartTime(MyLogicalRepWorker->subid);
 
        proc_exit(0);
    }
 
    /* Exit if the subscription was disabled. */
    if (!newsub->enabled)
    {
        ereport(LOG,
                (errmsg("logical replication worker for subscription \"%s\" will stop because the subscription was disabled",
                        MySubscription->name)));
 
        apply_worker_exit();
    }
 
    /* !slotname should never happen when enabled is true. */
    Assert(newsub->slotname);
 
    /* two-phase cannot be altered while the worker is running */
    Assert(newsub->twophasestate == MySubscription->twophasestate);
 
    /*
     * Exit if any parameter that affects the remote connection was changed.
     * The launcher will start a new worker but note that the parallel apply
     * worker won't restart if the streaming option's value is changed from
     * 'parallel' to any other value or the server decides not to stream the
     * in-progress transaction.
     */
    if (strcmp(newsub->conninfo, MySubscription->conninfo) != 0 ||
        strcmp(newsub->name, MySubscription->name) != 0 ||
        strcmp(newsub->slotname, MySubscription->slotname) != 0 ||
        newsub->binary != MySubscription->binary ||
        newsub->stream != MySubscription->stream ||
        newsub->passwordrequired != MySubscription->passwordrequired ||
        strcmp(newsub->origin, MySubscription->origin) != 0 ||
        newsub->owner != MySubscription->owner ||
        !equal(newsub->publications, MySubscription->publications))
    {
        if (am_parallel_apply_worker())
            ereport(LOG,
                    (errmsg("logical replication parallel apply worker for subscription \"%s\" will stop because of a parameter change",
                            MySubscription->name)));
        else
            ereport(LOG,
                    (errmsg("logical replication worker for subscription \"%s\" will restart because of a parameter change",
                            MySubscription->name)));
 
        apply_worker_exit();
    }
 
    /*
     * Exit if the subscription owner's superuser privileges have been
     * revoked.
     */
    if (!newsub->ownersuperuser && MySubscription->ownersuperuser)
    {
        if (am_parallel_apply_worker())
            ereport(LOG,
                    errmsg("logical replication parallel apply worker for subscription \"%s\" will stop because the subscription owner's superuser privileges have been revoked",
                           MySubscription->name));
        else
            ereport(LOG,
                    errmsg("logical replication worker for subscription \"%s\" will restart because the subscription owner's superuser privileges have been revoked",
                           MySubscription->name));
 
        apply_worker_exit();
    }
 
    /* Check for other changes that should never happen too. */
    if (newsub->dbid != MySubscription->dbid)
    {
        elog(ERROR, "subscription %u changed unexpectedly",
             MyLogicalRepWorker->subid);
    }
 
    /* Clean old subscription info and switch to new one. */
    MemoryContextDelete(MySubscription->cxt);
    MySubscription = newsub;
 
    /* Change synchronous commit according to the user's wishes */
    SetConfigOption("synchronous_commit", MySubscription->synccommit,
                    PGC_BACKEND, PGC_S_OVERRIDE);
 
    /* Change wal_receiver_timeout according to the user's wishes */
    set_wal_receiver_timeout();
 
    if (started_tx)
        CommitTransactionCommand();
 
    MySubscriptionValid = true;
}

References am_leader_apply_worker(), am_parallel_apply_worker(), apply_worker_exit(), ApplyContext, ApplyLauncherForgetWorkerStartTime(), Assert, Subscription::binary, CommitTransactionCommand(), Subscription::conninfo, Subscription::cxt, Subscription::dbid, elog, equal(), ereport, errmsg, ERROR, fb(), GetSubscription(), IsTransactionState(), LOG, MemoryContextDelete(), MemoryContextSetParent(), MyLogicalRepWorker, MySubscription, MySubscriptionValid, Subscription::name, newsub(), Subscription::origin, Subscription::owner, Subscription::ownersuperuser, Subscription::passwordrequired, PGC_BACKEND, PGC_S_OVERRIDE, proc_exit(), Subscription::publications, set_wal_receiver_timeout(), 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 6100 of file worker.c.

{
    Assert(!is_skipping_changes());
    Assert(!in_remote_transaction);
    Assert(!in_streamed_transaction);
 
    /*
     * Quick return if it's not requested to skip this transaction. This
     * function is called for every remote transaction and we assume that
     * skipping the transaction is not used often.
     */
    if (likely(!XLogRecPtrIsValid(MySubscription->skiplsn) ||
               MySubscription->skiplsn != finish_lsn))
        return;
 
    /* Start skipping all changes of this transaction */
    skip_xact_finish_lsn = finish_lsn;
 
    ereport(LOG,
            errmsg("logical replication starts skipping transaction at LSN %X/%08X",
                   LSN_FORMAT_ARGS(skip_xact_finish_lsn)));
}

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 XLogRecPtrIsValid.

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

on_exit_clear_xact_state()

static void on_exit_clear_xact_state ( int  code, Datum  arg  )

static

Definition at line 5954 of file worker.c.

{
    replorigin_xact_clear(true);
}

References replorigin_xact_clear().

Referenced by InitializeLogRepWorker().

process_rdt_phase_transition()

static void process_rdt_phase_transition ( RetainDeadTuplesData *  rdt_data, bool  status_received  )

static

Definition at line 4445 of file worker.c.

{
    switch (rdt_data->phase)
    {
        case RDT_GET_CANDIDATE_XID:
            get_candidate_xid(rdt_data);
            break;
        case RDT_REQUEST_PUBLISHER_STATUS:
            request_publisher_status(rdt_data);
            break;
        case RDT_WAIT_FOR_PUBLISHER_STATUS:
            wait_for_publisher_status(rdt_data, status_received);
            break;
        case RDT_WAIT_FOR_LOCAL_FLUSH:
            wait_for_local_flush(rdt_data);
            break;
        case RDT_STOP_CONFLICT_INFO_RETENTION:
            stop_conflict_info_retention(rdt_data);
            break;
        case RDT_RESUME_CONFLICT_INFO_RETENTION:
            resume_conflict_info_retention(rdt_data);
            break;
    }
}

References fb(), get_candidate_xid(), RDT_GET_CANDIDATE_XID, RDT_REQUEST_PUBLISHER_STATUS, RDT_RESUME_CONFLICT_INFO_RETENTION, RDT_STOP_CONFLICT_INFO_RETENTION, RDT_WAIT_FOR_LOCAL_FLUSH, RDT_WAIT_FOR_PUBLISHER_STATUS, request_publisher_status(), resume_conflict_info_retention(), stop_conflict_info_retention(), wait_for_local_flush(), and wait_for_publisher_status().

Referenced by get_candidate_xid(), maybe_advance_nonremovable_xid(), wait_for_local_flush(), and wait_for_publisher_status().

ReplicationOriginNameForLogicalRep()

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

Definition at line 648 of file worker.c.

{
    if (OidIsValid(relid))
    {
        /* Replication origin name for tablesync workers. */
        snprintf(originname, szoriginname, "pg_%u_%u", suboid, relid);
    }
    else
    {
        /* Replication origin name for non-tablesync workers. */
        snprintf(originname, szoriginname, "pg_%u", suboid);
    }
}

References fb(), OidIsValid, and snprintf.

Referenced by AlterSubscription(), AlterSubscription_refresh(), binary_upgrade_replorigin_advance(), CreateSubscription(), DropSubscription(), LogicalRepSyncTableStart(), ParallelApplyWorkerMain(), ProcessSyncingTablesForApply(), ProcessSyncingTablesForSync(), run_apply_worker(), and run_tablesync_worker().

request_publisher_status()

static void request_publisher_status ( RetainDeadTuplesData *  rdt_data )

static

Definition at line 4538 of file worker.c.

{
    static StringInfo request_message = NULL;
 
    if (!request_message)
    {
        MemoryContext oldctx = MemoryContextSwitchTo(ApplyContext);
 
        request_message = makeStringInfo();
        MemoryContextSwitchTo(oldctx);
    }
    else
        resetStringInfo(request_message);
 
    /*
     * Send the current time to update the remote walsender's latest reply
     * message received time.
     */
    pq_sendbyte(request_message, PqReplMsg_PrimaryStatusRequest);
    pq_sendint64(request_message, GetCurrentTimestamp());
 
    elog(DEBUG2, "sending publisher status request message");
 
    /* Send a request for the publisher status */
    walrcv_send(LogRepWorkerWalRcvConn,
                request_message->data, request_message->len);
 
    rdt_data->phase = RDT_WAIT_FOR_PUBLISHER_STATUS;
 
    /*
     * Skip calling maybe_advance_nonremovable_xid() since further transition
     * is possible only once we receive the publisher status message.
     */
}

References ApplyContext, DEBUG2, elog, fb(), GetCurrentTimestamp(), LogRepWorkerWalRcvConn, makeStringInfo(), MemoryContextSwitchTo(), pq_sendbyte(), pq_sendint64(), PqReplMsg_PrimaryStatusRequest, RDT_WAIT_FOR_PUBLISHER_STATUS, resetStringInfo(), and walrcv_send.

Referenced by process_rdt_phase_transition().

reset_apply_error_context_info()

static void reset_apply_error_context_info ( void  )

inlinestatic

Definition at line 6318 of file worker.c.

{
    apply_error_callback_arg.command = 0;
    apply_error_callback_arg.rel = NULL;
    apply_error_callback_arg.remote_attnum = -1;
    set_apply_error_context_xact(InvalidTransactionId, InvalidXLogRecPtr);
}

References apply_error_callback_arg, ApplyErrorCallbackArg::command, fb(), InvalidTransactionId, InvalidXLogRecPtr, ApplyErrorCallbackArg::rel, ApplyErrorCallbackArg::remote_attnum, and set_apply_error_context_xact().

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

reset_retention_data_fields()

static void reset_retention_data_fields ( RetainDeadTuplesData *  rdt_data )

static

Definition at line 4943 of file worker.c.

{
    rdt_data->phase = RDT_GET_CANDIDATE_XID;
    rdt_data->remote_lsn = InvalidXLogRecPtr;
    rdt_data->remote_oldestxid = InvalidFullTransactionId;
    rdt_data->remote_nextxid = InvalidFullTransactionId;
    rdt_data->reply_time = 0;
    rdt_data->remote_wait_for = InvalidFullTransactionId;
    rdt_data->candidate_xid = InvalidTransactionId;
    rdt_data->table_sync_wait_time = 0;
}

References fb(), InvalidFullTransactionId, InvalidTransactionId, InvalidXLogRecPtr, and RDT_GET_CANDIDATE_XID.

Referenced by stop_conflict_info_retention(), and wait_for_local_flush().

resume_conflict_info_retention()

static void resume_conflict_info_retention ( RetainDeadTuplesData *  rdt_data )

static

Definition at line 4864 of file worker.c.

{
    /* We can't resume retention without updating retention status. */
    if (!update_retention_status(true))
        return;
 
    ereport(LOG,
            errmsg("logical replication worker for subscription \"%s\" will resume retaining the information for detecting conflicts",
                   MySubscription->name),
            MySubscription->maxretention
            ? errdetail("Retention is re-enabled because the apply process has caught up with the publisher within the configured max_retention_duration.")
            : errdetail("Retention is re-enabled because max_retention_duration has been set to unlimited."));
 
    /*
     * Restart the worker to let the launcher initialize
     * oldest_nonremovable_xid at startup.
     *
     * While it's technically possible to derive this value on-the-fly using
     * the conflict detection slot's xmin, doing so risks a race condition:
     * the launcher might clean slot.xmin just after retention resumes. This
     * would make oldest_nonremovable_xid unreliable, especially during xid
     * wraparound.
     *
     * Although this can be prevented by introducing heavy weight locking, the
     * complexity it will bring doesn't seem worthwhile given how rarely
     * retention is resumed.
     */
    apply_worker_exit();
}

References apply_worker_exit(), ereport, errdetail(), errmsg, LOG, Subscription::maxretention, MySubscription, Subscription::name, and update_retention_status().

Referenced by process_rdt_phase_transition().

run_apply_worker()

static void run_apply_worker ( void  )

static

Definition at line 5682 of file worker.c.

{
    char        originname[NAMEDATALEN];
    XLogRecPtr  origin_startpos = InvalidXLogRecPtr;
    char       *slotname = NULL;
    WalRcvStreamOptions options;
    ReplOriginId originid;
    TimeLineID  startpointTLI;
    char       *err;
    bool        must_use_password;
 
    slotname = MySubscription->slotname;
 
    /*
     * This shouldn't happen if the subscription is enabled, but guard against
     * DDL bugs or manual catalog changes.  (libpqwalreceiver will crash if
     * slot is NULL.)
     */
    if (!slotname)
        ereport(ERROR,
                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                 errmsg("subscription has no replication slot set")));
 
    /* Setup replication origin tracking. */
    ReplicationOriginNameForLogicalRep(MySubscription->oid, InvalidOid,
                                       originname, sizeof(originname));
    StartTransactionCommand();
    originid = replorigin_by_name(originname, true);
    if (!OidIsValid(originid))
        originid = replorigin_create(originname);
    replorigin_session_setup(originid, 0);
    replorigin_xact_state.origin = originid;
    origin_startpos = replorigin_session_get_progress(false);
    CommitTransactionCommand();
 
    /* Is the use of a password mandatory? */
    must_use_password = MySubscription->passwordrequired &&
        !MySubscription->ownersuperuser;
 
    LogRepWorkerWalRcvConn = walrcv_connect(MySubscription->conninfo, true,
                                            true, must_use_password,
                                            MySubscription->name, &err);
 
    if (LogRepWorkerWalRcvConn == NULL)
        ereport(ERROR,
                (errcode(ERRCODE_CONNECTION_FAILURE),
                 errmsg("apply worker for subscription \"%s\" could not connect to the publisher: %s",
                        MySubscription->name, err)));
 
    /*
     * We don't really use the output identify_system for anything but it does
     * some initializations on the upstream so let's still call it.
     */
    (void) walrcv_identify_system(LogRepWorkerWalRcvConn, &startpointTLI);
 
    set_apply_error_context_origin(originname);
 
    set_stream_options(&options, slotname, &origin_startpos);
 
    /*
     * Even when the two_phase mode is requested by the user, it remains as
     * the tri-state PENDING until all tablesyncs have reached READY state.
     * Only then, can it become ENABLED.
     *
     * Note: If the subscription has no tables then leave the state as
     * PENDING, which allows ALTER SUBSCRIPTION ... REFRESH PUBLICATION to
     * work.
     */
    if (MySubscription->twophasestate == LOGICALREP_TWOPHASE_STATE_PENDING &&
        AllTablesyncsReady())
    {
        /* Start streaming with two_phase enabled */
        options.proto.logical.twophase = true;
        walrcv_startstreaming(LogRepWorkerWalRcvConn, &options);
 
        StartTransactionCommand();
 
        /*
         * Updating pg_subscription might involve TOAST table access, so
         * ensure we have a valid snapshot.
         */
        PushActiveSnapshot(GetTransactionSnapshot());
 
        UpdateTwoPhaseState(MySubscription->oid, LOGICALREP_TWOPHASE_STATE_ENABLED);
        MySubscription->twophasestate = LOGICALREP_TWOPHASE_STATE_ENABLED;
        PopActiveSnapshot();
        CommitTransactionCommand();
    }
    else
    {
        walrcv_startstreaming(LogRepWorkerWalRcvConn, &options);
    }
 
    ereport(DEBUG1,
            (errmsg_internal("logical replication apply worker for subscription \"%s\" two_phase is %s",
                             MySubscription->name,
                             MySubscription->twophasestate == LOGICALREP_TWOPHASE_STATE_DISABLED ? "DISABLED" :
                             MySubscription->twophasestate == LOGICALREP_TWOPHASE_STATE_PENDING ? "PENDING" :
                             MySubscription->twophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED ? "ENABLED" :
                             "?")));
 
    /* Run the main loop. */
    start_apply(origin_startpos);
}

References AllTablesyncsReady(), CommitTransactionCommand(), Subscription::conninfo, DEBUG1, ereport, err(), errcode(), errmsg, errmsg_internal(), ERROR, fb(), GetTransactionSnapshot(), InvalidOid, InvalidXLogRecPtr, LogRepWorkerWalRcvConn, MySubscription, Subscription::name, NAMEDATALEN, Subscription::oid, OidIsValid, ReplOriginXactState::origin, Subscription::ownersuperuser, Subscription::passwordrequired, PopActiveSnapshot(), PushActiveSnapshot(), ReplicationOriginNameForLogicalRep(), replorigin_by_name(), replorigin_create(), replorigin_session_get_progress(), replorigin_session_setup(), replorigin_xact_state, set_apply_error_context_origin(), set_stream_options(), Subscription::slotname, start_apply(), StartTransactionCommand(), Subscription::twophasestate, UpdateTwoPhaseState(), walrcv_connect, walrcv_identify_system, and walrcv_startstreaming.

Referenced by ApplyWorkerMain().

send_feedback()

static void send_feedback ( XLogRecPtr  recvpos, bool  force, bool  requestReply  )

static

Definition at line 4319 of file worker.c.

{
    static StringInfo reply_message = NULL;
    static TimestampTz send_time = 0;
 
    static XLogRecPtr last_recvpos = InvalidXLogRecPtr;
    static XLogRecPtr last_writepos = InvalidXLogRecPtr;
 
    XLogRecPtr  writepos;
    XLogRecPtr  flushpos;
    TimestampTz now;
    bool        have_pending_txes;
 
    /*
     * If the user doesn't want status to be reported to the publisher, be
     * sure to exit before doing anything at all.
     */
    if (!force && wal_receiver_status_interval <= 0)
        return;
 
    /* It's legal to not pass a recvpos */
    if (recvpos < last_recvpos)
        recvpos = last_recvpos;
 
    get_flush_position(&writepos, &flushpos, &have_pending_txes);
 
    /*
     * No outstanding transactions to flush, we can report the latest received
     * position. This is important for synchronous replication.
     */
    if (!have_pending_txes)
        flushpos = writepos = recvpos;
 
    if (writepos < last_writepos)
        writepos = last_writepos;
 
    if (flushpos < last_flushpos)
        flushpos = last_flushpos;
 
    now = GetCurrentTimestamp();
 
    /* if we've already reported everything we're good */
    if (!force &&
        writepos == last_writepos &&
        flushpos == last_flushpos &&
        !TimestampDifferenceExceeds(send_time, now,
                                    wal_receiver_status_interval * 1000))
        return;
    send_time = now;
 
    if (!reply_message)
    {
        MemoryContext oldctx = MemoryContextSwitchTo(ApplyContext);
 
        reply_message = makeStringInfo();
        MemoryContextSwitchTo(oldctx);
    }
    else
        resetStringInfo(reply_message);
 
    pq_sendbyte(reply_message, PqReplMsg_StandbyStatusUpdate);
    pq_sendint64(reply_message, recvpos);   /* write */
    pq_sendint64(reply_message, flushpos);  /* flush */
    pq_sendint64(reply_message, writepos);  /* apply */
    pq_sendint64(reply_message, now);   /* sendTime */
    pq_sendbyte(reply_message, requestReply);   /* replyRequested */
 
    elog(DEBUG2, "sending feedback (force %d) to recv %X/%08X, write %X/%08X, flush %X/%08X",
         force,
         LSN_FORMAT_ARGS(recvpos),
         LSN_FORMAT_ARGS(writepos),
         LSN_FORMAT_ARGS(flushpos));
 
    walrcv_send(LogRepWorkerWalRcvConn,
                reply_message->data, reply_message->len);
 
    if (recvpos > last_recvpos)
        last_recvpos = recvpos;
    if (writepos > last_writepos)
        last_writepos = writepos;
    if (flushpos > last_flushpos)
        last_flushpos = flushpos;
}

References ApplyContext, StringInfoData::data, DEBUG2, elog, fb(), get_flush_position(), GetCurrentTimestamp(), InvalidXLogRecPtr, last_flushpos, StringInfoData::len, LogRepWorkerWalRcvConn, LSN_FORMAT_ARGS, makeStringInfo(), MemoryContextSwitchTo(), now(), pq_sendbyte(), pq_sendint64(), PqReplMsg_StandbyStatusUpdate, reply_message, resetStringInfo(), TimestampDifferenceExceeds(), wal_receiver_status_interval, and walrcv_send.

Referenced by LogicalRepApplyLoop().

set_apply_error_context_origin()

void set_apply_error_context_origin

(

char *

originname

)

Definition at line 6380 of file worker.c.

{
    apply_error_callback_arg.origin_name = MemoryContextStrdup(ApplyContext,
                                                               originname);
}

References apply_error_callback_arg, ApplyContext, fb(), MemoryContextStrdup(), and ApplyErrorCallbackArg::origin_name.

Referenced by ParallelApplyWorkerMain(), run_apply_worker(), and run_tablesync_worker().

set_apply_error_context_xact()

static void set_apply_error_context_xact ( TransactionId  xid, XLogRecPtr  lsn  )

inlinestatic

Definition at line 6310 of file worker.c.

{
    apply_error_callback_arg.remote_xid = xid;
    apply_error_callback_arg.finish_lsn = lsn;
}

References apply_error_callback_arg, ApplyErrorCallbackArg::finish_lsn, and ApplyErrorCallbackArg::remote_xid.

Referenced by apply_handle_begin(), apply_handle_begin_prepare(), apply_handle_commit_prepared(), apply_handle_rollback_prepared(), apply_handle_stream_abort(), apply_handle_stream_commit(), apply_handle_stream_prepare(), apply_handle_stream_start(), and reset_apply_error_context_info().

set_stream_options()

void set_stream_options	(	WalRcvStreamOptions *	options,
		char *	slotname,
		XLogRecPtr *	origin_startpos
	)

Definition at line 5573 of file worker.c.

{
    int         server_version;
 
    options->logical = true;
    options->startpoint = *origin_startpos;
    options->slotname = slotname;
 
    server_version = walrcv_server_version(LogRepWorkerWalRcvConn);
    options->proto.logical.proto_version =
        server_version >= 160000 ? LOGICALREP_PROTO_STREAM_PARALLEL_VERSION_NUM :
        server_version >= 150000 ? LOGICALREP_PROTO_TWOPHASE_VERSION_NUM :
        server_version >= 140000 ? LOGICALREP_PROTO_STREAM_VERSION_NUM :
        LOGICALREP_PROTO_VERSION_NUM;
 
    options->proto.logical.publication_names = MySubscription->publications;
    options->proto.logical.binary = MySubscription->binary;
 
    /*
     * Assign the appropriate option value for streaming option according to
     * the 'streaming' mode and the publisher's ability to support that mode.
     */
    if (server_version >= 160000 &&
        MySubscription->stream == LOGICALREP_STREAM_PARALLEL)
    {
        options->proto.logical.streaming_str = "parallel";
        MyLogicalRepWorker->parallel_apply = true;
    }
    else if (server_version >= 140000 &&
             MySubscription->stream != LOGICALREP_STREAM_OFF)
    {
        options->proto.logical.streaming_str = "on";
        MyLogicalRepWorker->parallel_apply = false;
    }
    else
    {
        options->proto.logical.streaming_str = NULL;
        MyLogicalRepWorker->parallel_apply = false;
    }
 
    options->proto.logical.twophase = false;
    options->proto.logical.origin = pstrdup(MySubscription->origin);
}

References Subscription::binary, fb(), LOGICALREP_PROTO_STREAM_PARALLEL_VERSION_NUM, LOGICALREP_PROTO_STREAM_VERSION_NUM, LOGICALREP_PROTO_TWOPHASE_VERSION_NUM, LOGICALREP_PROTO_VERSION_NUM, LogRepWorkerWalRcvConn, MyLogicalRepWorker, MySubscription, Subscription::origin, LogicalRepWorker::parallel_apply, pstrdup(), Subscription::publications, server_version, Subscription::stream, and walrcv_server_version.

Referenced by run_apply_worker(), and run_tablesync_worker().

set_wal_receiver_timeout()

static void set_wal_receiver_timeout ( void  )

static

Definition at line 5191 of file worker.c.

{
    bool        parsed;
    int         val;
    int         prev_timeout = wal_receiver_timeout;
 
    /*
     * Set the wal_receiver_timeout GUC to MySubscription->walrcvtimeout,
     * which comes from the subscription's wal_receiver_timeout option. If the
     * value is -1, reset the GUC to its default, meaning it will inherit from
     * the server config, command line, or role/database settings.
     */
    parsed = parse_int(MySubscription->walrcvtimeout, &val, 0, NULL);
    if (parsed && val == -1)
        SetConfigOption("wal_receiver_timeout", NULL,
                        PGC_BACKEND, PGC_S_SESSION);
    else
        SetConfigOption("wal_receiver_timeout", MySubscription->walrcvtimeout,
                        PGC_BACKEND, PGC_S_SESSION);
 
    /*
     * Log the wal_receiver_timeout setting (in milliseconds) as a debug
     * message when it changes, to verify it was set correctly.
     */
    if (prev_timeout != wal_receiver_timeout)
        elog(DEBUG1, "logical replication worker for subscription \"%s\" wal_receiver_timeout: %d ms",
             MySubscription->name, wal_receiver_timeout);
}

References DEBUG1, elog, fb(), MySubscription, Subscription::name, parse_int(), PGC_BACKEND, PGC_S_SESSION, SetConfigOption(), val, wal_receiver_timeout, and Subscription::walrcvtimeout.

Referenced by InitializeLogRepWorker(), and maybe_reread_subscription().

SetupApplyOrSyncWorker()

void SetupApplyOrSyncWorker

(

int

worker_slot

)

Definition at line 5963 of file worker.c.

{
    /* Attach to slot */
    logicalrep_worker_attach(worker_slot);
 
    Assert(am_tablesync_worker() || am_sequencesync_worker() || am_leader_apply_worker());
 
    /* Setup signal handling */
    pqsignal(SIGHUP, SignalHandlerForConfigReload);
    BackgroundWorkerUnblockSignals();
 
    /*
     * We don't currently need any ResourceOwner in a walreceiver process, but
     * if we did, we could call CreateAuxProcessResourceOwner here.
     */
 
    /* Initialise stats to a sanish value */
    MyLogicalRepWorker->last_send_time = MyLogicalRepWorker->last_recv_time =
        MyLogicalRepWorker->reply_time = GetCurrentTimestamp();
 
    /* Load the libpq-specific functions */
    load_file("libpqwalreceiver", false);
 
    InitializeLogRepWorker();
 
    /* Connect to the origin and start the replication. */
    elog(DEBUG1, "connecting to publisher using connection string \"%s\"",
         MySubscription->conninfo);
 
    /*
     * Setup callback for syscache so that we know when something changes in
     * the subscription relation state.
     */
    CacheRegisterSyscacheCallback(SUBSCRIPTIONRELMAP,
                                  InvalidateSyncingRelStates,
                                  (Datum) 0);
}

References am_leader_apply_worker(), am_sequencesync_worker(), am_tablesync_worker(), Assert, BackgroundWorkerUnblockSignals(), CacheRegisterSyscacheCallback(), Subscription::conninfo, DEBUG1, elog, fb(), GetCurrentTimestamp(), InitializeLogRepWorker(), InvalidateSyncingRelStates(), LogicalRepWorker::last_recv_time, LogicalRepWorker::last_send_time, load_file(), logicalrep_worker_attach(), MyLogicalRepWorker, MySubscription, pqsignal, LogicalRepWorker::reply_time, SIGHUP, and SignalHandlerForConfigReload().

Referenced by ApplyWorkerMain(), SequenceSyncWorkerMain(), and TableSyncWorkerMain().

should_apply_changes_for_rel()

static bool should_apply_changes_for_rel ( LogicalRepRelMapEntry *  rel )

static

Definition at line 688 of file worker.c.

{
    switch (MyLogicalRepWorker->type)
    {
        case WORKERTYPE_TABLESYNC:
            return MyLogicalRepWorker->relid == rel->localreloid;
 
        case WORKERTYPE_PARALLEL_APPLY:
            /* We don't synchronize rel's that are in unknown state. */
            if (rel->state != SUBREL_STATE_READY &&
                rel->state != SUBREL_STATE_UNKNOWN)
                ereport(ERROR,
                        (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                         errmsg("logical replication parallel apply worker for subscription \"%s\" will stop",
                                MySubscription->name),
                         errdetail("Cannot handle streamed replication transactions using parallel apply workers until all tables have been synchronized.")));
 
            return rel->state == SUBREL_STATE_READY;
 
        case WORKERTYPE_APPLY:
            return (rel->state == SUBREL_STATE_READY ||
                    (rel->state == SUBREL_STATE_SYNCDONE &&
                     rel->statelsn <= remote_final_lsn));
 
        case WORKERTYPE_SEQUENCESYNC:
            /* Should never happen. */
            elog(ERROR, "sequence synchronization worker is not expected to apply changes");
            break;
 
        case WORKERTYPE_UNKNOWN:
            /* Should never happen. */
            elog(ERROR, "Unknown worker type");
    }
 
    return false;               /* dummy for compiler */
}

References elog, ereport, errcode(), errdetail(), errmsg, ERROR, fb(), LogicalRepRelMapEntry::localreloid, MyLogicalRepWorker, MySubscription, Subscription::name, LogicalRepWorker::relid, remote_final_lsn, LogicalRepRelMapEntry::state, LogicalRepRelMapEntry::statelsn, LogicalRepWorker::type, WORKERTYPE_APPLY, WORKERTYPE_PARALLEL_APPLY, WORKERTYPE_SEQUENCESYNC, WORKERTYPE_TABLESYNC, and WORKERTYPE_UNKNOWN.

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

should_stop_conflict_info_retention()

static bool should_stop_conflict_info_retention ( RetainDeadTuplesData *  rdt_data )

static

Definition at line 4791 of file worker.c.

{
    TimestampTz now;
 
    Assert(TransactionIdIsValid(rdt_data->candidate_xid));
    Assert(rdt_data->phase == RDT_WAIT_FOR_PUBLISHER_STATUS ||
           rdt_data->phase == RDT_WAIT_FOR_LOCAL_FLUSH);
 
    if (!MySubscription->maxretention)
        return false;
 
    /*
     * Use last_recv_time when applying changes in the loop to avoid
     * unnecessary system time retrieval. If last_recv_time is not available,
     * obtain the current timestamp.
     */
    now = rdt_data->last_recv_time ? rdt_data->last_recv_time : GetCurrentTimestamp();
 
    /*
     * Return early if the wait time has not exceeded the configured maximum
     * (max_retention_duration). Time spent waiting for table synchronization
     * is excluded from this calculation, as it occurs infrequently.
     */
    if (!TimestampDifferenceExceeds(rdt_data->candidate_xid_time, now,
                                    MySubscription->maxretention +
                                    rdt_data->table_sync_wait_time))
        return false;
 
    return true;
}

References Assert, fb(), GetCurrentTimestamp(), Subscription::maxretention, MySubscription, now(), RDT_WAIT_FOR_LOCAL_FLUSH, RDT_WAIT_FOR_PUBLISHER_STATUS, TimestampDifferenceExceeds(), and TransactionIdIsValid.

Referenced by wait_for_local_flush(), and wait_for_publisher_status().

slot_fill_defaults()

static void slot_fill_defaults ( LogicalRepRelMapEntry *  rel, EState *  estate, TupleTableSlot *  slot  )

static

Definition at line 966 of file worker.c.

{
    TupleDesc   desc = RelationGetDescr(rel->localrel);
    int         num_phys_attrs = desc->natts;
    int         i;
    int         attnum,
                num_defaults = 0;
    int        *defmap;
    ExprState **defexprs;
    ExprContext *econtext;
 
    econtext = GetPerTupleExprContext(estate);
 
    /* We got all the data via replication, no need to evaluate anything. */
    if (num_phys_attrs == rel->remoterel.natts)
        return;
 
    defmap = palloc_array(int, num_phys_attrs);
    defexprs = palloc_array(ExprState *, num_phys_attrs);
 
    Assert(rel->attrmap->maplen == num_phys_attrs);
    for (attnum = 0; attnum < num_phys_attrs; attnum++)
    {
        CompactAttribute *cattr = TupleDescCompactAttr(desc, attnum);
        Expr       *defexpr;
 
        if (cattr->attisdropped || cattr->attgenerated)
            continue;
 
        if (rel->attrmap->attnums[attnum] >= 0)
            continue;
 
        defexpr = (Expr *) build_column_default(rel->localrel, attnum + 1);
 
        if (defexpr != NULL)
        {
            /* Run the expression through planner */
            defexpr = expression_planner(defexpr);
 
            /* Initialize executable expression in copycontext */
            defexprs[num_defaults] = ExecInitExpr(defexpr, NULL);
            defmap[num_defaults] = attnum;
            num_defaults++;
        }
    }
 
    for (i = 0; i < num_defaults; i++)
        slot->tts_values[defmap[i]] =
            ExecEvalExpr(defexprs[i], econtext, &slot->tts_isnull[defmap[i]]);
}

References Assert, attnum, AttrMap::attnums, LogicalRepRelMapEntry::attrmap, build_column_default(), ExecEvalExpr(), ExecInitExpr(), expression_planner(), fb(), GetPerTupleExprContext, i, LogicalRepRelMapEntry::localrel, AttrMap::maplen, TupleDescData::natts, LogicalRepRelation::natts, palloc_array, RelationGetDescr, LogicalRepRelMapEntry::remoterel, TupleTableSlot::tts_isnull, TupleTableSlot::tts_values, and TupleDescCompactAttr().

Referenced by apply_handle_insert().

slot_modify_data()

static void slot_modify_data ( TupleTableSlot *  slot, TupleTableSlot *  srcslot, LogicalRepRelMapEntry *  rel, LogicalRepTupleData *  tupleData  )

static

Definition at line 1131 of file worker.c.

{
    int         natts = slot->tts_tupleDescriptor->natts;
    int         i;
 
    /* We'll fill "slot" with a virtual tuple, so we must start with ... */
    ExecClearTuple(slot);
 
    /*
     * Copy all the column data from srcslot, so that we'll have valid values
     * for unreplaced columns.
     */
    Assert(natts == srcslot->tts_tupleDescriptor->natts);
    slot_getallattrs(srcslot);
    memcpy(slot->tts_values, srcslot->tts_values, natts * sizeof(Datum));
    memcpy(slot->tts_isnull, srcslot->tts_isnull, natts * sizeof(bool));
 
    /* Call the "in" function for each replaced attribute */
    Assert(natts == rel->attrmap->maplen);
    for (i = 0; i < natts; i++)
    {
        Form_pg_attribute att = TupleDescAttr(slot->tts_tupleDescriptor, i);
        int         remoteattnum = rel->attrmap->attnums[i];
 
        if (remoteattnum < 0)
            continue;
 
        if (remoteattnum >= tupleData->ncols)
            ereport(ERROR,
                    (errcode(ERRCODE_PROTOCOL_VIOLATION),
                     errmsg("logical replication column %d not found in tuple: only %d column(s) received",
                            remoteattnum + 1, tupleData->ncols)));
 
        if (tupleData->colstatus[remoteattnum] != LOGICALREP_COLUMN_UNCHANGED)
        {
            StringInfo  colvalue = &tupleData->colvalues[remoteattnum];
 
            /* Set attnum for error callback */
            apply_error_callback_arg.remote_attnum = remoteattnum;
 
            if (tupleData->colstatus[remoteattnum] == LOGICALREP_COLUMN_TEXT)
            {
                Oid         typinput;
                Oid         typioparam;
 
                getTypeInputInfo(att->atttypid, &typinput, &typioparam);
                slot->tts_values[i] =
                    OidInputFunctionCall(typinput, colvalue->data,
                                         typioparam, att->atttypmod);
                slot->tts_isnull[i] = false;
            }
            else if (tupleData->colstatus[remoteattnum] == LOGICALREP_COLUMN_BINARY)
            {
                Oid         typreceive;
                Oid         typioparam;
 
                /*
                 * In some code paths we may be asked to re-parse the same
                 * tuple data.  Reset the StringInfo's cursor so that works.
                 */
                colvalue->cursor = 0;
 
                getTypeBinaryInputInfo(att->atttypid, &typreceive, &typioparam);
                slot->tts_values[i] =
                    OidReceiveFunctionCall(typreceive, colvalue,
                                           typioparam, att->atttypmod);
 
                /* Trouble if it didn't eat the whole buffer */
                if (colvalue->cursor != colvalue->len)
                    ereport(ERROR,
                            (errcode(ERRCODE_INVALID_BINARY_REPRESENTATION),
                             errmsg("incorrect binary data format in logical replication column %d",
                                    remoteattnum + 1)));
                slot->tts_isnull[i] = false;
            }
            else
            {
                /* must be LOGICALREP_COLUMN_NULL */
                slot->tts_values[i] = (Datum) 0;
                slot->tts_isnull[i] = true;
            }
 
            /* Reset attnum for error callback */
            apply_error_callback_arg.remote_attnum = -1;
        }
    }
 
    /* And finally, declare that "slot" contains a valid virtual tuple */
    ExecStoreVirtualTuple(slot);
}

References apply_error_callback_arg, Assert, AttrMap::attnums, LogicalRepRelMapEntry::attrmap, ereport, errcode(), ERRCODE_PROTOCOL_VIOLATION, errmsg, ERROR, ExecClearTuple(), ExecStoreVirtualTuple(), fb(), getTypeBinaryInputInfo(), getTypeInputInfo(), i, LOGICALREP_COLUMN_BINARY, LOGICALREP_COLUMN_TEXT, LOGICALREP_COLUMN_UNCHANGED, AttrMap::maplen, memcpy(), TupleDescData::natts, OidInputFunctionCall(), OidReceiveFunctionCall(), ApplyErrorCallbackArg::remote_attnum, slot_getallattrs(), TupleTableSlot::tts_isnull, TupleTableSlot::tts_tupleDescriptor, TupleTableSlot::tts_values, and TupleDescAttr().

Referenced by apply_handle_tuple_routing(), and apply_handle_update_internal().

slot_store_data()

static void slot_store_data ( TupleTableSlot *  slot, LogicalRepRelMapEntry *  rel, LogicalRepTupleData *  tupleData  )

static

Definition at line 1024 of file worker.c.

{
    int         natts = slot->tts_tupleDescriptor->natts;
    int         i;
 
    ExecClearTuple(slot);
 
    /* Call the "in" function for each non-dropped, non-null attribute */
    Assert(natts == rel->attrmap->maplen);
    for (i = 0; i < natts; i++)
    {
        Form_pg_attribute att = TupleDescAttr(slot->tts_tupleDescriptor, i);
        int         remoteattnum = rel->attrmap->attnums[i];
 
        if (!att->attisdropped && remoteattnum >= 0)
        {
            StringInfo  colvalue;
 
            if (remoteattnum >= tupleData->ncols)
                ereport(ERROR,
                        (errcode(ERRCODE_PROTOCOL_VIOLATION),
                         errmsg("logical replication column %d not found in tuple: only %d column(s) received",
                                remoteattnum + 1, tupleData->ncols)));
 
            colvalue = &tupleData->colvalues[remoteattnum];
 
            /* Set attnum for error callback */
            apply_error_callback_arg.remote_attnum = remoteattnum;
 
            if (tupleData->colstatus[remoteattnum] == LOGICALREP_COLUMN_TEXT)
            {
                Oid         typinput;
                Oid         typioparam;
 
                getTypeInputInfo(att->atttypid, &typinput, &typioparam);
                slot->tts_values[i] =
                    OidInputFunctionCall(typinput, colvalue->data,
                                         typioparam, att->atttypmod);
                slot->tts_isnull[i] = false;
            }
            else if (tupleData->colstatus[remoteattnum] == LOGICALREP_COLUMN_BINARY)
            {
                Oid         typreceive;
                Oid         typioparam;
 
                /*
                 * In some code paths we may be asked to re-parse the same
                 * tuple data.  Reset the StringInfo's cursor so that works.
                 */
                colvalue->cursor = 0;
 
                getTypeBinaryInputInfo(att->atttypid, &typreceive, &typioparam);
                slot->tts_values[i] =
                    OidReceiveFunctionCall(typreceive, colvalue,
                                           typioparam, att->atttypmod);
 
                /* Trouble if it didn't eat the whole buffer */
                if (colvalue->cursor != colvalue->len)
                    ereport(ERROR,
                            (errcode(ERRCODE_INVALID_BINARY_REPRESENTATION),
                             errmsg("incorrect binary data format in logical replication column %d",
                                    remoteattnum + 1)));
                slot->tts_isnull[i] = false;
            }
            else
            {
                /*
                 * NULL value from remote.  (We don't expect to see
                 * LOGICALREP_COLUMN_UNCHANGED here, but if we do, treat it as
                 * NULL.)
                 */
                slot->tts_values[i] = (Datum) 0;
                slot->tts_isnull[i] = true;
            }
 
            /* Reset attnum for error callback */
            apply_error_callback_arg.remote_attnum = -1;
        }
        else
        {
            /*
             * We assign NULL to dropped attributes and missing values
             * (missing values should be later filled using
             * slot_fill_defaults).
             */
            slot->tts_values[i] = (Datum) 0;
            slot->tts_isnull[i] = true;
        }
    }
 
    ExecStoreVirtualTuple(slot);
}

References apply_error_callback_arg, Assert, AttrMap::attnums, LogicalRepRelMapEntry::attrmap, ereport, errcode(), ERRCODE_PROTOCOL_VIOLATION, errmsg, ERROR, ExecClearTuple(), ExecStoreVirtualTuple(), fb(), getTypeBinaryInputInfo(), getTypeInputInfo(), i, LOGICALREP_COLUMN_BINARY, LOGICALREP_COLUMN_TEXT, AttrMap::maplen, TupleDescData::natts, OidInputFunctionCall(), OidReceiveFunctionCall(), ApplyErrorCallbackArg::remote_attnum, TupleTableSlot::tts_isnull, TupleTableSlot::tts_tupleDescriptor, TupleTableSlot::tts_values, and TupleDescAttr().

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

start_apply()

void start_apply

(

XLogRecPtr

origin_startpos

)

Definition at line 5642 of file worker.c.

{
    PG_TRY();
    {
        LogicalRepApplyLoop(origin_startpos);
    }
    PG_CATCH();
    {
        /*
         * Reset the origin state to prevent the advancement of origin
         * progress if we fail to apply. Otherwise, this will result in
         * transaction loss as that transaction won't be sent again by the
         * server.
         */
        replorigin_xact_clear(true);
 
        if (MySubscription->disableonerr)
            DisableSubscriptionAndExit();
        else
        {
            /*
             * Report the worker failed while applying changes. Abort the
             * current transaction so that the stats message is sent in an
             * idle state.
             */
            AbortOutOfAnyTransaction();
            pgstat_report_subscription_error(MySubscription->oid);
 
            PG_RE_THROW();
        }
    }
    PG_END_TRY();
}

References AbortOutOfAnyTransaction(), Subscription::disableonerr, DisableSubscriptionAndExit(), fb(), LogicalRepApplyLoop(), MySubscription, Subscription::oid, PG_CATCH, PG_END_TRY, PG_RE_THROW, PG_TRY, pgstat_report_subscription_error(), and replorigin_xact_clear().

Referenced by run_apply_worker(), and run_tablesync_worker().

stop_conflict_info_retention()

static void stop_conflict_info_retention ( RetainDeadTuplesData *  rdt_data )

static

Definition at line 4826 of file worker.c.

{
    /* Stop retention if not yet */
    if (MySubscription->retentionactive)
    {
        /*
         * If the retention status cannot be updated (e.g., due to active
         * transaction), skip further processing to avoid inconsistent
         * retention behavior.
         */
        if (!update_retention_status(false))
            return;
 
        SpinLockAcquire(&MyLogicalRepWorker->relmutex);
        MyLogicalRepWorker->oldest_nonremovable_xid = InvalidTransactionId;
        SpinLockRelease(&MyLogicalRepWorker->relmutex);
 
        ereport(LOG,
                errmsg("logical replication worker for subscription \"%s\" has stopped retaining the information for detecting conflicts",
                       MySubscription->name),
                errdetail("Retention is stopped because the apply process has not caught up with the publisher within the configured max_retention_duration."));
    }
 
    Assert(!TransactionIdIsValid(MyLogicalRepWorker->oldest_nonremovable_xid));
 
    /*
     * If retention has been stopped, reset to the initial phase to retry
     * resuming retention. This reset is required to recalculate the current
     * wait time and resume retention if the time falls within
     * max_retention_duration.
     */
    reset_retention_data_fields(rdt_data);
}

References Assert, ereport, errdetail(), errmsg, fb(), InvalidTransactionId, LOG, MyLogicalRepWorker, MySubscription, Subscription::name, LogicalRepWorker::oldest_nonremovable_xid, LogicalRepWorker::relmutex, reset_retention_data_fields(), Subscription::retentionactive, SpinLockAcquire(), SpinLockRelease(), TransactionIdIsValid, and update_retention_status().

Referenced by process_rdt_phase_transition().

stop_skipping_changes()

static void stop_skipping_changes ( void  )

static

Definition at line 6127 of file worker.c.

{
    if (!is_skipping_changes())
        return;
 
    ereport(LOG,
            errmsg("logical replication completed skipping transaction at LSN %X/%08X",
                   LSN_FORMAT_ARGS(skip_xact_finish_lsn)));
 
    /* Stop skipping changes */
    skip_xact_finish_lsn = InvalidXLogRecPtr;
}

References ereport, errmsg, InvalidXLogRecPtr, is_skipping_changes, LOG, LSN_FORMAT_ARGS, and skip_xact_finish_lsn.

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

store_flush_position()

void store_flush_position	(	XLogRecPtr	remote_lsn,
		XLogRecPtr	local_lsn
	)

Definition at line 3961 of file worker.c.

{
    FlushPosition *flushpos;
 
    /*
     * Skip for parallel apply workers, because the lsn_mapping is maintained
     * by the leader apply worker.
     */
    if (am_parallel_apply_worker())
        return;
 
    /* Need to do this in permanent context */
    MemoryContextSwitchTo(ApplyContext);
 
    /* Track commit lsn  */
    flushpos = palloc_object(FlushPosition);
    flushpos->local_end = local_lsn;
    flushpos->remote_end = remote_lsn;
 
    dlist_push_tail(&lsn_mapping, &flushpos->node);
    MemoryContextSwitchTo(ApplyMessageContext);
}

References am_parallel_apply_worker(), ApplyContext, ApplyMessageContext, dlist_push_tail(), fb(), lsn_mapping, MemoryContextSwitchTo(), and palloc_object.

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

stream_abort_internal()

static void stream_abort_internal ( TransactionId  xid, TransactionId  subxid  )

static

Definition at line 2005 of file worker.c.

{
    /*
     * If the two XIDs are the same, it's in fact abort of toplevel xact, so
     * just delete the files with serialized info.
     */
    if (xid == subxid)
        stream_cleanup_files(MyLogicalRepWorker->subid, xid);
    else
    {
        /*
         * OK, so it's a subxact. We need to read the subxact file for the
         * toplevel transaction, determine the offset tracked for the subxact,
         * and truncate the file with changes. We also remove the subxacts
         * with higher offsets (or rather higher XIDs).
         *
         * We intentionally scan the array from the tail, because we're likely
         * aborting a change for the most recent subtransactions.
         *
         * We can't use the binary search here as subxact XIDs won't
         * necessarily arrive in sorted order, consider the case where we have
         * released the savepoint for multiple subtransactions and then
         * performed rollback to savepoint for one of the earlier
         * sub-transaction.
         */
        int64       i;
        int64       subidx;
        BufFile    *fd;
        bool        found = false;
        char        path[MAXPGPATH];
 
        subidx = -1;
        begin_replication_step();
        subxact_info_read(MyLogicalRepWorker->subid, xid);
 
        for (i = subxact_data.nsubxacts; i > 0; i--)
        {
            if (subxact_data.subxacts[i - 1].xid == subxid)
            {
                subidx = (i - 1);
                found = true;
                break;
            }
        }
 
        /*
         * If it's an empty sub-transaction then we will not find the subxid
         * here so just cleanup the subxact info and return.
         */
        if (!found)
        {
            /* Cleanup the subxact info */
            cleanup_subxact_info();
            end_replication_step();
            CommitTransactionCommand();
            return;
        }
 
        /* open the changes file */
        changes_filename(path, MyLogicalRepWorker->subid, xid);
        fd = BufFileOpenFileSet(MyLogicalRepWorker->stream_fileset, path,
                                O_RDWR, false);
 
        /* OK, truncate the file at the right offset */
        BufFileTruncateFileSet(fd, subxact_data.subxacts[subidx].fileno,
                               subxact_data.subxacts[subidx].offset);
        BufFileClose(fd);
 
        /* discard the subxacts added later */
        subxact_data.nsubxacts = subidx;
 
        /* write the updated subxact list */
        subxact_info_write(MyLogicalRepWorker->subid, xid);
 
        end_replication_step();
        CommitTransactionCommand();
    }
}

References begin_replication_step(), BufFileClose(), BufFileOpenFileSet(), BufFileTruncateFileSet(), changes_filename(), cleanup_subxact_info(), CommitTransactionCommand(), end_replication_step(), fb(), fd(), SubXactInfo::fileno, i, MAXPGPATH, MyLogicalRepWorker, ApplySubXactData::nsubxacts, SubXactInfo::offset, stream_cleanup_files(), LogicalRepWorker::stream_fileset, LogicalRepWorker::subid, subxact_data, subxact_info_read(), subxact_info_write(), ApplySubXactData::subxacts, and SubXactInfo::xid.

Referenced by apply_handle_stream_abort().

stream_cleanup_files()

void stream_cleanup_files	(	Oid	subid,
		TransactionId	xid
	)

Definition at line 5440 of file worker.c.

{
    char        path[MAXPGPATH];
 
    /* Delete the changes file. */
    changes_filename(path, subid, xid);
    BufFileDeleteFileSet(MyLogicalRepWorker->stream_fileset, path, false);
 
    /* Delete the subxact file, if it exists. */
    subxact_filename(path, subid, xid);
    BufFileDeleteFileSet(MyLogicalRepWorker->stream_fileset, path, true);
}

References BufFileDeleteFileSet(), changes_filename(), MAXPGPATH, MyLogicalRepWorker, LogicalRepWorker::stream_fileset, and subxact_filename().

Referenced by apply_handle_stream_commit(), apply_handle_stream_prepare(), pa_free_worker_info(), and stream_abort_internal().

stream_close_file()

static void stream_close_file ( void  )

static

Definition at line 5509 of file worker.c.

{
    Assert(stream_fd != NULL);
 
    BufFileClose(stream_fd);
 
    stream_fd = NULL;
}

References Assert, BufFileClose(), fb(), and stream_fd.

Referenced by apply_handle_stream_abort(), apply_handle_stream_commit(), apply_handle_stream_prepare(), apply_spooled_messages(), and stream_stop_internal().

stream_open_and_write_change()

static void stream_open_and_write_change ( TransactionId  xid, char  action, StringInfo  s  )

static

Definition at line 5557 of file worker.c.

{
    Assert(!in_streamed_transaction);
 
    if (!stream_fd)
        stream_start_internal(xid, false);
 
    stream_write_change(action, s);
    stream_stop_internal(xid);
}

References Assert, in_streamed_transaction, stream_fd, stream_start_internal(), stream_stop_internal(), and stream_write_change().

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

stream_open_file()

static void stream_open_file ( Oid  subid, TransactionId  xid, bool  first_segment  )

static

Definition at line 5464 of file worker.c.

{
    char        path[MAXPGPATH];
    MemoryContext oldcxt;
 
    Assert(OidIsValid(subid));
    Assert(TransactionIdIsValid(xid));
    Assert(stream_fd == NULL);
 
 
    changes_filename(path, subid, xid);
    elog(DEBUG1, "opening file \"%s\" for streamed changes", path);
 
    /*
     * Create/open the buffiles under the logical streaming context so that we
     * have those files until stream stop.
     */
    oldcxt = MemoryContextSwitchTo(LogicalStreamingContext);
 
    /*
     * If this is the first streamed segment, create the changes file.
     * Otherwise, just open the file for writing, in append mode.
     */
    if (first_segment)
        stream_fd = BufFileCreateFileSet(MyLogicalRepWorker->stream_fileset,
                                         path);
    else
    {
        /*
         * Open the file and seek to the end of the file because we always
         * append the changes file.
         */
        stream_fd = BufFileOpenFileSet(MyLogicalRepWorker->stream_fileset,
                                       path, O_RDWR, false);
        BufFileSeek(stream_fd, 0, 0, SEEK_END);
    }
 
    MemoryContextSwitchTo(oldcxt);
}

References Assert, BufFileCreateFileSet(), BufFileOpenFileSet(), BufFileSeek(), changes_filename(), DEBUG1, elog, fb(), LogicalStreamingContext, MAXPGPATH, MemoryContextSwitchTo(), MyLogicalRepWorker, OidIsValid, stream_fd, LogicalRepWorker::stream_fileset, and TransactionIdIsValid.

Referenced by stream_start_internal().

stream_start_internal()

void stream_start_internal	(	TransactionId	xid,
		bool	first_segment
	)

Definition at line 1704 of file worker.c.

{
    begin_replication_step();
 
    /*
     * Initialize the worker's stream_fileset if we haven't yet. This will be
     * used for the entire duration of the worker so create it in a permanent
     * context. We create this on the very first streaming message from any
     * transaction and then use it for this and other streaming transactions.
     * Now, we could create a fileset at the start of the worker as well but
     * then we won't be sure that it will ever be used.
     */
    if (!MyLogicalRepWorker->stream_fileset)
    {
        MemoryContext oldctx;
 
        oldctx = MemoryContextSwitchTo(ApplyContext);
 
        MyLogicalRepWorker->stream_fileset = palloc_object(FileSet);
        FileSetInit(MyLogicalRepWorker->stream_fileset);
 
        MemoryContextSwitchTo(oldctx);
    }
 
    /* Open the spool file for this transaction. */
    stream_open_file(MyLogicalRepWorker->subid, xid, first_segment);
 
    /* If this is not the first segment, open existing subxact file. */
    if (!first_segment)
        subxact_info_read(MyLogicalRepWorker->subid, xid);
 
    end_replication_step();
}

References ApplyContext, begin_replication_step(), end_replication_step(), fb(), FileSetInit(), MemoryContextSwitchTo(), MyLogicalRepWorker, palloc_object, LogicalRepWorker::stream_fileset, stream_open_file(), LogicalRepWorker::subid, and subxact_info_read().

Referenced by apply_handle_stream_start(), pa_switch_to_partial_serialize(), and stream_open_and_write_change().

stream_stop_internal()

void stream_stop_internal

(

TransactionId

xid

)

Definition at line 1879 of file worker.c.

{
    /*
     * Serialize information about subxacts for the toplevel transaction, then
     * close the stream messages spool file.
     */
    subxact_info_write(MyLogicalRepWorker->subid, xid);
    stream_close_file();
 
    /* We must be in a valid transaction state */
    Assert(IsTransactionState());
 
    /* Commit the per-stream transaction */
    CommitTransactionCommand();
 
    /* Reset per-stream context */
    MemoryContextReset(LogicalStreamingContext);
}

References Assert, CommitTransactionCommand(), IsTransactionState(), LogicalStreamingContext, MemoryContextReset(), MyLogicalRepWorker, stream_close_file(), LogicalRepWorker::subid, and subxact_info_write().

Referenced by apply_handle_stream_stop(), and stream_open_and_write_change().

stream_write_change()

static void stream_write_change ( char  action, StringInfo  s  )

static

Definition at line 5527 of file worker.c.

{
    int         len;
 
    Assert(stream_fd != NULL);
 
    /* total on-disk size, including the action type character */
    len = (s->len - s->cursor) + sizeof(char);
 
    /* first write the size */
    BufFileWrite(stream_fd, &len, sizeof(len));
 
    /* then the action */
    BufFileWrite(stream_fd, &action, sizeof(action));
 
    /* and finally the remaining part of the buffer (after the XID) */
    len = (s->len - s->cursor);
 
    BufFileWrite(stream_fd, &s->data[s->cursor], len);
}

References Assert, BufFileWrite(), StringInfoData::cursor, StringInfoData::data, fb(), StringInfoData::len, len, and stream_fd.

Referenced by apply_handle_stream_start(), apply_handle_stream_stop(), handle_streamed_transaction(), and stream_open_and_write_change().

subscription_change_cb()

static void subscription_change_cb ( Datum  arg, SysCacheIdentifier  cacheid, uint32  hashvalue  )

static

Definition at line 5226 of file worker.c.

{
    MySubscriptionValid = false;
}

References MySubscriptionValid.

Referenced by InitializeLogRepWorker().

subxact_filename()

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

inlinestatic

Definition at line 5419 of file worker.c.

{
    snprintf(path, MAXPGPATH, "%u-%u.subxacts", subid, xid);
}

References MAXPGPATH, and snprintf.

Referenced by stream_cleanup_files(), subxact_info_read(), and subxact_info_write().

subxact_info_add()

static void subxact_info_add ( TransactionId  xid )

static

Definition at line 5341 of file worker.c.

{
    SubXactInfo *subxacts = subxact_data.subxacts;
    int64       i;
 
    /* We must have a valid top level stream xid and a stream fd. */
    Assert(TransactionIdIsValid(stream_xid));
    Assert(stream_fd != NULL);
 
    /*
     * If the XID matches the toplevel transaction, we don't want to add it.
     */
    if (stream_xid == xid)
        return;
 
    /*
     * In most cases we're checking the same subxact as we've already seen in
     * the last call, so make sure to ignore it (this change comes later).
     */
    if (subxact_data.subxact_last == xid)
        return;
 
    /* OK, remember we're processing this XID. */
    subxact_data.subxact_last = xid;
 
    /*
     * Check if the transaction is already present in the array of subxact. We
     * intentionally scan the array from the tail, because we're likely adding
     * a change for the most recent subtransactions.
     *
     * XXX Can we rely on the subxact XIDs arriving in sorted order? That
     * would allow us to use binary search here.
     */
    for (i = subxact_data.nsubxacts; i > 0; i--)
    {
        /* found, so we're done */
        if (subxacts[i - 1].xid == xid)
            return;
    }
 
    /* This is a new subxact, so we need to add it to the array. */
    if (subxact_data.nsubxacts == 0)
    {
        MemoryContext oldctx;
 
        subxact_data.nsubxacts_max = 128;
 
        /*
         * Allocate this memory for subxacts in per-stream context, see
         * subxact_info_read.
         */
        oldctx = MemoryContextSwitchTo(LogicalStreamingContext);
        subxacts = palloc_array(SubXactInfo, subxact_data.nsubxacts_max);
        MemoryContextSwitchTo(oldctx);
    }
    else if (subxact_data.nsubxacts == subxact_data.nsubxacts_max)
    {
        subxact_data.nsubxacts_max *= 2;
        subxacts = repalloc_array(subxacts, SubXactInfo,
                                  subxact_data.nsubxacts_max);
    }
 
    subxacts[subxact_data.nsubxacts].xid = xid;
 
    /*
     * Get the current offset of the stream file and store it as offset of
     * this subxact.
     */
    BufFileTell(stream_fd,
                &subxacts[subxact_data.nsubxacts].fileno,
                &subxacts[subxact_data.nsubxacts].offset);
 
    subxact_data.nsubxacts++;
    subxact_data.subxacts = subxacts;
}

References Assert, BufFileTell(), fb(), SubXactInfo::fileno, i, LogicalStreamingContext, MemoryContextSwitchTo(), ApplySubXactData::nsubxacts, ApplySubXactData::nsubxacts_max, SubXactInfo::offset, palloc_array, repalloc_array, stream_fd, stream_xid, subxact_data, ApplySubXactData::subxact_last, ApplySubXactData::subxacts, TransactionIdIsValid, and SubXactInfo::xid.

Referenced by handle_streamed_transaction().

subxact_info_read()

static void subxact_info_read ( Oid  subid, TransactionId  xid  )

static

Definition at line 5290 of file worker.c.

{
    char        path[MAXPGPATH];
    Size        len;
    BufFile    *fd;
    MemoryContext oldctx;
 
    Assert(!subxact_data.subxacts);
    Assert(subxact_data.nsubxacts == 0);
    Assert(subxact_data.nsubxacts_max == 0);
 
    /*
     * If the subxact file doesn't exist that means we don't have any subxact
     * info.
     */
    subxact_filename(path, subid, xid);
    fd = BufFileOpenFileSet(MyLogicalRepWorker->stream_fileset, path, O_RDONLY,
                            true);
    if (fd == NULL)
        return;
 
    /* read number of subxact items */
    BufFileReadExact(fd, &subxact_data.nsubxacts, sizeof(subxact_data.nsubxacts));
 
    len = sizeof(SubXactInfo) * subxact_data.nsubxacts;
 
    /* we keep the maximum as a power of 2 */
    subxact_data.nsubxacts_max = 1 << pg_ceil_log2_32(subxact_data.nsubxacts);
 
    /*
     * Allocate subxact information in the logical streaming context. We need
     * this information during the complete stream so that we can add the sub
     * transaction info to this. On stream stop we will flush this information
     * to the subxact file and reset the logical streaming context.
     */
    oldctx = MemoryContextSwitchTo(LogicalStreamingContext);
    subxact_data.subxacts = palloc_array(SubXactInfo,
                                         subxact_data.nsubxacts_max);
    MemoryContextSwitchTo(oldctx);
 
    if (len > 0)
        BufFileReadExact(fd, subxact_data.subxacts, len);
 
    BufFileClose(fd);
}

References Assert, BufFileClose(), BufFileOpenFileSet(), BufFileReadExact(), fb(), fd(), len, LogicalStreamingContext, MAXPGPATH, MemoryContextSwitchTo(), MyLogicalRepWorker, ApplySubXactData::nsubxacts, ApplySubXactData::nsubxacts_max, palloc_array, pg_ceil_log2_32(), LogicalRepWorker::stream_fileset, subxact_data, subxact_filename(), and ApplySubXactData::subxacts.

Referenced by stream_abort_internal(), and stream_start_internal().

subxact_info_write()

static void subxact_info_write ( Oid  subid, TransactionId  xid  )

static

Definition at line 5241 of file worker.c.

{
    char        path[MAXPGPATH];
    Size        len;
    BufFile    *fd;
 
    Assert(TransactionIdIsValid(xid));
 
    /* construct the subxact filename */
    subxact_filename(path, subid, xid);
 
    /* Delete the subxacts file, if exists. */
    if (subxact_data.nsubxacts == 0)
    {
        cleanup_subxact_info();
        BufFileDeleteFileSet(MyLogicalRepWorker->stream_fileset, path, true);
 
        return;
    }
 
    /*
     * Create the subxact file if it not already created, otherwise open the
     * existing file.
     */
    fd = BufFileOpenFileSet(MyLogicalRepWorker->stream_fileset, path, O_RDWR,
                            true);
    if (fd == NULL)
        fd = BufFileCreateFileSet(MyLogicalRepWorker->stream_fileset, path);
 
    len = sizeof(SubXactInfo) * subxact_data.nsubxacts;
 
    /* Write the subxact count and subxact info */
    BufFileWrite(fd, &subxact_data.nsubxacts, sizeof(subxact_data.nsubxacts));
    BufFileWrite(fd, subxact_data.subxacts, len);
 
    BufFileClose(fd);
 
    /* free the memory allocated for subxact info */
    cleanup_subxact_info();
}

References Assert, BufFileClose(), BufFileCreateFileSet(), BufFileDeleteFileSet(), BufFileOpenFileSet(), BufFileWrite(), cleanup_subxact_info(), fb(), fd(), len, MAXPGPATH, MyLogicalRepWorker, ApplySubXactData::nsubxacts, LogicalRepWorker::stream_fileset, subxact_data, subxact_filename(), ApplySubXactData::subxacts, and TransactionIdIsValid.

Referenced by stream_abort_internal(), and stream_stop_internal().

TargetPrivilegesCheck()

static void TargetPrivilegesCheck ( Relation  rel, AclMode  mode  )

static

Definition at line 2618 of file worker.c.

{
    Oid         relid;
    AclResult   aclresult;
 
    relid = RelationGetRelid(rel);
    aclresult = pg_class_aclcheck(relid, GetUserId(), mode);
    if (aclresult != ACLCHECK_OK)
        aclcheck_error(aclresult,
                       get_relkind_objtype(rel->rd_rel->relkind),
                       get_rel_name(relid));
 
    /*
     * We lack the infrastructure to honor RLS policies.  It might be possible
     * to add such infrastructure here, but tablesync workers lack it, too, so
     * we don't bother.  RLS does not ordinarily apply to TRUNCATE commands,
     * but it seems dangerous to replicate a TRUNCATE and then refuse to
     * replicate subsequent INSERTs, so we forbid all commands the same.
     */
    if (check_enable_rls(relid, InvalidOid, false) == RLS_ENABLED)
        ereport(ERROR,
                (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                 errmsg("user \"%s\" cannot replicate into relation with row-level security enabled: \"%s\"",
                        GetUserNameFromId(GetUserId(), true),
                        RelationGetRelationName(rel))));
}

References aclcheck_error(), ACLCHECK_OK, check_enable_rls(), ereport, errcode(), errmsg, ERROR, fb(), get_rel_name(), get_relkind_objtype(), GetUserId(), GetUserNameFromId(), InvalidOid, mode, pg_class_aclcheck(), RelationData::rd_rel, RelationGetRelationName, RelationGetRelid, and RLS_ENABLED.

Referenced by apply_handle_delete_internal(), apply_handle_insert_internal(), apply_handle_truncate(), apply_handle_tuple_routing(), apply_handle_update_internal(), and FindReplTupleInLocalRel().

update_retention_status()

static bool update_retention_status ( bool  active )

static

Definition at line 4904 of file worker.c.

{
    /*
     * Do not update the catalog during an active transaction. The transaction
     * may be started during change application, leading to a possible
     * rollback of catalog updates if the application fails subsequently.
     */
    if (IsTransactionState())
        return false;
 
    StartTransactionCommand();
 
    /*
     * Updating pg_subscription might involve TOAST table access, so ensure we
     * have a valid snapshot.
     */
    PushActiveSnapshot(GetTransactionSnapshot());
 
    /* Update pg_subscription.subretentionactive */
    UpdateDeadTupleRetentionStatus(MySubscription->oid, active);
 
    PopActiveSnapshot();
    CommitTransactionCommand();
 
    /* Notify launcher to update the conflict slot */
    ApplyLauncherWakeup();
 
    MySubscription->retentionactive = active;
 
    return true;
}

References ApplyLauncherWakeup(), CommitTransactionCommand(), GetTransactionSnapshot(), IsTransactionState(), MySubscription, Subscription::oid, PopActiveSnapshot(), PushActiveSnapshot(), Subscription::retentionactive, StartTransactionCommand(), and UpdateDeadTupleRetentionStatus().

Referenced by resume_conflict_info_retention(), and stop_conflict_info_retention().

UpdateWorkerStats()

static void UpdateWorkerStats ( XLogRecPtr  last_lsn, TimestampTz  send_time, bool  reply  )

static

Definition at line 3987 of file worker.c.

{
    MyLogicalRepWorker->last_lsn = last_lsn;
    MyLogicalRepWorker->last_send_time = send_time;
    MyLogicalRepWorker->last_recv_time = GetCurrentTimestamp();
    if (reply)
    {
        MyLogicalRepWorker->reply_lsn = last_lsn;
        MyLogicalRepWorker->reply_time = send_time;
    }
}

References fb(), GetCurrentTimestamp(), LogicalRepWorker::last_lsn, LogicalRepWorker::last_recv_time, LogicalRepWorker::last_send_time, MyLogicalRepWorker, LogicalRepWorker::reply_lsn, and LogicalRepWorker::reply_time.

Referenced by LogicalRepApplyLoop().

wait_for_local_flush()

static void wait_for_local_flush ( RetainDeadTuplesData *  rdt_data )

static

Definition at line 4636 of file worker.c.

{
    Assert(XLogRecPtrIsValid(rdt_data->remote_lsn) &&
           TransactionIdIsValid(rdt_data->candidate_xid));
 
    /*
     * We expect the publisher and subscriber clocks to be in sync using time
     * sync service like NTP. Otherwise, we will advance this worker's
     * oldest_nonremovable_xid prematurely, leading to the removal of rows
     * required to detect update_deleted reliably. This check primarily
     * addresses scenarios where the publisher's clock falls behind; if the
     * publisher's clock is ahead, subsequent transactions will naturally bear
     * later commit timestamps, conforming to the design outlined atop
     * worker.c.
     *
     * XXX Consider waiting for the publisher's clock to catch up with the
     * subscriber's before proceeding to the next phase.
     */
    if (TimestampDifferenceExceeds(rdt_data->reply_time,
                                   rdt_data->candidate_xid_time, 0))
        ereport(ERROR,
                errmsg_internal("oldest_nonremovable_xid transaction ID could be advanced prematurely"),
                errdetail_internal("The clock on the publisher is behind that of the subscriber."));
 
    /*
     * Do not attempt to advance the non-removable transaction ID when table
     * sync is in progress. During this time, changes from a single
     * transaction may be applied by multiple table sync workers corresponding
     * to the target tables. So, it's necessary for all table sync workers to
     * apply and flush the corresponding changes before advancing the
     * transaction ID, otherwise, dead tuples that are still needed for
     * conflict detection in table sync workers could be removed prematurely.
     * However, confirming the apply and flush progress across all table sync
     * workers is complex and not worth the effort, so we simply return if not
     * all tables are in the READY state.
     *
     * Advancing the transaction ID is necessary even when no tables are
     * currently subscribed, to avoid retaining dead tuples unnecessarily.
     * While it might seem safe to skip all phases and directly assign
     * candidate_xid to oldest_nonremovable_xid during the
     * RDT_GET_CANDIDATE_XID phase in such cases, this is unsafe. If users
     * concurrently add tables to the subscription, the apply worker may not
     * process invalidations in time. Consequently,
     * HasSubscriptionTablesCached() might miss the new tables, leading to
     * premature advancement of oldest_nonremovable_xid.
     *
     * Performing the check during RDT_WAIT_FOR_LOCAL_FLUSH is safe, as
     * invalidations are guaranteed to be processed before applying changes
     * from newly added tables while waiting for the local flush to reach
     * remote_lsn.
     *
     * Additionally, even if we check for subscription tables during
     * RDT_GET_CANDIDATE_XID, they might be dropped before reaching
     * RDT_WAIT_FOR_LOCAL_FLUSH. Therefore, it's still necessary to verify
     * subscription tables at this stage to prevent unnecessary tuple
     * retention.
     */
    if (HasSubscriptionTablesCached() && !AllTablesyncsReady())
    {
        TimestampTz now;
 
        now = rdt_data->last_recv_time
            ? rdt_data->last_recv_time : GetCurrentTimestamp();
 
        /*
         * Record the time spent waiting for table sync, it is needed for the
         * timeout check in should_stop_conflict_info_retention().
         */
        rdt_data->table_sync_wait_time =
            TimestampDifferenceMilliseconds(rdt_data->candidate_xid_time, now);
 
        return;
    }
 
    /*
     * We don't need to maintain oldest_nonremovable_xid if we decide to stop
     * retaining conflict information for this worker.
     */
    if (should_stop_conflict_info_retention(rdt_data))
    {
        rdt_data->phase = RDT_STOP_CONFLICT_INFO_RETENTION;
        return;
    }
 
    /*
     * Update and check the remote flush position if we are applying changes
     * in a loop. This is done at most once per WalWriterDelay to avoid
     * performing costly operations in get_flush_position() too frequently
     * during change application.
     */
    if (last_flushpos < rdt_data->remote_lsn && rdt_data->last_recv_time &&
        TimestampDifferenceExceeds(rdt_data->flushpos_update_time,
                                   rdt_data->last_recv_time, WalWriterDelay))
    {
        XLogRecPtr  writepos;
        XLogRecPtr  flushpos;
        bool        have_pending_txes;
 
        /* Fetch the latest remote flush position */
        get_flush_position(&writepos, &flushpos, &have_pending_txes);
 
        if (flushpos > last_flushpos)
            last_flushpos = flushpos;
 
        rdt_data->flushpos_update_time = rdt_data->last_recv_time;
    }
 
    /* Return to wait for the changes to be applied */
    if (last_flushpos < rdt_data->remote_lsn)
        return;
 
    /*
     * Reaching this point implies should_stop_conflict_info_retention()
     * returned false earlier, meaning that the most recent duration for
     * advancing the non-removable transaction ID is within the
     * max_retention_duration or max_retention_duration is set to 0.
     *
     * Therefore, if conflict info retention was previously stopped due to a
     * timeout, it is now safe to resume retention.
     */
    if (!MySubscription->retentionactive)
    {
        rdt_data->phase = RDT_RESUME_CONFLICT_INFO_RETENTION;
        return;
    }
 
    /*
     * Reaching here means the remote WAL position has been received, and all
     * transactions up to that position on the publisher have been applied and
     * flushed locally. So, we can advance the non-removable transaction ID.
     */
    SpinLockAcquire(&MyLogicalRepWorker->relmutex);
    MyLogicalRepWorker->oldest_nonremovable_xid = rdt_data->candidate_xid;
    SpinLockRelease(&MyLogicalRepWorker->relmutex);
 
    elog(DEBUG2, "confirmed flush up to remote lsn %X/%08X: new oldest_nonremovable_xid %u",
         LSN_FORMAT_ARGS(rdt_data->remote_lsn),
         rdt_data->candidate_xid);
 
    /* Notify launcher to update the xmin of the conflict slot */
    ApplyLauncherWakeup();
 
    reset_retention_data_fields(rdt_data);
 
    /* process the next phase */
    process_rdt_phase_transition(rdt_data, false);
}

References AllTablesyncsReady(), ApplyLauncherWakeup(), Assert, DEBUG2, elog, ereport, errdetail_internal(), errmsg_internal(), ERROR, fb(), get_flush_position(), GetCurrentTimestamp(), HasSubscriptionTablesCached(), last_flushpos, LSN_FORMAT_ARGS, MyLogicalRepWorker, MySubscription, now(), LogicalRepWorker::oldest_nonremovable_xid, process_rdt_phase_transition(), RDT_RESUME_CONFLICT_INFO_RETENTION, RDT_STOP_CONFLICT_INFO_RETENTION, LogicalRepWorker::relmutex, reset_retention_data_fields(), Subscription::retentionactive, should_stop_conflict_info_retention(), SpinLockAcquire(), SpinLockRelease(), TimestampDifferenceExceeds(), TimestampDifferenceMilliseconds(), TransactionIdIsValid, WalWriterDelay, and XLogRecPtrIsValid.

Referenced by process_rdt_phase_transition().

wait_for_publisher_status()

static void wait_for_publisher_status ( RetainDeadTuplesData *  rdt_data, bool  status_received  )

static

Definition at line 4577 of file worker.c.

{
    /*
     * Return if we have requested but not yet received the publisher status.
     */
    if (!status_received)
        return;
 
    /*
     * We don't need to maintain oldest_nonremovable_xid if we decide to stop
     * retaining conflict information for this worker.
     */
    if (should_stop_conflict_info_retention(rdt_data))
    {
        rdt_data->phase = RDT_STOP_CONFLICT_INFO_RETENTION;
        return;
    }
 
    if (!FullTransactionIdIsValid(rdt_data->remote_wait_for))
        rdt_data->remote_wait_for = rdt_data->remote_nextxid;
 
    /*
     * Check if all remote concurrent transactions that were active at the
     * first status request have now completed. If completed, proceed to the
     * next phase; otherwise, continue checking the publisher status until
     * these transactions finish.
     *
     * It's possible that transactions in the commit phase during the last
     * cycle have now finished committing, but remote_oldestxid remains older
     * than remote_wait_for. This can happen if some old transaction came in
     * the commit phase when we requested status in this cycle. We do not
     * handle this case explicitly as it's rare and the benefit doesn't
     * justify the required complexity. Tracking would require either caching
     * all xids at the publisher or sending them to subscribers. The condition
     * will resolve naturally once the remaining transactions are finished.
     *
     * Directly advancing the non-removable transaction ID is possible if
     * there are no activities on the publisher since the last advancement
     * cycle. However, it requires maintaining two fields, last_remote_nextxid
     * and last_remote_lsn, within the structure for comparison with the
     * current cycle's values. Considering the minimal cost of continuing in
     * RDT_WAIT_FOR_LOCAL_FLUSH without awaiting changes, we opted not to
     * advance the transaction ID here.
     */
    if (FullTransactionIdPrecedesOrEquals(rdt_data->remote_wait_for,
                                          rdt_data->remote_oldestxid))
        rdt_data->phase = RDT_WAIT_FOR_LOCAL_FLUSH;
    else
        rdt_data->phase = RDT_REQUEST_PUBLISHER_STATUS;
 
    /* process the next phase */
    process_rdt_phase_transition(rdt_data, false);
}

References fb(), FullTransactionIdIsValid, FullTransactionIdPrecedesOrEquals, process_rdt_phase_transition(), RDT_REQUEST_PUBLISHER_STATUS, RDT_STOP_CONFLICT_INFO_RETENTION, RDT_WAIT_FOR_LOCAL_FLUSH, and should_stop_conflict_info_retention().

Referenced by process_rdt_phase_transition().

Variable Documentation

apply_error_callback_arg

ApplyErrorCallbackArg apply_error_callback_arg

static

Initial value:

=
{
    .command = 0,
    .rel = NULL,
    .remote_attnum = -1,
    .remote_xid = InvalidTransactionId,
    .finish_lsn = InvalidXLogRecPtr,
    .origin_name = NULL,
}

Definition at line 464 of file worker.c.

{
    .command = 0,
    .rel = NULL,
    .remote_attnum = -1,
    .remote_xid = InvalidTransactionId,
    .finish_lsn = InvalidXLogRecPtr,
    .origin_name = NULL,
};

Referenced by apply_dispatch(), apply_error_callback(), apply_handle_delete(), apply_handle_insert(), apply_handle_update(), reset_apply_error_context_info(), set_apply_error_context_origin(), set_apply_error_context_xact(), slot_modify_data(), and slot_store_data().

apply_error_context_stack

ErrorContextCallback* apply_error_context_stack = NULL

Definition at line 474 of file worker.c.

Referenced by LogicalRepApplyLoop(), and ProcessParallelApplyMessage().

ApplyContext

MemoryContext ApplyContext = NULL

Definition at line 477 of file worker.c.

Referenced by copy_sequences(), InitializeLogRepWorker(), LogicalParallelApplyLoop(), LogicalRepApplyLoop(), LogicalRepSyncSequences(), maybe_reread_subscription(), pa_allocate_worker(), pa_launch_parallel_worker(), request_publisher_status(), send_feedback(), set_apply_error_context_origin(), start_table_sync(), store_flush_position(), and stream_start_internal().

ApplyMessageContext

MemoryContext ApplyMessageContext = NULL

Definition at line 476 of file worker.c.

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

in_remote_transaction

bool in_remote_transaction = false

Definition at line 489 of file worker.c.

Referenced by apply_handle_begin(), apply_handle_begin_prepare(), apply_handle_commit_internal(), apply_handle_commit_prepared(), apply_handle_origin(), apply_handle_prepare(), apply_handle_rollback_prepared(), apply_handle_stream_prepare(), apply_spooled_messages(), LogicalRepApplyLoop(), and maybe_start_skipping_changes().

in_streamed_transaction

bool in_streamed_transaction = false

static

Definition at line 493 of file worker.c.

Referenced by apply_handle_origin(), apply_handle_stream_abort(), apply_handle_stream_commit(), apply_handle_stream_prepare(), apply_handle_stream_start(), apply_handle_stream_stop(), get_transaction_apply_action(), LogicalRepApplyLoop(), maybe_start_skipping_changes(), and stream_open_and_write_change().

InitializingApplyWorker

bool InitializingApplyWorker = false

Definition at line 504 of file worker.c.

Referenced by ApplyWorkerMain(), logicalrep_worker_onexit(), and ParallelApplyWorkerMain().

last_flushpos

XLogRecPtr last_flushpos = InvalidXLogRecPtr

static

Definition at line 532 of file worker.c.

Referenced by send_feedback(), and wait_for_local_flush().

LogicalStreamingContext

MemoryContext LogicalStreamingContext = NULL

static

Definition at line 480 of file worker.c.

Referenced by LogicalRepApplyLoop(), stream_open_file(), stream_stop_internal(), subxact_info_add(), and subxact_info_read().

LogRepWorkerWalRcvConn

WalReceiverConn* LogRepWorkerWalRcvConn = NULL

Definition at line 482 of file worker.c.

Referenced by copy_read_data(), copy_table(), fetch_remote_table_info(), logicalrep_worker_onexit(), LogicalRepApplyLoop(), LogicalRepSyncSequences(), LogicalRepSyncTableStart(), ProcessSyncingTablesForSync(), request_publisher_status(), run_apply_worker(), run_tablesync_worker(), send_feedback(), and set_stream_options().

lsn_mapping

dlist_head lsn_mapping = DLIST_STATIC_INIT(lsn_mapping)

static

Definition at line 313 of file worker.c.

Referenced by get_flush_position(), LogicalRepApplyLoop(), and store_flush_position().

MySubscription

Subscription* MySubscription = NULL

Definition at line 484 of file worker.c.

Referenced by adjust_xid_advance_interval(), apply_handle_commit_prepared(), apply_handle_delete(), apply_handle_insert(), apply_handle_prepare_internal(), apply_handle_rollback_prepared(), apply_handle_truncate(), apply_handle_update(), can_advance_nonremovable_xid(), clear_subscription_skip_lsn(), copy_sequence(), copy_sequences(), copy_table(), DisableSubscriptionAndExit(), fetch_remote_table_info(), FetchRelationStates(), FindDeletedTupleInLocalRel(), FinishSyncWorker(), InitializeLogRepWorker(), launch_sync_worker(), logicalrep_rel_open(), LogicalRepApplyLoop(), LogicalRepSyncSequences(), LogicalRepSyncTableStart(), maybe_reread_subscription(), maybe_start_skipping_changes(), pa_can_start(), pa_launch_parallel_worker(), pa_start_subtrans(), pa_stream_abort(), ParallelApplyWorkerMain(), ProcessParallelApplyInterrupts(), ProcessSyncingTablesForApply(), report_sequence_errors(), ReportApplyConflict(), resume_conflict_info_retention(), run_apply_worker(), run_tablesync_worker(), set_stream_options(), set_wal_receiver_timeout(), SetupApplyOrSyncWorker(), should_apply_changes_for_rel(), should_stop_conflict_info_retention(), start_apply(), start_sequence_sync(), start_table_sync(), stop_conflict_info_retention(), update_retention_status(), and wait_for_local_flush().

MySubscriptionValid

bool MySubscriptionValid = false

static

Definition at line 485 of file worker.c.

Referenced by InitializeLogRepWorker(), maybe_reread_subscription(), and subscription_change_cb().

on_commit_wakeup_workers_subids

List* on_commit_wakeup_workers_subids = NIL

static

Definition at line 487 of file worker.c.

Referenced by AtEOXact_LogicalRepWorkers(), and LogicalRepWorkersWakeupAtCommit().

parallel_stream_nchanges

uint32 parallel_stream_nchanges = 0

static

Definition at line 501 of file worker.c.

Referenced by apply_handle_stream_start(), apply_handle_stream_stop(), and handle_streamed_transaction().

remote_final_lsn

XLogRecPtr remote_final_lsn = InvalidXLogRecPtr

static

Definition at line 490 of file worker.c.

Referenced by apply_handle_begin(), apply_handle_begin_prepare(), apply_handle_commit(), apply_handle_prepare(), apply_spooled_messages(), and should_apply_changes_for_rel().

skip_xact_finish_lsn

XLogRecPtr skip_xact_finish_lsn = InvalidXLogRecPtr

static

Definition at line 521 of file worker.c.

Referenced by maybe_start_skipping_changes(), and stop_skipping_changes().

stream_fd

BufFile* stream_fd = NULL

static

Definition at line 525 of file worker.c.

Referenced by apply_handle_stream_abort(), apply_handle_stream_commit(), apply_handle_stream_prepare(), apply_spooled_messages(), handle_streamed_transaction(), stream_close_file(), stream_open_and_write_change(), stream_open_file(), stream_write_change(), and subxact_info_add().

stream_xid

TransactionId stream_xid = InvalidTransactionId

static

Definition at line 495 of file worker.c.

Referenced by apply_handle_begin(), apply_handle_begin_prepare(), apply_handle_stream_start(), apply_handle_stream_stop(), handle_streamed_transaction(), and subxact_info_add().

subxact_data

ApplySubXactData subxact_data = {0, 0, InvalidTransactionId, NULL}

static

Definition at line 550 of file worker.c.

{0, 0, InvalidTransactionId, NULL};

Referenced by cleanup_subxact_info(), stream_abort_internal(), subxact_info_add(), subxact_info_read(), and subxact_info_write().