56 bool transactional,
const char *prefix,
57 Size sz,
const char *message);
106 #define NUM_ROWFILTER_PUBACTIONS (PUBACTION_DELETE+1)
280 bool protocol_version_given =
false;
281 bool publication_names_given =
false;
282 bool binary_option_given =
false;
283 bool messages_option_given =
false;
284 bool streaming_given =
false;
285 bool two_phase_option_given =
false;
286 bool origin_option_given =
false;
288 data->binary =
false;
290 data->messages =
false;
291 data->two_phase =
false;
300 if (strcmp(defel->
defname,
"proto_version") == 0)
302 unsigned long parsed;
305 if (protocol_version_given)
307 (
errcode(ERRCODE_SYNTAX_ERROR),
308 errmsg(
"conflicting or redundant options")));
309 protocol_version_given =
true;
312 parsed = strtoul(
strVal(defel->
arg), &endptr, 10);
313 if (errno != 0 || *endptr !=
'\0')
315 (
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
316 errmsg(
"invalid proto_version")));
320 (
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
321 errmsg(
"proto_version \"%s\" out of range",
326 else if (strcmp(defel->
defname,
"publication_names") == 0)
328 if (publication_names_given)
330 (
errcode(ERRCODE_SYNTAX_ERROR),
331 errmsg(
"conflicting or redundant options")));
332 publication_names_given =
true;
335 &
data->publication_names))
337 (
errcode(ERRCODE_INVALID_NAME),
338 errmsg(
"invalid publication_names syntax")));
340 else if (strcmp(defel->
defname,
"binary") == 0)
342 if (binary_option_given)
344 (
errcode(ERRCODE_SYNTAX_ERROR),
345 errmsg(
"conflicting or redundant options")));
346 binary_option_given =
true;
350 else if (strcmp(defel->
defname,
"messages") == 0)
352 if (messages_option_given)
354 (
errcode(ERRCODE_SYNTAX_ERROR),
355 errmsg(
"conflicting or redundant options")));
356 messages_option_given =
true;
360 else if (strcmp(defel->
defname,
"streaming") == 0)
364 (
errcode(ERRCODE_SYNTAX_ERROR),
365 errmsg(
"conflicting or redundant options")));
366 streaming_given =
true;
370 else if (strcmp(defel->
defname,
"two_phase") == 0)
372 if (two_phase_option_given)
374 (
errcode(ERRCODE_SYNTAX_ERROR),
375 errmsg(
"conflicting or redundant options")));
376 two_phase_option_given =
true;
380 else if (strcmp(defel->
defname,
"origin") == 0)
384 if (origin_option_given)
387 errmsg(
"conflicting or redundant options"));
388 origin_option_given =
true;
392 data->publish_no_origin =
true;
394 data->publish_no_origin =
false;
397 errcode(ERRCODE_INVALID_PARAMETER_VALUE),
398 errmsg(
"unrecognized origin value: \"%s\"", 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")));
489 if (!
data->two_phase)
493 (
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
494 errmsg(
"requested proto_version=%d does not support two-phase commit, need %d or higher",
498 (
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
499 errmsg(
"two-phase commit requested, but not supported by output plugin")));
511 if (!publication_callback_registered)
516 publication_callback_registered =
true;
571 send_replication_origin);
599 elog(
DEBUG1,
"skipped replication of an empty transaction with XID: %u", txn->
xid);
620 send_replication_origin);
712 if (
data->in_streaming)
736 if (
data->in_streaming)
765 if (att->attisdropped || att->attgenerated)
829 elog(
DEBUG3,
"row filter evaluates to %s (isnull: %s)",
830 isnull ?
"false" :
DatumGetBool(ret) ?
"true" :
"false",
831 isnull ?
"true" :
"false");
854 "entry private context",
870 bool no_filter[] = {
false,
false,
false};
873 bool has_filter =
true;
891 foreach(lc, publications)
896 bool pub_no_filter =
true;
920 Anum_pg_publication_rel_prqual,
994 foreach(lc, rfnodes[
idx])
1033 foreach(lc, publications)
1047 bool pub_no_list =
true;
1064 Anum_pg_publication_rel_prattrs,
1084 if (att->attisdropped || att->attgenerated)
1112 errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1113 errmsg(
"cannot use different column lists for table \"%s.%s\" in different publications",
1236 static const int map_changetype_pubaction[] = {
1246 Assert(new_slot || old_slot);
1249 filter_exprstate = entry->
exprstate[map_changetype_pubaction[*
action]];
1252 if (!filter_exprstate)
1255 elog(
DEBUG3,
"table \"%s.%s\" has row filter",
1278 if (!new_slot || !old_slot)
1295 tmp_new_slot = NULL;
1318 if (att->attlen == -1 &&
1330 desc->
natts *
sizeof(
bool));
1355 if (!old_matched && !new_matched)
1367 if (!old_matched && new_matched)
1372 *new_slot_ptr = tmp_new_slot;
1382 else if (old_matched && !new_matched)
1422 if (
data->in_streaming)
1447 if (!change->
data.
tp.oldtuple)
1449 elog(
DEBUG1,
"didn't send DELETE change because of missing oldtuple");
1465 targetrel = ancestor;
1468 if (change->
data.
tp.oldtuple)
1483 if (change->
data.
tp.newtuple)
1514 if (txndata && !txndata->sent_begin_txn)
1571 if (
data->in_streaming)
1579 for (
i = 0;
i < nrelations;
i++)
1596 if (relation->
rd_rel->relispartition &&
1600 relids[nrelids++] = relid;
1603 if (txndata && !txndata->sent_begin_txn)
1627 XLogRecPtr message_lsn,
bool transactional,
const char *prefix,
Size sz,
1628 const char *message)
1640 if (
data->in_streaming)
1708 foreach(lc, pubnames)
1710 char *pubname = (
char *)
lfirst(lc);
1713 result =
lappend(result, pub);
1754 send_replication_origin =
false;
1760 send_replication_origin);
1765 data->in_streaming =
true;
1785 data->in_streaming =
false;
1877 static bool relation_callbacks_registered =
false;
1886 ctl.
hcxt = cachectx;
1895 if (relation_callbacks_registered)
1921 relation_callbacks_registered =
true;
2004 Oid publish_as_relid = relid;
2005 int publish_ancestor_level = 0;
2014 if (
data->publications)
2071 foreach(lc,
data->publications)
2074 bool publish =
false;
2081 Oid pub_relid = relid;
2082 int ancestor_level = 0;
2102 bool ancestor_published =
false;
2122 ancestor_published =
true;
2125 pub_relid = ancestor;
2126 ancestor_level = level;
2146 (relkind != RELKIND_PARTITIONED_TABLE || pub->
pubviaroot))
2160 if (publish_ancestor_level > ancestor_level)
2168 if (publish_ancestor_level < ancestor_level)
2170 publish_as_relid = pub_relid;
2171 publish_ancestor_level = ancestor_level;
2174 rel_publications =
NIL;
2179 Assert(publish_as_relid == pub_relid);
2183 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 PG_USED_FOR_ASSERTS_ONLY
#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)
#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 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::@100::@101 tp
ReorderBufferChangeType action
struct ReorderBufferTXN * txn
struct ReorderBufferChange::@100::@102 truncate
union ReorderBufferChange::@100 data
void * output_plugin_private
union ReorderBufferTXN::@106 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