worker.c File Reference

#include "postgres.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "funcapi.h"
#include "access/xact.h"
#include "access/xlog_internal.h"
#include "catalog/catalog.h"
#include "catalog/namespace.h"
#include "catalog/pg_subscription.h"
#include "catalog/pg_subscription_rel.h"
#include "commands/tablecmds.h"
#include "commands/trigger.h"
#include "executor/executor.h"
#include "executor/nodeModifyTable.h"
#include "libpq/pqformat.h"
#include "libpq/pqsignal.h"
#include "mb/pg_wchar.h"
#include "nodes/makefuncs.h"
#include "optimizer/planner.h"
#include "parser/parse_relation.h"
#include "postmaster/bgworker.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/reorderbuffer.h"
#include "replication/origin.h"
#include "replication/snapbuild.h"
#include "replication/walreceiver.h"
#include "replication/worker_internal.h"
#include "rewrite/rewriteHandler.h"
#include "storage/bufmgr.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/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/timeout.h"
#include "utils/tqual.h"
#include "utils/syscache.h"

Include dependency graph for worker.c:

Go to the source code of this file.

Data Structures
struct	FlushPosition
struct	SlotErrCallbackArg

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

Typedefs
typedef struct FlushPosition	FlushPosition
typedef struct SlotErrCallbackArg	SlotErrCallbackArg

Functions
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 bool	should_apply_changes_for_rel (LogicalRepRelMapEntry *rel)
static bool	ensure_transaction (void)
static EState *	create_estate_for_relation (LogicalRepRelMapEntry *rel)
static void	slot_fill_defaults (LogicalRepRelMapEntry *rel, EState *estate, TupleTableSlot *slot)
static void	slot_store_error_callback (void *arg)
static void	slot_store_cstrings (TupleTableSlot *slot, LogicalRepRelMapEntry *rel, char **values)
static void	slot_modify_cstrings (TupleTableSlot *slot, LogicalRepRelMapEntry *rel, char **values, bool *replaces)
static void	apply_handle_begin (StringInfo s)
static void	apply_handle_commit (StringInfo s)
static void	apply_handle_origin (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	apply_dispatch (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)
static void	logicalrep_worker_sighup (SIGNAL_ARGS)
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
WalReceiverConn *	wrconn = NULL
Subscription *	MySubscription = NULL
bool	MySubscriptionValid = false
bool	in_remote_transaction = false
static XLogRecPtr	remote_final_lsn = InvalidXLogRecPtr
static volatile sig_atomic_t	got_SIGHUP = false

Macro Definition Documentation

NAPTIME_PER_CYCLE

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

Definition at line 93 of file worker.c.

Referenced by LogicalRepApplyLoop().

Typedef Documentation

FlushPosition

typedef struct FlushPosition FlushPosition

SlotErrCallbackArg

typedef struct SlotErrCallbackArg SlotErrCallbackArg

Function Documentation

apply_dispatch()

static void apply_dispatch ( StringInfo  s )

static

Definition at line 960 of file worker.c.

References generate_unaccent_rules::action, apply_handle_begin(), apply_handle_commit(), apply_handle_delete(), apply_handle_insert(), apply_handle_origin(), apply_handle_relation(), apply_handle_truncate(), apply_handle_type(), apply_handle_update(), ereport, errcode(), errmsg(), ERROR, and pq_getmsgbyte().

Referenced by LogicalRepApplyLoop().

{
    char        action = pq_getmsgbyte(s);

    switch (action)
    {
            /* BEGIN */
        case 'B':
            apply_handle_begin(s);
            break;
            /* COMMIT */
        case 'C':
            apply_handle_commit(s);
            break;
            /* INSERT */
        case 'I':
            apply_handle_insert(s);
            break;
            /* UPDATE */
        case 'U':
            apply_handle_update(s);
            break;
            /* DELETE */
        case 'D':
            apply_handle_delete(s);
            break;
            /* TRUNCATE */
        case 'T':
            apply_handle_truncate(s);
            break;
            /* RELATION */
        case 'R':
            apply_handle_relation(s);
            break;
            /* TYPE */
        case 'Y':
            apply_handle_type(s);
            break;
            /* ORIGIN */
        case 'O':
            apply_handle_origin(s);
            break;
        default:
            ereport(ERROR,
                    (errcode(ERRCODE_PROTOCOL_VIOLATION),
                     errmsg("invalid logical replication message type \"%c\"", action)));
    }
}

apply_handle_begin()

static void apply_handle_begin ( StringInfo  s )

static

Definition at line 455 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_commit()

static void apply_handle_commit ( StringInfo  s )

static

Definition at line 474 of file worker.c.

References AcceptInvalidationMessages(), am_tablesync_worker(), Assert, LogicalRepCommitData::commit_lsn, LogicalRepCommitData::committime, CommitTransactionCommand(), LogicalRepCommitData::end_lsn, in_remote_transaction, IsTransactionState(), logicalrep_read_commit(), maybe_reread_subscription(), pgstat_report_activity(), pgstat_report_stat(), process_syncing_tables(), remote_final_lsn, replorigin_session_origin_lsn, replorigin_session_origin_timestamp, STATE_IDLE, and store_flush_position().

Referenced by apply_dispatch().

{
    LogicalRepCommitData commit_data;

    logicalrep_read_commit(s, &commit_data);

    Assert(commit_data.commit_lsn == remote_final_lsn);

    /* The synchronization worker runs in single transaction. */
    if (IsTransactionState() && !am_tablesync_worker())
    {
        /*
         * 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;

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

    pgstat_report_activity(STATE_IDLE, NULL);
}

apply_handle_delete()

static void apply_handle_delete ( StringInfo  s )

static

Definition at line 800 of file worker.c.

References AfterTriggerEndQuery(), Assert, check_relation_updatable(), CommandCounterIncrement(), create_estate_for_relation(), DEBUG1, elog, ensure_transaction(), EState::es_result_relation_info, EState::es_tupleTable, EvalPlanQualEnd(), EvalPlanQualInit(), EvalPlanQualSetSlot, ExecCloseIndices(), ExecInitExtraTupleSlot(), ExecOpenIndices(), ExecResetTupleTable(), ExecSimpleRelationDelete(), FreeExecutorState(), GetPerTupleMemoryContext, GetRelationIdentityOrPK(), GetTransactionSnapshot(), LogicalRepRelMapEntry::localrel, LockTupleExclusive, logicalrep_read_delete(), logicalrep_rel_close(), logicalrep_rel_open(), MemoryContextSwitchTo(), NIL, NoLock, OidIsValid, PopActiveSnapshot(), PushActiveSnapshot(), RelationFindReplTupleByIndex(), RelationFindReplTupleSeq(), RelationGetDescr, RelationGetRelationName, LogicalRepRelMapEntry::remoterel, LogicalRepRelation::replident, RowExclusiveLock, should_apply_changes_for_rel(), slot_store_cstrings(), and LogicalRepTupleData::values.

Referenced by apply_dispatch().

{
    LogicalRepRelMapEntry *rel;
    LogicalRepTupleData oldtup;
    LogicalRepRelId relid;
    Oid         idxoid;
    EState     *estate;
    EPQState    epqstate;
    TupleTableSlot *remoteslot;
    TupleTableSlot *localslot;
    bool        found;
    MemoryContext oldctx;

    ensure_transaction();

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

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

    /* Initialize the executor state. */
    estate = create_estate_for_relation(rel);
    remoteslot = ExecInitExtraTupleSlot(estate,
                                        RelationGetDescr(rel->localrel));
    localslot = ExecInitExtraTupleSlot(estate,
                                       RelationGetDescr(rel->localrel));
    EvalPlanQualInit(&epqstate, estate, NULL, NIL, -1);

    PushActiveSnapshot(GetTransactionSnapshot());
    ExecOpenIndices(estate->es_result_relation_info, false);

    /* Find the tuple using the replica identity index. */
    oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
    slot_store_cstrings(remoteslot, rel, oldtup.values);
    MemoryContextSwitchTo(oldctx);

    /*
     * Try to find tuple using either replica identity index, primary key or
     * if needed, sequential scan.
     */
    idxoid = GetRelationIdentityOrPK(rel->localrel);
    Assert(OidIsValid(idxoid) ||
           (rel->remoterel.replident == REPLICA_IDENTITY_FULL));

    if (OidIsValid(idxoid))
        found = RelationFindReplTupleByIndex(rel->localrel, idxoid,
                                             LockTupleExclusive,
                                             remoteslot, localslot);
    else
        found = RelationFindReplTupleSeq(rel->localrel, LockTupleExclusive,
                                         remoteslot, localslot);
    /* If found delete it. */
    if (found)
    {
        EvalPlanQualSetSlot(&epqstate, localslot);

        /* Do the actual delete. */
        ExecSimpleRelationDelete(estate, &epqstate, localslot);
    }
    else
    {
        /* The tuple to be deleted could not be found. */
        elog(DEBUG1,
             "logical replication could not find row for delete "
             "in replication target relation \"%s\"",
             RelationGetRelationName(rel->localrel));
    }

    /* Cleanup. */
    ExecCloseIndices(estate->es_result_relation_info);
    PopActiveSnapshot();

    /* Handle queued AFTER triggers. */
    AfterTriggerEndQuery(estate);

    EvalPlanQualEnd(&epqstate);
    ExecResetTupleTable(estate->es_tupleTable, false);
    FreeExecutorState(estate);

    logicalrep_rel_close(rel, NoLock);

    CommandCounterIncrement();
}

apply_handle_insert()

static void apply_handle_insert ( StringInfo  s )

static

Definition at line 585 of file worker.c.

References AfterTriggerEndQuery(), CommandCounterIncrement(), create_estate_for_relation(), ensure_transaction(), EState::es_result_relation_info, EState::es_tupleTable, ExecCloseIndices(), ExecInitExtraTupleSlot(), ExecOpenIndices(), ExecResetTupleTable(), ExecSimpleRelationInsert(), FreeExecutorState(), GetPerTupleMemoryContext, GetTransactionSnapshot(), LogicalRepRelMapEntry::localrel, logicalrep_read_insert(), logicalrep_rel_close(), logicalrep_rel_open(), MemoryContextSwitchTo(), NoLock, PopActiveSnapshot(), PushActiveSnapshot(), RelationGetDescr, RowExclusiveLock, should_apply_changes_for_rel(), slot_fill_defaults(), slot_store_cstrings(), and LogicalRepTupleData::values.

Referenced by apply_dispatch().

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

    ensure_transaction();

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

    /* Initialize the executor state. */
    estate = create_estate_for_relation(rel);
    remoteslot = ExecInitExtraTupleSlot(estate,
                                        RelationGetDescr(rel->localrel));

    /* Process and store remote tuple in the slot */
    oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
    slot_store_cstrings(remoteslot, rel, newtup.values);
    slot_fill_defaults(rel, estate, remoteslot);
    MemoryContextSwitchTo(oldctx);

    PushActiveSnapshot(GetTransactionSnapshot());
    ExecOpenIndices(estate->es_result_relation_info, false);

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

    /* Cleanup. */
    ExecCloseIndices(estate->es_result_relation_info);
    PopActiveSnapshot();

    /* Handle queued AFTER triggers. */
    AfterTriggerEndQuery(estate);

    ExecResetTupleTable(estate->es_tupleTable, false);
    FreeExecutorState(estate);

    logicalrep_rel_close(rel, NoLock);

    CommandCounterIncrement();
}

apply_handle_origin()

static void apply_handle_origin ( StringInfo  s )

static

Definition at line 518 of file worker.c.

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

Referenced by apply_dispatch().

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

apply_handle_relation()

static void apply_handle_relation ( StringInfo  s )

static

Definition at line 540 of file worker.c.

References logicalrep_read_rel(), and logicalrep_relmap_update().

Referenced by apply_dispatch().

{
    LogicalRepRelation *rel;

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

apply_handle_truncate()

static void apply_handle_truncate ( StringInfo  s )

static

Definition at line 900 of file worker.c.

References CommandCounterIncrement(), DROP_RESTRICT, ensure_transaction(), ExecuteTruncateGuts(), lappend(), lappend_oid(), lfirst, lfirst_oid, LogicalRepRelMapEntry::localrel, LogicalRepRelMapEntry::localreloid, logicalrep_read_truncate(), logicalrep_rel_close(), logicalrep_rel_open(), NIL, NoLock, RelationIsLogicallyLogged, RowExclusiveLock, and should_apply_changes_for_rel().

Referenced by apply_dispatch().

{
    bool        cascade = false;
    bool        restart_seqs = false;
    List       *remote_relids = NIL;
    List       *remote_rels = NIL;
    List       *rels = NIL;
    List       *relids = NIL;
    List       *relids_logged = NIL;
    ListCell   *lc;

    ensure_transaction();

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

    /*
     * Even if we used CASCADE on the upstream master 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);
    }

    CommandCounterIncrement();
}

apply_handle_type()

static void apply_handle_type ( StringInfo  s )

static

Definition at line 555 of file worker.c.

References logicalrep_read_typ(), and logicalrep_typmap_update().

Referenced by apply_dispatch().

{
    LogicalRepTyp typ;

    logicalrep_read_typ(s, &typ);
    logicalrep_typmap_update(&typ);
}

apply_handle_update()

static void apply_handle_update ( StringInfo  s )

static

Definition at line 679 of file worker.c.

References AfterTriggerEndQuery(), Assert, LogicalRepTupleData::changed, check_relation_updatable(), CommandCounterIncrement(), create_estate_for_relation(), DEBUG1, elog, ensure_transaction(), EState::es_result_relation_info, EState::es_tupleTable, EvalPlanQualEnd(), EvalPlanQualInit(), EvalPlanQualSetSlot, ExecClearTuple(), ExecCloseIndices(), ExecInitExtraTupleSlot(), ExecOpenIndices(), ExecResetTupleTable(), ExecSimpleRelationUpdate(), ExecStoreTuple(), FreeExecutorState(), GetPerTupleMemoryContext, GetRelationIdentityOrPK(), GetTransactionSnapshot(), InvalidBuffer, LogicalRepRelMapEntry::localrel, LockTupleExclusive, logicalrep_read_update(), logicalrep_rel_close(), logicalrep_rel_open(), MemoryContextSwitchTo(), NIL, NoLock, OidIsValid, PopActiveSnapshot(), PushActiveSnapshot(), RelationFindReplTupleByIndex(), RelationFindReplTupleSeq(), RelationGetDescr, RelationGetRelationName, LogicalRepRelMapEntry::remoterel, LogicalRepRelation::replident, RowExclusiveLock, should_apply_changes_for_rel(), slot_modify_cstrings(), slot_store_cstrings(), TupleTableSlot::tts_tuple, and LogicalRepTupleData::values.

Referenced by apply_dispatch().

{
    LogicalRepRelMapEntry *rel;
    LogicalRepRelId relid;
    Oid         idxoid;
    EState     *estate;
    EPQState    epqstate;
    LogicalRepTupleData oldtup;
    LogicalRepTupleData newtup;
    bool        has_oldtup;
    TupleTableSlot *localslot;
    TupleTableSlot *remoteslot;
    bool        found;
    MemoryContext oldctx;

    ensure_transaction();

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

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

    /* Initialize the executor state. */
    estate = create_estate_for_relation(rel);
    remoteslot = ExecInitExtraTupleSlot(estate,
                                        RelationGetDescr(rel->localrel));
    localslot = ExecInitExtraTupleSlot(estate,
                                       RelationGetDescr(rel->localrel));
    EvalPlanQualInit(&epqstate, estate, NULL, NIL, -1);

    PushActiveSnapshot(GetTransactionSnapshot());
    ExecOpenIndices(estate->es_result_relation_info, false);

    /* Build the search tuple. */
    oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
    slot_store_cstrings(remoteslot, rel,
                        has_oldtup ? oldtup.values : newtup.values);
    MemoryContextSwitchTo(oldctx);

    /*
     * Try to find tuple using either replica identity index, primary key or
     * if needed, sequential scan.
     */
    idxoid = GetRelationIdentityOrPK(rel->localrel);
    Assert(OidIsValid(idxoid) ||
           (rel->remoterel.replident == REPLICA_IDENTITY_FULL && has_oldtup));

    if (OidIsValid(idxoid))
        found = RelationFindReplTupleByIndex(rel->localrel, idxoid,
                                             LockTupleExclusive,
                                             remoteslot, localslot);
    else
        found = RelationFindReplTupleSeq(rel->localrel, LockTupleExclusive,
                                         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));
        ExecStoreTuple(localslot->tts_tuple, remoteslot, InvalidBuffer, false);
        slot_modify_cstrings(remoteslot, rel, newtup.values, newtup.changed);
        MemoryContextSwitchTo(oldctx);

        EvalPlanQualSetSlot(&epqstate, remoteslot);

        /* Do the actual update. */
        ExecSimpleRelationUpdate(estate, &epqstate, localslot, remoteslot);
    }
    else
    {
        /*
         * The tuple to be updated could not be found.
         *
         * TODO what to do here, change the log level to LOG perhaps?
         */
        elog(DEBUG1,
             "logical replication did not find row for update "
             "in replication target relation \"%s\"",
             RelationGetRelationName(rel->localrel));
    }

    /* Cleanup. */
    ExecCloseIndices(estate->es_result_relation_info);
    PopActiveSnapshot();

    /* Handle queued AFTER triggers. */
    AfterTriggerEndQuery(estate);

    EvalPlanQualEnd(&epqstate);
    ExecResetTupleTable(estate->es_tupleTable, false);
    FreeExecutorState(estate);

    logicalrep_rel_close(rel, NoLock);

    CommandCounterIncrement();
}

ApplyWorkerMain()

void ApplyWorkerMain

(

Datum

main_arg

)

Definition at line 1581 of file worker.c.

References ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate, am_tablesync_worker(), Assert, BackgroundWorkerInitializeConnectionByOid(), BackgroundWorkerUnblockSignals(), CacheRegisterSyscacheCallback(), CommitTransactionCommand(), Subscription::conninfo, CurrentResourceOwner, DatumGetInt32, LogicalRepWorker::dbid, DEBUG1, die(), elog, Subscription::enabled, ereport, 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_VERSION_NUM, logicalrep_worker_attach(), logicalrep_worker_sighup(), LogicalRepApplyLoop(), LogicalRepSyncTableStart(), MemoryContextSwitchTo(), MyLogicalRepWorker, MySubscriptionValid, Subscription::name, NAMEDATALEN, Subscription::oid, OidIsValid, options, pfree(), PGC_BACKEND, PGC_S_OVERRIDE, PGC_SUSET, pqsignal(), proc_exit(), WalRcvStreamOptions::proto, pstrdup(), Subscription::publications, LogicalRepWorker::relid, replorigin_by_name(), replorigin_create(), replorigin_session_get_progress(), replorigin_session_origin, replorigin_session_setup(), LogicalRepWorker::reply_time, ResourceOwnerCreate(), server_version, SetConfigOption(), SIGHUP, Subscription::slotname, WalRcvStreamOptions::slotname, snprintf(), WalRcvStreamOptions::startpoint, StartTransactionCommand(), LogicalRepWorker::subid, subscription_change_cb(), SUBSCRIPTIONOID, SUBSCRIPTIONRELMAP, Subscription::synccommit, TopMemoryContext, LogicalRepWorker::userid, walrcv_connect, walrcv_identify_system, and walrcv_startstreaming.

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

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

    /* Setup signal handling */
    pqsignal(SIGHUP, logicalrep_worker_sighup);
    pqsignal(SIGTERM, die);
    BackgroundWorkerUnblockSignals();

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

    Assert(CurrentResourceOwner == NULL);
    CurrentResourceOwner = ResourceOwnerCreate(NULL,
                                               "logical replication apply");

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

    /* 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 synchroniation worker, call initial sync. */
        syncslotname = LogicalRepSyncTableStart(&origin_startpos);

        /* The slot name needs to be allocated in permanent memory context. */
        oldctx = MemoryContextSwitchTo(ApplyContext);
        myslotname = pstrdup(syncslotname);
        MemoryContextSwitchTo(oldctx);

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

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

        wrconn = walrcv_connect(MySubscription->conninfo, true, MySubscription->name,
                                &err);
        if (wrconn == NULL)
            ereport(ERROR,
                    (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(wrconn, &startpointTLI,
                                      &server_version);

    }

    /*
     * 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;
    options.proto.logical.proto_version = LOGICALREP_PROTO_VERSION_NUM;
    options.proto.logical.publication_names = MySubscription->publications;

    /* Start normal logical streaming replication. */
    walrcv_startstreaming(wrconn, &options);

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

    proc_exit(0);
}

check_relation_updatable()

static void check_relation_updatable ( LogicalRepRelMapEntry *  rel )

static

Definition at line 645 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)));
}

create_estate_for_relation()

static EState* create_estate_for_relation ( LogicalRepRelMapEntry *  rel )

static

Definition at line 190 of file worker.c.

References AfterTriggerBeginQuery(), CreateExecutorState(), EState::es_num_result_relations, EState::es_output_cid, EState::es_range_table, EState::es_result_relation_info, EState::es_result_relations, EState::es_trig_tuple_slot, ExecInitExtraTupleSlot(), GetCurrentCommandId(), InitResultRelInfo(), list_make1, LogicalRepRelMapEntry::localrel, makeNode, RelationData::rd_rel, RelationGetRelid, RangeTblEntry::relid, RangeTblEntry::relkind, ResultRelInfo::ri_TrigDesc, RTE_RELATION, and RangeTblEntry::rtekind.

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

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

    estate = CreateExecutorState();

    rte = makeNode(RangeTblEntry);
    rte->rtekind = RTE_RELATION;
    rte->relid = RelationGetRelid(rel->localrel);
    rte->relkind = rel->localrel->rd_rel->relkind;
    estate->es_range_table = list_make1(rte);

    resultRelInfo = makeNode(ResultRelInfo);
    InitResultRelInfo(resultRelInfo, rel->localrel, 1, NULL, 0);

    estate->es_result_relations = resultRelInfo;
    estate->es_num_result_relations = 1;
    estate->es_result_relation_info = resultRelInfo;

    estate->es_output_cid = GetCurrentCommandId(true);

    /* Triggers might need a slot */
    if (resultRelInfo->ri_TrigDesc)
        estate->es_trig_tuple_slot = ExecInitExtraTupleSlot(estate, NULL);

    /* Prepare to catch AFTER triggers. */
    AfterTriggerBeginQuery();

    return estate;
}

ensure_transaction()

static bool ensure_transaction ( void  )

static

Definition at line 161 of file worker.c.

References CurrentMemoryContext, IsTransactionState(), maybe_reread_subscription(), MemoryContextSwitchTo(), SetCurrentStatementStartTimestamp(), and StartTransactionCommand().

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

{
    if (IsTransactionState())
    {
        SetCurrentStatementStartTimestamp();

        if (CurrentMemoryContext != ApplyMessageContext)
            MemoryContextSwitchTo(ApplyMessageContext);

        return false;
    }

    SetCurrentStatementStartTimestamp();
    StartTransactionCommand();

    maybe_reread_subscription();

    MemoryContextSwitchTo(ApplyMessageContext);
    return true;
}

get_flush_position()

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

static

Definition at line 1023 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 569 of file worker.c.

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

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

{
    Oid         idxoid;

    idxoid = RelationGetReplicaIndex(rel);

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

    return idxoid;
}

IsLogicalWorker()

bool IsLogicalWorker

(

void

)

Definition at line 1759 of file worker.c.

References MyLogicalRepWorker.

Referenced by ProcessInterrupts().

{
    return MyLogicalRepWorker != NULL;
}

logicalrep_worker_sighup()

static void logicalrep_worker_sighup ( SIGNAL_ARGS  )

static

Definition at line 1567 of file worker.c.

References got_SIGHUP, MyLatch, and SetLatch().

Referenced by ApplyWorkerMain().

{
    int         save_errno = errno;

    got_SIGHUP = true;

    /* Waken anything waiting on the process latch */
    SetLatch(MyLatch);

    errno = save_errno;
}

LogicalRepApplyLoop()

static void LogicalRepApplyLoop ( XLogRecPtr  last_received )

static

Definition at line 1102 of file worker.c.

References AcceptInvalidationMessages(), ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate, apply_dispatch(), buf, CHECK_FOR_INTERRUPTS, StringInfoData::cursor, StringInfoData::data, dlist_is_empty(), ereport, errmsg(), ERROR, fd(), GetCurrentTimestamp(), got_SIGHUP, in_remote_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(), proc_exit(), 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_LATCH_SET, WL_POSTMASTER_DEATH, WL_SOCKET_READABLE, and WL_TIMEOUT.

Referenced by ApplyWorkerMain().

{
    /*
     * Init the ApplyMessageContext which we clean up after each replication
     * protocol message.
     */
    ApplyMessageContext = AllocSetContextCreate(ApplyContext,
                                                "ApplyMessageContext",
                                                ALLOCSET_DEFAULT_SIZES);

    /* mark as idle, before starting to loop */
    pgstat_report_activity(STATE_IDLE, NULL);

    for (;;)
    {
        pgsocket    fd = PGINVALID_SOCKET;
        int         rc;
        int         len;
        char       *buf = NULL;
        bool        endofstream = false;
        TimestampTz last_recv_timestamp = GetCurrentTimestamp();
        bool        ping_sent = false;
        long        wait_time;

        CHECK_FOR_INTERRUPTS();

        MemoryContextSwitchTo(ApplyMessageContext);

        len = walrcv_receive(wrconn, &buf, &fd);

        if (len != 0)
        {
            /* Process the data */
            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(wrconn, &buf, &fd);
            }
        }

        /* confirm all writes so far */
        send_feedback(last_received, false, false);

        if (!in_remote_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)
        {
            TimeLineID  tli;

            walrcv_endstreaming(wrconn, &tli);
            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_POSTMASTER_DEATH,
                               fd, wait_time,
                               WAIT_EVENT_LOGICAL_APPLY_MAIN);

        /* Emergency bailout if postmaster has died */
        if (rc & WL_POSTMASTER_DEATH)
            proc_exit(1);

        if (rc & WL_LATCH_SET)
        {
            ResetLatch(MyLatch);
            CHECK_FOR_INTERRUPTS();
        }

        if (got_SIGHUP)
        {
            got_SIGHUP = 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 master anyway, to report any
             * progress in applying WAL.
             */
            bool        requestReply = false;

            /*
             * Check if time since last receive from standby 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,
                            (errmsg("terminating logical replication worker due to timeout")));

                /*
                 * We didn't receive anything new, for half of receiver
                 * replication timeout. Ping the server.
                 */
                if (!ping_sent)
                {
                    timeout = TimestampTzPlusMilliseconds(last_recv_timestamp,
                                                          (wal_receiver_timeout / 2));
                    if (now >= timeout)
                    {
                        requestReply = true;
                        ping_sent = true;
                    }
                }
            }

            send_feedback(last_received, requestReply, requestReply);
        }
    }
}

maybe_reread_subscription()

static void maybe_reread_subscription ( void  )

static

Definition at line 1424 of file worker.c.

References Assert, 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(), LogicalRepWorker::subid, and Subscription::synccommit.

Referenced by apply_handle_commit(), ensure_transaction(), 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);
    }

    /*
     * Exit if connection string was changed. The launcher will start new
     * worker.
     */
    if (strcmp(newsub->conninfo, MySubscription->conninfo) != 0)
    {
        ereport(LOG,
                (errmsg("logical replication apply worker for subscription \"%s\" will "
                        "restart because the connection information was changed",
                        MySubscription->name)));

        proc_exit(0);
    }

    /*
     * Exit if subscription name was changed (it's used for
     * fallback_application_name). The launcher will start new worker.
     */
    if (strcmp(newsub->name, MySubscription->name) != 0)
    {
        ereport(LOG,
                (errmsg("logical replication apply worker for subscription \"%s\" will "
                        "restart because subscription was renamed",
                        MySubscription->name)));

        proc_exit(0);
    }

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

    /*
     * We need to make new connection to new slot if slot name has changed so
     * exit here as well if that's the case.
     */
    if (strcmp(newsub->slotname, MySubscription->slotname) != 0)
    {
        ereport(LOG,
                (errmsg("logical replication apply worker for subscription \"%s\" will "
                        "restart because the replication slot name was changed",
                        MySubscription->name)));

        proc_exit(0);
    }

    /*
     * Exit if publication list was changed. The launcher will start new
     * worker.
     */
    if (!equal(newsub->publications, MySubscription->publications))
    {
        ereport(LOG,
                (errmsg("logical replication apply worker for subscription \"%s\" will "
                        "restart because subscription's publications were changed",
                        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 1335 of file worker.c.

References StringInfoData::data, DEBUG2, elog, get_flush_position(), GetCurrentTimestamp(), InvalidXLogRecPtr, StringInfoData::len, 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,
         (uint32) (recvpos >> 32), (uint32) recvpos,
         (uint32) (writepos >> 32), (uint32) writepos,
         (uint32) (flushpos >> 32), (uint32) flushpos
        );

    walrcv_send(wrconn, 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 145 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 231 of file worker.c.

References attnum, LogicalRepRelMapEntry::attrmap, build_column_default(), ExecEvalExpr(), ExecInitExpr(), expression_planner(), GetPerTupleExprContext, i, LogicalRepRelMapEntry::localrel, LogicalRepRelation::natts, tupleDesc::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 *));

    for (attnum = 0; attnum < num_phys_attrs; attnum++)
    {
        Expr       *defexpr;

        if (TupleDescAttr(desc, attnum)->attisdropped)
            continue;

        if (rel->attrmap[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_cstrings()

static void slot_modify_cstrings ( TupleTableSlot *  slot, LogicalRepRelMapEntry *  rel, char **  values, bool *  replaces  )

static

Definition at line 389 of file worker.c.

References ErrorContextCallback::arg, LogicalRepRelMapEntry::attrmap, ErrorContextCallback::callback, error_context_stack, ExecClearTuple(), ExecStoreVirtualTuple(), getTypeInputInfo(), i, SlotErrCallbackArg::local_attnum, tupleDesc::natts, OidInputFunctionCall(), 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_update().

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

    slot_getallattrs(slot);
    ExecClearTuple(slot);

    /* Push callback + info on the error context stack */
    errarg.rel = rel;
    errarg.local_attnum = -1;
    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 */
    for (i = 0; i < natts; i++)
    {
        Form_pg_attribute att = TupleDescAttr(slot->tts_tupleDescriptor, i);
        int         remoteattnum = rel->attrmap[i];

        if (remoteattnum < 0)
            continue;

        if (!replaces[remoteattnum])
            continue;

        if (values[remoteattnum] != NULL)
        {
            Oid         typinput;
            Oid         typioparam;

            errarg.local_attnum = i;
            errarg.remote_attnum = remoteattnum;

            getTypeInputInfo(att->atttypid, &typinput, &typioparam);
            slot->tts_values[i] =
                OidInputFunctionCall(typinput, values[remoteattnum],
                                     typioparam, att->atttypmod);
            slot->tts_isnull[i] = false;

            errarg.local_attnum = -1;
            errarg.remote_attnum = -1;
        }
        else
        {
            slot->tts_values[i] = (Datum) 0;
            slot->tts_isnull[i] = true;
        }
    }

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

    ExecStoreVirtualTuple(slot);
}

slot_store_cstrings()

static void slot_store_cstrings ( TupleTableSlot *  slot, LogicalRepRelMapEntry *  rel, char **  values  )

static

Definition at line 321 of file worker.c.

References ErrorContextCallback::arg, LogicalRepRelMapEntry::attrmap, ErrorContextCallback::callback, error_context_stack, ExecClearTuple(), ExecStoreVirtualTuple(), getTypeInputInfo(), i, SlotErrCallbackArg::local_attnum, tupleDesc::natts, OidInputFunctionCall(), 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.local_attnum = -1;
    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 attribute */
    for (i = 0; i < natts; i++)
    {
        Form_pg_attribute att = TupleDescAttr(slot->tts_tupleDescriptor, i);
        int         remoteattnum = rel->attrmap[i];

        if (!att->attisdropped && remoteattnum >= 0 &&
            values[remoteattnum] != NULL)
        {
            Oid         typinput;
            Oid         typioparam;

            errarg.local_attnum = i;
            errarg.remote_attnum = remoteattnum;

            getTypeInputInfo(att->atttypid, &typinput, &typioparam);
            slot->tts_values[i] =
                OidInputFunctionCall(typinput, values[remoteattnum],
                                     typioparam, att->atttypmod);
            slot->tts_isnull[i] = false;

            errarg.local_attnum = -1;
            errarg.remote_attnum = -1;
        }
        else
        {
            /*
             * We assign NULL to dropped attributes, NULL values, 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);
}

slot_store_error_callback()

static void slot_store_error_callback ( void *  arg )

static

Definition at line 286 of file worker.c.

References LogicalRepRelation::attnames, LogicalRepRelation::atttyps, errcontext, format_type_be(), get_atttype(), SlotErrCallbackArg::local_attnum, LogicalRepRelMapEntry::localreloid, logicalrep_typmap_gettypname(), LogicalRepRelation::nspname, SlotErrCallbackArg::rel, LogicalRepRelation::relname, SlotErrCallbackArg::remote_attnum, and LogicalRepRelMapEntry::remoterel.

Referenced by slot_modify_cstrings(), and slot_store_cstrings().

{
    SlotErrCallbackArg *errarg = (SlotErrCallbackArg *) arg;
    LogicalRepRelMapEntry *rel;
    char       *remotetypname;
    Oid         remotetypoid,
                localtypoid;

    /* Nothing to do if remote attribute number is not set */
    if (errarg->remote_attnum < 0)
        return;

    rel = errarg->rel;
    remotetypoid = rel->remoterel.atttyps[errarg->remote_attnum];

    /* Fetch remote type name from the LogicalRepTypMap cache */
    remotetypname = logicalrep_typmap_gettypname(remotetypoid);

    /* Fetch local type OID from the local sys cache */
    localtypoid = get_atttype(rel->localreloid, errarg->local_attnum + 1);

    errcontext("processing remote data for replication target relation \"%s.%s\" column \"%s\", "
               "remote type %s, local type %s",
               rel->remoterel.nspname, rel->remoterel.relname,
               rel->remoterel.attnames[errarg->remote_attnum],
               remotetypname,
               format_type_be(localtypoid));
}

store_flush_position()

static void store_flush_position ( XLogRecPtr  remote_lsn )

static

Definition at line 1067 of file worker.c.

References dlist_push_tail(), FlushPosition::local_end, MemoryContextSwitchTo(), FlushPosition::node, palloc(), FlushPosition::remote_end, and XactLastCommitEnd.

Referenced by apply_handle_commit().

{
    FlushPosition *flushpos;

    /* Need to do this in permanent context */
    MemoryContextSwitchTo(ApplyContext);

    /* Track commit lsn  */
    flushpos = (FlushPosition *) palloc(sizeof(FlushPosition));
    flushpos->local_end = XactLastCommitEnd;
    flushpos->remote_end = remote_lsn;

    dlist_push_tail(&lsn_mapping, &flushpos->node);
    MemoryContextSwitchTo(ApplyMessageContext);
}

subscription_change_cb()

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

static

Definition at line 1560 of file worker.c.

References MySubscriptionValid.

Referenced by ApplyWorkerMain().

{
    MySubscriptionValid = false;
}

UpdateWorkerStats()

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

static

Definition at line 1086 of file worker.c.

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

Referenced by LogicalRepApplyLoop().

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

Variable Documentation

ApplyContext

MemoryContext ApplyContext = NULL

Definition at line 112 of file worker.c.

ApplyMessageContext

MemoryContext ApplyMessageContext = NULL

static

Definition at line 111 of file worker.c.

got_SIGHUP

volatile sig_atomic_t got_SIGHUP = false

static

Definition at line 129 of file worker.c.

Referenced by logicalrep_worker_sighup(), and LogicalRepApplyLoop().

in_remote_transaction

bool in_remote_transaction = false

Definition at line 119 of file worker.c.

Referenced by apply_handle_begin(), apply_handle_commit(), apply_handle_origin(), and LogicalRepApplyLoop().

lsn_mapping

dlist_head lsn_mapping = DLIST_STATIC_INIT(lsn_mapping)

static

Definition at line 102 of file worker.c.

MySubscription

Subscription* MySubscription = NULL

Definition at line 116 of file worker.c.

Referenced by logicalrep_rel_open(), LogicalRepSyncTableStart(), pg_attribute_noreturn(), and process_syncing_tables_for_apply().

MySubscriptionValid

bool MySubscriptionValid = false

Definition at line 117 of file worker.c.

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

remote_final_lsn

XLogRecPtr remote_final_lsn = InvalidXLogRecPtr

static

Definition at line 120 of file worker.c.

Referenced by apply_handle_begin(), apply_handle_commit(), and should_apply_changes_for_rel().

wrconn

WalReceiverConn* wrconn = NULL

Definition at line 114 of file worker.c.

Referenced by AlterSubscription_refresh(), copy_read_data(), copy_table(), CreateSubscription(), DropSubscription(), fetch_remote_table_info(), logicalrep_worker_onexit(), LogicalRepSyncTableStart(), and process_syncing_tables_for_sync().