55 bool transactional,
const char *prefix,
56 Size sz,
const char *message);
108 #define NUM_ROWFILTER_PUBACTIONS (PUBACTION_DELETE+1)
284 bool protocol_version_given =
false;
285 bool publication_names_given =
false;
286 bool binary_option_given =
false;
287 bool messages_option_given =
false;
288 bool streaming_given =
false;
289 bool two_phase_option_given =
false;
291 data->binary =
false;
292 data->streaming =
false;
293 data->messages =
false;
294 data->two_phase =
false;
303 if (strcmp(defel->
defname,
"proto_version") == 0)
305 unsigned long parsed;
308 if (protocol_version_given)
310 (
errcode(ERRCODE_SYNTAX_ERROR),
311 errmsg(
"conflicting or redundant options")));
312 protocol_version_given =
true;
315 parsed = strtoul(
strVal(defel->
arg), &endptr, 10);
316 if (errno != 0 || *endptr !=
'\0')
318 (
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
319 errmsg(
"invalid proto_version")));
323 (
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
324 errmsg(
"proto_version \"%s\" out of range",
329 else if (strcmp(defel->
defname,
"publication_names") == 0)
331 if (publication_names_given)
333 (
errcode(ERRCODE_SYNTAX_ERROR),
334 errmsg(
"conflicting or redundant options")));
335 publication_names_given =
true;
338 &
data->publication_names))
340 (
errcode(ERRCODE_INVALID_NAME),
341 errmsg(
"invalid publication_names syntax")));
343 else if (strcmp(defel->
defname,
"binary") == 0)
345 if (binary_option_given)
347 (
errcode(ERRCODE_SYNTAX_ERROR),
348 errmsg(
"conflicting or redundant options")));
349 binary_option_given =
true;
353 else if (strcmp(defel->
defname,
"messages") == 0)
355 if (messages_option_given)
357 (
errcode(ERRCODE_SYNTAX_ERROR),
358 errmsg(
"conflicting or redundant options")));
359 messages_option_given =
true;
363 else if (strcmp(defel->
defname,
"streaming") == 0)
367 (
errcode(ERRCODE_SYNTAX_ERROR),
368 errmsg(
"conflicting or redundant options")));
369 streaming_given =
true;
373 else if (strcmp(defel->
defname,
"two_phase") == 0)
375 if (two_phase_option_given)
377 (
errcode(ERRCODE_SYNTAX_ERROR),
378 errmsg(
"conflicting or redundant options")));
379 two_phase_option_given =
true;
399 "logical replication output context",
403 "logical replication cache context",
424 (
errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
425 errmsg(
"client sent proto_version=%d but we only support protocol %d or lower",
430 (
errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
431 errmsg(
"client sent proto_version=%d but we only support protocol %d or higher",
436 (
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
437 errmsg(
"publication_names parameter missing")));
445 if (!
data->streaming)
449 (
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
450 errmsg(
"requested proto_version=%d does not support streaming, need %d or higher",
454 (
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
455 errmsg(
"streaming requested, but not supported by output plugin")));
467 if (!
data->two_phase)
471 (
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
472 errmsg(
"requested proto_version=%d does not support two-phase commit, need %d or higher",
476 (
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
477 errmsg(
"two-phase commit requested, but not supported by output plugin")));
540 send_replication_origin);
568 elog(
DEBUG1,
"skipped replication of an empty transaction with XID: %u", txn->
xid);
589 send_replication_origin);
733 if (att->attisdropped || att->attgenerated)
793 elog(
DEBUG3,
"row filter evaluates to %s (isnull: %s)",
794 isnull ?
"false" :
DatumGetBool(ret) ?
"true" :
"false",
795 isnull ?
"true" :
"false");
818 "entry private context",
834 bool no_filter[] = {
false,
false,
false};
837 bool has_filter =
true;
854 foreach(lc, publications)
859 bool pub_no_filter =
false;
868 pub_no_filter =
true;
883 Anum_pg_publication_rel_prqual,
888 pub_no_filter =
true;
961 foreach(lc, rfnodes[
idx])
1001 foreach(lc, publications)
1015 bool pub_no_list =
true;
1032 Anum_pg_publication_rel_prattrs,
1065 errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1066 errmsg(
"cannot use different column lists for table \"%s.%s\" in different publications",
1189 static const int map_changetype_pubaction[] = {
1199 Assert(new_slot || old_slot);
1202 filter_exprstate = entry->
exprstate[map_changetype_pubaction[*
action]];
1205 if (!filter_exprstate)
1208 elog(
DEBUG3,
"table \"%s.%s\" has row filter",
1231 if (!new_slot || !old_slot)
1248 tmp_new_slot = NULL;
1271 if (att->attlen == -1 &&
1283 desc->
natts *
sizeof(
bool));
1308 if (!old_matched && !new_matched)
1320 if (!old_matched && new_matched)
1325 *new_slot_ptr = tmp_new_slot;
1335 else if (old_matched && !new_matched)
1417 targetrel = ancestor;
1441 if (txndata && !txndata->sent_begin_txn)
1456 if (change->
data.
tp.oldtuple)
1472 targetrel = ancestor;
1495 if (txndata && !txndata->sent_begin_txn)
1510 new_slot,
data->binary,
1515 old_slot, new_slot,
data->binary,
1520 old_slot,
data->binary);
1529 if (change->
data.
tp.oldtuple)
1541 targetrel = ancestor;
1559 if (txndata && !txndata->sent_begin_txn)
1566 old_slot,
data->binary);
1570 elog(
DEBUG1,
"didn't send DELETE change because of missing oldtuple");
1611 for (
i = 0;
i < nrelations;
i++)
1628 if (relation->
rd_rel->relispartition &&
1632 relids[nrelids++] = relid;
1635 if (txndata && !txndata->sent_begin_txn)
1659 XLogRecPtr message_lsn,
bool transactional,
const char *prefix,
Size sz,
1660 const char *message)
1667 if (!
data->messages)
1736 foreach(lc, pubnames)
1738 char *pubname = (
char *)
lfirst(lc);
1741 result =
lappend(result, pub);
1781 send_replication_origin =
false;
1787 send_replication_origin);
1903 ctl.
hcxt = cachectx;
2008 Oid publish_as_relid = relid;
2009 int publish_ancestor_level = 0;
2018 if (
data->publications)
2075 foreach(lc,
data->publications)
2078 bool publish =
false;
2085 Oid pub_relid = relid;
2086 int ancestor_level = 0;
2106 bool ancestor_published =
false;
2126 ancestor_published =
true;
2129 pub_relid = ancestor;
2130 ancestor_level = level;
2150 (relkind != RELKIND_PARTITIONED_TABLE || pub->
pubviaroot))
2164 if (publish_ancestor_level > ancestor_level)
2172 if (publish_ancestor_level < ancestor_level)
2174 publish_as_relid = pub_relid;
2175 publish_ancestor_level = ancestor_level;
2178 rel_publications =
NIL;
2183 Assert(publish_as_relid == pub_relid);
2187 rel_publications =
lappend(rel_publications, pub);
2385 static int changes_count = 0;
2393 #define CHANGES_THRESHOLD 100
Datum idx(PG_FUNCTION_ARGS)
AttrMap * build_attrmap_by_name_if_req(TupleDesc indesc, TupleDesc outdesc)
void free_attrmap(AttrMap *map)
bool bms_equal(const Bitmapset *a, const Bitmapset *b)
void bms_free(Bitmapset *a)
int bms_num_members(const Bitmapset *a)
bool bms_is_member(int x, const Bitmapset *a)
#define TextDatumGetCString(d)
#define AssertVariableIsOfType(varname, typename)
#define OidIsValid(objectId)
bool defGetBoolean(DefElem *def)
void hash_destroy(HTAB *hashp)
void * hash_search(HTAB *hashp, const void *keyPtr, HASHACTION action, bool *foundPtr)
HTAB * hash_create(const char *tabname, long nelem, const HASHCTL *info, int flags)
void * hash_seq_search(HASH_SEQ_STATUS *status)
void hash_seq_init(HASH_SEQ_STATUS *status, HTAB *hashp)
int errcode(int sqlerrcode)
int errmsg(const char *fmt,...)
#define ereport(elevel,...)
ExprState * ExecPrepareExpr(Expr *node, EState *estate)
TupleTableSlot * MakeTupleTableSlot(TupleDesc tupleDesc, const TupleTableSlotOps *tts_ops)
const TupleTableSlotOps TTSOpsVirtual
TupleTableSlot * ExecStoreVirtualTuple(TupleTableSlot *slot)
void ExecDropSingleTupleTableSlot(TupleTableSlot *slot)
TupleTableSlot * ExecStoreHeapTuple(HeapTuple tuple, TupleTableSlot *slot, bool shouldFree)
const TupleTableSlotOps TTSOpsHeapTuple
TupleTableSlot * MakeSingleTupleTableSlot(TupleDesc tupdesc, const TupleTableSlotOps *tts_ops)
void ExecInitRangeTable(EState *estate, List *rangeTable)
EState * CreateExecutorState(void)
#define ResetPerTupleExprContext(estate)
#define GetPerTupleExprContext(estate)
static Datum ExecEvalExprSwitchContext(ExprState *state, ExprContext *econtext, bool *isNull)
#define HeapTupleIsValid(tuple)
void CacheRegisterRelcacheCallback(RelcacheCallbackFunction func, Datum arg)
void CacheRegisterSyscacheCallback(int cacheid, SyscacheCallbackFunction func, Datum arg)
Assert(fmt[strlen(fmt) - 1] !='\n')
List * lappend(List *list, void *datum)
List * lappend_int(List *list, int datum)
void list_free(List *list)
bool list_member_oid(const List *list, Oid datum)
void list_free_deep(List *list)
void OutputPluginWrite(struct LogicalDecodingContext *ctx, bool last_write)
void OutputPluginUpdateProgress(struct LogicalDecodingContext *ctx, bool skipped_xact)
void OutputPluginPrepareWrite(struct LogicalDecodingContext *ctx, bool last_write)
#define LOGICALREP_PROTO_MIN_VERSION_NUM
#define LOGICALREP_PROTO_STREAM_VERSION_NUM
#define LOGICALREP_PROTO_TWOPHASE_VERSION_NUM
#define LOGICALREP_PROTO_MAX_VERSION_NUM
bool get_rel_relispartition(Oid relid)
char * get_namespace_name(Oid nspid)
char get_rel_relkind(Oid relid)
Oid get_rel_namespace(Oid relid)
Expr * make_orclause(List *orclauses)
void MemoryContextReset(MemoryContext context)
void pfree(void *pointer)
void * palloc0(Size size)
void * MemoryContextAllocZero(MemoryContext context, Size size)
MemoryContext CacheMemoryContext
void MemoryContextDelete(MemoryContext context)
#define AllocSetContextCreate
#define ALLOCSET_DEFAULT_SIZES
#define ALLOCSET_SMALL_SIZES
#define MemoryContextCopyAndSetIdentifier(cxt, id)
#define IsA(nodeptr, _type_)
bool replorigin_by_oid(RepOriginId roident, bool missing_ok, char **roname)
#define InvalidRepOriginId
@ OUTPUT_PLUGIN_BINARY_OUTPUT
void(* LogicalOutputPluginInit)(struct OutputPluginCallbacks *cb)
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
List * get_partition_ancestors(Oid relid)
FormData_pg_attribute * Form_pg_attribute
static int list_length(const List *l)
#define foreach_delete_current(lst, cell)
Publication * GetPublicationByName(const char *pubname, bool missing_ok)
List * GetSchemaPublications(Oid schemaid)
List * GetRelationPublications(Oid relid)
Oid GetTopMostAncestorInPublication(Oid puboid, List *ancestors, int *ancestor_level)
Bitmapset * pub_collist_to_bitmapset(Bitmapset *columns, Datum pubcols, MemoryContext mcxt)
bool is_publishable_relation(Relation rel)
static void static void status(const char *fmt,...) pg_attribute_printf(1
static List * LoadPublications(List *pubnames)
static void pgoutput_send_begin(LogicalDecodingContext *ctx, ReorderBufferTXN *txn)
static void rel_sync_cache_publication_cb(Datum arg, int cacheid, uint32 hashvalue)
struct RelationSyncEntry RelationSyncEntry
static void pgoutput_ensure_entry_cxt(PGOutputData *data, RelationSyncEntry *entry)
static void parse_output_parameters(List *options, PGOutputData *data)
static void init_tuple_slot(PGOutputData *data, Relation relation, RelationSyncEntry *entry)
static bool pgoutput_row_filter_exec_expr(ExprState *state, ExprContext *econtext)
static void update_replication_progress(LogicalDecodingContext *ctx, bool skipped_xact)
#define NUM_ROWFILTER_PUBACTIONS
static void pgoutput_begin_prepare_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn)
static void init_rel_sync_cache(MemoryContext decoding_context)
struct PGOutputTxnData PGOutputTxnData
static void pgoutput_startup(LogicalDecodingContext *ctx, OutputPluginOptions *opt, bool is_init)
static void pgoutput_truncate(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, int nrelations, Relation relations[], ReorderBufferChange *change)
static void pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, Relation rel, ReorderBufferChange *change)
static void rel_sync_cache_relation_cb(Datum arg, Oid relid)
static void pgoutput_prepare_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr prepare_lsn)
static RelationSyncEntry * get_rel_sync_entry(PGOutputData *data, Relation relation)
static bool pgoutput_origin_filter(LogicalDecodingContext *ctx, RepOriginId origin_id)
static void send_repl_origin(LogicalDecodingContext *ctx, RepOriginId origin_id, XLogRecPtr origin_lsn, bool send_origin)
static void pgoutput_rollback_prepared_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr prepare_end_lsn, TimestampTz prepare_time)
static void pgoutput_shutdown(LogicalDecodingContext *ctx)
static void cleanup_rel_sync_cache(TransactionId xid, bool is_commit)
static void pgoutput_stream_abort(struct LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr abort_lsn)
static void pgoutput_stream_prepare_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr prepare_lsn)
static void maybe_send_schema(LogicalDecodingContext *ctx, ReorderBufferChange *change, Relation relation, RelationSyncEntry *relentry)
static void pgoutput_begin_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn)
static void send_relation_and_attrs(Relation relation, TransactionId xid, LogicalDecodingContext *ctx, Bitmapset *columns)
static HTAB * RelationSyncCache
static void pgoutput_row_filter_init(PGOutputData *data, List *publications, RelationSyncEntry *entry)
static void pgoutput_stream_commit(struct LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr commit_lsn)
#define CHANGES_THRESHOLD
static void pgoutput_commit_prepared_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr commit_lsn)
static bool pgoutput_row_filter(Relation relation, TupleTableSlot *old_slot, TupleTableSlot **new_slot_ptr, RelationSyncEntry *entry, ReorderBufferChangeType *action)
static void set_schema_sent_in_streamed_txn(RelationSyncEntry *entry, TransactionId xid)
static void pgoutput_column_list_init(PGOutputData *data, List *publications, RelationSyncEntry *entry)
static void pgoutput_stream_stop(struct LogicalDecodingContext *ctx, ReorderBufferTXN *txn)
static void pgoutput_stream_start(struct LogicalDecodingContext *ctx, ReorderBufferTXN *txn)
static void publication_invalidation_cb(Datum arg, int cacheid, uint32 hashvalue)
void _PG_output_plugin_init(OutputPluginCallbacks *cb)
static void pgoutput_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr message_lsn, bool transactional, const char *prefix, Size sz, const char *message)
static void pgoutput_commit_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr commit_lsn)
static bool publications_valid
static bool get_schema_sent_in_streamed_txn(RelationSyncEntry *entry, TransactionId xid)
static EState * create_estate_for_relation(Relation rel)
#define VARATT_IS_EXTERNAL_ONDISK(PTR)
#define ObjectIdGetDatum(X)
void logicalrep_write_commit(StringInfo out, ReorderBufferTXN *txn, XLogRecPtr commit_lsn)
void logicalrep_write_rollback_prepared(StringInfo out, ReorderBufferTXN *txn, XLogRecPtr prepare_end_lsn, TimestampTz prepare_time)
void logicalrep_write_origin(StringInfo out, const char *origin, XLogRecPtr origin_lsn)
void logicalrep_write_rel(StringInfo out, TransactionId xid, Relation rel, Bitmapset *columns)
void logicalrep_write_delete(StringInfo out, TransactionId xid, Relation rel, TupleTableSlot *oldslot, bool binary)
void logicalrep_write_message(StringInfo out, TransactionId xid, XLogRecPtr lsn, bool transactional, const char *prefix, Size sz, const char *message)
void logicalrep_write_update(StringInfo out, TransactionId xid, Relation rel, TupleTableSlot *oldslot, TupleTableSlot *newslot, bool binary, Bitmapset *columns)
void logicalrep_write_prepare(StringInfo out, ReorderBufferTXN *txn, XLogRecPtr prepare_lsn)
void logicalrep_write_insert(StringInfo out, TransactionId xid, Relation rel, TupleTableSlot *newslot, bool binary, Bitmapset *columns)
void logicalrep_write_typ(StringInfo out, TransactionId xid, Oid typoid)
void logicalrep_write_truncate(StringInfo out, TransactionId xid, int nrelids, Oid relids[], bool cascade, bool restart_seqs)
void logicalrep_write_begin(StringInfo out, ReorderBufferTXN *txn)
void logicalrep_write_commit_prepared(StringInfo out, ReorderBufferTXN *txn, XLogRecPtr commit_lsn)
void logicalrep_write_stream_commit(StringInfo out, ReorderBufferTXN *txn, XLogRecPtr commit_lsn)
void logicalrep_write_stream_prepare(StringInfo out, ReorderBufferTXN *txn, XLogRecPtr prepare_lsn)
void logicalrep_write_stream_abort(StringInfo out, TransactionId xid, TransactionId subxid)
void logicalrep_write_begin_prepare(StringInfo out, ReorderBufferTXN *txn)
void logicalrep_write_stream_start(StringInfo out, TransactionId xid, bool first_segment)
void logicalrep_write_stream_stop(StringInfo out)
void * stringToNode(const char *str)
#define RelationGetRelid(relation)
#define RelationGetDescr(relation)
#define RelationGetNumberOfAttributes(relation)
#define RelationGetRelationName(relation)
#define RelationIsValid(relation)
#define RelationGetNamespace(relation)
Relation RelationIdGetRelation(Oid relationId)
void RelationClose(Relation relation)
#define rbtxn_is_streamed(txn)
@ REORDER_BUFFER_CHANGE_INSERT
@ REORDER_BUFFER_CHANGE_DELETE
@ REORDER_BUFFER_CHANGE_UPDATE
TupleTableSlot * ecxt_scantuple
void * output_plugin_private
List * output_plugin_options
LogicalDecodeStreamChangeCB stream_change_cb
LogicalDecodeMessageCB message_cb
LogicalDecodeStreamTruncateCB stream_truncate_cb
LogicalDecodeStreamMessageCB stream_message_cb
LogicalDecodeFilterByOriginCB filter_by_origin_cb
LogicalDecodeTruncateCB truncate_cb
LogicalDecodeStreamStopCB stream_stop_cb
LogicalDecodeStreamCommitCB stream_commit_cb
LogicalDecodeRollbackPreparedCB rollback_prepared_cb
LogicalDecodeStreamPrepareCB stream_prepare_cb
LogicalDecodeCommitPreparedCB commit_prepared_cb
LogicalDecodeStreamStartCB stream_start_cb
LogicalDecodePrepareCB prepare_cb
LogicalDecodeStartupCB startup_cb
LogicalDecodeCommitCB commit_cb
LogicalDecodeBeginCB begin_cb
LogicalDecodeStreamAbortCB stream_abort_cb
LogicalDecodeBeginPrepareCB begin_prepare_cb
LogicalDecodeChangeCB change_cb
LogicalDecodeShutdownCB shutdown_cb
OutputPluginOutputType output_type
PublicationActions pubactions
ExprState * exprstate[NUM_ROWFILTER_PUBACTIONS]
PublicationActions pubactions
TupleTableSlot * old_slot
TupleTableSlot * new_slot
ReorderBufferChangeType action
struct ReorderBufferChange::@105::@107 truncate
struct ReorderBufferChange::@105::@106 tp
union ReorderBufferChange::@105 data
struct ReorderBufferTXN * txn
struct ReorderBufferTXN * toptxn
void * output_plugin_private
void ReleaseSysCache(HeapTuple tuple)
Datum SysCacheGetAttr(int cacheId, HeapTuple tup, AttrNumber attributeNumber, bool *isNull)
HeapTuple SearchSysCache2(int cacheId, Datum key1, Datum key2)
@ PUBLICATIONNAMESPACEMAP
#define InvalidTransactionId
#define FirstGenbkiObjectId
TupleTableSlot * execute_attr_map_slot(AttrMap *attrMap, TupleTableSlot *in_slot, TupleTableSlot *out_slot)
TupleDesc CreateTupleDescCopy(TupleDesc tupdesc)
#define TupleDescAttr(tupdesc, i)
static TupleTableSlot * ExecClearTuple(TupleTableSlot *slot)
static void slot_getallattrs(TupleTableSlot *slot)
bool SplitIdentifierString(char *rawstring, char separator, List **namelist)
CommandId GetCurrentCommandId(bool used)
#define InvalidXLogRecPtr