sequencesync.c File Reference

#include "postgres.h"
#include "access/genam.h"
#include "access/table.h"
#include "catalog/pg_sequence.h"
#include "catalog/pg_subscription_rel.h"
#include "commands/sequence.h"
#include "pgstat.h"
#include "postmaster/interrupt.h"
#include "replication/logicalworker.h"
#include "replication/worker_internal.h"
#include "storage/lwlock.h"
#include "utils/acl.h"
#include "utils/builtins.h"
#include "utils/fmgroids.h"
#include "utils/guc.h"
#include "utils/inval.h"
#include "utils/lsyscache.h"
#include "utils/memutils.h"
#include "utils/pg_lsn.h"
#include "utils/syscache.h"
#include "utils/usercontext.h"

Include dependency graph for sequencesync.c:

Go to the source code of this file.

Macros
#define	REMOTE_SEQ_COL_COUNT   11
#define	MAX_SEQUENCES_SYNC_PER_BATCH   100

Typedefs
typedef enum CopySeqResult	CopySeqResult

Enumerations
enum	CopySeqResult {   COPYSEQ_SUCCESS , COPYSEQ_MISMATCH , COPYSEQ_SUBSCRIBER_INSUFFICIENT_PERM , COPYSEQ_PUBLISHER_INSUFFICIENT_PERM ,   COPYSEQ_SKIPPED }

Functions
void	ProcessSequencesForSync (void)
static void	get_sequences_string (List *seqindexes, StringInfo buf)
static void	report_sequence_errors (List *mismatched_seqs_idx, List *sub_insuffperm_seqs_idx, List *pub_insuffperm_seqs_idx, List *missing_seqs_idx)
static CopySeqResult	get_and_validate_seq_info (TupleTableSlot *slot, Relation *sequence_rel, LogicalRepSequenceInfo **seqinfo, int *seqidx)
static CopySeqResult	copy_sequence (LogicalRepSequenceInfo *seqinfo, Oid seqowner)
static void	copy_sequences (WalReceiverConn *conn)
static void	LogicalRepSyncSequences (void)
static void	start_sequence_sync (void)
void	SequenceSyncWorkerMain (Datum main_arg)

Variables
static List *	seqinfos = NIL

Macro Definition Documentation

MAX_SEQUENCES_SYNC_PER_BATCH

#define MAX_SEQUENCES_SYNC_PER_BATCH   100

REMOTE_SEQ_COL_COUNT

#define REMOTE_SEQ_COL_COUNT   11

Definition at line 75 of file sequencesync.c.

Typedef Documentation

CopySeqResult

typedef enum CopySeqResult CopySeqResult

Enumeration Type Documentation

CopySeqResult

enum CopySeqResult

Enumerator
COPYSEQ_SUCCESS
COPYSEQ_MISMATCH
COPYSEQ_SUBSCRIBER_INSUFFICIENT_PERM
COPYSEQ_PUBLISHER_INSUFFICIENT_PERM
COPYSEQ_SKIPPED

Definition at line 77 of file sequencesync.c.

{
    COPYSEQ_SUCCESS,
    COPYSEQ_MISMATCH,
    COPYSEQ_SUBSCRIBER_INSUFFICIENT_PERM,
    COPYSEQ_PUBLISHER_INSUFFICIENT_PERM,
    COPYSEQ_SKIPPED
} CopySeqResult;

Function Documentation

copy_sequence()

static CopySeqResult copy_sequence ( LogicalRepSequenceInfo *  seqinfo, Oid  seqowner  )

static

Definition at line 375 of file sequencesync.c.

{
    UserContext ucxt;
    AclResult   aclresult;
    bool        run_as_owner = MySubscription->runasowner;
    Oid         seqoid = seqinfo->localrelid;
 
    /*
     * If the user did not opt to run as the owner of the subscription
     * ('run_as_owner'), then copy the sequence as the owner of the sequence.
     */
    if (!run_as_owner)
        SwitchToUntrustedUser(seqowner, &ucxt);
 
    aclresult = pg_class_aclcheck(seqoid, GetUserId(), ACL_UPDATE);
 
    if (aclresult != ACLCHECK_OK)
    {
        if (!run_as_owner)
            RestoreUserContext(&ucxt);
 
        return COPYSEQ_SUBSCRIBER_INSUFFICIENT_PERM;
    }
 
    /*
     * The log counter (log_cnt) tracks how many sequence values are still
     * unused locally. It is only relevant to the local node and managed
     * internally by nextval() when allocating new ranges. Since log_cnt does
     * not affect the visible sequence state (like last_value or is_called)
     * and is only used for local caching, it need not be copied to the
     * subscriber during synchronization.
     */
    SetSequence(seqoid, seqinfo->last_value, seqinfo->is_called);
 
    if (!run_as_owner)
        RestoreUserContext(&ucxt);
 
    /*
     * Record the remote sequence's LSN in pg_subscription_rel and mark the
     * sequence as READY.
     */
    UpdateSubscriptionRelState(MySubscription->oid, seqoid, SUBREL_STATE_READY,
                               seqinfo->page_lsn, false);
 
    return COPYSEQ_SUCCESS;
}

References ACL_UPDATE, ACLCHECK_OK, COPYSEQ_SUBSCRIBER_INSUFFICIENT_PERM, COPYSEQ_SUCCESS, fb(), GetUserId(), MySubscription, Subscription::oid, pg_class_aclcheck(), RestoreUserContext(), Subscription::runasowner, SetSequence(), SwitchToUntrustedUser(), and UpdateSubscriptionRelState().

Referenced by copy_sequences().

copy_sequences()

static void copy_sequences ( WalReceiverConn *  conn )

static

Definition at line 426 of file sequencesync.c.

{
    int         cur_batch_base_index = 0;
    int         n_seqinfos = list_length(seqinfos);
    List       *mismatched_seqs_idx = NIL;
    List       *missing_seqs_idx = NIL;
    List       *sub_insuffperm_seqs_idx = NIL;
    List       *pub_insuffperm_seqs_idx = NIL;
    StringInfoData seqstr;
    StringInfoData cmd;
    MemoryContext oldctx;
 
    initStringInfo(&seqstr);
    initStringInfo(&cmd);
 
#define MAX_SEQUENCES_SYNC_PER_BATCH 100
 
    elog(DEBUG1,
         "logical replication sequence synchronization for subscription \"%s\" - total unsynchronized: %d",
         MySubscription->name, n_seqinfos);
 
    while (cur_batch_base_index < n_seqinfos)
    {
        Oid         seqRow[REMOTE_SEQ_COL_COUNT] = {INT8OID, BOOLOID, INT8OID,
        BOOLOID, LSNOID, OIDOID, INT8OID, INT8OID, INT8OID, INT8OID, BOOLOID};
        int         batch_size = 0;
        int         batch_succeeded_count = 0;
        int         batch_mismatched_count = 0;
        int         batch_skipped_count = 0;
        int         batch_sub_insuffperm_count = 0;
        int         batch_pub_insuffperm_count = 0;
        int         batch_missing_count;
 
        WalRcvExecResult *res;
        TupleTableSlot *slot;
 
        StartTransactionCommand();
 
        for (int idx = cur_batch_base_index; idx < n_seqinfos; idx++)
        {
            char       *nspname_literal;
            char       *seqname_literal;
 
            LogicalRepSequenceInfo *seqinfo =
                (LogicalRepSequenceInfo *) list_nth(seqinfos, idx);
 
            if (seqstr.len > 0)
                appendStringInfoString(&seqstr, ", ");
 
            nspname_literal = quote_literal_cstr(seqinfo->nspname);
            seqname_literal = quote_literal_cstr(seqinfo->seqname);
 
            appendStringInfo(&seqstr, "(%s, %s, %d)",
                             nspname_literal, seqname_literal, idx);
 
            if (++batch_size == MAX_SEQUENCES_SYNC_PER_BATCH)
                break;
        }
 
        /*
         * We deliberately avoid acquiring a local lock on the sequence before
         * querying the publisher to prevent potential distributed deadlocks
         * in bi-directional replication setups.
         *
         * Example scenario:
         *
         * - On each node, a background worker acquires a lock on a sequence
         * as part of a sync operation.
         *
         * - Concurrently, a user transaction attempts to alter the same
         * sequence, waiting on the background worker's lock.
         *
         * - Meanwhile, a query from the other node tries to access metadata
         * that depends on the completion of the alter operation.
         *
         * - This creates a circular wait across nodes:
         *
         * Node-1: Query -> waits on Alter -> waits on Sync Worker
         *
         * Node-2: Query -> waits on Alter -> waits on Sync Worker
         *
         * Since each node only sees part of the wait graph, the deadlock may
         * go undetected, leading to indefinite blocking.
         *
         * Note: Each entry in VALUES includes an index 'seqidx' that
         * represents the sequence's position in the local 'seqinfos' list.
         * This index is propagated to the query results and later used to
         * directly map the fetched publisher sequence rows back to their
         * corresponding local entries without relying on result order or name
         * matching.
         */
        appendStringInfo(&cmd,
                         "SELECT s.seqidx, has_sequence_privilege(c.oid, 'SELECT'),\n"
                         "       ps.*, seq.seqtypid,\n"
                         "       seq.seqstart, seq.seqincrement, seq.seqmin,\n"
                         "       seq.seqmax, seq.seqcycle\n"
                         "FROM ( VALUES %s ) AS s (schname, seqname, seqidx)\n"
                         "JOIN pg_namespace n ON n.nspname = s.schname\n"
                         "JOIN pg_class c ON c.relnamespace = n.oid AND c.relname = s.seqname\n"
                         "JOIN pg_sequence seq ON seq.seqrelid = c.oid\n"
                         "JOIN LATERAL pg_get_sequence_data(seq.seqrelid) AS ps ON true\n",
                         seqstr.data);
 
        res = walrcv_exec(conn, cmd.data, lengthof(seqRow), seqRow);
        if (res->status != WALRCV_OK_TUPLES)
            ereport(ERROR,
                    errcode(ERRCODE_CONNECTION_FAILURE),
                    errmsg("could not fetch sequence information from the publisher: %s",
                           res->err));
 
        slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple);
        while (tuplestore_gettupleslot(res->tuplestore, true, false, slot))
        {
            CopySeqResult sync_status;
            LogicalRepSequenceInfo *seqinfo;
            Relation    sequence_rel = NULL;
            int         seqidx;
 
            CHECK_FOR_INTERRUPTS();
 
            if (ConfigReloadPending)
            {
                ConfigReloadPending = false;
                ProcessConfigFile(PGC_SIGHUP);
            }
 
            sync_status = get_and_validate_seq_info(slot, &sequence_rel,
                                                    &seqinfo, &seqidx);
            if (sync_status == COPYSEQ_SUCCESS)
                sync_status = copy_sequence(seqinfo,
                                            sequence_rel->rd_rel->relowner);
 
            switch (sync_status)
            {
                case COPYSEQ_SUCCESS:
                    elog(DEBUG1,
                         "logical replication synchronization for subscription \"%s\", sequence \"%s.%s\" has finished",
                         MySubscription->name, seqinfo->nspname,
                         seqinfo->seqname);
                    batch_succeeded_count++;
                    break;
                case COPYSEQ_MISMATCH:
 
                    /*
                     * Remember mismatched sequences in a long-lived memory
                     * context since these will be used after the transaction
                     * is committed.
                     */
                    oldctx = MemoryContextSwitchTo(ApplyContext);
                    mismatched_seqs_idx = lappend_int(mismatched_seqs_idx,
                                                      seqidx);
                    MemoryContextSwitchTo(oldctx);
                    batch_mismatched_count++;
                    break;
                case COPYSEQ_SUBSCRIBER_INSUFFICIENT_PERM:
 
                    /*
                     * Remember sequences with insufficient privileges in a
                     * long-lived memory context since these will be used
                     * after the transaction is committed.
                     */
                    oldctx = MemoryContextSwitchTo(ApplyContext);
                    sub_insuffperm_seqs_idx = lappend_int(sub_insuffperm_seqs_idx,
                                                          seqidx);
                    MemoryContextSwitchTo(oldctx);
                    batch_sub_insuffperm_count++;
                    break;
                case COPYSEQ_PUBLISHER_INSUFFICIENT_PERM:
 
                    /*
                     * Remember sequences for which the publisher lacks the
                     * privileges required by pg_get_sequence_data().
                     */
                    oldctx = MemoryContextSwitchTo(ApplyContext);
                    pub_insuffperm_seqs_idx = lappend_int(pub_insuffperm_seqs_idx,
                                                          seqidx);
                    MemoryContextSwitchTo(oldctx);
                    batch_pub_insuffperm_count++;
                    break;
                case COPYSEQ_SKIPPED:
 
                    /*
                     * Concurrent removal of a sequence on the subscriber is
                     * treated as success, since the only viable action is to
                     * skip the corresponding sequence data. Missing sequences
                     * on the publisher are treated as ERROR.
                     */
                    if (seqinfo->found_on_pub)
                    {
                        ereport(LOG,
                                errmsg("skip synchronization of sequence \"%s.%s\" because it has been dropped concurrently",
                                       seqinfo->nspname,
                                       seqinfo->seqname));
                        batch_skipped_count++;
                    }
                    break;
            }
 
            if (sequence_rel)
                table_close(sequence_rel, NoLock);
        }
 
        ExecDropSingleTupleTableSlot(slot);
        walrcv_clear_result(res);
        resetStringInfo(&seqstr);
        resetStringInfo(&cmd);
 
        batch_missing_count = batch_size - (batch_succeeded_count +
                                            batch_mismatched_count +
                                            batch_sub_insuffperm_count +
                                            batch_pub_insuffperm_count +
                                            batch_skipped_count);
 
        elog(DEBUG1,
             "logical replication sequence synchronization for subscription \"%s\" - batch #%d = %d attempted, %d succeeded, %d mismatched, %d subscriber insufficient permission, %d publisher insufficient permission, %d missing from publisher, %d skipped",
             MySubscription->name,
             (cur_batch_base_index / MAX_SEQUENCES_SYNC_PER_BATCH) + 1,
             batch_size, batch_succeeded_count, batch_mismatched_count,
             batch_sub_insuffperm_count, batch_pub_insuffperm_count, batch_missing_count, batch_skipped_count);
 
        /* Commit this batch, and prepare for next batch */
        CommitTransactionCommand();
 
        if (batch_missing_count)
        {
            for (int idx = cur_batch_base_index; idx < cur_batch_base_index + batch_size; idx++)
            {
                LogicalRepSequenceInfo *seqinfo =
                    (LogicalRepSequenceInfo *) list_nth(seqinfos, idx);
 
                /* If the sequence was not found on publisher, record it */
                if (!seqinfo->found_on_pub)
                    missing_seqs_idx = lappend_int(missing_seqs_idx, idx);
            }
        }
 
        /*
         * cur_batch_base_index is not incremented sequentially because some
         * sequences may be missing, and the number of fetched rows may not
         * match the batch size.
         */
        cur_batch_base_index += batch_size;
    }
 
    /* Report mismatches, permission issues, or missing sequences */
    report_sequence_errors(mismatched_seqs_idx, sub_insuffperm_seqs_idx,
                           pub_insuffperm_seqs_idx, missing_seqs_idx);
}

References appendStringInfo(), appendStringInfoString(), ApplyContext, CHECK_FOR_INTERRUPTS, CommitTransactionCommand(), ConfigReloadPending, conn, copy_sequence(), COPYSEQ_MISMATCH, COPYSEQ_PUBLISHER_INSUFFICIENT_PERM, COPYSEQ_SKIPPED, COPYSEQ_SUBSCRIBER_INSUFFICIENT_PERM, COPYSEQ_SUCCESS, StringInfoData::data, DEBUG1, elog, ereport, WalRcvExecResult::err, errcode(), errmsg, ERROR, ExecDropSingleTupleTableSlot(), fb(), get_and_validate_seq_info(), idx(), initStringInfo(), lappend_int(), lengthof, list_length(), list_nth(), LOG, MakeSingleTupleTableSlot(), MAX_SEQUENCES_SYNC_PER_BATCH, MemoryContextSwitchTo(), MySubscription, Subscription::name, NIL, NoLock, PGC_SIGHUP, ProcessConfigFile(), quote_literal_cstr(), REMOTE_SEQ_COL_COUNT, report_sequence_errors(), resetStringInfo(), seqinfos, StartTransactionCommand(), WalRcvExecResult::status, table_close(), TTSOpsMinimalTuple, WalRcvExecResult::tupledesc, WalRcvExecResult::tuplestore, tuplestore_gettupleslot(), walrcv_clear_result(), walrcv_exec, and WALRCV_OK_TUPLES.

Referenced by LogicalRepSyncSequences().

get_and_validate_seq_info()

static CopySeqResult get_and_validate_seq_info ( TupleTableSlot *  slot, Relation *  sequence_rel, LogicalRepSequenceInfo **  seqinfo, int *  seqidx  )

static

Definition at line 267 of file sequencesync.c.

{
    bool        isnull;
    int         col = 0;
    Datum       datum;
    bool        remote_has_select_priv;
    Oid         remote_typid;
    int64       remote_start;
    int64       remote_increment;
    int64       remote_min;
    int64       remote_max;
    bool        remote_cycle;
    CopySeqResult result = COPYSEQ_SUCCESS;
    HeapTuple   tup;
    Form_pg_sequence local_seq;
    LogicalRepSequenceInfo *seqinfo_local;
 
    *seqidx = DatumGetInt32(slot_getattr(slot, ++col, &isnull));
    Assert(!isnull);
 
    /* Identify the corresponding local sequence for the given index. */
    *seqinfo = seqinfo_local =
        (LogicalRepSequenceInfo *) list_nth(seqinfos, *seqidx);
 
    /*
     * The remote sequence state can be NULL if the publisher lacks the
     * required privileges or if the sequence was dropped concurrently after
     * it was identified in the catalog snapshot (see pg_get_sequence_data()).
     */
    remote_has_select_priv = DatumGetBool(slot_getattr(slot, ++col, &isnull));
    Assert(!isnull);
 
    datum = slot_getattr(slot, ++col, &isnull);
    if (isnull)
        return remote_has_select_priv ? COPYSEQ_SKIPPED :
            COPYSEQ_PUBLISHER_INSUFFICIENT_PERM;
 
    seqinfo_local->last_value = DatumGetInt64(datum);
 
    seqinfo_local->is_called = DatumGetBool(slot_getattr(slot, ++col, &isnull));
    Assert(!isnull);
 
    seqinfo_local->page_lsn = DatumGetLSN(slot_getattr(slot, ++col, &isnull));
    Assert(!isnull);
 
    remote_typid = DatumGetObjectId(slot_getattr(slot, ++col, &isnull));
    Assert(!isnull);
 
    remote_start = DatumGetInt64(slot_getattr(slot, ++col, &isnull));
    Assert(!isnull);
 
    remote_increment = DatumGetInt64(slot_getattr(slot, ++col, &isnull));
    Assert(!isnull);
 
    remote_min = DatumGetInt64(slot_getattr(slot, ++col, &isnull));
    Assert(!isnull);
 
    remote_max = DatumGetInt64(slot_getattr(slot, ++col, &isnull));
    Assert(!isnull);
 
    remote_cycle = DatumGetBool(slot_getattr(slot, ++col, &isnull));
    Assert(!isnull);
 
    /* Sanity check */
    Assert(col == REMOTE_SEQ_COL_COUNT);
 
    seqinfo_local->found_on_pub = true;
 
    *sequence_rel = try_table_open(seqinfo_local->localrelid, RowExclusiveLock);
 
    /* Sequence was concurrently dropped? */
    if (!*sequence_rel)
        return COPYSEQ_SKIPPED;
 
    tup = SearchSysCache1(SEQRELID, ObjectIdGetDatum(seqinfo_local->localrelid));
 
    /* Sequence was concurrently dropped? */
    if (!HeapTupleIsValid(tup))
        elog(ERROR, "cache lookup failed for sequence %u",
             seqinfo_local->localrelid);
 
    local_seq = (Form_pg_sequence) GETSTRUCT(tup);
 
    /* Sequence parameters for remote/local are the same? */
    if (local_seq->seqtypid != remote_typid ||
        local_seq->seqstart != remote_start ||
        local_seq->seqincrement != remote_increment ||
        local_seq->seqmin != remote_min ||
        local_seq->seqmax != remote_max ||
        local_seq->seqcycle != remote_cycle)
        result = COPYSEQ_MISMATCH;
 
    /* Sequence was concurrently renamed? */
    if (strcmp(seqinfo_local->nspname,
               get_namespace_name(RelationGetNamespace(*sequence_rel))) ||
        strcmp(seqinfo_local->seqname, RelationGetRelationName(*sequence_rel)))
        result = COPYSEQ_MISMATCH;
 
    ReleaseSysCache(tup);
    return result;
}

References Assert, COPYSEQ_MISMATCH, COPYSEQ_PUBLISHER_INSUFFICIENT_PERM, COPYSEQ_SKIPPED, COPYSEQ_SUCCESS, DatumGetBool(), DatumGetInt32(), DatumGetInt64(), DatumGetLSN(), DatumGetObjectId(), elog, ERROR, fb(), Form_pg_sequence, get_namespace_name(), GETSTRUCT(), HeapTupleIsValid, list_nth(), ObjectIdGetDatum(), RelationGetNamespace, RelationGetRelationName, ReleaseSysCache(), REMOTE_SEQ_COL_COUNT, result, RowExclusiveLock, SearchSysCache1(), seqinfos, slot_getattr(), and try_table_open().

Referenced by copy_sequences().

get_sequences_string()

static void get_sequences_string ( List *  seqindexes, StringInfo  buf  )

static

Definition at line 149 of file sequencesync.c.

{
    resetStringInfo(buf);
    foreach_int(seqidx, seqindexes)
    {
        LogicalRepSequenceInfo *seqinfo =
            (LogicalRepSequenceInfo *) list_nth(seqinfos, seqidx);
 
        if (buf->len > 0)
            appendStringInfoString(buf, ", ");
 
        appendStringInfo(buf, "\"%s.%s\"", seqinfo->nspname, seqinfo->seqname);
    }
}

References appendStringInfo(), appendStringInfoString(), buf, fb(), foreach_int, list_nth(), resetStringInfo(), and seqinfos.

Referenced by report_sequence_errors().

LogicalRepSyncSequences()

static void LogicalRepSyncSequences ( void  )

static

Definition at line 680 of file sequencesync.c.

{
    char       *err;
    bool        must_use_password;
    Relation    rel;
    HeapTuple   tup;
    ScanKeyData skey[2];
    SysScanDesc scan;
    Oid         subid = MyLogicalRepWorker->subid;
    StringInfoData app_name;
 
    StartTransactionCommand();
 
    rel = table_open(SubscriptionRelRelationId, AccessShareLock);
 
    ScanKeyInit(&skey[0],
                Anum_pg_subscription_rel_srsubid,
                BTEqualStrategyNumber, F_OIDEQ,
                ObjectIdGetDatum(subid));
 
    ScanKeyInit(&skey[1],
                Anum_pg_subscription_rel_srsubstate,
                BTEqualStrategyNumber, F_CHAREQ,
                CharGetDatum(SUBREL_STATE_INIT));
 
    scan = systable_beginscan(rel, InvalidOid, false,
                              NULL, 2, skey);
    while (HeapTupleIsValid(tup = systable_getnext(scan)))
    {
        Form_pg_subscription_rel subrel;
        LogicalRepSequenceInfo *seq;
        Relation    sequence_rel;
        MemoryContext oldctx;
 
        CHECK_FOR_INTERRUPTS();
 
        subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
 
        sequence_rel = try_table_open(subrel->srrelid, RowExclusiveLock);
 
        /* Skip if sequence was dropped concurrently */
        if (!sequence_rel)
            continue;
 
        /* Skip if the relation is not a sequence */
        if (sequence_rel->rd_rel->relkind != RELKIND_SEQUENCE)
        {
            table_close(sequence_rel, NoLock);
            continue;
        }
 
        /*
         * Worker needs to process sequences across transaction boundary, so
         * allocate them under long-lived context.
         */
        oldctx = MemoryContextSwitchTo(ApplyContext);
 
        seq = palloc0_object(LogicalRepSequenceInfo);
        seq->localrelid = subrel->srrelid;
        seq->nspname = get_namespace_name(RelationGetNamespace(sequence_rel));
        seq->seqname = pstrdup(RelationGetRelationName(sequence_rel));
        seqinfos = lappend(seqinfos, seq);
 
        MemoryContextSwitchTo(oldctx);
 
        table_close(sequence_rel, NoLock);
    }
 
    /* Cleanup */
    systable_endscan(scan);
    table_close(rel, AccessShareLock);
 
    CommitTransactionCommand();
 
    /*
     * Exit early if no catalog entries found, likely due to concurrent drops.
     */
    if (!seqinfos)
        return;
 
    /* Is the use of a password mandatory? */
    must_use_password = MySubscription->passwordrequired &&
        !MySubscription->ownersuperuser;
 
    initStringInfo(&app_name);
    appendStringInfo(&app_name, "pg_%u_sequence_sync_" UINT64_FORMAT,
                     MySubscription->oid, GetSystemIdentifier());
 
    /*
     * Establish the connection to the publisher for sequence synchronization.
     */
    LogRepWorkerWalRcvConn =
        walrcv_connect(MySubscription->conninfo, true, true,
                       must_use_password,
                       app_name.data, &err);
    if (LogRepWorkerWalRcvConn == NULL)
        ereport(ERROR,
                errcode(ERRCODE_CONNECTION_FAILURE),
                errmsg("sequencesync worker for subscription \"%s\" could not connect to the publisher: %s",
                       MySubscription->name, err));
 
    pfree(app_name.data);
 
    copy_sequences(LogRepWorkerWalRcvConn);
}

References AccessShareLock, appendStringInfo(), ApplyContext, BTEqualStrategyNumber, CharGetDatum(), CHECK_FOR_INTERRUPTS, CommitTransactionCommand(), Subscription::conninfo, copy_sequences(), ereport, err(), errcode(), errmsg, ERROR, fb(), Form_pg_subscription_rel, get_namespace_name(), GETSTRUCT(), GetSystemIdentifier(), HeapTupleIsValid, initStringInfo(), InvalidOid, lappend(), LogRepWorkerWalRcvConn, MemoryContextSwitchTo(), MyLogicalRepWorker, MySubscription, Subscription::name, NoLock, ObjectIdGetDatum(), Subscription::oid, Subscription::ownersuperuser, palloc0_object, Subscription::passwordrequired, pfree(), pstrdup(), RelationGetNamespace, RelationGetRelationName, RowExclusiveLock, ScanKeyInit(), seqinfos, StartTransactionCommand(), LogicalRepWorker::subid, systable_beginscan(), systable_endscan(), systable_getnext(), table_close(), table_open(), try_table_open(), UINT64_FORMAT, and walrcv_connect.

Referenced by start_sequence_sync().

ProcessSequencesForSync()

void ProcessSequencesForSync

(

void

)

Definition at line 97 of file sequencesync.c.

{
    LogicalRepWorker *sequencesync_worker;
    int         nsyncworkers;
    bool        has_pending_sequences;
    bool        started_tx;
 
    FetchRelationStates(NULL, &has_pending_sequences, &started_tx);
 
    if (started_tx)
    {
        CommitTransactionCommand();
        pgstat_report_stat(true);
    }
 
    if (!has_pending_sequences)
        return;
 
    LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
 
    /* Check if there is a sequencesync worker already running? */
    sequencesync_worker = logicalrep_worker_find(WORKERTYPE_SEQUENCESYNC,
                                                 MyLogicalRepWorker->subid,
                                                 InvalidOid, true);
    if (sequencesync_worker)
    {
        LWLockRelease(LogicalRepWorkerLock);
        return;
    }
 
    /*
     * Count running sync workers for this subscription, while we have the
     * lock.
     */
    nsyncworkers = logicalrep_sync_worker_count(MyLogicalRepWorker->subid);
    LWLockRelease(LogicalRepWorkerLock);
 
    /*
     * It is okay to read/update last_seqsync_start_time here in apply worker
     * as we have already ensured that sync worker doesn't exist.
     */
    launch_sync_worker(WORKERTYPE_SEQUENCESYNC, nsyncworkers, InvalidOid,
                       &MyLogicalRepWorker->last_seqsync_start_time);
}

References CommitTransactionCommand(), fb(), FetchRelationStates(), InvalidOid, LogicalRepWorker::last_seqsync_start_time, launch_sync_worker(), logicalrep_sync_worker_count(), logicalrep_worker_find(), LW_SHARED, LWLockAcquire(), LWLockRelease(), MyLogicalRepWorker, pgstat_report_stat(), LogicalRepWorker::subid, and WORKERTYPE_SEQUENCESYNC.

Referenced by ProcessSyncingRelations().

report_sequence_errors()

static void report_sequence_errors ( List *  mismatched_seqs_idx, List *  sub_insuffperm_seqs_idx, List *  pub_insuffperm_seqs_idx, List *  missing_seqs_idx  )

static

Definition at line 176 of file sequencesync.c.

{
    StringInfoData seqstr;
 
    /* Quick exit if there are no errors to report */
    if (!mismatched_seqs_idx && !sub_insuffperm_seqs_idx &&
        !pub_insuffperm_seqs_idx && !missing_seqs_idx)
        return;
 
    initStringInfo(&seqstr);
 
    if (mismatched_seqs_idx)
    {
        get_sequences_string(mismatched_seqs_idx, &seqstr);
        ereport(WARNING,
                errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                errmsg_plural("mismatched or renamed sequence on subscriber (%s)",
                              "mismatched or renamed sequences on subscriber (%s)",
                              list_length(mismatched_seqs_idx),
                              seqstr.data));
    }
 
    if (sub_insuffperm_seqs_idx)
    {
        get_sequences_string(sub_insuffperm_seqs_idx, &seqstr);
 
        /*
         * With run_as_owner enabled, sequence synchronization runs as the
         * subscription owner, so a missing UPDATE privilege should be granted
         * to that role. Otherwise, the worker switches to the sequence owner
         * before checking privileges, so no useful GRANT hint can be
         * provided.
         */
        ereport(WARNING,
                errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                errmsg_plural("insufficient privileges on subscriber sequence (%s)",
                              "insufficient privileges on subscriber sequences (%s)",
                              list_length(sub_insuffperm_seqs_idx),
                              seqstr.data),
                MySubscription->runasowner ?
                errhint_plural("Grant UPDATE on the sequence to the subscription "
                               "owner on the subscriber.",
                               "Grant UPDATE on the sequences to the subscription "
                               "owner on the subscriber.",
                               list_length(sub_insuffperm_seqs_idx)) : 0);
    }
 
    if (pub_insuffperm_seqs_idx)
    {
        get_sequences_string(pub_insuffperm_seqs_idx, &seqstr);
        ereport(WARNING,
                errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                errmsg_plural("insufficient privileges on publisher sequence (%s)",
                              "insufficient privileges on publisher sequences (%s)",
                              list_length(pub_insuffperm_seqs_idx),
                              seqstr.data),
                errhint_plural("Grant SELECT on the sequence to the role used for "
                               "the replication connection on the publisher.",
                               "Grant SELECT on the sequences to the role used for "
                               "the replication connection on the publisher.",
                               list_length(pub_insuffperm_seqs_idx)));
    }
 
    if (missing_seqs_idx)
    {
        get_sequences_string(missing_seqs_idx, &seqstr);
        ereport(WARNING,
                errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                errmsg_plural("missing sequence on publisher (%s)",
                              "missing sequences on publisher (%s)",
                              list_length(missing_seqs_idx),
                              seqstr.data));
    }
 
    ereport(ERROR,
            errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
            errmsg("logical replication sequence synchronization failed for subscription \"%s\"",
                   MySubscription->name));
}

References ereport, errcode(), errhint_plural(), errmsg, errmsg_plural(), ERROR, fb(), get_sequences_string(), initStringInfo(), list_length(), MySubscription, Subscription::name, Subscription::runasowner, and WARNING.

Referenced by copy_sequences().

SequenceSyncWorkerMain()

void SequenceSyncWorkerMain

(

Datum

main_arg

)

Definition at line 825 of file sequencesync.c.

{
    int         worker_slot = DatumGetInt32(main_arg);
 
    SetupApplyOrSyncWorker(worker_slot);
 
    start_sequence_sync();
 
    FinishSyncWorker();
}

References DatumGetInt32(), fb(), FinishSyncWorker(), SetupApplyOrSyncWorker(), and start_sequence_sync().

start_sequence_sync()

static void start_sequence_sync ( void  )

static

Definition at line 794 of file sequencesync.c.

{
    Assert(am_sequencesync_worker());
 
    PG_TRY();
    {
        /* Call initial sync. */
        LogicalRepSyncSequences();
    }
    PG_CATCH();
    {
        if (MySubscription->disableonerr)
            DisableSubscriptionAndExit();
        else
        {
            /*
             * Report the worker failed during sequence synchronization. Abort
             * the current transaction so that the stats message is sent in an
             * idle state.
             */
            AbortOutOfAnyTransaction();
            pgstat_report_subscription_error(MySubscription->oid);
 
            PG_RE_THROW();
        }
    }
    PG_END_TRY();
}

References AbortOutOfAnyTransaction(), am_sequencesync_worker(), Assert, Subscription::disableonerr, DisableSubscriptionAndExit(), LogicalRepSyncSequences(), MySubscription, Subscription::oid, PG_CATCH, PG_END_TRY, PG_RE_THROW, PG_TRY, and pgstat_report_subscription_error().

Referenced by SequenceSyncWorkerMain().

Variable Documentation

seqinfos

List* seqinfos = NIL

static

Definition at line 86 of file sequencesync.c.

Referenced by copy_sequences(), get_and_validate_seq_info(), get_sequences_string(), and LogicalRepSyncSequences().