worker.c File Reference

#include "postgres.h"
#include <sys/stat.h>
#include <unistd.h>
#include "access/table.h"
#include "access/tableam.h"
#include "access/twophase.h"
#include "access/xact.h"
#include "access/xlog_internal.h"
#include "catalog/catalog.h"
#include "catalog/namespace.h"
#include "catalog/partition.h"
#include "catalog/pg_inherits.h"
#include "catalog/pg_subscription.h"
#include "catalog/pg_subscription_rel.h"
#include "catalog/pg_tablespace.h"
#include "commands/tablecmds.h"
#include "commands/tablespace.h"
#include "commands/trigger.h"
#include "executor/executor.h"
#include "executor/execPartition.h"
#include "executor/nodeModifyTable.h"
#include "funcapi.h"
#include "libpq/pqformat.h"
#include "libpq/pqsignal.h"
#include "mb/pg_wchar.h"
#include "miscadmin.h"
#include "nodes/makefuncs.h"
#include "optimizer/optimizer.h"
#include "pgstat.h"
#include "postmaster/bgworker.h"
#include "postmaster/interrupt.h"
#include "postmaster/postmaster.h"
#include "postmaster/walwriter.h"
#include "replication/decode.h"
#include "replication/logical.h"
#include "replication/logicalproto.h"
#include "replication/logicalrelation.h"
#include "replication/logicalworker.h"
#include "replication/origin.h"
#include "replication/reorderbuffer.h"
#include "replication/snapbuild.h"
#include "replication/walreceiver.h"
#include "replication/worker_internal.h"
#include "rewrite/rewriteHandler.h"
#include "storage/buffile.h"
#include "storage/bufmgr.h"
#include "storage/fd.h"
#include "storage/ipc.h"
#include "storage/lmgr.h"
#include "storage/proc.h"
#include "storage/procarray.h"
#include "tcop/tcopprot.h"
#include "utils/builtins.h"
#include "utils/catcache.h"
#include "utils/dynahash.h"
#include "utils/datum.h"
#include "utils/fmgroids.h"
#include "utils/guc.h"
#include "utils/inval.h"
#include "utils/lsyscache.h"
#include "utils/memutils.h"
#include "utils/rel.h"
#include "utils/syscache.h"
#include "utils/timeout.h"

Include dependency graph for worker.c:

Go to the source code of this file.

Data Structures
struct	FlushPosition
struct	SlotErrCallbackArg
struct	ApplyExecutionData
struct	StreamXidHash
struct	SubXactInfo
struct	ApplySubXactData

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

Typedefs
typedef struct FlushPosition	FlushPosition
typedef struct SlotErrCallbackArg	SlotErrCallbackArg
typedef struct ApplyExecutionData	ApplyExecutionData
typedef struct StreamXidHash	StreamXidHash
typedef struct SubXactInfo	SubXactInfo
typedef struct ApplySubXactData	ApplySubXactData

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_cleanup_files (Oid subid, TransactionId xid)
static void	stream_open_file (Oid subid, TransactionId xid, bool first)
static void	stream_write_change (char action, StringInfo s)
static void	stream_close_file (void)
static void	send_feedback (XLogRecPtr recvpos, bool force, bool requestReply)
static void	store_flush_position (XLogRecPtr remote_lsn)
static void	maybe_reread_subscription (void)
static void	apply_dispatch (StringInfo s)
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)
static void	apply_handle_delete_internal (ApplyExecutionData *edata, ResultRelInfo *relinfo, TupleTableSlot *remoteslot)
static bool	FindReplTupleInLocalRel (EState *estate, Relation localrel, LogicalRepRelation *remoterel, TupleTableSlot *remoteslot, TupleTableSlot **localslot)
static void	apply_handle_tuple_routing (ApplyExecutionData *edata, TupleTableSlot *remoteslot, LogicalRepTupleData *newtup, CmdType operation)
static void	TwoPhaseTransactionGid (Oid subid, TransactionId xid, char *gid, int szgid)
static void	apply_spooled_messages (TransactionId xid, XLogRecPtr lsn)
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_error_callback (void *arg)
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)
static void	apply_handle_stream_start (StringInfo s)
static void	apply_handle_stream_stop (StringInfo s)
static void	apply_handle_stream_abort (StringInfo s)
static void	apply_handle_stream_commit (StringInfo s)
static void	apply_handle_relation (StringInfo s)
static void	apply_handle_type (StringInfo s)
static Oid	GetRelationIdentityOrPK (Relation rel)
static void	apply_handle_insert (StringInfo s)
static void	check_relation_updatable (LogicalRepRelMapEntry *rel)
static void	apply_handle_update (StringInfo s)
static void	apply_handle_delete (StringInfo s)
static void	apply_handle_truncate (StringInfo s)
static void	get_flush_position (XLogRecPtr *write, XLogRecPtr *flush, bool *have_pending_txes)
static void	UpdateWorkerStats (XLogRecPtr last_lsn, TimestampTz send_time, bool reply)
static void	LogicalRepApplyLoop (XLogRecPtr last_received)
static void	subscription_change_cb (Datum arg, int cacheid, uint32 hashvalue)
void	ApplyWorkerMain (Datum main_arg)
bool	IsLogicalWorker (void)

Variables
static dlist_head	lsn_mapping = DLIST_STATIC_INIT(lsn_mapping)
static MemoryContext	ApplyMessageContext = NULL
MemoryContext	ApplyContext = NULL
static MemoryContext	LogicalStreamingContext = NULL
WalReceiverConn *	LogRepWorkerWalRcvConn = NULL
Subscription *	MySubscription = NULL
bool	MySubscriptionValid = false
bool	in_remote_transaction = false
static XLogRecPtr	remote_final_lsn = InvalidXLogRecPtr
static bool	in_streamed_transaction = false
static TransactionId	stream_xid = InvalidTransactionId
static HTAB *	xidhash = NULL
static BufFile *	stream_fd = NULL
static ApplySubXactData	subxact_data = {0, 0, InvalidTransactionId, NULL}

Macro Definition Documentation

NAPTIME_PER_CYCLE

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

Definition at line 195 of file worker.c.

Referenced by LogicalRepApplyLoop().

Typedef Documentation

ApplyExecutionData

typedef struct ApplyExecutionData ApplyExecutionData

ApplySubXactData

typedef struct ApplySubXactData ApplySubXactData

FlushPosition

typedef struct FlushPosition FlushPosition

SlotErrCallbackArg

typedef struct SlotErrCallbackArg SlotErrCallbackArg

StreamXidHash

typedef struct StreamXidHash StreamXidHash

SubXactInfo

typedef struct SubXactInfo SubXactInfo

Function Documentation

apply_dispatch()

static void apply_dispatch ( StringInfo  s )

static

Definition at line 2328 of file worker.c.

References generate_unaccent_rules::action, 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(), ereport, errcode(), errmsg_internal(), ERROR, LOGICAL_REP_MSG_BEGIN, LOGICAL_REP_MSG_BEGIN_PREPARE, LOGICAL_REP_MSG_COMMIT, LOGICAL_REP_MSG_COMMIT_PREPARED, LOGICAL_REP_MSG_DELETE, LOGICAL_REP_MSG_INSERT, LOGICAL_REP_MSG_MESSAGE, LOGICAL_REP_MSG_ORIGIN, LOGICAL_REP_MSG_PREPARE, LOGICAL_REP_MSG_RELATION, LOGICAL_REP_MSG_ROLLBACK_PREPARED, LOGICAL_REP_MSG_STREAM_ABORT, LOGICAL_REP_MSG_STREAM_COMMIT, LOGICAL_REP_MSG_STREAM_END, LOGICAL_REP_MSG_STREAM_PREPARE, LOGICAL_REP_MSG_STREAM_START, LOGICAL_REP_MSG_TRUNCATE, LOGICAL_REP_MSG_TYPE, LOGICAL_REP_MSG_UPDATE, and pq_getmsgbyte().

Referenced by apply_spooled_messages(), and LogicalRepApplyLoop().

{
    LogicalRepMsgType action = pq_getmsgbyte(s);

    switch (action)
    {
        case LOGICAL_REP_MSG_BEGIN:
            apply_handle_begin(s);
            return;

        case LOGICAL_REP_MSG_COMMIT:
            apply_handle_commit(s);
            return;

        case LOGICAL_REP_MSG_INSERT:
            apply_handle_insert(s);
            return;

        case LOGICAL_REP_MSG_UPDATE:
            apply_handle_update(s);
            return;

        case LOGICAL_REP_MSG_DELETE:
            apply_handle_delete(s);
            return;

        case LOGICAL_REP_MSG_TRUNCATE:
            apply_handle_truncate(s);
            return;

        case LOGICAL_REP_MSG_RELATION:
            apply_handle_relation(s);
            return;

        case LOGICAL_REP_MSG_TYPE:
            apply_handle_type(s);
            return;

        case LOGICAL_REP_MSG_ORIGIN:
            apply_handle_origin(s);
            return;

        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.
             */
            return;

        case LOGICAL_REP_MSG_STREAM_START:
            apply_handle_stream_start(s);
            return;

        case LOGICAL_REP_MSG_STREAM_END:
            apply_handle_stream_stop(s);
            return;

        case LOGICAL_REP_MSG_STREAM_ABORT:
            apply_handle_stream_abort(s);
            return;

        case LOGICAL_REP_MSG_STREAM_COMMIT:
            apply_handle_stream_commit(s);
            return;

        case LOGICAL_REP_MSG_BEGIN_PREPARE:
            apply_handle_begin_prepare(s);
            return;

        case LOGICAL_REP_MSG_PREPARE:
            apply_handle_prepare(s);
            return;

        case LOGICAL_REP_MSG_COMMIT_PREPARED:
            apply_handle_commit_prepared(s);
            return;

        case LOGICAL_REP_MSG_ROLLBACK_PREPARED:
            apply_handle_rollback_prepared(s);
            return;

        case LOGICAL_REP_MSG_STREAM_PREPARE:
            apply_handle_stream_prepare(s);
            return;
    }

    ereport(ERROR,
            (errcode(ERRCODE_PROTOCOL_VIOLATION),
             errmsg_internal("invalid logical replication message type \"%c\"",
                             action)));
}

apply_handle_begin()

static void apply_handle_begin ( StringInfo  s )

static

Definition at line 825 of file worker.c.

References LogicalRepBeginData::final_lsn, in_remote_transaction, logicalrep_read_begin(), pgstat_report_activity(), remote_final_lsn, and STATE_RUNNING.

Referenced by apply_dispatch().

{
    LogicalRepBeginData begin_data;

    logicalrep_read_begin(s, &begin_data);

    remote_final_lsn = begin_data.final_lsn;

    in_remote_transaction = true;

    pgstat_report_activity(STATE_RUNNING, NULL);
}

apply_handle_begin_prepare()

static void apply_handle_begin_prepare ( StringInfo  s )

static

Definition at line 869 of file worker.c.

References am_tablesync_worker(), ereport, errcode(), errmsg_internal(), ERROR, in_remote_transaction, logicalrep_read_begin_prepare(), pgstat_report_activity(), LogicalRepPreparedTxnData::prepare_lsn, remote_final_lsn, and STATE_RUNNING.

Referenced by apply_dispatch().

{
    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")));

    logicalrep_read_begin_prepare(s, &begin_data);

    remote_final_lsn = begin_data.prepare_lsn;

    in_remote_transaction = true;

    pgstat_report_activity(STATE_RUNNING, NULL);
}

apply_handle_commit()

static void apply_handle_commit ( StringInfo  s )

static

Definition at line 844 of file worker.c.

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

Referenced by apply_dispatch().

{
    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/%X in commit message (expected %X/%X)",
                                 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. */
    process_syncing_tables(commit_data.end_lsn);

    pgstat_report_activity(STATE_IDLE, NULL);
}

apply_handle_commit_internal()

static void apply_handle_commit_internal ( LogicalRepCommitData *  commit_data )

static

Definition at line 1482 of file worker.c.

References AcceptInvalidationMessages(), LogicalRepCommitData::committime, CommitTransactionCommand(), LogicalRepCommitData::end_lsn, in_remote_transaction, IsTransactionState(), maybe_reread_subscription(), pgstat_report_stat(), replorigin_session_origin_lsn, replorigin_session_origin_timestamp, and store_flush_position().

Referenced by apply_handle_commit(), and apply_handle_stream_commit().

{
    if (IsTransactionState())
    {
        /*
         * Update origin state so we can restart streaming from correct
         * position in case of crash.
         */
        replorigin_session_origin_lsn = commit_data->end_lsn;
        replorigin_session_origin_timestamp = commit_data->committime;

        CommitTransactionCommand();
        pgstat_report_stat(false);

        store_flush_position(commit_data->end_lsn);
    }
    else
    {
        /* Process any invalidation messages that might have accumulated. */
        AcceptInvalidationMessages();
        maybe_reread_subscription();
    }

    in_remote_transaction = false;
}

apply_handle_commit_prepared()

static void apply_handle_commit_prepared ( StringInfo  s )

static

Definition at line 971 of file worker.c.

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

Referenced by apply_dispatch().

{
    LogicalRepCommitPreparedTxnData prepare_data;
    char        gid[GIDSIZE];

    logicalrep_read_commit_prepared(s, &prepare_data);

    /* 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_session_origin_lsn = prepare_data.end_lsn;
    replorigin_session_origin_timestamp = prepare_data.commit_time;

    FinishPreparedTransaction(gid, true);
    end_replication_step();
    CommitTransactionCommand();
    pgstat_report_stat(false);

    store_flush_position(prepare_data.end_lsn);
    in_remote_transaction = false;

    /* Process any tables that are being synchronized in parallel. */
    process_syncing_tables(prepare_data.end_lsn);

    pgstat_report_activity(STATE_IDLE, NULL);
}

apply_handle_delete()

static void apply_handle_delete ( StringInfo  s )

static

Definition at line 1844 of file worker.c.

References apply_handle_delete_internal(), apply_handle_tuple_routing(), begin_replication_step(), check_relation_updatable(), CMD_DELETE, create_edata_for_relation(), end_replication_step(), ApplyExecutionData::estate, ExecInitExtraTupleSlot(), finish_edata(), GetPerTupleMemoryContext, handle_streamed_transaction(), LogicalRepRelMapEntry::localrel, LOGICAL_REP_MSG_DELETE, logicalrep_read_delete(), logicalrep_rel_close(), logicalrep_rel_open(), MemoryContextSwitchTo(), NoLock, RelationData::rd_rel, RelationGetDescr, RowExclusiveLock, should_apply_changes_for_rel(), slot_store_data(), ApplyExecutionData::targetRelInfo, and TTSOpsVirtual.

Referenced by apply_dispatch().

{
    LogicalRepRelMapEntry *rel;
    LogicalRepTupleData oldtup;
    LogicalRepRelId relid;
    ApplyExecutionData *edata;
    EState     *estate;
    TupleTableSlot *remoteslot;
    MemoryContext oldctx;

    if (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;
    }

    /* Check if we can do the delete. */
    check_relation_updatable(rel);

    /* 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
        apply_handle_delete_internal(edata, edata->targetRelInfo,
                                     remoteslot);

    finish_edata(edata);

    logicalrep_rel_close(rel, NoLock);

    end_replication_step();
}

apply_handle_delete_internal()

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

static

Definition at line 1908 of file worker.c.

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

Referenced by apply_handle_delete(), and apply_handle_tuple_routing().

{
    EState     *estate = edata->estate;
    Relation    localrel = relinfo->ri_RelationDesc;
    LogicalRepRelation *remoterel = &edata->targetRel->remoterel;
    EPQState    epqstate;
    TupleTableSlot *localslot;
    bool        found;

    EvalPlanQualInit(&epqstate, estate, NULL, NIL, -1);
    ExecOpenIndices(relinfo, false);

    found = FindReplTupleInLocalRel(estate, localrel, remoterel,
                                    remoteslot, &localslot);

    /* If found delete it. */
    if (found)
    {
        EvalPlanQualSetSlot(&epqstate, localslot);

        /* Do the actual delete. */
        ExecSimpleRelationDelete(relinfo, estate, &epqstate, localslot);
    }
    else
    {
        /*
         * The tuple to be deleted could not be found.  Do nothing except for
         * emitting a log message.
         *
         * XXX should this be promoted to ereport(LOG) perhaps?
         */
        elog(DEBUG1,
             "logical replication did not find row to be deleted "
             "in replication target relation \"%s\"",
             RelationGetRelationName(localrel));
    }

    /* Cleanup. */
    ExecCloseIndices(relinfo);
    EvalPlanQualEnd(&epqstate);
}

apply_handle_insert()

static void apply_handle_insert ( StringInfo  s )

static

Definition at line 1567 of file worker.c.

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

Referenced by apply_dispatch().

{
    LogicalRepRelMapEntry *rel;
    LogicalRepTupleData newtup;
    LogicalRepRelId relid;
    ApplyExecutionData *edata;
    EState     *estate;
    TupleTableSlot *remoteslot;
    MemoryContext oldctx;

    if (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;
    }

    /* 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
        apply_handle_insert_internal(edata, edata->targetRelInfo,
                                     remoteslot);

    finish_edata(edata);

    logicalrep_rel_close(rel, NoLock);

    end_replication_step();
}

apply_handle_insert_internal()

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

static

Definition at line 1629 of file worker.c.

References ApplyExecutionData::estate, ExecCloseIndices(), ExecOpenIndices(), and ExecSimpleRelationInsert().

Referenced by apply_handle_insert(), and apply_handle_tuple_routing().

{
    EState     *estate = edata->estate;

    /* We must open indexes here. */
    ExecOpenIndices(relinfo, false);

    /* Do the insert. */
    ExecSimpleRelationInsert(relinfo, estate, remoteslot);

    /* Cleanup. */
    ExecCloseIndices(relinfo);
}

apply_handle_origin()

static void apply_handle_origin ( StringInfo  s )

static

Definition at line 1111 of file worker.c.

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

Referenced by apply_dispatch().

{
    /*
     * 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")));
}

apply_handle_prepare()

static void apply_handle_prepare ( StringInfo  s )

static

Definition at line 926 of file worker.c.

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

Referenced by apply_dispatch().

{
    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/%X in prepare message (expected %X/%X)",
                                 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. It is done this way because at
     * commit prepared time, we won't know whether we have skipped preparing a
     * transaction because of no change.
     *
     * 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);

    store_flush_position(prepare_data.end_lsn);

    in_remote_transaction = false;

    /* Process any tables that are being synchronized in parallel. */
    process_syncing_tables(prepare_data.end_lsn);

    pgstat_report_activity(STATE_IDLE, NULL);
}

apply_handle_prepare_internal()

static void apply_handle_prepare_internal ( LogicalRepPreparedTxnData *  prepare_data )

static

Definition at line 892 of file worker.c.

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

Referenced by apply_handle_prepare(), and apply_handle_stream_prepare().

{
    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.
     */
    BeginTransactionBlock();
    CommitTransactionCommand(); /* Completes the preceding Begin command. */

    /*
     * Update origin state so we can restart streaming from correct position
     * in case of crash.
     */
    replorigin_session_origin_lsn = prepare_data->end_lsn;
    replorigin_session_origin_timestamp = prepare_data->prepare_time;

    PrepareTransactionBlock(gid);
}

apply_handle_relation()

static void apply_handle_relation ( StringInfo  s )

static

Definition at line 1517 of file worker.c.

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

Referenced by apply_dispatch().

{
    LogicalRepRelation *rel;

    if (handle_streamed_transaction(LOGICAL_REP_MSG_RELATION, s))
        return;

    rel = logicalrep_read_rel(s);
    logicalrep_relmap_update(rel);
}

apply_handle_rollback_prepared()

static void apply_handle_rollback_prepared ( StringInfo  s )

static

Definition at line 1010 of file worker.c.

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

Referenced by apply_dispatch().

{
    LogicalRepRollbackPreparedTxnData rollback_data;
    char        gid[GIDSIZE];

    logicalrep_read_rollback_prepared(s, &rollback_data);

    /* 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_session_origin_lsn = rollback_data.rollback_end_lsn;
        replorigin_session_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();
    }

    pgstat_report_stat(false);

    store_flush_position(rollback_data.rollback_end_lsn);
    in_remote_transaction = false;

    /* Process any tables that are being synchronized in parallel. */
    process_syncing_tables(rollback_data.rollback_end_lsn);

    pgstat_report_activity(STATE_IDLE, NULL);
}

apply_handle_stream_abort()

static void apply_handle_stream_abort ( StringInfo  s )

static

Definition at line 1221 of file worker.c.

References begin_replication_step(), BufFileClose(), BufFileOpenShared(), BufFileTruncateShared(), changes_filename(), cleanup_subxact_info(), CommitTransactionCommand(), end_replication_step(), ereport, errcode(), errmsg_internal(), ERROR, fd(), SubXactInfo::fileno, HASH_FIND, hash_search(), i, in_streamed_transaction, logicalrep_read_stream_abort(), MAXPGPATH, MyLogicalRepWorker, ApplySubXactData::nsubxacts, SubXactInfo::offset, stream_cleanup_files(), StreamXidHash::stream_fileset, LogicalRepWorker::subid, subxact_info_read(), subxact_info_write(), ApplySubXactData::subxacts, and SubXactInfo::xid.

Referenced by apply_dispatch().

{
    TransactionId xid;
    TransactionId subxid;

    if (in_streamed_transaction)
        ereport(ERROR,
                (errcode(ERRCODE_PROTOCOL_VIOLATION),
                 errmsg_internal("STREAM ABORT message without STREAM STOP")));

    logicalrep_read_stream_abort(s, &xid, &subxid);

    /*
     * 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];
        StreamXidHash *ent;

        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;
        }

        ent = (StreamXidHash *) hash_search(xidhash,
                                            (void *) &xid,
                                            HASH_FIND,
                                            NULL);
        if (!ent)
            ereport(ERROR,
                    (errcode(ERRCODE_PROTOCOL_VIOLATION),
                     errmsg_internal("transaction %u not found in stream XID hash table",
                                     xid)));

        /* open the changes file */
        changes_filename(path, MyLogicalRepWorker->subid, xid);
        fd = BufFileOpenShared(ent->stream_fileset, path, O_RDWR);

        /* OK, truncate the file at the right offset */
        BufFileTruncateShared(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();
    }
}

apply_handle_stream_commit()

static void apply_handle_stream_commit ( StringInfo  s )

static

Definition at line 1451 of file worker.c.

References apply_handle_commit_internal(), apply_spooled_messages(), LogicalRepCommitData::commit_lsn, DEBUG1, elog, LogicalRepCommitData::end_lsn, ereport, errcode(), errmsg_internal(), ERROR, in_streamed_transaction, logicalrep_read_stream_commit(), MyLogicalRepWorker, pgstat_report_activity(), process_syncing_tables(), STATE_IDLE, stream_cleanup_files(), and LogicalRepWorker::subid.

Referenced by apply_dispatch().

{
    TransactionId xid;
    LogicalRepCommitData commit_data;

    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);

    elog(DEBUG1, "received commit for streamed transaction %u", xid);

    apply_spooled_messages(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);

    /* Process any tables that are being synchronized in parallel. */
    process_syncing_tables(commit_data.end_lsn);

    pgstat_report_activity(STATE_IDLE, NULL);
}

apply_handle_stream_prepare()

static void apply_handle_stream_prepare ( StringInfo  s )

static

Definition at line 1063 of file worker.c.

References am_tablesync_worker(), apply_handle_prepare_internal(), apply_spooled_messages(), CommitTransactionCommand(), DEBUG1, elog, LogicalRepPreparedTxnData::end_lsn, ereport, errcode(), errmsg_internal(), ERROR, in_remote_transaction, in_streamed_transaction, logicalrep_read_stream_prepare(), MyLogicalRepWorker, pgstat_report_activity(), pgstat_report_stat(), LogicalRepPreparedTxnData::prepare_lsn, process_syncing_tables(), STATE_IDLE, store_flush_position(), stream_cleanup_files(), LogicalRepWorker::subid, and LogicalRepPreparedTxnData::xid.

Referenced by apply_dispatch().

{
    LogicalRepPreparedTxnData prepare_data;

    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);

    elog(DEBUG1, "received prepare for streamed transaction %u", prepare_data.xid);

    /* Replay all the spooled operations. */
    apply_spooled_messages(prepare_data.xid, prepare_data.prepare_lsn);

    /* Mark the transaction as prepared. */
    apply_handle_prepare_internal(&prepare_data);

    CommitTransactionCommand();

    pgstat_report_stat(false);

    store_flush_position(prepare_data.end_lsn);

    in_remote_transaction = false;

    /* unlink the files with serialized changes and subxact info. */
    stream_cleanup_files(MyLogicalRepWorker->subid, prepare_data.xid);

    /* Process any tables that are being synchronized in parallel. */
    process_syncing_tables(prepare_data.end_lsn);

    pgstat_report_activity(STATE_IDLE, NULL);
}

apply_handle_stream_start()

static void apply_handle_stream_start ( StringInfo  s )

static

Definition at line 1129 of file worker.c.

References ApplyContext, begin_replication_step(), end_replication_step(), HASHCTL::entrysize, ereport, errcode(), errmsg_internal(), ERROR, HASH_BLOBS, HASH_CONTEXT, hash_create(), HASH_ELEM, HASHCTL::hcxt, in_streamed_transaction, HASHCTL::keysize, logicalrep_read_stream_start(), MyLogicalRepWorker, pgstat_report_activity(), STATE_RUNNING, stream_open_file(), stream_xid, LogicalRepWorker::subid, subxact_info_read(), and TransactionIdIsValid.

Referenced by apply_dispatch().

{
    bool        first_segment;
    HASHCTL     hash_ctl;

    if (in_streamed_transaction)
        ereport(ERROR,
                (errcode(ERRCODE_PROTOCOL_VIOLATION),
                 errmsg_internal("duplicate STREAM START message")));

    /*
     * Start a transaction on stream start, 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 the
     * transaction for handling the buffile, used for serializing the
     * streaming data and subxact info.
     */
    begin_replication_step();

    /* 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")));

    /*
     * Initialize the xidhash table if we haven't yet. This will be used for
     * the entire duration of the apply worker so create it in permanent
     * context.
     */
    if (xidhash == NULL)
    {
        hash_ctl.keysize = sizeof(TransactionId);
        hash_ctl.entrysize = sizeof(StreamXidHash);
        hash_ctl.hcxt = ApplyContext;
        xidhash = hash_create("StreamXidHash", 1024, &hash_ctl,
                              HASH_ELEM | HASH_BLOBS | HASH_CONTEXT);
    }

    /* open the spool file for this transaction */
    stream_open_file(MyLogicalRepWorker->subid, stream_xid, first_segment);

    /* if this is not the first segment, open existing subxact file */
    if (!first_segment)
        subxact_info_read(MyLogicalRepWorker->subid, stream_xid);

    pgstat_report_activity(STATE_RUNNING, NULL);

    end_replication_step();
}

apply_handle_stream_stop()

static void apply_handle_stream_stop ( StringInfo  s )

static

Definition at line 1189 of file worker.c.

References Assert, CommitTransactionCommand(), ereport, errcode(), errmsg_internal(), ERROR, in_streamed_transaction, IsTransactionState(), MemoryContextReset(), MyLogicalRepWorker, pgstat_report_activity(), STATE_IDLE, stream_close_file(), stream_xid, LogicalRepWorker::subid, and subxact_info_write().

Referenced by apply_dispatch().

{
    if (!in_streamed_transaction)
        ereport(ERROR,
                (errcode(ERRCODE_PROTOCOL_VIOLATION),
                 errmsg_internal("STREAM STOP message without STREAM START")));

    /*
     * Close the file with serialized changes, and serialize information about
     * subxacts for the toplevel transaction.
     */
    subxact_info_write(MyLogicalRepWorker->subid, stream_xid);
    stream_close_file();

    /* We must be in a valid transaction state */
    Assert(IsTransactionState());

    /* Commit the per-stream transaction */
    CommitTransactionCommand();

    in_streamed_transaction = false;

    /* Reset per-stream context */
    MemoryContextReset(LogicalStreamingContext);

    pgstat_report_activity(STATE_IDLE, NULL);
}

apply_handle_truncate()

static void apply_handle_truncate ( StringInfo  s )

static

Definition at line 2213 of file worker.c.

References AccessExclusiveLock, begin_replication_step(), DROP_RESTRICT, end_replication_step(), ExecuteTruncateGuts(), find_all_inheritors(), handle_streamed_transaction(), 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(), NIL, NoLock, RelationData::rd_rel, RELATION_IS_OTHER_TEMP, RelationIsLogicallyLogged, should_apply_changes_for_rel(), table_close(), and table_open().

Referenced by apply_dispatch().

{
    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;

    if (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);
        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;
                }

                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.
     */
    ExecuteTruncateGuts(rels,
                        relids,
                        relids_logged,
                        DROP_RESTRICT,
                        restart_seqs);
    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();
}

apply_handle_tuple_routing()

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

static

Definition at line 1989 of file worker.c.

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

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

{
    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;

    /* 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;

    /*
     * 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 = partrelinfo->ri_RootToPartitionMap;
    if (map != NULL)
        remoteslot_part = execute_attr_map_slot(map->attrMap, remoteslot,
                                                remoteslot_part);
    else
    {
        remoteslot_part = ExecCopySlot(remoteslot_part, remoteslot);
        slot_getallattrs(remoteslot_part);
    }
    MemoryContextSwitchTo(oldctx);

    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);
            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.
             */
            {
                AttrMap    *attrmap = map ? map->attrMap : NULL;
                LogicalRepRelMapEntry *part_entry;
                TupleTableSlot *localslot;
                ResultRelInfo *partrelinfo_new;
                bool        found;

                part_entry = logicalrep_partition_open(relmapentry, partrel,
                                                       attrmap);

                /* Get the matching local tuple from the partition. */
                found = FindReplTupleInLocalRel(estate, partrel,
                                                &part_entry->remoterel,
                                                remoteslot_part, &localslot);
                if (!found)
                {
                    /*
                     * The tuple to be updated could not be found.  Do nothing
                     * except for emitting a log message.
                     *
                     * XXX should this be promoted to ereport(LOG) perhaps?
                     */
                    elog(DEBUG1,
                         "logical replication did not find row to be updated "
                         "in replication target relation's partition \"%s\"",
                         RelationGetRelationName(partrel));
                    return;
                }

                /*
                 * 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);

                /*
                 * 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.
                     */
                    EPQState    epqstate;

                    EvalPlanQualInit(&epqstate, estate, NULL, NIL, -1);
                    ExecOpenIndices(partrelinfo, false);

                    EvalPlanQualSetSlot(&epqstate, remoteslot_part);
                    ExecSimpleRelationUpdate(partrelinfo, estate, &epqstate,
                                             localslot, remoteslot_part);
                    ExecCloseIndices(partrelinfo);
                    EvalPlanQualEnd(&epqstate);
                }
                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);

                    /* DELETE old tuple found in the old partition. */
                    apply_handle_delete_internal(edata, partrelinfo,
                                                 localslot);

                    /* INSERT new tuple into the new partition. */

                    /*
                     * Convert the replacement tuple to match the destination
                     * partition rowtype.
                     */
                    oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
                    partrel = partrelinfo_new->ri_RelationDesc;
                    remoteslot_part = partrelinfo_new->ri_PartitionTupleSlot;
                    if (remoteslot_part == NULL)
                        remoteslot_part = table_slot_create(partrel,
                                                            &estate->es_tupleTable);
                    map = partrelinfo_new->ri_RootToPartitionMap;
                    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);
                }
            }
            break;

        default:
            elog(ERROR, "unrecognized CmdType: %d", (int) operation);
            break;
    }
}

apply_handle_type()

static void apply_handle_type ( StringInfo  s )

static

Definition at line 1534 of file worker.c.

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

Referenced by apply_dispatch().

{
    LogicalRepTyp typ;

    if (handle_streamed_transaction(LOGICAL_REP_MSG_TYPE, s))
        return;

    logicalrep_read_typ(s, &typ);
}

apply_handle_update()

static void apply_handle_update ( StringInfo  s )

static

Definition at line 1684 of file worker.c.

References apply_handle_tuple_routing(), apply_handle_update_internal(), Assert, AttrMap::attnums, LogicalRepRelMapEntry::attrmap, begin_replication_step(), bms_add_member(), check_relation_updatable(), CMD_UPDATE, LogicalRepTupleData::colstatus, create_edata_for_relation(), end_replication_step(), EState::es_range_table, ApplyExecutionData::estate, ExecInitExtraTupleSlot(), fill_extraUpdatedCols(), finish_edata(), FirstLowInvalidHeapAttributeNumber, GetPerTupleMemoryContext, handle_streamed_transaction(), i, list_nth(), LogicalRepRelMapEntry::localrel, LOGICAL_REP_MSG_UPDATE, LOGICALREP_COLUMN_UNCHANGED, logicalrep_read_update(), logicalrep_rel_close(), logicalrep_rel_open(), MemoryContextSwitchTo(), TupleDescData::natts, LogicalRepTupleData::ncols, NoLock, RelationData::rd_rel, RelationGetDescr, RowExclusiveLock, should_apply_changes_for_rel(), slot_store_data(), ApplyExecutionData::targetRelInfo, TupleTableSlot::tts_tupleDescriptor, TTSOpsVirtual, TupleDescAttr, and RangeTblEntry::updatedCols.

Referenced by apply_dispatch().

{
    LogicalRepRelMapEntry *rel;
    LogicalRepRelId relid;
    ApplyExecutionData *edata;
    EState     *estate;
    LogicalRepTupleData oldtup;
    LogicalRepTupleData newtup;
    bool        has_oldtup;
    TupleTableSlot *remoteslot;
    RangeTblEntry *target_rte;
    MemoryContext oldctx;

    if (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;
    }

    /* Check if we can do the update. */
    check_relation_updatable(rel);

    /* 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_rte = list_nth(estate->es_range_table, 0);
    for (int i = 0; i < remoteslot->tts_tupleDescriptor->natts; i++)
    {
        Form_pg_attribute att = TupleDescAttr(remoteslot->tts_tupleDescriptor, i);
        int         remoteattnum = rel->attrmap->attnums[i];

        if (!att->attisdropped && remoteattnum >= 0)
        {
            Assert(remoteattnum < newtup.ncols);
            if (newtup.colstatus[remoteattnum] != LOGICALREP_COLUMN_UNCHANGED)
                target_rte->updatedCols =
                    bms_add_member(target_rte->updatedCols,
                                   i + 1 - FirstLowInvalidHeapAttributeNumber);
        }
    }

    /* Also populate extraUpdatedCols, in case we have generated columns */
    fill_extraUpdatedCols(target_rte, rel->localrel);

    /* 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);

    finish_edata(edata);

    logicalrep_rel_close(rel, NoLock);

    end_replication_step();
}

apply_handle_update_internal()

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

static

Definition at line 1780 of file worker.c.

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

Referenced by apply_handle_update().

{
    EState     *estate = edata->estate;
    LogicalRepRelMapEntry *relmapentry = edata->targetRel;
    Relation    localrel = relinfo->ri_RelationDesc;
    EPQState    epqstate;
    TupleTableSlot *localslot;
    bool        found;
    MemoryContext oldctx;

    EvalPlanQualInit(&epqstate, estate, NULL, NIL, -1);
    ExecOpenIndices(relinfo, false);

    found = FindReplTupleInLocalRel(estate, localrel,
                                    &relmapentry->remoterel,
                                    remoteslot, &localslot);
    ExecClearTuple(remoteslot);

    /*
     * Tuple found.
     *
     * Note this will fail if there are other conflicting unique indexes.
     */
    if (found)
    {
        /* Process and store remote tuple in the slot */
        oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
        slot_modify_data(remoteslot, localslot, relmapentry, newtup);
        MemoryContextSwitchTo(oldctx);

        EvalPlanQualSetSlot(&epqstate, remoteslot);

        /* Do the actual update. */
        ExecSimpleRelationUpdate(relinfo, estate, &epqstate, localslot,
                                 remoteslot);
    }
    else
    {
        /*
         * The tuple to be updated could not be found.  Do nothing except for
         * emitting a log message.
         *
         * XXX should this be promoted to ereport(LOG) perhaps?
         */
        elog(DEBUG1,
             "logical replication did not find row to be updated "
             "in replication target relation \"%s\"",
             RelationGetRelationName(localrel));
    }

    /* Cleanup. */
    ExecCloseIndices(relinfo);
    EvalPlanQualEnd(&epqstate);
}

apply_spooled_messages()

static void apply_spooled_messages ( TransactionId  xid, XLogRecPtr  lsn  )

static

Definition at line 1324 of file worker.c.

References appendBinaryStringInfo(), apply_dispatch(), begin_replication_step(), BufFileClose(), BufFileOpenShared(), BufFileRead(), changes_filename(), CHECK_FOR_INTERRUPTS, StringInfoData::data, DEBUG1, elog, end_replication_step(), ereport, errcode(), errcode_for_file_access(), errmsg(), errmsg_internal(), ERROR, fd(), HASH_FIND, hash_search(), in_remote_transaction, initStringInfo(), MAXPGPATH, MemoryContextReset(), MemoryContextSwitchTo(), MyLogicalRepWorker, palloc(), pfree(), pgstat_report_activity(), remote_final_lsn, repalloc(), resetStringInfo(), s2, STATE_RUNNING, StreamXidHash::stream_fileset, LogicalRepWorker::subid, and TopTransactionContext.

Referenced by apply_handle_stream_commit(), and apply_handle_stream_prepare().

{
    StringInfoData s2;
    int         nchanges;
    char        path[MAXPGPATH];
    char       *buffer = NULL;
    StreamXidHash *ent;
    MemoryContext oldcxt;
    BufFile    *fd;

    /* 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);

    ent = (StreamXidHash *) hash_search(xidhash,
                                        (void *) &xid,
                                        HASH_FIND,
                                        NULL);
    if (!ent)
        ereport(ERROR,
                (errcode(ERRCODE_PROTOCOL_VIOLATION),
                 errmsg_internal("transaction %u not found in stream XID hash table",
                                 xid)));

    fd = BufFileOpenShared(ent->stream_fileset, path, O_RDONLY);

    buffer = palloc(BLCKSZ);
    initStringInfo(&s2);

    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)
    {
        int         nbytes;
        int         len;

        CHECK_FOR_INTERRUPTS();

        /* read length of the on-disk record */
        nbytes = BufFileRead(fd, &len, sizeof(len));

        /* have we reached end of the file? */
        if (nbytes == 0)
            break;

        /* do we have a correct length? */
        if (nbytes != sizeof(len))
            ereport(ERROR,
                    (errcode_for_file_access(),
                     errmsg("could not read from streaming transaction's changes file \"%s\": %m",
                            path)));

        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 */
        if (BufFileRead(fd, buffer, len) != len)
            ereport(ERROR,
                    (errcode_for_file_access(),
                     errmsg("could not read from streaming transaction's changes file \"%s\": %m",
                            path)));

        /* copy the buffer to the stringinfo and call apply_dispatch */
        resetStringInfo(&s2);
        appendBinaryStringInfo(&s2, buffer, len);

        /* Ensure we are reading the data into our memory context. */
        oldcxt = MemoryContextSwitchTo(ApplyMessageContext);

        apply_dispatch(&s2);

        MemoryContextReset(ApplyMessageContext);

        MemoryContextSwitchTo(oldcxt);

        nchanges++;

        if (nchanges % 1000 == 0)
            elog(DEBUG1, "replayed %d changes from file \"%s\"",
                 nchanges, path);
    }

    BufFileClose(fd);

    pfree(buffer);
    pfree(s2.data);

    elog(DEBUG1, "replayed %d (all) changes from file \"%s\"",
         nchanges, path);

    return;
}

ApplyWorkerMain()

void ApplyWorkerMain

(

Datum

main_arg

)

Definition at line 3417 of file worker.c.

References ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate, AllTablesyncsReady(), am_tablesync_worker(), BackgroundWorkerInitializeConnectionByOid(), BackgroundWorkerUnblockSignals(), Subscription::binary, CacheRegisterSyscacheCallback(), CommitTransactionCommand(), Subscription::conninfo, DatumGetInt32, LogicalRepWorker::dbid, DEBUG1, die, elog, Subscription::enabled, ereport, errcode(), errmsg(), ERROR, get_rel_name(), GetCurrentTimestamp(), GetSubscription(), invalidate_syncing_table_states(), LogicalRepWorker::last_recv_time, LogicalRepWorker::last_send_time, load_file(), LOG, WalRcvStreamOptions::logical, LOGICALREP_PROTO_STREAM_VERSION_NUM, LOGICALREP_PROTO_TWOPHASE_VERSION_NUM, LOGICALREP_PROTO_VERSION_NUM, LOGICALREP_TWOPHASE_STATE_DISABLED, LOGICALREP_TWOPHASE_STATE_ENABLED, LOGICALREP_TWOPHASE_STATE_PENDING, logicalrep_worker_attach(), LogicalRepApplyLoop(), LogicalRepSyncTableStart(), MemoryContextStrdup(), MemoryContextSwitchTo(), MyLogicalRepWorker, MySubscriptionValid, Subscription::name, NAMEDATALEN, Subscription::oid, OidIsValid, options, pfree(), PGC_BACKEND, PGC_S_OVERRIDE, PGC_SUSET, pqsignal(), proc_exit(), WalRcvStreamOptions::proto, Subscription::publications, LogicalRepWorker::relid, replorigin_by_name(), replorigin_create(), replorigin_session_get_progress(), replorigin_session_origin, replorigin_session_setup(), LogicalRepWorker::reply_time, server_version, SetConfigOption(), SIGHUP, SignalHandlerForConfigReload(), Subscription::slotname, WalRcvStreamOptions::slotname, snprintf, WalRcvStreamOptions::startpoint, StartTransactionCommand(), Subscription::stream, LogicalRepWorker::subid, subscription_change_cb(), SUBSCRIPTIONOID, SUBSCRIPTIONRELMAP, Subscription::synccommit, TopMemoryContext, Subscription::twophasestate, UpdateTwoPhaseState(), LogicalRepWorker::userid, walrcv_connect, walrcv_identify_system, walrcv_server_version, and walrcv_startstreaming.

{
    int         worker_slot = DatumGetInt32(main_arg);
    MemoryContext oldctx;
    char        originname[NAMEDATALEN];
    XLogRecPtr  origin_startpos;
    char       *myslotname;
    WalRcvStreamOptions options;
    int         server_version;

    /* Attach to slot */
    logicalrep_worker_attach(worker_slot);

    /* Setup signal handling */
    pqsignal(SIGHUP, SignalHandlerForConfigReload);
    pqsignal(SIGTERM, die);
    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);

    /* 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);

    /* Load the subscription into persistent memory context. */
    ApplyContext = AllocSetContextCreate(TopMemoryContext,
                                         "ApplyContext",
                                         ALLOCSET_DEFAULT_SIZES);
    StartTransactionCommand();
    oldctx = MemoryContextSwitchTo(ApplyContext);

    MySubscription = GetSubscription(MyLogicalRepWorker->subid, true);
    if (!MySubscription)
    {
        ereport(LOG,
                (errmsg("logical replication apply worker for subscription %u will not "
                        "start because the subscription was removed during startup",
                        MyLogicalRepWorker->subid)));
        proc_exit(0);
    }

    MySubscriptionValid = true;
    MemoryContextSwitchTo(oldctx);

    if (!MySubscription->enabled)
    {
        ereport(LOG,
                (errmsg("logical replication apply worker for subscription \"%s\" will not "
                        "start because the subscription was disabled during startup",
                        MySubscription->name)));

        proc_exit(0);
    }

    /* Setup synchronous commit according to the user's wishes */
    SetConfigOption("synchronous_commit", MySubscription->synccommit,
                    PGC_BACKEND, PGC_S_OVERRIDE);

    /* Keep us informed about subscription changes. */
    CacheRegisterSyscacheCallback(SUBSCRIPTIONOID,
                                  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
        ereport(LOG,
                (errmsg("logical replication apply worker for subscription \"%s\" has started",
                        MySubscription->name)));

    CommitTransactionCommand();

    /* Connect to the origin and start the replication. */
    elog(DEBUG1, "connecting to publisher using connection string \"%s\"",
         MySubscription->conninfo);

    if (am_tablesync_worker())
    {
        char       *syncslotname;

        /* This is table synchronization worker, call initial sync. */
        syncslotname = LogicalRepSyncTableStart(&origin_startpos);

        /* allocate slot name in long-lived context */
        myslotname = MemoryContextStrdup(ApplyContext, syncslotname);

        pfree(syncslotname);
    }
    else
    {
        /* This is main apply worker */
        RepOriginId originid;
        TimeLineID  startpointTLI;
        char       *err;

        myslotname = 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 (!myslotname)
            ereport(ERROR,
                    (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                     errmsg("subscription has no replication slot set")));

        /* Setup replication origin tracking. */
        StartTransactionCommand();
        snprintf(originname, sizeof(originname), "pg_%u", MySubscription->oid);
        originid = replorigin_by_name(originname, true);
        if (!OidIsValid(originid))
            originid = replorigin_create(originname);
        replorigin_session_setup(originid);
        replorigin_session_origin = originid;
        origin_startpos = replorigin_session_get_progress(false);
        CommitTransactionCommand();

        LogRepWorkerWalRcvConn = walrcv_connect(MySubscription->conninfo, true,
                                                MySubscription->name, &err);
        if (LogRepWorkerWalRcvConn == NULL)
            ereport(ERROR,
                    (errcode(ERRCODE_CONNECTION_FAILURE),
                     errmsg("could not connect to the publisher: %s", 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);
    }

    /*
     * Setup callback for syscache so that we know when something changes in
     * the subscription relation state.
     */
    CacheRegisterSyscacheCallback(SUBSCRIPTIONRELMAP,
                                  invalidate_syncing_table_states,
                                  (Datum) 0);

    /* Build logical replication streaming options. */
    options.logical = true;
    options.startpoint = origin_startpos;
    options.slotname = myslotname;

    server_version = walrcv_server_version(LogRepWorkerWalRcvConn);
    options.proto.logical.proto_version =
        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;
    options.proto.logical.streaming = MySubscription->stream;
    options.proto.logical.twophase = false;

    if (!am_tablesync_worker())
    {
        /*
         * 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();
            UpdateTwoPhaseState(MySubscription->oid, LOGICALREP_TWOPHASE_STATE_ENABLED);
            MySubscription->twophasestate = LOGICALREP_TWOPHASE_STATE_ENABLED;
            CommitTransactionCommand();
        }
        else
        {
            walrcv_startstreaming(LogRepWorkerWalRcvConn, &options);
        }

        ereport(DEBUG1,
                (errmsg("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" :
                        "?")));
    }
    else
    {
        /* Start normal logical streaming replication. */
        walrcv_startstreaming(LogRepWorkerWalRcvConn, &options);
    }

    /* Run the main loop. */
    LogicalRepApplyLoop(origin_startpos);

    proc_exit(0);
}

begin_replication_step()

static void begin_replication_step ( void  )

static

Definition at line 370 of file worker.c.

References 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_abort(), apply_handle_stream_start(), apply_handle_truncate(), apply_handle_update(), and apply_spooled_messages().

{
    SetCurrentStatementStartTimestamp();

    if (!IsTransactionState())
    {
        StartTransactionCommand();
        maybe_reread_subscription();
    }

    PushActiveSnapshot(GetTransactionSnapshot());

    MemoryContextSwitchTo(ApplyMessageContext);
}

changes_filename()

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

inlinestatic

Definition at line 3191 of file worker.c.

References MAXPGPATH, and snprintf.

Referenced by apply_handle_stream_abort(), apply_spooled_messages(), stream_cleanup_files(), and stream_open_file().

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

check_relation_updatable()

static void check_relation_updatable ( LogicalRepRelMapEntry *  rel )

static

Definition at line 1650 of file worker.c.

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

Referenced by apply_handle_delete(), and apply_handle_update().

{
    /* 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)));
}

cleanup_subxact_info()

static void cleanup_subxact_info ( void  )

inlinestatic

Definition at line 3386 of file worker.c.

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

Referenced by apply_handle_stream_abort(), and subxact_info_write().

{
    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;
}

create_edata_for_relation()

static ApplyExecutionData* create_edata_for_relation ( LogicalRepRelMapEntry *  rel )

static

Definition at line 447 of file worker.c.

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

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

{
    ApplyExecutionData *edata;
    EState     *estate;
    RangeTblEntry *rte;
    ResultRelInfo *resultRelInfo;

    edata = (ApplyExecutionData *) palloc0(sizeof(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;
    ExecInitRangeTable(estate, list_make1(rte));

    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;
}

end_replication_step()

static void end_replication_step ( void  )

static

Definition at line 393 of file worker.c.

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_abort(), apply_handle_stream_start(), apply_handle_truncate(), apply_handle_update(), and apply_spooled_messages().

{
    PopActiveSnapshot();

    CommandCounterIncrement();
}

FindReplTupleInLocalRel()

static bool FindReplTupleInLocalRel ( EState *  estate, Relation  localrel, LogicalRepRelation *  remoterel, TupleTableSlot *  remoteslot, TupleTableSlot **  localslot  )

static

Definition at line 1960 of file worker.c.

References Assert, EState::es_tupleTable, GetRelationIdentityOrPK(), LockTupleExclusive, OidIsValid, RelationFindReplTupleByIndex(), RelationFindReplTupleSeq(), LogicalRepRelation::replident, and table_slot_create().

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

{
    Oid         idxoid;
    bool        found;

    *localslot = table_slot_create(localrel, &estate->es_tupleTable);

    idxoid = GetRelationIdentityOrPK(localrel);
    Assert(OidIsValid(idxoid) ||
           (remoterel->replident == REPLICA_IDENTITY_FULL));

    if (OidIsValid(idxoid))
        found = RelationFindReplTupleByIndex(localrel, idxoid,
                                             LockTupleExclusive,
                                             remoteslot, *localslot);
    else
        found = RelationFindReplTupleSeq(localrel, LockTupleExclusive,
                                         remoteslot, *localslot);

    return found;
}

finish_edata()

static void finish_edata ( ApplyExecutionData *  edata )

static

Definition at line 500 of file worker.c.

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

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

{
    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);
}

get_flush_position()

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

static

Definition at line 2436 of file worker.c.

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

Referenced by send_feedback().

{
    dlist_mutable_iter iter;
    XLogRecPtr  local_flush = GetFlushRecPtr();

    *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);
}

GetRelationIdentityOrPK()

static Oid GetRelationIdentityOrPK ( Relation  rel )

static

Definition at line 1550 of file worker.c.

References OidIsValid, RelationGetPrimaryKeyIndex(), and RelationGetReplicaIndex().

Referenced by check_relation_updatable(), and FindReplTupleInLocalRel().

{
    Oid         idxoid;

    idxoid = RelationGetReplicaIndex(rel);

    if (!OidIsValid(idxoid))
        idxoid = RelationGetPrimaryKeyIndex(rel);

    return idxoid;
}

handle_streamed_transaction()

static bool handle_streamed_transaction ( LogicalRepMsgType  action, StringInfo  s  )

static

Definition at line 409 of file worker.c.

References Assert, ereport, errcode(), errmsg_internal(), ERROR, in_streamed_transaction, pq_getmsgint(), stream_write_change(), stream_xid, subxact_info_add(), and TransactionIdIsValid.

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

{
    TransactionId xid;

    /* not in streaming mode */
    if (!in_streamed_transaction)
        return false;

    Assert(stream_fd != NULL);
    Assert(TransactionIdIsValid(stream_xid));

    /*
     * We should have received XID of the subxact as the first part of the
     * message, so extract it.
     */
    xid = pq_getmsgint(s, 4);

    if (!TransactionIdIsValid(xid))
        ereport(ERROR,
                (errcode(ERRCODE_PROTOCOL_VIOLATION),
                 errmsg_internal("invalid transaction ID in streamed replication transaction")));

    /* Add the new subxact to the array (unless already there). */
    subxact_info_add(xid);

    /* write the change to the current file */
    stream_write_change(action, s);

    return true;
}

IsLogicalWorker()

bool IsLogicalWorker

(

void

)

Definition at line 3648 of file worker.c.

References MyLogicalRepWorker.

Referenced by ProcessInterrupts().

{
    return MyLogicalRepWorker != NULL;
}

LogicalRepApplyLoop()

static void LogicalRepApplyLoop ( XLogRecPtr  last_received )

static

Definition at line 2515 of file worker.c.

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

Referenced by ApplyWorkerMain().

{
    TimestampTz last_recv_timestamp = GetCurrentTimestamp();
    bool        ping_sent = false;
    TimeLineID  tli;

    /*
     * 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);

    /* 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;

                    /* Reset timeout. */
                    last_recv_timestamp = GetCurrentTimestamp();
                    ping_sent = false;

                    /* Ensure we are reading the data into our memory context. */
                    MemoryContextSwitchTo(ApplyMessageContext);

                    s.data = buf;
                    s.len = len;
                    s.cursor = 0;
                    s.maxlen = -1;

                    c = pq_getmsgbyte(&s);

                    if (c == 'w')
                    {
                        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);
                    }
                    else if (c == 'k')
                    {
                        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);
                        UpdateWorkerStats(last_received, timestamp, true);
                    }
                    /* 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);

        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 table synchronization changes. */
            process_syncing_tables(last_received);
        }

        /* Cleanup the memory. */
        MemoryContextResetAndDeleteChildren(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;

        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);
        }
    }

    /* All done */
    walrcv_endstreaming(LogRepWorkerWalRcvConn, &tli);
}

maybe_reread_subscription()

static void maybe_reread_subscription ( void  )

static

Definition at line 2840 of file worker.c.

References Assert, Subscription::binary, CommitTransactionCommand(), Subscription::conninfo, Subscription::dbid, elog, Subscription::enabled, equal(), ereport, errmsg(), ERROR, FreeSubscription(), GetSubscription(), IsTransactionState(), LOG, MemoryContextSwitchTo(), MyLogicalRepWorker, MySubscriptionValid, Subscription::name, newsub(), PGC_BACKEND, PGC_S_OVERRIDE, proc_exit(), Subscription::publications, SetConfigOption(), Subscription::slotname, StartTransactionCommand(), Subscription::stream, LogicalRepWorker::subid, Subscription::synccommit, and Subscription::twophasestate.

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

{
    MemoryContext oldctx;
    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;
    }

    /* Ensure allocations in permanent context. */
    oldctx = MemoryContextSwitchTo(ApplyContext);

    newsub = GetSubscription(MyLogicalRepWorker->subid, true);

    /*
     * Exit if the subscription was removed. This normally should not happen
     * as the worker gets killed during DROP SUBSCRIPTION.
     */
    if (!newsub)
    {
        ereport(LOG,
                (errmsg("logical replication apply worker for subscription \"%s\" will "
                        "stop because the subscription was removed",
                        MySubscription->name)));

        proc_exit(0);
    }

    /*
     * Exit if the subscription was disabled. This normally should not happen
     * as the worker gets killed during ALTER SUBSCRIPTION ... DISABLE.
     */
    if (!newsub->enabled)
    {
        ereport(LOG,
                (errmsg("logical replication apply worker for subscription \"%s\" will "
                        "stop because the subscription was disabled",
                        MySubscription->name)));

        proc_exit(0);
    }

    /* !slotname should never happen when enabled is true. */
    Assert(newsub->slotname);

    /* two-phase should not be altered */
    Assert(newsub->twophasestate == MySubscription->twophasestate);

    /*
     * Exit if any parameter that affects the remote connection was changed.
     * The launcher will start a new worker.
     */
    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 ||
        !equal(newsub->publications, MySubscription->publications))
    {
        ereport(LOG,
                (errmsg("logical replication apply worker for subscription \"%s\" will restart because of a parameter change",
                        MySubscription->name)));

        proc_exit(0);
    }

    /* 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. */
    FreeSubscription(MySubscription);
    MySubscription = newsub;

    MemoryContextSwitchTo(oldctx);

    /* Change synchronous commit according to the user's wishes */
    SetConfigOption("synchronous_commit", MySubscription->synccommit,
                    PGC_BACKEND, PGC_S_OVERRIDE);

    if (started_tx)
        CommitTransactionCommand();

    MySubscriptionValid = true;
}

send_feedback()

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

static

Definition at line 2751 of file worker.c.

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

Referenced by LogicalRepApplyLoop().

{
    static StringInfo reply_message = NULL;
    static TimestampTz send_time = 0;

    static XLogRecPtr last_recvpos = InvalidXLogRecPtr;
    static XLogRecPtr last_writepos = InvalidXLogRecPtr;
    static XLogRecPtr last_flushpos = 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, 'r');
    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/%X, write %X/%X, flush %X/%X",
         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;
}

should_apply_changes_for_rel()

static bool should_apply_changes_for_rel ( LogicalRepRelMapEntry *  rel )

static

Definition at line 352 of file worker.c.

References am_tablesync_worker(), LogicalRepRelMapEntry::localreloid, MyLogicalRepWorker, LogicalRepWorker::relid, remote_final_lsn, LogicalRepRelMapEntry::state, and LogicalRepRelMapEntry::statelsn.

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

{
    if (am_tablesync_worker())
        return MyLogicalRepWorker->relid == rel->localreloid;
    else
        return (rel->state == SUBREL_STATE_READY ||
                (rel->state == SUBREL_STATE_SYNCDONE &&
                 rel->statelsn <= remote_final_lsn));
}

slot_fill_defaults()

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

static

Definition at line 531 of file worker.c.

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

Referenced by apply_handle_insert().

{
    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 = (int *) palloc(num_phys_attrs * sizeof(int));
    defexprs = (ExprState **) palloc(num_phys_attrs * sizeof(ExprState *));

    Assert(rel->attrmap->maplen == num_phys_attrs);
    for (attnum = 0; attnum < num_phys_attrs; attnum++)
    {
        Expr       *defexpr;

        if (TupleDescAttr(desc, attnum)->attisdropped || TupleDescAttr(desc, attnum)->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]]);
}

slot_modify_data()

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

static

Definition at line 721 of file worker.c.

References ErrorContextCallback::arg, Assert, AttrMap::attnums, LogicalRepRelMapEntry::attrmap, ErrorContextCallback::callback, LogicalRepTupleData::colstatus, LogicalRepTupleData::colvalues, StringInfoData::cursor, StringInfoData::data, ereport, errcode(), errmsg(), ERROR, error_context_stack, ExecClearTuple(), ExecStoreVirtualTuple(), getTypeBinaryInputInfo(), getTypeInputInfo(), i, StringInfoData::len, LOGICALREP_COLUMN_BINARY, LOGICALREP_COLUMN_TEXT, LOGICALREP_COLUMN_UNCHANGED, AttrMap::maplen, TupleDescData::natts, OidInputFunctionCall(), OidReceiveFunctionCall(), ErrorContextCallback::previous, SlotErrCallbackArg::rel, SlotErrCallbackArg::remote_attnum, slot_getallattrs(), slot_store_error_callback(), TupleTableSlot::tts_isnull, TupleTableSlot::tts_tupleDescriptor, TupleTableSlot::tts_values, and TupleDescAttr.

Referenced by apply_handle_tuple_routing(), and apply_handle_update_internal().

{
    int         natts = slot->tts_tupleDescriptor->natts;
    int         i;
    SlotErrCallbackArg errarg;
    ErrorContextCallback errcallback;

    /* 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));

    /* For error reporting, push callback + info on the error context stack */
    errarg.rel = rel;
    errarg.remote_attnum = -1;
    errcallback.callback = slot_store_error_callback;
    errcallback.arg = (void *) &errarg;
    errcallback.previous = error_context_stack;
    error_context_stack = &errcallback;

    /* 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;

        Assert(remoteattnum < tupleData->ncols);

        if (tupleData->colstatus[remoteattnum] != LOGICALREP_COLUMN_UNCHANGED)
        {
            StringInfo  colvalue = &tupleData->colvalues[remoteattnum];

            errarg.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;
            }

            errarg.remote_attnum = -1;
        }
    }

    /* Pop the error context stack */
    error_context_stack = errcallback.previous;

    /* And finally, declare that "slot" contains a valid virtual tuple */
    ExecStoreVirtualTuple(slot);
}

slot_store_data()

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

static

Definition at line 609 of file worker.c.

References ErrorContextCallback::arg, Assert, AttrMap::attnums, LogicalRepRelMapEntry::attrmap, ErrorContextCallback::callback, LogicalRepTupleData::colstatus, LogicalRepTupleData::colvalues, StringInfoData::cursor, StringInfoData::data, ereport, errcode(), errmsg(), ERROR, error_context_stack, ExecClearTuple(), ExecStoreVirtualTuple(), getTypeBinaryInputInfo(), getTypeInputInfo(), i, StringInfoData::len, LOGICALREP_COLUMN_BINARY, LOGICALREP_COLUMN_TEXT, AttrMap::maplen, TupleDescData::natts, OidInputFunctionCall(), OidReceiveFunctionCall(), ErrorContextCallback::previous, SlotErrCallbackArg::rel, SlotErrCallbackArg::remote_attnum, slot_store_error_callback(), TupleTableSlot::tts_isnull, TupleTableSlot::tts_tupleDescriptor, TupleTableSlot::tts_values, and TupleDescAttr.

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

{
    int         natts = slot->tts_tupleDescriptor->natts;
    int         i;
    SlotErrCallbackArg errarg;
    ErrorContextCallback errcallback;

    ExecClearTuple(slot);

    /* Push callback + info on the error context stack */
    errarg.rel = rel;
    errarg.remote_attnum = -1;
    errcallback.callback = slot_store_error_callback;
    errcallback.arg = (void *) &errarg;
    errcallback.previous = error_context_stack;
    error_context_stack = &errcallback;

    /* 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 = &tupleData->colvalues[remoteattnum];

            Assert(remoteattnum < tupleData->ncols);

            errarg.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;
            }

            errarg.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;
        }
    }

    /* Pop the error context stack */
    error_context_stack = errcallback.previous;

    ExecStoreVirtualTuple(slot);
}