pgoutput.c

Go to the documentation of this file.

/*-------------------------------------------------------------------------
 *
 * pgoutput.c
 *      Logical Replication output plugin
 *
 * Copyright (c) 2012-2026, PostgreSQL Global Development Group
 *
 * IDENTIFICATION
 *        src/backend/replication/pgoutput/pgoutput.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"
 
#include "access/tupconvert.h"
#include "catalog/partition.h"
#include "catalog/pg_publication.h"
#include "catalog/pg_publication_rel.h"
#include "catalog/pg_subscription.h"
#include "commands/defrem.h"
#include "commands/subscriptioncmds.h"
#include "executor/executor.h"
#include "fmgr.h"
#include "nodes/makefuncs.h"
#include "parser/parse_relation.h"
#include "replication/logical.h"
#include "replication/logicalproto.h"
#include "replication/origin.h"
#include "replication/pgoutput.h"
#include "rewrite/rewriteHandler.h"
#include "utils/builtins.h"
#include "utils/inval.h"
#include "utils/lsyscache.h"
#include "utils/memutils.h"
#include "utils/rel.h"
#include "utils/syscache.h"
#include "utils/varlena.h"
 
PG_MODULE_MAGIC_EXT(
                    .name = "pgoutput",
                    .version = PG_VERSION
);
 
static void pgoutput_startup(LogicalDecodingContext *ctx,
                             OutputPluginOptions *opt, bool is_init);
static void pgoutput_shutdown(LogicalDecodingContext *ctx);
static void pgoutput_begin_txn(LogicalDecodingContext *ctx,
                               ReorderBufferTXN *txn);
static void pgoutput_commit_txn(LogicalDecodingContext *ctx,
                                ReorderBufferTXN *txn, XLogRecPtr commit_lsn);
static void pgoutput_change(LogicalDecodingContext *ctx,
                            ReorderBufferTXN *txn, Relation relation,
                            ReorderBufferChange *change);
static void pgoutput_truncate(LogicalDecodingContext *ctx,
                              ReorderBufferTXN *txn, int nrelations, Relation relations[],
                              ReorderBufferChange *change);
static void pgoutput_message(LogicalDecodingContext *ctx,
                             ReorderBufferTXN *txn, XLogRecPtr message_lsn,
                             bool transactional, const char *prefix,
                             Size sz, const char *message);
static bool pgoutput_origin_filter(LogicalDecodingContext *ctx,
                                   RepOriginId origin_id);
static void pgoutput_begin_prepare_txn(LogicalDecodingContext *ctx,
                                       ReorderBufferTXN *txn);
static void pgoutput_prepare_txn(LogicalDecodingContext *ctx,
                                 ReorderBufferTXN *txn, XLogRecPtr prepare_lsn);
static void pgoutput_commit_prepared_txn(LogicalDecodingContext *ctx,
                                         ReorderBufferTXN *txn, XLogRecPtr commit_lsn);
static void pgoutput_rollback_prepared_txn(LogicalDecodingContext *ctx,
                                           ReorderBufferTXN *txn,
                                           XLogRecPtr prepare_end_lsn,
                                           TimestampTz prepare_time);
static void pgoutput_stream_start(struct LogicalDecodingContext *ctx,
                                  ReorderBufferTXN *txn);
static void pgoutput_stream_stop(struct LogicalDecodingContext *ctx,
                                 ReorderBufferTXN *txn);
static void pgoutput_stream_abort(struct LogicalDecodingContext *ctx,
                                  ReorderBufferTXN *txn,
                                  XLogRecPtr abort_lsn);
static void pgoutput_stream_commit(struct LogicalDecodingContext *ctx,
                                   ReorderBufferTXN *txn,
                                   XLogRecPtr commit_lsn);
static void pgoutput_stream_prepare_txn(LogicalDecodingContext *ctx,
                                        ReorderBufferTXN *txn, XLogRecPtr prepare_lsn);
 
static bool publications_valid;
 
static List *LoadPublications(List *pubnames);
static void publication_invalidation_cb(Datum arg, int cacheid,
                                        uint32 hashvalue);
static void send_repl_origin(LogicalDecodingContext *ctx,
                             RepOriginId origin_id, XLogRecPtr origin_lsn,
                             bool send_origin);
 
/*
 * Only 3 publication actions are used for row filtering ("insert", "update",
 * "delete"). See RelationSyncEntry.exprstate[].
 */
enum RowFilterPubAction
{
    PUBACTION_INSERT,
    PUBACTION_UPDATE,
    PUBACTION_DELETE,
};
 
#define NUM_ROWFILTER_PUBACTIONS (PUBACTION_DELETE+1)
 
/*
 * Entry in the map used to remember which relation schemas we sent.
 *
 * The schema_sent flag determines if the current schema record for the
 * relation (and for its ancestor if publish_as_relid is set) was already
 * sent to the subscriber (in which case we don't need to send it again).
 *
 * The schema cache on downstream is however updated only at commit time,
 * and with streamed transactions the commit order may be different from
 * the order the transactions are sent in. Also, the (sub) transactions
 * might get aborted so we need to send the schema for each (sub) transaction
 * so that we don't lose the schema information on abort. For handling this,
 * we maintain the list of xids (streamed_txns) for those we have already sent
 * the schema.
 *
 * For partitions, 'pubactions' considers not only the table's own
 * publications, but also those of all of its ancestors.
 */
typedef struct RelationSyncEntry
{
    Oid         relid;          /* relation oid */
 
    bool        replicate_valid;    /* overall validity flag for entry */
 
    bool        schema_sent;
 
    /*
     * This will be PUBLISH_GENCOLS_STORED if the relation contains generated
     * columns and the 'publish_generated_columns' parameter is set to
     * PUBLISH_GENCOLS_STORED. Otherwise, it will be PUBLISH_GENCOLS_NONE,
     * indicating that no generated columns should be published, unless
     * explicitly specified in the column list.
     */
    PublishGencolsType include_gencols_type;
    List       *streamed_txns;  /* streamed toplevel transactions with this
                                 * schema */
 
    /* are we publishing this rel? */
    PublicationActions pubactions;
 
    /*
     * ExprState array for row filter. Different publication actions don't
     * allow multiple expressions to always be combined into one, because
     * updates or deletes restrict the column in expression to be part of the
     * replica identity index whereas inserts do not have this restriction, so
     * there is one ExprState per publication action.
     */
    ExprState  *exprstate[NUM_ROWFILTER_PUBACTIONS];
    EState     *estate;         /* executor state used for row filter */
    TupleTableSlot *new_slot;   /* slot for storing new tuple */
    TupleTableSlot *old_slot;   /* slot for storing old tuple */
 
    /*
     * OID of the relation to publish changes as.  For a partition, this may
     * be set to one of its ancestors whose schema will be used when
     * replicating changes, if publish_via_partition_root is set for the
     * publication.
     */
    Oid         publish_as_relid;
 
    /*
     * Map used when replicating using an ancestor's schema to convert tuples
     * from partition's type to the ancestor's; NULL if publish_as_relid is
     * same as 'relid' or if unnecessary due to partition and the ancestor
     * having identical TupleDesc.
     */
    AttrMap    *attrmap;
 
    /*
     * Columns included in the publication, or NULL if all columns are
     * included implicitly.  Note that the attnums in this bitmap are not
     * shifted by FirstLowInvalidHeapAttributeNumber.
     */
    Bitmapset  *columns;
 
    /*
     * Private context to store additional data for this entry - state for the
     * row filter expressions, column list, etc.
     */
    MemoryContext entry_cxt;
} RelationSyncEntry;
 
/*
 * Maintain a per-transaction level variable to track whether the transaction
 * has sent BEGIN. BEGIN is only sent when the first change in a transaction
 * is processed. This makes it possible to skip sending a pair of BEGIN/COMMIT
 * messages for empty transactions which saves network bandwidth.
 *
 * This optimization is not used for prepared transactions because if the
 * WALSender restarts after prepare of a transaction and before commit prepared
 * of the same transaction then we won't be able to figure out if we have
 * skipped sending BEGIN/PREPARE of a transaction as it was empty. This is
 * because we would have lost the in-memory txndata information that was
 * present prior to the restart. This will result in sending a spurious
 * COMMIT PREPARED without a corresponding prepared transaction at the
 * downstream which would lead to an error when it tries to process it.
 *
 * XXX We could achieve this optimization by changing protocol to send
 * additional information so that downstream can detect that the corresponding
 * prepare has not been sent. However, adding such a check for every
 * transaction in the downstream could be costly so we might want to do it
 * optionally.
 *
 * We also don't have this optimization for streamed transactions because
 * they can contain prepared transactions.
 */
typedef struct PGOutputTxnData
{
    bool        sent_begin_txn; /* flag indicating whether BEGIN has been sent */
} PGOutputTxnData;
 
/* Map used to remember which relation schemas we sent. */
static HTAB *RelationSyncCache = NULL;
 
static void init_rel_sync_cache(MemoryContext cachectx);
static void cleanup_rel_sync_cache(TransactionId xid, bool is_commit);
static RelationSyncEntry *get_rel_sync_entry(PGOutputData *data,
                                             Relation relation);
static void send_relation_and_attrs(Relation relation, TransactionId xid,
                                    LogicalDecodingContext *ctx,
                                    RelationSyncEntry *relentry);
static void rel_sync_cache_relation_cb(Datum arg, Oid relid);
static void rel_sync_cache_publication_cb(Datum arg, int cacheid,
                                          uint32 hashvalue);
static void set_schema_sent_in_streamed_txn(RelationSyncEntry *entry,
                                            TransactionId xid);
static bool get_schema_sent_in_streamed_txn(RelationSyncEntry *entry,
                                            TransactionId xid);
static void init_tuple_slot(PGOutputData *data, Relation relation,
                            RelationSyncEntry *entry);
static void pgoutput_memory_context_reset(void *arg);
 
/* row filter routines */
static EState *create_estate_for_relation(Relation rel);
static void pgoutput_row_filter_init(PGOutputData *data,
                                     List *publications,
                                     RelationSyncEntry *entry);
static bool pgoutput_row_filter_exec_expr(ExprState *state,
                                          ExprContext *econtext);
static bool pgoutput_row_filter(Relation relation, TupleTableSlot *old_slot,
                                TupleTableSlot **new_slot_ptr,
                                RelationSyncEntry *entry,
                                ReorderBufferChangeType *action);
 
/* column list routines */
static void pgoutput_column_list_init(PGOutputData *data,
                                      List *publications,
                                      RelationSyncEntry *entry);
 
/*
 * Specify output plugin callbacks
 */
void
_PG_output_plugin_init(OutputPluginCallbacks *cb)
{
    cb->startup_cb = pgoutput_startup;
    cb->begin_cb = pgoutput_begin_txn;
    cb->change_cb = pgoutput_change;
    cb->truncate_cb = pgoutput_truncate;
    cb->message_cb = pgoutput_message;
    cb->commit_cb = pgoutput_commit_txn;
 
    cb->begin_prepare_cb = pgoutput_begin_prepare_txn;
    cb->prepare_cb = pgoutput_prepare_txn;
    cb->commit_prepared_cb = pgoutput_commit_prepared_txn;
    cb->rollback_prepared_cb = pgoutput_rollback_prepared_txn;
    cb->filter_by_origin_cb = pgoutput_origin_filter;
    cb->shutdown_cb = pgoutput_shutdown;
 
    /* transaction streaming */
    cb->stream_start_cb = pgoutput_stream_start;
    cb->stream_stop_cb = pgoutput_stream_stop;
    cb->stream_abort_cb = pgoutput_stream_abort;
    cb->stream_commit_cb = pgoutput_stream_commit;
    cb->stream_change_cb = pgoutput_change;
    cb->stream_message_cb = pgoutput_message;
    cb->stream_truncate_cb = pgoutput_truncate;
    /* transaction streaming - two-phase commit */
    cb->stream_prepare_cb = pgoutput_stream_prepare_txn;
}
 
static void
parse_output_parameters(List *options, PGOutputData *data)
{
    ListCell   *lc;
    bool        protocol_version_given = false;
    bool        publication_names_given = false;
    bool        binary_option_given = false;
    bool        messages_option_given = false;
    bool        streaming_given = false;
    bool        two_phase_option_given = false;
    bool        origin_option_given = false;
 
    /* Initialize optional parameters to defaults */
    data->binary = false;
    data->streaming = LOGICALREP_STREAM_OFF;
    data->messages = false;
    data->two_phase = false;
    data->publish_no_origin = false;
 
    foreach(lc, options)
    {
        DefElem    *defel = (DefElem *) lfirst(lc);
 
        Assert(defel->arg == NULL || IsA(defel->arg, String));
 
        /* Check each param, whether or not we recognize it */
        if (strcmp(defel->defname, "proto_version") == 0)
        {
            unsigned long parsed;
            char       *endptr;
 
            if (protocol_version_given)
                ereport(ERROR,
                        (errcode(ERRCODE_SYNTAX_ERROR),
                         errmsg("conflicting or redundant options")));
            protocol_version_given = true;
 
            errno = 0;
            parsed = strtoul(strVal(defel->arg), &endptr, 10);
            if (errno != 0 || *endptr != '\0')
                ereport(ERROR,
                        (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                         errmsg("invalid proto_version")));
 
            if (parsed > PG_UINT32_MAX)
                ereport(ERROR,
                        (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                         errmsg("proto_version \"%s\" out of range",
                                strVal(defel->arg))));
 
            data->protocol_version = (uint32) parsed;
        }
        else if (strcmp(defel->defname, "publication_names") == 0)
        {
            if (publication_names_given)
                ereport(ERROR,
                        (errcode(ERRCODE_SYNTAX_ERROR),
                         errmsg("conflicting or redundant options")));
            publication_names_given = true;
 
            /*
             * Pass a copy of the DefElem->arg since SplitIdentifierString
             * modifies its input.
             */
            if (!SplitIdentifierString(pstrdup(strVal(defel->arg)), ',',
                                       &data->publication_names))
                ereport(ERROR,
                        (errcode(ERRCODE_INVALID_NAME),
                         errmsg("invalid publication_names syntax")));
        }
        else if (strcmp(defel->defname, "binary") == 0)
        {
            if (binary_option_given)
                ereport(ERROR,
                        (errcode(ERRCODE_SYNTAX_ERROR),
                         errmsg("conflicting or redundant options")));
            binary_option_given = true;
 
            data->binary = defGetBoolean(defel);
        }
        else if (strcmp(defel->defname, "messages") == 0)
        {
            if (messages_option_given)
                ereport(ERROR,
                        (errcode(ERRCODE_SYNTAX_ERROR),
                         errmsg("conflicting or redundant options")));
            messages_option_given = true;
 
            data->messages = defGetBoolean(defel);
        }
        else if (strcmp(defel->defname, "streaming") == 0)
        {
            if (streaming_given)
                ereport(ERROR,
                        (errcode(ERRCODE_SYNTAX_ERROR),
                         errmsg("conflicting or redundant options")));
            streaming_given = true;
 
            data->streaming = defGetStreamingMode(defel);
        }
        else if (strcmp(defel->defname, "two_phase") == 0)
        {
            if (two_phase_option_given)
                ereport(ERROR,
                        (errcode(ERRCODE_SYNTAX_ERROR),
                         errmsg("conflicting or redundant options")));
            two_phase_option_given = true;
 
            data->two_phase = defGetBoolean(defel);
        }
        else if (strcmp(defel->defname, "origin") == 0)
        {
            char       *origin;
 
            if (origin_option_given)
                ereport(ERROR,
                        errcode(ERRCODE_SYNTAX_ERROR),
                        errmsg("conflicting or redundant options"));
            origin_option_given = true;
 
            origin = defGetString(defel);
            if (pg_strcasecmp(origin, LOGICALREP_ORIGIN_NONE) == 0)
                data->publish_no_origin = true;
            else if (pg_strcasecmp(origin, LOGICALREP_ORIGIN_ANY) == 0)
                data->publish_no_origin = false;
            else
                ereport(ERROR,
                        errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                        errmsg("unrecognized origin value: \"%s\"", origin));
        }
        else
            elog(ERROR, "unrecognized pgoutput option: %s", defel->defname);
    }
 
    /* Check required options */
    if (!protocol_version_given)
        ereport(ERROR,
                errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                errmsg("option \"%s\" missing", "proto_version"));
    if (!publication_names_given)
        ereport(ERROR,
                errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                errmsg("option \"%s\" missing", "publication_names"));
}
 
/*
 * Memory context reset callback of PGOutputData->context.
 */
static void
pgoutput_memory_context_reset(void *arg)
{
    if (RelationSyncCache)
    {
        hash_destroy(RelationSyncCache);
        RelationSyncCache = NULL;
    }
}
 
/*
 * Initialize this plugin
 */
static void
pgoutput_startup(LogicalDecodingContext *ctx, OutputPluginOptions *opt,
                 bool is_init)
{
    PGOutputData *data = palloc0_object(PGOutputData);
    static bool publication_callback_registered = false;
    MemoryContextCallback *mcallback;
 
    /* Create our memory context for private allocations. */
    data->context = AllocSetContextCreate(ctx->context,
                                          "logical replication output context",
                                          ALLOCSET_DEFAULT_SIZES);
 
    data->cachectx = AllocSetContextCreate(ctx->context,
                                           "logical replication cache context",
                                           ALLOCSET_DEFAULT_SIZES);
 
    data->pubctx = AllocSetContextCreate(ctx->context,
                                         "logical replication publication list context",
                                         ALLOCSET_SMALL_SIZES);
 
    /*
     * Ensure to cleanup RelationSyncCache even when logical decoding invoked
     * via SQL interface ends up with an error.
     */
    mcallback = palloc0_object(MemoryContextCallback);
    mcallback->func = pgoutput_memory_context_reset;
    MemoryContextRegisterResetCallback(ctx->context, mcallback);
 
    ctx->output_plugin_private = data;
 
    /* This plugin uses binary protocol. */
    opt->output_type = OUTPUT_PLUGIN_BINARY_OUTPUT;
 
    /*
     * This is replication start and not slot initialization.
     *
     * Parse and validate options passed by the client.
     */
    if (!is_init)
    {
        /* Parse the params and ERROR if we see any we don't recognize */
        parse_output_parameters(ctx->output_plugin_options, data);
 
        /* Check if we support requested protocol */
        if (data->protocol_version > LOGICALREP_PROTO_MAX_VERSION_NUM)
            ereport(ERROR,
                    (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                     errmsg("client sent proto_version=%d but server only supports protocol %d or lower",
                            data->protocol_version, LOGICALREP_PROTO_MAX_VERSION_NUM)));
 
        if (data->protocol_version < LOGICALREP_PROTO_MIN_VERSION_NUM)
            ereport(ERROR,
                    (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                     errmsg("client sent proto_version=%d but server only supports protocol %d or higher",
                            data->protocol_version, LOGICALREP_PROTO_MIN_VERSION_NUM)));
 
        /*
         * Decide whether to enable streaming. It is disabled by default, in
         * which case we just update the flag in decoding context. Otherwise
         * we only allow it with sufficient version of the protocol, and when
         * the output plugin supports it.
         */
        if (data->streaming == LOGICALREP_STREAM_OFF)
            ctx->streaming = false;
        else if (data->streaming == LOGICALREP_STREAM_ON &&
                 data->protocol_version < LOGICALREP_PROTO_STREAM_VERSION_NUM)
            ereport(ERROR,
                    (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                     errmsg("requested proto_version=%d does not support streaming, need %d or higher",
                            data->protocol_version, LOGICALREP_PROTO_STREAM_VERSION_NUM)));
        else if (data->streaming == LOGICALREP_STREAM_PARALLEL &&
                 data->protocol_version < LOGICALREP_PROTO_STREAM_PARALLEL_VERSION_NUM)
            ereport(ERROR,
                    (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                     errmsg("requested proto_version=%d does not support parallel streaming, need %d or higher",
                            data->protocol_version, LOGICALREP_PROTO_STREAM_PARALLEL_VERSION_NUM)));
        else if (!ctx->streaming)
            ereport(ERROR,
                    (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                     errmsg("streaming requested, but not supported by output plugin")));
 
        /*
         * Here, we just check whether the two-phase option is passed by
         * plugin and decide whether to enable it at later point of time. It
         * remains enabled if the previous start-up has done so. But we only
         * allow the option to be passed in with sufficient version of the
         * protocol, and when the output plugin supports it.
         */
        if (!data->two_phase)
            ctx->twophase_opt_given = false;
        else if (data->protocol_version < LOGICALREP_PROTO_TWOPHASE_VERSION_NUM)
            ereport(ERROR,
                    (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                     errmsg("requested proto_version=%d does not support two-phase commit, need %d or higher",
                            data->protocol_version, LOGICALREP_PROTO_TWOPHASE_VERSION_NUM)));
        else if (!ctx->twophase)
            ereport(ERROR,
                    (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                     errmsg("two-phase commit requested, but not supported by output plugin")));
        else
            ctx->twophase_opt_given = true;
 
        /* Init publication state. */
        data->publications = NIL;
        publications_valid = false;
 
        /*
         * Register callback for pg_publication if we didn't already do that
         * during some previous call in this process.
         */
        if (!publication_callback_registered)
        {
            CacheRegisterSyscacheCallback(PUBLICATIONOID,
                                          publication_invalidation_cb,
                                          (Datum) 0);
            CacheRegisterRelSyncCallback(rel_sync_cache_relation_cb,
                                         (Datum) 0);
            publication_callback_registered = true;
        }
 
        /* Initialize relation schema cache. */
        init_rel_sync_cache(CacheMemoryContext);
    }
    else
    {
        /*
         * Disable the streaming and prepared transactions during the slot
         * initialization mode.
         */
        ctx->streaming = false;
        ctx->twophase = false;
    }
}
 
/*
 * BEGIN callback.
 *
 * Don't send the BEGIN message here instead postpone it until the first
 * change. In logical replication, a common scenario is to replicate a set of
 * tables (instead of all tables) and transactions whose changes were on
 * the table(s) that are not published will produce empty transactions. These
 * empty transactions will send BEGIN and COMMIT messages to subscribers,
 * using bandwidth on something with little/no use for logical replication.
 */
static void
pgoutput_begin_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn)
{
    PGOutputTxnData *txndata = MemoryContextAllocZero(ctx->context,
                                                      sizeof(PGOutputTxnData));
 
    txn->output_plugin_private = txndata;
}
 
/*
 * Send BEGIN.
 *
 * This is called while processing the first change of the transaction.
 */
static void
pgoutput_send_begin(LogicalDecodingContext *ctx, ReorderBufferTXN *txn)
{
    bool        send_replication_origin = txn->origin_id != InvalidRepOriginId;
    PGOutputTxnData *txndata = (PGOutputTxnData *) txn->output_plugin_private;
 
    Assert(txndata);
    Assert(!txndata->sent_begin_txn);
 
    OutputPluginPrepareWrite(ctx, !send_replication_origin);
    logicalrep_write_begin(ctx->out, txn);
    txndata->sent_begin_txn = true;
 
    send_repl_origin(ctx, txn->origin_id, txn->origin_lsn,
                     send_replication_origin);
 
    OutputPluginWrite(ctx, true);
}
 
/*
 * COMMIT callback
 */
static void
pgoutput_commit_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
                    XLogRecPtr commit_lsn)
{
    PGOutputTxnData *txndata = (PGOutputTxnData *) txn->output_plugin_private;
    bool        sent_begin_txn;
 
    Assert(txndata);
 
    /*
     * We don't need to send the commit message unless some relevant change
     * from this transaction has been sent to the downstream.
     */
    sent_begin_txn = txndata->sent_begin_txn;
    OutputPluginUpdateProgress(ctx, !sent_begin_txn);
    pfree(txndata);
    txn->output_plugin_private = NULL;
 
    if (!sent_begin_txn)
    {
        elog(DEBUG1, "skipped replication of an empty transaction with XID: %u", txn->xid);
        return;
    }
 
    OutputPluginPrepareWrite(ctx, true);
    logicalrep_write_commit(ctx->out, txn, commit_lsn);
    OutputPluginWrite(ctx, true);
}
 
/*
 * BEGIN PREPARE callback
 */
static void
pgoutput_begin_prepare_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn)
{
    bool        send_replication_origin = txn->origin_id != InvalidRepOriginId;
 
    OutputPluginPrepareWrite(ctx, !send_replication_origin);
    logicalrep_write_begin_prepare(ctx->out, txn);
 
    send_repl_origin(ctx, txn->origin_id, txn->origin_lsn,
                     send_replication_origin);
 
    OutputPluginWrite(ctx, true);
}
 
/*
 * PREPARE callback
 */
static void
pgoutput_prepare_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
                     XLogRecPtr prepare_lsn)
{
    OutputPluginUpdateProgress(ctx, false);
 
    OutputPluginPrepareWrite(ctx, true);
    logicalrep_write_prepare(ctx->out, txn, prepare_lsn);
    OutputPluginWrite(ctx, true);
}
 
/*
 * COMMIT PREPARED callback
 */
static void
pgoutput_commit_prepared_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
                             XLogRecPtr commit_lsn)
{
    OutputPluginUpdateProgress(ctx, false);
 
    OutputPluginPrepareWrite(ctx, true);
    logicalrep_write_commit_prepared(ctx->out, txn, commit_lsn);
    OutputPluginWrite(ctx, true);
}
 
/*
 * ROLLBACK PREPARED callback
 */
static void
pgoutput_rollback_prepared_txn(LogicalDecodingContext *ctx,
                               ReorderBufferTXN *txn,
                               XLogRecPtr prepare_end_lsn,
                               TimestampTz prepare_time)
{
    OutputPluginUpdateProgress(ctx, false);
 
    OutputPluginPrepareWrite(ctx, true);
    logicalrep_write_rollback_prepared(ctx->out, txn, prepare_end_lsn,
                                       prepare_time);
    OutputPluginWrite(ctx, true);
}
 
/*
 * Write the current schema of the relation and its ancestor (if any) if not
 * done yet.
 */
static void
maybe_send_schema(LogicalDecodingContext *ctx,
                  ReorderBufferChange *change,
                  Relation relation, RelationSyncEntry *relentry)
{
    PGOutputData *data = (PGOutputData *) ctx->output_plugin_private;
    bool        schema_sent;
    TransactionId xid = InvalidTransactionId;
    TransactionId topxid = InvalidTransactionId;
 
    /*
     * Remember XID of the (sub)transaction for the change. We don't care if
     * it's top-level transaction or not (we have already sent that XID in
     * start of the current streaming block).
     *
     * If we're not in a streaming block, just use InvalidTransactionId and
     * the write methods will not include it.
     */
    if (data->in_streaming)
        xid = change->txn->xid;
 
    if (rbtxn_is_subtxn(change->txn))
        topxid = rbtxn_get_toptxn(change->txn)->xid;
    else
        topxid = xid;
 
    /*
     * Do we need to send the schema? We do track streamed transactions
     * separately, because those may be applied later (and the regular
     * transactions won't see their effects until then) and in an order that
     * we don't know at this point.
     *
     * XXX There is a scope of optimization here. Currently, we always send
     * the schema first time in a streaming transaction but we can probably
     * avoid that by checking 'relentry->schema_sent' flag. However, before
     * doing that we need to study its impact on the case where we have a mix
     * of streaming and non-streaming transactions.
     */
    if (data->in_streaming)
        schema_sent = get_schema_sent_in_streamed_txn(relentry, topxid);
    else
        schema_sent = relentry->schema_sent;
 
    /* Nothing to do if we already sent the schema. */
    if (schema_sent)
        return;
 
    /*
     * Send the schema.  If the changes will be published using an ancestor's
     * schema, not the relation's own, send that ancestor's schema before
     * sending relation's own (XXX - maybe sending only the former suffices?).
     */
    if (relentry->publish_as_relid != RelationGetRelid(relation))
    {
        Relation    ancestor = RelationIdGetRelation(relentry->publish_as_relid);
 
        send_relation_and_attrs(ancestor, xid, ctx, relentry);
        RelationClose(ancestor);
    }
 
    send_relation_and_attrs(relation, xid, ctx, relentry);
 
    if (data->in_streaming)
        set_schema_sent_in_streamed_txn(relentry, topxid);
    else
        relentry->schema_sent = true;
}
 
/*
 * Sends a relation
 */
static void
send_relation_and_attrs(Relation relation, TransactionId xid,
                        LogicalDecodingContext *ctx,
                        RelationSyncEntry *relentry)
{
    TupleDesc   desc = RelationGetDescr(relation);
    Bitmapset  *columns = relentry->columns;
    PublishGencolsType include_gencols_type = relentry->include_gencols_type;
    int         i;
 
    /*
     * Write out type info if needed.  We do that only for user-created types.
     * We use FirstGenbkiObjectId as the cutoff, so that we only consider
     * objects with hand-assigned OIDs to be "built in", not for instance any
     * function or type defined in the information_schema. This is important
     * because only hand-assigned OIDs can be expected to remain stable across
     * major versions.
     */
    for (i = 0; i < desc->natts; i++)
    {
        Form_pg_attribute att = TupleDescAttr(desc, i);
 
        if (!logicalrep_should_publish_column(att, columns,
                                              include_gencols_type))
            continue;
 
        if (att->atttypid < FirstGenbkiObjectId)
            continue;
 
        OutputPluginPrepareWrite(ctx, false);
        logicalrep_write_typ(ctx->out, xid, att->atttypid);
        OutputPluginWrite(ctx, false);
    }
 
    OutputPluginPrepareWrite(ctx, false);
    logicalrep_write_rel(ctx->out, xid, relation, columns,
                         include_gencols_type);
    OutputPluginWrite(ctx, false);
}
 
/*
 * Executor state preparation for evaluation of row filter expressions for the
 * specified relation.
 */
static EState *
create_estate_for_relation(Relation rel)
{
    EState     *estate;
    RangeTblEntry *rte;
    List       *perminfos = NIL;
 
    estate = CreateExecutorState();
 
    rte = makeNode(RangeTblEntry);
    rte->rtekind = RTE_RELATION;
    rte->relid = RelationGetRelid(rel);
    rte->relkind = rel->rd_rel->relkind;
    rte->rellockmode = AccessShareLock;
 
    addRTEPermissionInfo(&perminfos, rte);
 
    ExecInitRangeTable(estate, list_make1(rte), perminfos,
                       bms_make_singleton(1));
 
    estate->es_output_cid = GetCurrentCommandId(false);
 
    return estate;
}
 
/*
 * Evaluates row filter.
 *
 * If the row filter evaluates to NULL, it is taken as false i.e. the change
 * isn't replicated.
 */
static bool
pgoutput_row_filter_exec_expr(ExprState *state, ExprContext *econtext)
{
    Datum       ret;
    bool        isnull;
 
    Assert(state != NULL);
 
    ret = ExecEvalExprSwitchContext(state, econtext, &isnull);
 
    elog(DEBUG3, "row filter evaluates to %s (isnull: %s)",
         isnull ? "false" : DatumGetBool(ret) ? "true" : "false",
         isnull ? "true" : "false");
 
    if (isnull)
        return false;
 
    return DatumGetBool(ret);
}
 
/*
 * Make sure the per-entry memory context exists.
 */
static void
pgoutput_ensure_entry_cxt(PGOutputData *data, RelationSyncEntry *entry)
{
    Relation    relation;
 
    /* The context may already exist, in which case bail out. */
    if (entry->entry_cxt)
        return;
 
    relation = RelationIdGetRelation(entry->publish_as_relid);
 
    entry->entry_cxt = AllocSetContextCreate(data->cachectx,
                                             "entry private context",
                                             ALLOCSET_SMALL_SIZES);
 
    MemoryContextCopyAndSetIdentifier(entry->entry_cxt,
                                      RelationGetRelationName(relation));
}
 
/*
 * Initialize the row filter.
 */
static void
pgoutput_row_filter_init(PGOutputData *data, List *publications,
                         RelationSyncEntry *entry)
{
    ListCell   *lc;
    List       *rfnodes[] = {NIL, NIL, NIL};    /* One per pubaction */
    bool        no_filter[] = {false, false, false};    /* One per pubaction */
    MemoryContext oldctx;
    int         idx;
    bool        has_filter = true;
    Oid         schemaid = get_rel_namespace(entry->publish_as_relid);
 
    /*
     * Find if there are any row filters for this relation. If there are, then
     * prepare the necessary ExprState and cache it in entry->exprstate. To
     * build an expression state, we need to ensure the following:
     *
     * All the given publication-table mappings must be checked.
     *
     * Multiple publications might have multiple row filters for this
     * relation. Since row filter usage depends on the DML operation, there
     * are multiple lists (one for each operation) to which row filters will
     * be appended.
     *
     * FOR ALL TABLES and FOR TABLES IN SCHEMA implies "don't use row filter
     * expression" so it takes precedence.
     */
    foreach(lc, publications)
    {
        Publication *pub = lfirst(lc);
        HeapTuple   rftuple = NULL;
        Datum       rfdatum = 0;
        bool        pub_no_filter = true;
 
        /*
         * If the publication is FOR ALL TABLES, or the publication includes a
         * FOR TABLES IN SCHEMA where the table belongs to the referred
         * schema, then it is treated the same as if there are no row filters
         * (even if other publications have a row filter).
         */
        if (!pub->alltables &&
            !SearchSysCacheExists2(PUBLICATIONNAMESPACEMAP,
                                   ObjectIdGetDatum(schemaid),
                                   ObjectIdGetDatum(pub->oid)))
        {
            /*
             * Check for the presence of a row filter in this publication.
             */
            rftuple = SearchSysCache2(PUBLICATIONRELMAP,
                                      ObjectIdGetDatum(entry->publish_as_relid),
                                      ObjectIdGetDatum(pub->oid));
 
            if (HeapTupleIsValid(rftuple))
            {
                /* Null indicates no filter. */
                rfdatum = SysCacheGetAttr(PUBLICATIONRELMAP, rftuple,
                                          Anum_pg_publication_rel_prqual,
                                          &pub_no_filter);
            }
        }
 
        if (pub_no_filter)
        {
            if (rftuple)
                ReleaseSysCache(rftuple);
 
            no_filter[PUBACTION_INSERT] |= pub->pubactions.pubinsert;
            no_filter[PUBACTION_UPDATE] |= pub->pubactions.pubupdate;
            no_filter[PUBACTION_DELETE] |= pub->pubactions.pubdelete;
 
            /*
             * Quick exit if all the DML actions are publicized via this
             * publication.
             */
            if (no_filter[PUBACTION_INSERT] &&
                no_filter[PUBACTION_UPDATE] &&
                no_filter[PUBACTION_DELETE])
            {
                has_filter = false;
                break;
            }
 
            /* No additional work for this publication. Next one. */
            continue;
        }
 
        /* Form the per pubaction row filter lists. */
        if (pub->pubactions.pubinsert && !no_filter[PUBACTION_INSERT])
            rfnodes[PUBACTION_INSERT] = lappend(rfnodes[PUBACTION_INSERT],
                                                TextDatumGetCString(rfdatum));
        if (pub->pubactions.pubupdate && !no_filter[PUBACTION_UPDATE])
            rfnodes[PUBACTION_UPDATE] = lappend(rfnodes[PUBACTION_UPDATE],
                                                TextDatumGetCString(rfdatum));
        if (pub->pubactions.pubdelete && !no_filter[PUBACTION_DELETE])
            rfnodes[PUBACTION_DELETE] = lappend(rfnodes[PUBACTION_DELETE],
                                                TextDatumGetCString(rfdatum));
 
        ReleaseSysCache(rftuple);
    }                           /* loop all subscribed publications */
 
    /* Clean the row filter */
    for (idx = 0; idx < NUM_ROWFILTER_PUBACTIONS; idx++)
    {
        if (no_filter[idx])
        {
            list_free_deep(rfnodes[idx]);
            rfnodes[idx] = NIL;
        }
    }
 
    if (has_filter)
    {
        Relation    relation = RelationIdGetRelation(entry->publish_as_relid);
 
        pgoutput_ensure_entry_cxt(data, entry);
 
        /*
         * Now all the filters for all pubactions are known. Combine them when
         * their pubactions are the same.
         */
        oldctx = MemoryContextSwitchTo(entry->entry_cxt);
        entry->estate = create_estate_for_relation(relation);
        for (idx = 0; idx < NUM_ROWFILTER_PUBACTIONS; idx++)
        {
            List       *filters = NIL;
            Expr       *rfnode;
 
            if (rfnodes[idx] == NIL)
                continue;
 
            foreach(lc, rfnodes[idx])
                filters = lappend(filters, expand_generated_columns_in_expr(stringToNode((char *) lfirst(lc)), relation, 1));
 
            /* combine the row filter and cache the ExprState */
            rfnode = make_orclause(filters);
            entry->exprstate[idx] = ExecPrepareExpr(rfnode, entry->estate);
        }                       /* for each pubaction */
        MemoryContextSwitchTo(oldctx);
 
        RelationClose(relation);
    }
}
 
/*
 * If the table contains a generated column, check for any conflicting
 * values of 'publish_generated_columns' parameter in the publications.
 */
static void
check_and_init_gencol(PGOutputData *data, List *publications,
                      RelationSyncEntry *entry)
{
    Relation    relation = RelationIdGetRelation(entry->publish_as_relid);
    TupleDesc   desc = RelationGetDescr(relation);
    bool        gencolpresent = false;
    bool        first = true;
 
    /* Check if there is any generated column present. */
    for (int i = 0; i < desc->natts; i++)
    {
        CompactAttribute *att = TupleDescCompactAttr(desc, i);
 
        if (att->attgenerated)
        {
            gencolpresent = true;
            break;
        }
    }
 
    /* There are no generated columns to be published. */
    if (!gencolpresent)
    {
        entry->include_gencols_type = PUBLISH_GENCOLS_NONE;
        return;
    }
 
    /*
     * There may be a conflicting value for 'publish_generated_columns'
     * parameter in the publications.
     */
    foreach_ptr(Publication, pub, publications)
    {
        /*
         * The column list takes precedence over the
         * 'publish_generated_columns' parameter. Those will be checked later,
         * see pgoutput_column_list_init.
         */
        if (check_and_fetch_column_list(pub, entry->publish_as_relid, NULL, NULL))
            continue;
 
        if (first)
        {
            entry->include_gencols_type = pub->pubgencols_type;
            first = false;
        }
        else if (entry->include_gencols_type != pub->pubgencols_type)
            ereport(ERROR,
                    errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                    errmsg("cannot use different values of publish_generated_columns for table \"%s.%s\" in different publications",
                           get_namespace_name(RelationGetNamespace(relation)),
                           RelationGetRelationName(relation)));
    }
}
 
/*
 * Initialize the column list.
 */
static void
pgoutput_column_list_init(PGOutputData *data, List *publications,
                          RelationSyncEntry *entry)
{
    ListCell   *lc;
    bool        first = true;
    Relation    relation = RelationIdGetRelation(entry->publish_as_relid);
    bool        found_pub_collist = false;
    Bitmapset  *relcols = NULL;
 
    pgoutput_ensure_entry_cxt(data, entry);
 
    /*
     * Find if there are any column lists for this relation. If there are,
     * build a bitmap using the column lists.
     *
     * Multiple publications might have multiple column lists for this
     * relation.
     *
     * Note that we don't support the case where the column list is different
     * for the same table when combining publications. See comments atop
     * fetch_relation_list. But one can later change the publication so we
     * still need to check all the given publication-table mappings and report
     * an error if any publications have a different column list.
     */
    foreach(lc, publications)
    {
        Publication *pub = lfirst(lc);
        Bitmapset  *cols = NULL;
 
        /* Retrieve the bitmap of columns for a column list publication. */
        found_pub_collist |= check_and_fetch_column_list(pub,
                                                         entry->publish_as_relid,
                                                         entry->entry_cxt, &cols);
 
        /*
         * For non-column list publications — e.g. TABLE (without a column
         * list), ALL TABLES, or ALL TABLES IN SCHEMA, we consider all columns
         * of the table (including generated columns when
         * 'publish_generated_columns' parameter is true).
         */
        if (!cols)
        {
            /*
             * Cache the table columns for the first publication with no
             * specified column list to detect publication with a different
             * column list.
             */
            if (!relcols && (list_length(publications) > 1))
            {
                MemoryContext oldcxt = MemoryContextSwitchTo(entry->entry_cxt);
 
                relcols = pub_form_cols_map(relation,
                                            entry->include_gencols_type);
                MemoryContextSwitchTo(oldcxt);
            }
 
            cols = relcols;
        }
 
        if (first)
        {
            entry->columns = cols;
            first = false;
        }
        else if (!bms_equal(entry->columns, cols))
            ereport(ERROR,
                    errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                    errmsg("cannot use different column lists for table \"%s.%s\" in different publications",
                           get_namespace_name(RelationGetNamespace(relation)),
                           RelationGetRelationName(relation)));
    }                           /* loop all subscribed publications */
 
    /*
     * If no column list publications exist, columns to be published will be
     * computed later according to the 'publish_generated_columns' parameter.
     */
    if (!found_pub_collist)
        entry->columns = NULL;
 
    RelationClose(relation);
}
 
/*
 * Initialize the slot for storing new and old tuples, and build the map that
 * will be used to convert the relation's tuples into the ancestor's format.
 */
static void
init_tuple_slot(PGOutputData *data, Relation relation,
                RelationSyncEntry *entry)
{
    MemoryContext oldctx;
    TupleDesc   oldtupdesc;
    TupleDesc   newtupdesc;
 
    oldctx = MemoryContextSwitchTo(data->cachectx);
 
    /*
     * Create tuple table slots. Create a copy of the TupleDesc as it needs to
     * live as long as the cache remains.
     */
    oldtupdesc = CreateTupleDescCopyConstr(RelationGetDescr(relation));
    newtupdesc = CreateTupleDescCopyConstr(RelationGetDescr(relation));
 
    entry->old_slot = MakeSingleTupleTableSlot(oldtupdesc, &TTSOpsHeapTuple);
    entry->new_slot = MakeSingleTupleTableSlot(newtupdesc, &TTSOpsHeapTuple);
 
    MemoryContextSwitchTo(oldctx);
 
    /*
     * Cache the map that will be used to convert the relation's tuples into
     * the ancestor's format, if needed.
     */
    if (entry->publish_as_relid != RelationGetRelid(relation))
    {
        Relation    ancestor = RelationIdGetRelation(entry->publish_as_relid);
        TupleDesc   indesc = RelationGetDescr(relation);
        TupleDesc   outdesc = RelationGetDescr(ancestor);
 
        /* Map must live as long as the logical decoding context. */
        oldctx = MemoryContextSwitchTo(data->cachectx);
 
        entry->attrmap = build_attrmap_by_name_if_req(indesc, outdesc, false);
 
        MemoryContextSwitchTo(oldctx);
        RelationClose(ancestor);
    }
}
 
/*
 * Change is checked against the row filter if any.
 *
 * Returns true if the change is to be replicated, else false.
 *
 * For inserts, evaluate the row filter for new tuple.
 * For deletes, evaluate the row filter for old tuple.
 * For updates, evaluate the row filter for old and new tuple.
 *
 * For updates, if both evaluations are true, we allow sending the UPDATE and
 * if both the evaluations are false, it doesn't replicate the UPDATE. Now, if
 * only one of the tuples matches the row filter expression, we transform
 * UPDATE to DELETE or INSERT to avoid any data inconsistency based on the
 * following rules:
 *
 * Case 1: old-row (no match)    new-row (no match)  -> (drop change)
 * Case 2: old-row (no match)    new row (match)     -> INSERT
 * Case 3: old-row (match)       new-row (no match)  -> DELETE
 * Case 4: old-row (match)       new row (match)     -> UPDATE
 *
 * The new action is updated in the action parameter.
 *
 * The new slot could be updated when transforming the UPDATE into INSERT,
 * because the original new tuple might not have column values from the replica
 * identity.
 *
 * Examples:
 * Let's say the old tuple satisfies the row filter but the new tuple doesn't.
 * Since the old tuple satisfies, the initial table synchronization copied this
 * row (or another method was used to guarantee that there is data
 * consistency).  However, after the UPDATE the new tuple doesn't satisfy the
 * row filter, so from a data consistency perspective, that row should be
 * removed on the subscriber. The UPDATE should be transformed into a DELETE
 * statement and be sent to the subscriber. Keeping this row on the subscriber
 * is undesirable because it doesn't reflect what was defined in the row filter
 * expression on the publisher. This row on the subscriber would likely not be
 * modified by replication again. If someone inserted a new row with the same
 * old identifier, replication could stop due to a constraint violation.
 *
 * Let's say the old tuple doesn't match the row filter but the new tuple does.
 * Since the old tuple doesn't satisfy, the initial table synchronization
 * probably didn't copy this row. However, after the UPDATE the new tuple does
 * satisfy the row filter, so from a data consistency perspective, that row
 * should be inserted on the subscriber. Otherwise, subsequent UPDATE or DELETE
 * statements have no effect (it matches no row -- see
 * apply_handle_update_internal()). So, the UPDATE should be transformed into a
 * INSERT statement and be sent to the subscriber. However, this might surprise
 * someone who expects the data set to satisfy the row filter expression on the
 * provider.
 */
static bool
pgoutput_row_filter(Relation relation, TupleTableSlot *old_slot,
                    TupleTableSlot **new_slot_ptr, RelationSyncEntry *entry,
                    ReorderBufferChangeType *action)
{
    TupleDesc   desc;
    int         i;
    bool        old_matched,
                new_matched,
                result;
    TupleTableSlot *tmp_new_slot;
    TupleTableSlot *new_slot = *new_slot_ptr;
    ExprContext *ecxt;
    ExprState  *filter_exprstate;
 
    /*
     * We need this map to avoid relying on ReorderBufferChangeType enums
     * having specific values.
     */
    static const int map_changetype_pubaction[] = {
        [REORDER_BUFFER_CHANGE_INSERT] = PUBACTION_INSERT,
        [REORDER_BUFFER_CHANGE_UPDATE] = PUBACTION_UPDATE,
        [REORDER_BUFFER_CHANGE_DELETE] = PUBACTION_DELETE
    };
 
    Assert(*action == REORDER_BUFFER_CHANGE_INSERT ||
           *action == REORDER_BUFFER_CHANGE_UPDATE ||
           *action == REORDER_BUFFER_CHANGE_DELETE);
 
    Assert(new_slot || old_slot);
 
    /* Get the corresponding row filter */
    filter_exprstate = entry->exprstate[map_changetype_pubaction[*action]];
 
    /* Bail out if there is no row filter */
    if (!filter_exprstate)
        return true;
 
    elog(DEBUG3, "table \"%s.%s\" has row filter",
         get_namespace_name(RelationGetNamespace(relation)),
         RelationGetRelationName(relation));
 
    ResetPerTupleExprContext(entry->estate);
 
    ecxt = GetPerTupleExprContext(entry->estate);
 
    /*
     * For the following occasions where there is only one tuple, we can
     * evaluate the row filter for that tuple and return.
     *
     * For inserts, we only have the new tuple.
     *
     * For updates, we can have only a new tuple when none of the replica
     * identity columns changed and none of those columns have external data
     * but we still need to evaluate the row filter for the new tuple as the
     * existing values of those columns might not match the filter. Also,
     * users can use constant expressions in the row filter, so we anyway need
     * to evaluate it for the new tuple.
     *
     * For deletes, we only have the old tuple.
     */
    if (!new_slot || !old_slot)
    {
        ecxt->ecxt_scantuple = new_slot ? new_slot : old_slot;
        result = pgoutput_row_filter_exec_expr(filter_exprstate, ecxt);
 
        return result;
    }
 
    /*
     * Both the old and new tuples must be valid only for updates and need to
     * be checked against the row filter.
     */
    Assert(map_changetype_pubaction[*action] == PUBACTION_UPDATE);
 
    slot_getallattrs(new_slot);
    slot_getallattrs(old_slot);
 
    tmp_new_slot = NULL;
    desc = RelationGetDescr(relation);
 
    /*
     * The new tuple might not have all the replica identity columns, in which
     * case it needs to be copied over from the old tuple.
     */
    for (i = 0; i < desc->natts; i++)
    {
        CompactAttribute *att = TupleDescCompactAttr(desc, i);
 
        /*
         * if the column in the new tuple or old tuple is null, nothing to do
         */
        if (new_slot->tts_isnull[i] || old_slot->tts_isnull[i])
            continue;
 
        /*
         * Unchanged toasted replica identity columns are only logged in the
         * old tuple. Copy this over to the new tuple. The changed (or WAL
         * Logged) toast values are always assembled in memory and set as
         * VARTAG_INDIRECT. See ReorderBufferToastReplace.
         */
        if (att->attlen == -1 &&
            VARATT_IS_EXTERNAL_ONDISK(DatumGetPointer(new_slot->tts_values[i])) &&
            !VARATT_IS_EXTERNAL_ONDISK(DatumGetPointer(old_slot->tts_values[i])))
        {
            if (!tmp_new_slot)
            {
                tmp_new_slot = MakeSingleTupleTableSlot(desc, &TTSOpsVirtual);
                ExecClearTuple(tmp_new_slot);
 
                memcpy(tmp_new_slot->tts_values, new_slot->tts_values,
                       desc->natts * sizeof(Datum));
                memcpy(tmp_new_slot->tts_isnull, new_slot->tts_isnull,
                       desc->natts * sizeof(bool));
            }
 
            tmp_new_slot->tts_values[i] = old_slot->tts_values[i];
            tmp_new_slot->tts_isnull[i] = old_slot->tts_isnull[i];
        }
    }
 
    ecxt->ecxt_scantuple = old_slot;
    old_matched = pgoutput_row_filter_exec_expr(filter_exprstate, ecxt);
 
    if (tmp_new_slot)
    {
        ExecStoreVirtualTuple(tmp_new_slot);
        ecxt->ecxt_scantuple = tmp_new_slot;
    }
    else
        ecxt->ecxt_scantuple = new_slot;
 
    new_matched = pgoutput_row_filter_exec_expr(filter_exprstate, ecxt);
 
    /*
     * Case 1: if both tuples don't match the row filter, bailout. Send
     * nothing.
     */
    if (!old_matched && !new_matched)
        return false;
 
    /*
     * Case 2: if the old tuple doesn't satisfy the row filter but the new
     * tuple does, transform the UPDATE into INSERT.
     *
     * Use the newly transformed tuple that must contain the column values for
     * all the replica identity columns. This is required to ensure that the
     * while inserting the tuple in the downstream node, we have all the
     * required column values.
     */
    if (!old_matched && new_matched)
    {
        *action = REORDER_BUFFER_CHANGE_INSERT;
 
        if (tmp_new_slot)
            *new_slot_ptr = tmp_new_slot;
    }
 
    /*
     * Case 3: if the old tuple satisfies the row filter but the new tuple
     * doesn't, transform the UPDATE into DELETE.
     *
     * This transformation does not require another tuple. The Old tuple will
     * be used for DELETE.
     */
    else if (old_matched && !new_matched)
        *action = REORDER_BUFFER_CHANGE_DELETE;
 
    /*
     * Case 4: if both tuples match the row filter, transformation isn't
     * required. (*action is default UPDATE).
     */
 
    return true;
}
 
/*
 * Sends the decoded DML over wire.
 *
 * This is called both in streaming and non-streaming modes.
 */
static void
pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
                Relation relation, ReorderBufferChange *change)
{
    PGOutputData *data = (PGOutputData *) ctx->output_plugin_private;
    PGOutputTxnData *txndata = (PGOutputTxnData *) txn->output_plugin_private;
    MemoryContext old;
    RelationSyncEntry *relentry;
    TransactionId xid = InvalidTransactionId;
    Relation    ancestor = NULL;
    Relation    targetrel = relation;
    ReorderBufferChangeType action = change->action;
    TupleTableSlot *old_slot = NULL;
    TupleTableSlot *new_slot = NULL;
 
    if (!is_publishable_relation(relation))
        return;
 
    /*
     * Remember the xid for the change in streaming mode. We need to send xid
     * with each change in the streaming mode so that subscriber can make
     * their association and on aborts, it can discard the corresponding
     * changes.
     */
    if (data->in_streaming)
        xid = change->txn->xid;
 
    relentry = get_rel_sync_entry(data, relation);
 
    /* First check the table filter */
    switch (action)
    {
        case REORDER_BUFFER_CHANGE_INSERT:
            if (!relentry->pubactions.pubinsert)
                return;
            break;
        case REORDER_BUFFER_CHANGE_UPDATE:
            if (!relentry->pubactions.pubupdate)
                return;
            break;
        case REORDER_BUFFER_CHANGE_DELETE:
            if (!relentry->pubactions.pubdelete)
                return;
 
            /*
             * This is only possible if deletes are allowed even when replica
             * identity is not defined for a table. Since the DELETE action
             * can't be published, we simply return.
             */
            if (!change->data.tp.oldtuple)
            {
                elog(DEBUG1, "didn't send DELETE change because of missing oldtuple");
                return;
            }
            break;
        default:
            Assert(false);
    }
 
    /* Avoid leaking memory by using and resetting our own context */
    old = MemoryContextSwitchTo(data->context);
 
    /* Switch relation if publishing via root. */
    if (relentry->publish_as_relid != RelationGetRelid(relation))
    {
        Assert(relation->rd_rel->relispartition);
        ancestor = RelationIdGetRelation(relentry->publish_as_relid);
        targetrel = ancestor;
    }
 
    if (change->data.tp.oldtuple)
    {
        old_slot = relentry->old_slot;
        ExecStoreHeapTuple(change->data.tp.oldtuple, old_slot, false);
 
        /* Convert tuple if needed. */
        if (relentry->attrmap)
        {
            TupleTableSlot *slot = MakeTupleTableSlot(RelationGetDescr(targetrel),
                                                      &TTSOpsVirtual);
 
            old_slot = execute_attr_map_slot(relentry->attrmap, old_slot, slot);
        }
    }
 
    if (change->data.tp.newtuple)
    {
        new_slot = relentry->new_slot;
        ExecStoreHeapTuple(change->data.tp.newtuple, new_slot, false);
 
        /* Convert tuple if needed. */
        if (relentry->attrmap)
        {
            TupleTableSlot *slot = MakeTupleTableSlot(RelationGetDescr(targetrel),
                                                      &TTSOpsVirtual);
 
            new_slot = execute_attr_map_slot(relentry->attrmap, new_slot, slot);
        }
    }
 
    /*
     * Check row filter.
     *
     * Updates could be transformed to inserts or deletes based on the results
     * of the row filter for old and new tuple.
     */
    if (!pgoutput_row_filter(targetrel, old_slot, &new_slot, relentry, &action))
        goto cleanup;
 
    /*
     * Send BEGIN if we haven't yet.
     *
     * We send the BEGIN message after ensuring that we will actually send the
     * change. This avoids sending a pair of BEGIN/COMMIT messages for empty
     * transactions.
     */
    if (txndata && !txndata->sent_begin_txn)
        pgoutput_send_begin(ctx, txn);
 
    /*
     * Schema should be sent using the original relation because it also sends
     * the ancestor's relation.
     */
    maybe_send_schema(ctx, change, relation, relentry);
 
    OutputPluginPrepareWrite(ctx, true);
 
    /* Send the data */
    switch (action)
    {
        case REORDER_BUFFER_CHANGE_INSERT:
            logicalrep_write_insert(ctx->out, xid, targetrel, new_slot,
                                    data->binary, relentry->columns,
                                    relentry->include_gencols_type);
            break;
        case REORDER_BUFFER_CHANGE_UPDATE:
            logicalrep_write_update(ctx->out, xid, targetrel, old_slot,
                                    new_slot, data->binary, relentry->columns,
                                    relentry->include_gencols_type);
            break;
        case REORDER_BUFFER_CHANGE_DELETE:
            logicalrep_write_delete(ctx->out, xid, targetrel, old_slot,
                                    data->binary, relentry->columns,
                                    relentry->include_gencols_type);
            break;
        default:
            Assert(false);
    }
 
    OutputPluginWrite(ctx, true);
 
cleanup:
    if (RelationIsValid(ancestor))
    {
        RelationClose(ancestor);
        ancestor = NULL;
    }
 
    /* Drop the new slots that were used to store the converted tuples. */
    if (relentry->attrmap)
    {
        if (old_slot)
            ExecDropSingleTupleTableSlot(old_slot);
 
        if (new_slot)
            ExecDropSingleTupleTableSlot(new_slot);
    }
 
    MemoryContextSwitchTo(old);
    MemoryContextReset(data->context);
}
 
static void
pgoutput_truncate(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
                  int nrelations, Relation relations[], ReorderBufferChange *change)
{
    PGOutputData *data = (PGOutputData *) ctx->output_plugin_private;
    PGOutputTxnData *txndata = (PGOutputTxnData *) txn->output_plugin_private;
    MemoryContext old;
    RelationSyncEntry *relentry;
    int         i;
    int         nrelids;
    Oid        *relids;
    TransactionId xid = InvalidTransactionId;
 
    /* Remember the xid for the change in streaming mode. See pgoutput_change. */
    if (data->in_streaming)
        xid = change->txn->xid;
 
    old = MemoryContextSwitchTo(data->context);
 
    relids = palloc0(nrelations * sizeof(Oid));
    nrelids = 0;
 
    for (i = 0; i < nrelations; i++)
    {
        Relation    relation = relations[i];
        Oid         relid = RelationGetRelid(relation);
 
        if (!is_publishable_relation(relation))
            continue;
 
        relentry = get_rel_sync_entry(data, relation);
 
        if (!relentry->pubactions.pubtruncate)
            continue;
 
        /*
         * Don't send partitions if the publication wants to send only the
         * root tables through it.
         */
        if (relation->rd_rel->relispartition &&
            relentry->publish_as_relid != relid)
            continue;
 
        relids[nrelids++] = relid;
 
        /* Send BEGIN if we haven't yet */
        if (txndata && !txndata->sent_begin_txn)
            pgoutput_send_begin(ctx, txn);
 
        maybe_send_schema(ctx, change, relation, relentry);
    }
 
    if (nrelids > 0)
    {
        OutputPluginPrepareWrite(ctx, true);
        logicalrep_write_truncate(ctx->out,
                                  xid,
                                  nrelids,
                                  relids,
                                  change->data.truncate.cascade,
                                  change->data.truncate.restart_seqs);
        OutputPluginWrite(ctx, true);
    }
 
    MemoryContextSwitchTo(old);
    MemoryContextReset(data->context);
}
 
static void
pgoutput_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
                 XLogRecPtr message_lsn, bool transactional, const char *prefix, Size sz,
                 const char *message)
{
    PGOutputData *data = (PGOutputData *) ctx->output_plugin_private;
    TransactionId xid = InvalidTransactionId;
 
    if (!data->messages)
        return;
 
    /*
     * Remember the xid for the message in streaming mode. See
     * pgoutput_change.
     */
    if (data->in_streaming)
        xid = txn->xid;
 
    /*
     * Output BEGIN if we haven't yet. Avoid for non-transactional messages.
     */
    if (transactional)
    {
        PGOutputTxnData *txndata = (PGOutputTxnData *) txn->output_plugin_private;
 
        /* Send BEGIN if we haven't yet */
        if (txndata && !txndata->sent_begin_txn)
            pgoutput_send_begin(ctx, txn);
    }
 
    OutputPluginPrepareWrite(ctx, true);
    logicalrep_write_message(ctx->out,
                             xid,
                             message_lsn,
                             transactional,
                             prefix,
                             sz,
                             message);
    OutputPluginWrite(ctx, true);
}
 
/*
 * Return true if the data is associated with an origin and the user has
 * requested the changes that don't have an origin, false otherwise.
 */
static bool
pgoutput_origin_filter(LogicalDecodingContext *ctx,
                       RepOriginId origin_id)
{
    PGOutputData *data = (PGOutputData *) ctx->output_plugin_private;
 
    if (data->publish_no_origin && origin_id != InvalidRepOriginId)
        return true;
 
    return false;
}
 
/*
 * Shutdown the output plugin.
 *
 * Note, we don't need to clean the data->context, data->cachectx, and
 * data->pubctx as they are child contexts of the ctx->context so they
 * will be cleaned up by logical decoding machinery.
 */
static void
pgoutput_shutdown(LogicalDecodingContext *ctx)
{
    pgoutput_memory_context_reset(NULL);
}
 
/*
 * Load publications from the list of publication names.
 *
 * Here, we skip the publications that don't exist yet. This will allow us
 * to silently continue the replication in the absence of a missing publication.
 * This is required because we allow the users to create publications after they
 * have specified the required publications at the time of replication start.
 */
static List *
LoadPublications(List *pubnames)
{
    List       *result = NIL;
    ListCell   *lc;
 
    foreach(lc, pubnames)
    {
        char       *pubname = (char *) lfirst(lc);
        Publication *pub = GetPublicationByName(pubname, true);
 
        if (pub)
            result = lappend(result, pub);
        else
            ereport(WARNING,
                    errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                    errmsg("skipped loading publication \"%s\"", pubname),
                    errdetail("The publication does not exist at this point in the WAL."),
                    errhint("Create the publication if it does not exist."));
    }
 
    return result;
}
 
/*
 * Publication syscache invalidation callback.
 *
 * Called for invalidations on pg_publication.
 */
static void
publication_invalidation_cb(Datum arg, int cacheid, uint32 hashvalue)
{
    publications_valid = false;
}
 
/*
 * START STREAM callback
 */
static void
pgoutput_stream_start(struct LogicalDecodingContext *ctx,
                      ReorderBufferTXN *txn)
{
    PGOutputData *data = (PGOutputData *) ctx->output_plugin_private;
    bool        send_replication_origin = txn->origin_id != InvalidRepOriginId;
 
    /* we can't nest streaming of transactions */
    Assert(!data->in_streaming);
 
    /*
     * If we already sent the first stream for this transaction then don't
     * send the origin id in the subsequent streams.
     */
    if (rbtxn_is_streamed(txn))
        send_replication_origin = false;
 
    OutputPluginPrepareWrite(ctx, !send_replication_origin);
    logicalrep_write_stream_start(ctx->out, txn->xid, !rbtxn_is_streamed(txn));
 
    send_repl_origin(ctx, txn->origin_id, InvalidXLogRecPtr,
                     send_replication_origin);
 
    OutputPluginWrite(ctx, true);
 
    /* we're streaming a chunk of transaction now */
    data->in_streaming = true;
}
 
/*
 * STOP STREAM callback
 */
static void
pgoutput_stream_stop(struct LogicalDecodingContext *ctx,
                     ReorderBufferTXN *txn)
{
    PGOutputData *data = (PGOutputData *) ctx->output_plugin_private;
 
    /* we should be streaming a transaction */
    Assert(data->in_streaming);
 
    OutputPluginPrepareWrite(ctx, true);
    logicalrep_write_stream_stop(ctx->out);
    OutputPluginWrite(ctx, true);
 
    /* we've stopped streaming a transaction */
    data->in_streaming = false;
}
 
/*
 * Notify downstream to discard the streamed transaction (along with all
 * its subtransactions, if it's a toplevel transaction).
 */
static void
pgoutput_stream_abort(struct LogicalDecodingContext *ctx,
                      ReorderBufferTXN *txn,
                      XLogRecPtr abort_lsn)
{
    ReorderBufferTXN *toptxn;
    PGOutputData *data = (PGOutputData *) ctx->output_plugin_private;
    bool        write_abort_info = (data->streaming == LOGICALREP_STREAM_PARALLEL);
 
    /*
     * The abort should happen outside streaming block, even for streamed
     * transactions. The transaction has to be marked as streamed, though.
     */
    Assert(!data->in_streaming);
 
    /* determine the toplevel transaction */
    toptxn = rbtxn_get_toptxn(txn);
 
    Assert(rbtxn_is_streamed(toptxn));
 
    OutputPluginPrepareWrite(ctx, true);
    logicalrep_write_stream_abort(ctx->out, toptxn->xid, txn->xid, abort_lsn,
                                  txn->abort_time, write_abort_info);
 
    OutputPluginWrite(ctx, true);
 
    cleanup_rel_sync_cache(toptxn->xid, false);
}
 
/*
 * Notify downstream to apply the streamed transaction (along with all
 * its subtransactions).
 */
static void
pgoutput_stream_commit(struct LogicalDecodingContext *ctx,
                       ReorderBufferTXN *txn,
                       XLogRecPtr commit_lsn)
{
    PGOutputData *data PG_USED_FOR_ASSERTS_ONLY = (PGOutputData *) ctx->output_plugin_private;
 
    /*
     * The commit should happen outside streaming block, even for streamed
     * transactions. The transaction has to be marked as streamed, though.
     */
    Assert(!data->in_streaming);
    Assert(rbtxn_is_streamed(txn));
 
    OutputPluginUpdateProgress(ctx, false);
 
    OutputPluginPrepareWrite(ctx, true);
    logicalrep_write_stream_commit(ctx->out, txn, commit_lsn);
    OutputPluginWrite(ctx, true);
 
    cleanup_rel_sync_cache(txn->xid, true);
}
 
/*
 * PREPARE callback (for streaming two-phase commit).
 *
 * Notify the downstream to prepare the transaction.
 */
static void
pgoutput_stream_prepare_txn(LogicalDecodingContext *ctx,
                            ReorderBufferTXN *txn,
                            XLogRecPtr prepare_lsn)
{
    Assert(rbtxn_is_streamed(txn));
 
    OutputPluginUpdateProgress(ctx, false);
    OutputPluginPrepareWrite(ctx, true);
    logicalrep_write_stream_prepare(ctx->out, txn, prepare_lsn);
    OutputPluginWrite(ctx, true);
}
 
/*
 * Initialize the relation schema sync cache for a decoding session.
 *
 * The hash table is destroyed at the end of a decoding session. While
 * relcache invalidations still exist and will still be invoked, they
 * will just see the null hash table global and take no action.
 */
static void
init_rel_sync_cache(MemoryContext cachectx)
{
    HASHCTL     ctl;
    static bool relation_callbacks_registered = false;
 
    /* Nothing to do if hash table already exists */
    if (RelationSyncCache != NULL)
        return;
 
    /* Make a new hash table for the cache */
    ctl.keysize = sizeof(Oid);
    ctl.entrysize = sizeof(RelationSyncEntry);
    ctl.hcxt = cachectx;
 
    RelationSyncCache = hash_create("logical replication output relation cache",
                                    128, &ctl,
                                    HASH_ELEM | HASH_CONTEXT | HASH_BLOBS);
 
    Assert(RelationSyncCache != NULL);
 
    /* No more to do if we already registered callbacks */
    if (relation_callbacks_registered)
        return;
 
    /* We must update the cache entry for a relation after a relcache flush */
    CacheRegisterRelcacheCallback(rel_sync_cache_relation_cb, (Datum) 0);
 
    /*
     * Flush all cache entries after a pg_namespace change, in case it was a
     * schema rename affecting a relation being replicated.
     *
     * XXX: It is not a good idea to invalidate all the relation entries in
     * RelationSyncCache on schema rename. We can optimize it to invalidate
     * only the required relations by either having a specific invalidation
     * message containing impacted relations or by having schema information
     * in each RelationSyncCache entry and using hashvalue of pg_namespace.oid
     * passed to the callback.
     */
    CacheRegisterSyscacheCallback(NAMESPACEOID,
                                  rel_sync_cache_publication_cb,
                                  (Datum) 0);
 
    relation_callbacks_registered = true;
}
 
/*
 * We expect relatively small number of streamed transactions.
 */
static bool
get_schema_sent_in_streamed_txn(RelationSyncEntry *entry, TransactionId xid)
{
    return list_member_xid(entry->streamed_txns, xid);
}
 
/*
 * Add the xid in the rel sync entry for which we have already sent the schema
 * of the relation.
 */
static void
set_schema_sent_in_streamed_txn(RelationSyncEntry *entry, TransactionId xid)
{
    MemoryContext oldctx;
 
    oldctx = MemoryContextSwitchTo(CacheMemoryContext);
 
    entry->streamed_txns = lappend_xid(entry->streamed_txns, xid);
 
    MemoryContextSwitchTo(oldctx);
}
 
/*
 * Find or create entry in the relation schema cache.
 *
 * This looks up publications that the given relation is directly or
 * indirectly part of (the latter if it's really the relation's ancestor that
 * is part of a publication) and fills up the found entry with the information
 * about which operations to publish and whether to use an ancestor's schema
 * when publishing.
 */
static RelationSyncEntry *
get_rel_sync_entry(PGOutputData *data, Relation relation)
{
    RelationSyncEntry *entry;
    bool        found;
    MemoryContext oldctx;
    Oid         relid = RelationGetRelid(relation);
 
    Assert(RelationSyncCache != NULL);
 
    /* Find cached relation info, creating if not found */
    entry = (RelationSyncEntry *) hash_search(RelationSyncCache,
                                              &relid,
                                              HASH_ENTER, &found);
    Assert(entry != NULL);
 
    /* initialize entry, if it's new */
    if (!found)
    {
        entry->replicate_valid = false;
        entry->schema_sent = false;
        entry->include_gencols_type = PUBLISH_GENCOLS_NONE;
        entry->streamed_txns = NIL;
        entry->pubactions.pubinsert = entry->pubactions.pubupdate =
            entry->pubactions.pubdelete = entry->pubactions.pubtruncate = false;
        entry->new_slot = NULL;
        entry->old_slot = NULL;
        memset(entry->exprstate, 0, sizeof(entry->exprstate));
        entry->entry_cxt = NULL;
        entry->publish_as_relid = InvalidOid;
        entry->columns = NULL;
        entry->attrmap = NULL;
    }
 
    /* Validate the entry */
    if (!entry->replicate_valid)
    {
        Oid         schemaId = get_rel_namespace(relid);
        List       *pubids = GetRelationPublications(relid);
 
        /*
         * We don't acquire a lock on the namespace system table as we build
         * the cache entry using a historic snapshot and all the later changes
         * are absorbed while decoding WAL.
         */
        List       *schemaPubids = GetSchemaPublications(schemaId);
        ListCell   *lc;
        Oid         publish_as_relid = relid;
        int         publish_ancestor_level = 0;
        bool        am_partition = get_rel_relispartition(relid);
        char        relkind = get_rel_relkind(relid);
        List       *rel_publications = NIL;
 
        /* Reload publications if needed before use. */
        if (!publications_valid)
        {
            MemoryContextReset(data->pubctx);
 
            oldctx = MemoryContextSwitchTo(data->pubctx);
            data->publications = LoadPublications(data->publication_names);
            MemoryContextSwitchTo(oldctx);
            publications_valid = true;
        }
 
        /*
         * Reset schema_sent status as the relation definition may have
         * changed.  Also reset pubactions to empty in case rel was dropped
         * from a publication.  Also free any objects that depended on the
         * earlier definition.
         */
        entry->schema_sent = false;
        entry->include_gencols_type = PUBLISH_GENCOLS_NONE;
        list_free(entry->streamed_txns);
        entry->streamed_txns = NIL;
        bms_free(entry->columns);
        entry->columns = NULL;
        entry->pubactions.pubinsert = false;
        entry->pubactions.pubupdate = false;
        entry->pubactions.pubdelete = false;
        entry->pubactions.pubtruncate = false;
 
        /*
         * Tuple slots cleanups. (Will be rebuilt later if needed).
         */
        if (entry->old_slot)
        {
            TupleDesc   desc = entry->old_slot->tts_tupleDescriptor;
 
            Assert(desc->tdrefcount == -1);
 
            ExecDropSingleTupleTableSlot(entry->old_slot);
 
            /*
             * ExecDropSingleTupleTableSlot() would not free the TupleDesc, so
             * do it now to avoid any leaks.
             */
            FreeTupleDesc(desc);
        }
        if (entry->new_slot)
        {
            TupleDesc   desc = entry->new_slot->tts_tupleDescriptor;
 
            Assert(desc->tdrefcount == -1);
 
            ExecDropSingleTupleTableSlot(entry->new_slot);
 
            /*
             * ExecDropSingleTupleTableSlot() would not free the TupleDesc, so
             * do it now to avoid any leaks.
             */
            FreeTupleDesc(desc);
        }
 
        entry->old_slot = NULL;
        entry->new_slot = NULL;
 
        if (entry->attrmap)
            free_attrmap(entry->attrmap);
        entry->attrmap = NULL;
 
        /*
         * Row filter cache cleanups.
         */
        if (entry->entry_cxt)
            MemoryContextDelete(entry->entry_cxt);
 
        entry->entry_cxt = NULL;
        entry->estate = NULL;
        memset(entry->exprstate, 0, sizeof(entry->exprstate));
 
        /*
         * Build publication cache. We can't use one provided by relcache as
         * relcache considers all publications that the given relation is in,
         * but here we only need to consider ones that the subscriber
         * requested.
         */
        foreach(lc, data->publications)
        {
            Publication *pub = lfirst(lc);
            bool        publish = false;
 
            /*
             * Under what relid should we publish changes in this publication?
             * We'll use the top-most relid across all publications. Also
             * track the ancestor level for this publication.
             */
            Oid         pub_relid = relid;
            int         ancestor_level = 0;
 
            /*
             * If this is a FOR ALL TABLES publication, pick the partition
             * root and set the ancestor level accordingly.
             */
            if (pub->alltables)
            {
                publish = true;
                if (pub->pubviaroot && am_partition)
                {
                    List       *ancestors = get_partition_ancestors(relid);
 
                    pub_relid = llast_oid(ancestors);
                    ancestor_level = list_length(ancestors);
                }
            }
 
            if (!publish)
            {
                bool        ancestor_published = false;
 
                /*
                 * For a partition, check if any of the ancestors are
                 * published.  If so, note down the topmost ancestor that is
                 * published via this publication, which will be used as the
                 * relation via which to publish the partition's changes.
                 */
                if (am_partition)
                {
                    Oid         ancestor;
                    int         level;
                    List       *ancestors = get_partition_ancestors(relid);
 
                    ancestor = GetTopMostAncestorInPublication(pub->oid,
                                                               ancestors,
                                                               &level);
 
                    if (ancestor != InvalidOid)
                    {
                        ancestor_published = true;
                        if (pub->pubviaroot)
                        {
                            pub_relid = ancestor;
                            ancestor_level = level;
                        }
                    }
                }
 
                if (list_member_oid(pubids, pub->oid) ||
                    list_member_oid(schemaPubids, pub->oid) ||
                    ancestor_published)
                    publish = true;
            }
 
            /*
             * If the relation is to be published, determine actions to
             * publish, and list of columns, if appropriate.
             *
             * Don't publish changes for partitioned tables, because
             * publishing those of its partitions suffices, unless partition
             * changes won't be published due to pubviaroot being set.
             */
            if (publish &&
                (relkind != RELKIND_PARTITIONED_TABLE || pub->pubviaroot))
            {
                entry->pubactions.pubinsert |= pub->pubactions.pubinsert;
                entry->pubactions.pubupdate |= pub->pubactions.pubupdate;
                entry->pubactions.pubdelete |= pub->pubactions.pubdelete;
                entry->pubactions.pubtruncate |= pub->pubactions.pubtruncate;
 
                /*
                 * We want to publish the changes as the top-most ancestor
                 * across all publications. So we need to check if the already
                 * calculated level is higher than the new one. If yes, we can
                 * ignore the new value (as it's a child). Otherwise the new
                 * value is an ancestor, so we keep it.
                 */
                if (publish_ancestor_level > ancestor_level)
                    continue;
 
                /*
                 * If we found an ancestor higher up in the tree, discard the
                 * list of publications through which we replicate it, and use
                 * the new ancestor.
                 */
                if (publish_ancestor_level < ancestor_level)
                {
                    publish_as_relid = pub_relid;
                    publish_ancestor_level = ancestor_level;
 
                    /* reset the publication list for this relation */
                    rel_publications = NIL;
                }
                else
                {
                    /* Same ancestor level, has to be the same OID. */
                    Assert(publish_as_relid == pub_relid);
                }
 
                /* Track publications for this ancestor. */
                rel_publications = lappend(rel_publications, pub);
            }
        }
 
        entry->publish_as_relid = publish_as_relid;
 
        /*
         * Initialize the tuple slot, map, and row filter. These are only used
         * when publishing inserts, updates, or deletes.
         */
        if (entry->pubactions.pubinsert || entry->pubactions.pubupdate ||
            entry->pubactions.pubdelete)
        {
            /* Initialize the tuple slot and map */
            init_tuple_slot(data, relation, entry);
 
            /* Initialize the row filter */
            pgoutput_row_filter_init(data, rel_publications, entry);
 
            /* Check whether to publish generated columns. */
            check_and_init_gencol(data, rel_publications, entry);
 
            /* Initialize the column list */
            pgoutput_column_list_init(data, rel_publications, entry);
        }
 
        list_free(pubids);
        list_free(schemaPubids);
        list_free(rel_publications);
 
        entry->replicate_valid = true;
    }
 
    return entry;
}
 
/*
 * Cleanup list of streamed transactions and update the schema_sent flag.
 *
 * When a streamed transaction commits or aborts, we need to remove the
 * toplevel XID from the schema cache. If the transaction aborted, the
 * subscriber will simply throw away the schema records we streamed, so
 * we don't need to do anything else.
 *
 * If the transaction is committed, the subscriber will update the relation
 * cache - so tweak the schema_sent flag accordingly.
 */
static void
cleanup_rel_sync_cache(TransactionId xid, bool is_commit)
{
    HASH_SEQ_STATUS hash_seq;
    RelationSyncEntry *entry;
 
    Assert(RelationSyncCache != NULL);
 
    hash_seq_init(&hash_seq, RelationSyncCache);
    while ((entry = hash_seq_search(&hash_seq)) != NULL)
    {
        /*
         * We can set the schema_sent flag for an entry that has committed xid
         * in the list as that ensures that the subscriber would have the
         * corresponding schema and we don't need to send it unless there is
         * any invalidation for that relation.
         */
        foreach_xid(streamed_txn, entry->streamed_txns)
        {
            if (xid == streamed_txn)
            {
                if (is_commit)
                    entry->schema_sent = true;
 
                entry->streamed_txns =
                    foreach_delete_current(entry->streamed_txns, streamed_txn);
                break;
            }
        }
    }
}
 
/*
 * Relcache invalidation callback
 */
static void
rel_sync_cache_relation_cb(Datum arg, Oid relid)
{
    RelationSyncEntry *entry;
 
    /*
     * We can get here if the plugin was used in SQL interface as the
     * RelationSyncCache is destroyed when the decoding finishes, but there is
     * no way to unregister the relcache invalidation callback.
     */
    if (RelationSyncCache == NULL)
        return;
 
    /*
     * Nobody keeps pointers to entries in this hash table around outside
     * logical decoding callback calls - but invalidation events can come in
     * *during* a callback if we do any syscache access in the callback.
     * Because of that we must mark the cache entry as invalid but not damage
     * any of its substructure here.  The next get_rel_sync_entry() call will
     * rebuild it all.
     */
    if (OidIsValid(relid))
    {
        /*
         * Getting invalidations for relations that aren't in the table is
         * entirely normal.  So we don't care if it's found or not.
         */
        entry = (RelationSyncEntry *) hash_search(RelationSyncCache, &relid,
                                                  HASH_FIND, NULL);
        if (entry != NULL)
            entry->replicate_valid = false;
    }
    else
    {
        /* Whole cache must be flushed. */
        HASH_SEQ_STATUS status;
 
        hash_seq_init(&status, RelationSyncCache);
        while ((entry = (RelationSyncEntry *) hash_seq_search(&status)) != NULL)
        {
            entry->replicate_valid = false;
        }
    }
}
 
/*
 * Publication relation/schema map syscache invalidation callback
 *
 * Called for invalidations on pg_namespace.
 */
static void
rel_sync_cache_publication_cb(Datum arg, int cacheid, uint32 hashvalue)
{
    HASH_SEQ_STATUS status;
    RelationSyncEntry *entry;
 
    /*
     * We can get here if the plugin was used in SQL interface as the
     * RelationSyncCache is destroyed when the decoding finishes, but there is
     * no way to unregister the invalidation callbacks.
     */
    if (RelationSyncCache == NULL)
        return;
 
    /*
     * We have no easy way to identify which cache entries this invalidation
     * event might have affected, so just mark them all invalid.
     */
    hash_seq_init(&status, RelationSyncCache);
    while ((entry = (RelationSyncEntry *) hash_seq_search(&status)) != NULL)
    {
        entry->replicate_valid = false;
    }
}
 
/* Send Replication origin */
static void
send_repl_origin(LogicalDecodingContext *ctx, RepOriginId origin_id,
                 XLogRecPtr origin_lsn, bool send_origin)
{
    if (send_origin)
    {
        char       *origin;
 
        /*----------
         * XXX: which behaviour do we want here?
         *
         * Alternatives:
         *  - don't send origin message if origin name not found
         *    (that's what we do now)
         *  - throw error - that will break replication, not good
         *  - send some special "unknown" origin
         *----------
         */
        if (replorigin_by_oid(origin_id, true, &origin))
        {
            /* Message boundary */
            OutputPluginWrite(ctx, false);
            OutputPluginPrepareWrite(ctx, true);
 
            logicalrep_write_origin(ctx->out, origin, origin_lsn);
        }
    }
}