56 bool transactional,
const char *prefix,
57 Size sz,
const char *message);
108 #define NUM_ROWFILTER_PUBACTIONS (PUBACTION_DELETE+1)
282 bool protocol_version_given =
false;
283 bool publication_names_given =
false;
284 bool binary_option_given =
false;
285 bool messages_option_given =
false;
286 bool streaming_given =
false;
287 bool two_phase_option_given =
false;
288 bool origin_option_given =
false;
290 data->binary =
false;
292 data->messages =
false;
293 data->two_phase =
false;
302 if (strcmp(defel->
defname,
"proto_version") == 0)
304 unsigned long parsed;
307 if (protocol_version_given)
309 (
errcode(ERRCODE_SYNTAX_ERROR),
310 errmsg(
"conflicting or redundant options")));
311 protocol_version_given =
true;
314 parsed = strtoul(
strVal(defel->
arg), &endptr, 10);
315 if (errno != 0 || *endptr !=
'\0')
317 (
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
318 errmsg(
"invalid proto_version")));
322 (
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
323 errmsg(
"proto_version \"%s\" out of range",
328 else if (strcmp(defel->
defname,
"publication_names") == 0)
330 if (publication_names_given)
332 (
errcode(ERRCODE_SYNTAX_ERROR),
333 errmsg(
"conflicting or redundant options")));
334 publication_names_given =
true;
337 &
data->publication_names))
339 (
errcode(ERRCODE_INVALID_NAME),
340 errmsg(
"invalid publication_names syntax")));
342 else if (strcmp(defel->
defname,
"binary") == 0)
344 if (binary_option_given)
346 (
errcode(ERRCODE_SYNTAX_ERROR),
347 errmsg(
"conflicting or redundant options")));
348 binary_option_given =
true;
352 else if (strcmp(defel->
defname,
"messages") == 0)
354 if (messages_option_given)
356 (
errcode(ERRCODE_SYNTAX_ERROR),
357 errmsg(
"conflicting or redundant options")));
358 messages_option_given =
true;
362 else if (strcmp(defel->
defname,
"streaming") == 0)
366 (
errcode(ERRCODE_SYNTAX_ERROR),
367 errmsg(
"conflicting or redundant options")));
368 streaming_given =
true;
372 else if (strcmp(defel->
defname,
"two_phase") == 0)
374 if (two_phase_option_given)
376 (
errcode(ERRCODE_SYNTAX_ERROR),
377 errmsg(
"conflicting or redundant options")));
378 two_phase_option_given =
true;
382 else if (strcmp(defel->
defname,
"origin") == 0)
384 if (origin_option_given)
387 errmsg(
"conflicting or redundant options"));
388 origin_option_given =
true;
397 errcode(ERRCODE_INVALID_PARAMETER_VALUE),
398 errmsg(
"unrecognized origin value: \"%s\"",
data->origin));
413 static bool publication_callback_registered =
false;
417 "logical replication output context",
421 "logical replication cache context",
442 (
errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
443 errmsg(
"client sent proto_version=%d but server only supports protocol %d or lower",
448 (
errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
449 errmsg(
"client sent proto_version=%d but server only supports protocol %d or higher",
452 if (
data->publication_names ==
NIL)
454 (
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
455 errmsg(
"publication_names parameter missing")));
468 (
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
469 errmsg(
"requested proto_version=%d does not support streaming, need %d or higher",
474 (
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
475 errmsg(
"requested proto_version=%d does not support parallel streaming, need %d or higher",
479 (
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
480 errmsg(
"streaming requested, but not supported by output plugin")));
492 if (!
data->two_phase)
496 (
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
497 errmsg(
"requested proto_version=%d does not support two-phase commit, need %d or higher",
501 (
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
502 errmsg(
"two-phase commit requested, but not supported by output plugin")));
514 if (!publication_callback_registered)
519 publication_callback_registered =
true;
574 send_replication_origin);
602 elog(
DEBUG1,
"skipped replication of an empty transaction with XID: %u", txn->
xid);
623 send_replication_origin);
767 if (att->attisdropped || att->attgenerated)
831 elog(
DEBUG3,
"row filter evaluates to %s (isnull: %s)",
832 isnull ?
"false" :
DatumGetBool(ret) ?
"true" :
"false",
833 isnull ?
"true" :
"false");
856 "entry private context",
872 bool no_filter[] = {
false,
false,
false};
875 bool has_filter =
true;
893 foreach(lc, publications)
898 bool pub_no_filter =
true;
922 Anum_pg_publication_rel_prqual,
996 foreach(lc, rfnodes[
idx])
1035 foreach(lc, publications)
1049 bool pub_no_list =
true;
1066 Anum_pg_publication_rel_prattrs,
1086 if (att->attisdropped || att->attgenerated)
1114 errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1115 errmsg(
"cannot use different column lists for table \"%s.%s\" in different publications",
1238 static const int map_changetype_pubaction[] = {
1248 Assert(new_slot || old_slot);
1251 filter_exprstate = entry->
exprstate[map_changetype_pubaction[*
action]];
1254 if (!filter_exprstate)
1257 elog(
DEBUG3,
"table \"%s.%s\" has row filter",
1280 if (!new_slot || !old_slot)
1297 tmp_new_slot = NULL;
1320 if (att->attlen == -1 &&
1332 desc->
natts *
sizeof(
bool));
1357 if (!old_matched && !new_matched)
1369 if (!old_matched && new_matched)
1374 *new_slot_ptr = tmp_new_slot;
1384 else if (old_matched && !new_matched)
1449 if (!change->
data.
tp.oldtuple)
1451 elog(
DEBUG1,
"didn't send DELETE change because of missing oldtuple");
1467 targetrel = ancestor;
1470 if (change->
data.
tp.oldtuple)
1485 if (change->
data.
tp.newtuple)
1516 if (txndata && !txndata->sent_begin_txn)
1581 for (
i = 0;
i < nrelations;
i++)
1598 if (relation->
rd_rel->relispartition &&
1602 relids[nrelids++] = relid;
1605 if (txndata && !txndata->sent_begin_txn)
1629 XLogRecPtr message_lsn,
bool transactional,
const char *prefix,
Size sz,
1630 const char *message)
1708 foreach(lc, pubnames)
1710 char *pubname = (
char *)
lfirst(lc);
1713 result =
lappend(result, pub);
1753 send_replication_origin =
false;
1759 send_replication_origin);
1872 static bool relation_callbacks_registered =
false;
1881 ctl.
hcxt = cachectx;
1890 if (relation_callbacks_registered)
1916 relation_callbacks_registered =
true;
1999 Oid publish_as_relid = relid;
2000 int publish_ancestor_level = 0;
2009 if (
data->publications)
2066 foreach(lc,
data->publications)
2069 bool publish =
false;
2076 Oid pub_relid = relid;
2077 int ancestor_level = 0;
2097 bool ancestor_published =
false;
2117 ancestor_published =
true;
2120 pub_relid = ancestor;
2121 ancestor_level = level;
2141 (relkind != RELKIND_PARTITIONED_TABLE || pub->
pubviaroot))
2155 if (publish_ancestor_level > ancestor_level)
2163 if (publish_ancestor_level < ancestor_level)
2165 publish_as_relid = pub_relid;
2166 publish_ancestor_level = ancestor_level;
2169 rel_publications =
NIL;
2174 Assert(publish_as_relid == pub_relid);
2178 rel_publications =
lappend(rel_publications, pub);
Datum idx(PG_FUNCTION_ARGS)
void free_attrmap(AttrMap *map)
AttrMap * build_attrmap_by_name_if_req(TupleDesc indesc, TupleDesc outdesc, bool missing_ok)
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)
static void cleanup(void)
#define TextDatumGetCString(d)
#define OidIsValid(objectId)
elog(ERROR, "%s: %s", p2, msg)
bool defGetBoolean(DefElem *def)
char * defGetString(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, List *permInfos)
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)
if(TABLE==NULL||TABLE_index==NULL)
Assert(fmt[strlen(fmt) - 1] !='\n')
List * lappend_xid(List *list, TransactionId datum)
bool list_member_xid(const List *list, TransactionId datum)
List * lappend(List *list, void *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_STREAM_PARALLEL_VERSION_NUM
#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
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
RTEPermissionInfo * addRTEPermissionInfo(List **rteperminfos, RangeTblEntry *rte)
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
#define LOGICALREP_ORIGIN_NONE
#define LOGICALREP_STREAM_ON
#define LOGICALREP_ORIGIN_ANY
#define LOGICALREP_STREAM_OFF
#define LOGICALREP_STREAM_PARALLEL
static List * LoadPublications(List *pubnames)
static void pgoutput_send_begin(LogicalDecodingContext *ctx, ReorderBufferTXN *txn)
static bool publish_no_origin
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 pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, Relation relation, ReorderBufferChange *change)
#define NUM_ROWFILTER_PUBACTIONS
static void pgoutput_begin_prepare_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn)
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 init_rel_sync_cache(MemoryContext cachectx)
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)
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)
int pg_strcasecmp(const char *s1, const char *s2)
static bool DatumGetBool(Datum X)
static Datum ObjectIdGetDatum(Oid 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_stream_abort(StringInfo out, TransactionId xid, TransactionId subxid, XLogRecPtr abort_lsn, TimestampTz abort_time, bool write_abort_info)
void logicalrep_write_rel(StringInfo out, TransactionId xid, Relation rel, Bitmapset *columns)
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_delete(StringInfo out, TransactionId xid, Relation rel, TupleTableSlot *oldslot, bool binary, Bitmapset *columns)
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_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 RelationGetRelationName(relation)
#define RelationIsValid(relation)
#define RelationGetNamespace(relation)
Relation RelationIdGetRelation(Oid relationId)
void RelationClose(Relation relation)
#define rbtxn_is_streamed(txn)
#define rbtxn_get_toptxn(txn)
#define rbtxn_is_subtxn(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
struct ReorderBufferChange::@97::@99 truncate
ReorderBufferChangeType action
union ReorderBufferChange::@97 data
struct ReorderBufferTXN * txn
struct ReorderBufferChange::@97::@98 tp
void * output_plugin_private
union ReorderBufferTXN::@103 xact_time
char defGetStreamingMode(DefElem *def)
void ReleaseSysCache(HeapTuple tuple)
Datum SysCacheGetAttr(int cacheId, HeapTuple tup, AttrNumber attributeNumber, bool *isNull)
HeapTuple SearchSysCache2(int cacheId, Datum key1, Datum key2)
@ PUBLICATIONNAMESPACEMAP
#define SearchSysCacheExists2(cacheId, key1, key2)
#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)
#define VARATT_IS_EXTERNAL_ONDISK(PTR)
bool SplitIdentifierString(char *rawstring, char separator, List **namelist)
CommandId GetCurrentCommandId(bool used)
#define InvalidXLogRecPtr