/
githubmirror
/
postgres
Обзор
Документация
Войти
/
githubmirror
/
postgres
Код
Запросы
0
Пакеты
0
Релизы
0
Аналитика
Безопасность
master
src/backend/replication/pgoutput/pgoutput.c
2 522 строки
75 KB
Michael Paquier
Tighten proto_version parsing in pgoutput
06 авг 2026, 10:46
06 авг 2026, 10:46
290bb8a
Код
Авторство
О чём код?
/*------------------------------------------------------------------------- * * 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, ReplOriginId 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, SysCacheIdentifier cacheid, uint32 hashvalue); static void send_repl_origin(LogicalDecodingContext *ctx, ReplOriginId 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, SysCacheIdentifier 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; const char *val = strVal(defel->arg); if (protocol_version_given) ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), errmsg("conflicting or redundant options"))); protocol_version_given = true; errno = 0; parsed = strtoul(val, &endptr, 10); if (endptr == val || 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 != InvalidReplOriginId; 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 != InvalidReplOriginId; 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, 0); 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, 0); 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, ReplOriginId origin_id) { PGOutputData *data = (PGOutputData *) ctx->output_plugin_private; if (data->publish_no_origin && origin_id != InvalidReplOriginId) 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, SysCacheIdentifier 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 != InvalidReplOriginId; /* 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 = GetRelationIncludedPublications(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) { List *exceptpubids = NIL; if (am_partition) { List *ancestors = get_partition_ancestors(relid); Oid last_ancestor_relid = llast_oid(ancestors); /* * For a partition, changes are published via top-most * ancestor when pubviaroot is true, so populate pub_relid * accordingly. */ if (pub->pubviaroot) { pub_relid = last_ancestor_relid; ancestor_level = list_length(ancestors); } /* * Only the top-most ancestor can appear in the EXCEPT * clause. Therefore, for a partition, exclusion must be * evaluated at the top-most ancestor. */ exceptpubids = GetRelationExcludedPublications(last_ancestor_relid); } else { /* * For a regular table or a root partitioned table, check * exclusion on table itself. */ exceptpubids = GetRelationExcludedPublications(pub_relid); } if (!list_member_oid(exceptpubids, pub->oid)) publish = true; list_free(exceptpubids); if (!publish) continue; } 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, SysCacheIdentifier 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, ReplOriginId 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); } } }