56 bool transactional,
const char *prefix,
57 Size sz,
const char *message);
103#define NUM_ROWFILTER_PUBACTIONS (PUBACTION_DELETE+1)
289 bool protocol_version_given =
false;
290 bool publication_names_given =
false;
291 bool binary_option_given =
false;
292 bool messages_option_given =
false;
293 bool streaming_given =
false;
294 bool two_phase_option_given =
false;
295 bool origin_option_given =
false;
297 data->binary =
false;
298 data->streaming = LOGICALREP_STREAM_OFF;
299 data->messages =
false;
300 data->two_phase =
false;
309 if (strcmp(defel->
defname,
"proto_version") == 0)
311 unsigned long parsed;
314 if (protocol_version_given)
316 (
errcode(ERRCODE_SYNTAX_ERROR),
317 errmsg(
"conflicting or redundant options")));
318 protocol_version_given =
true;
321 parsed = strtoul(
strVal(defel->
arg), &endptr, 10);
322 if (errno != 0 || *endptr !=
'\0')
324 (
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
325 errmsg(
"invalid proto_version")));
329 (
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
330 errmsg(
"proto_version \"%s\" out of range",
335 else if (strcmp(defel->
defname,
"publication_names") == 0)
337 if (publication_names_given)
339 (
errcode(ERRCODE_SYNTAX_ERROR),
340 errmsg(
"conflicting or redundant options")));
341 publication_names_given =
true;
344 &
data->publication_names))
346 (
errcode(ERRCODE_INVALID_NAME),
347 errmsg(
"invalid publication_names syntax")));
349 else if (strcmp(defel->
defname,
"binary") == 0)
351 if (binary_option_given)
353 (
errcode(ERRCODE_SYNTAX_ERROR),
354 errmsg(
"conflicting or redundant options")));
355 binary_option_given =
true;
359 else if (strcmp(defel->
defname,
"messages") == 0)
361 if (messages_option_given)
363 (
errcode(ERRCODE_SYNTAX_ERROR),
364 errmsg(
"conflicting or redundant options")));
365 messages_option_given =
true;
369 else if (strcmp(defel->
defname,
"streaming") == 0)
373 (
errcode(ERRCODE_SYNTAX_ERROR),
374 errmsg(
"conflicting or redundant options")));
375 streaming_given =
true;
379 else if (strcmp(defel->
defname,
"two_phase") == 0)
381 if (two_phase_option_given)
383 (
errcode(ERRCODE_SYNTAX_ERROR),
384 errmsg(
"conflicting or redundant options")));
385 two_phase_option_given =
true;
389 else if (strcmp(defel->
defname,
"origin") == 0)
393 if (origin_option_given)
396 errmsg(
"conflicting or redundant options"));
397 origin_option_given =
true;
401 data->publish_no_origin =
true;
403 data->publish_no_origin =
false;
406 errcode(ERRCODE_INVALID_PARAMETER_VALUE),
407 errmsg(
"unrecognized origin value: \"%s\"", origin));
414 if (!protocol_version_given)
416 errcode(ERRCODE_INVALID_PARAMETER_VALUE),
417 errmsg(
"option \"%s\" missing",
"proto_version"));
418 if (!publication_names_given)
420 errcode(ERRCODE_INVALID_PARAMETER_VALUE),
421 errmsg(
"option \"%s\" missing",
"publication_names"));
432 static bool publication_callback_registered =
false;
436 "logical replication output context",
440 "logical replication cache context",
444 "logical replication publication list context",
465 (
errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
466 errmsg(
"client sent proto_version=%d but server only supports protocol %d or lower",
471 (
errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
472 errmsg(
"client sent proto_version=%d but server only supports protocol %d or higher",
481 if (
data->streaming == LOGICALREP_STREAM_OFF)
483 else if (
data->streaming == LOGICALREP_STREAM_ON &&
486 (
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
487 errmsg(
"requested proto_version=%d does not support streaming, need %d or higher",
489 else if (
data->streaming == LOGICALREP_STREAM_PARALLEL &&
492 (
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
493 errmsg(
"requested proto_version=%d does not support parallel streaming, need %d or higher",
497 (
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
498 errmsg(
"streaming requested, but not supported by output plugin")));
507 if (!
data->two_phase)
511 (
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
512 errmsg(
"requested proto_version=%d does not support two-phase commit, need %d or higher",
516 (
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
517 errmsg(
"two-phase commit requested, but not supported by output plugin")));
529 if (!publication_callback_registered)
534 publication_callback_registered =
true;
589 send_replication_origin);
617 elog(
DEBUG1,
"skipped replication of an empty transaction with XID: %u", txn->
xid);
638 send_replication_origin);
730 if (
data->in_streaming)
754 if (
data->in_streaming)
786 include_gencols_type))
799 include_gencols_type);
819 rte->relkind = rel->
rd_rel->relkind;
848 elog(
DEBUG3,
"row filter evaluates to %s (isnull: %s)",
849 isnull ?
"false" :
DatumGetBool(ret) ?
"true" :
"false",
850 isnull ?
"true" :
"false");
873 "entry private context",
889 bool no_filter[] = {
false,
false,
false};
892 bool has_filter =
true;
910 foreach(lc, publications)
915 bool pub_no_filter =
true;
939 Anum_pg_publication_rel_prqual,
1013 foreach(lc, rfnodes[
idx])
1036 bool gencolpresent =
false;
1040 for (
int i = 0;
i < desc->
natts;
i++)
1044 if (att->attgenerated)
1046 gencolpresent =
true;
1079 errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1080 errmsg(
"cannot use different values of publish_generated_columns for table \"%s.%s\" in different publications",
1096 bool found_pub_collist =
false;
1114 foreach(lc, publications)
1156 errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1157 errmsg(
"cannot use different column lists for table \"%s.%s\" in different publications",
1166 if (!found_pub_collist)
1287 static const int map_changetype_pubaction[] = {
1297 Assert(new_slot || old_slot);
1300 filter_exprstate = entry->
exprstate[map_changetype_pubaction[*
action]];
1303 if (!filter_exprstate)
1306 elog(
DEBUG3,
"table \"%s.%s\" has row filter",
1329 if (!new_slot || !old_slot)
1346 tmp_new_slot = NULL;
1381 desc->
natts *
sizeof(
bool));
1406 if (!old_matched && !new_matched)
1418 if (!old_matched && new_matched)
1423 *new_slot_ptr = tmp_new_slot;
1433 else if (old_matched && !new_matched)
1473 if (
data->in_streaming)
1500 elog(
DEBUG1,
"didn't send DELETE change because of missing oldtuple");
1516 targetrel = ancestor;
1565 if (txndata && !txndata->sent_begin_txn)
1635 if (
data->in_streaming)
1643 for (
i = 0;
i < nrelations;
i++)
1660 if (relation->
rd_rel->relispartition &&
1664 relids[nrelids++] = relid;
1667 if (txndata && !txndata->sent_begin_txn)
1691 XLogRecPtr message_lsn,
bool transactional,
const char *prefix,
Size sz,
1692 const char *message)
1704 if (
data->in_streaming)
1772 foreach(lc, pubnames)
1774 char *pubname = (
char *)
lfirst(lc);
1777 result =
lappend(result, pub);
1818 send_replication_origin =
false;
1824 send_replication_origin);
1829 data->in_streaming =
true;
1849 data->in_streaming =
false;
1863 bool write_abort_info = (
data->streaming == LOGICALREP_STREAM_PARALLEL);
1941 static bool relation_callbacks_registered =
false;
1948 ctl.keysize =
sizeof(
Oid);
1950 ctl.hcxt = cachectx;
1959 if (relation_callbacks_registered)
1985 relation_callbacks_registered =
true;
2069 Oid publish_as_relid = relid;
2070 int publish_ancestor_level = 0;
2158 foreach(lc,
data->publications)
2161 bool publish =
false;
2168 Oid pub_relid = relid;
2169 int ancestor_level = 0;
2189 bool ancestor_published =
false;
2209 ancestor_published =
true;
2212 pub_relid = ancestor;
2213 ancestor_level = level;
2233 (relkind != RELKIND_PARTITIONED_TABLE || pub->
pubviaroot))
2247 if (publish_ancestor_level > ancestor_level)
2255 if (publish_ancestor_level < ancestor_level)
2257 publish_as_relid = pub_relid;
2258 publish_ancestor_level = ancestor_level;
2261 rel_publications =
NIL;
2266 Assert(publish_as_relid == pub_relid);
2270 rel_publications =
lappend(rel_publications, pub);
2336 if (xid == streamed_txn)
Datum idx(PG_FUNCTION_ARGS)
void free_attrmap(AttrMap *map)
AttrMap * build_attrmap_by_name_if_req(TupleDesc indesc, TupleDesc outdesc, bool missing_ok)
Bitmapset * bms_make_singleton(int x)
bool bms_equal(const Bitmapset *a, const Bitmapset *b)
void bms_free(Bitmapset *a)
static void cleanup(void)
#define TextDatumGetCString(d)
#define PG_USED_FOR_ASSERTS_ONLY
#define Assert(condition)
#define OidIsValid(objectId)
char * defGetString(DefElem *def)
bool defGetBoolean(DefElem *def)
void * hash_search(HTAB *hashp, const void *keyPtr, HASHACTION action, bool *foundPtr)
void hash_destroy(HTAB *hashp)
void * hash_seq_search(HASH_SEQ_STATUS *status)
HTAB * hash_create(const char *tabname, long nelem, const HASHCTL *info, int flags)
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 * MakeSingleTupleTableSlot(TupleDesc tupdesc, const TupleTableSlotOps *tts_ops)
const TupleTableSlotOps TTSOpsVirtual
void ExecDropSingleTupleTableSlot(TupleTableSlot *slot)
TupleTableSlot * ExecStoreVirtualTuple(TupleTableSlot *slot)
const TupleTableSlotOps TTSOpsHeapTuple
TupleTableSlot * MakeTupleTableSlot(TupleDesc tupleDesc, const TupleTableSlotOps *tts_ops)
TupleTableSlot * ExecStoreHeapTuple(HeapTuple tuple, TupleTableSlot *slot, bool shouldFree)
void ExecInitRangeTable(EState *estate, List *rangeTable, List *permInfos, Bitmapset *unpruned_relids)
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)
List * lappend(List *list, void *datum)
List * lappend_xid(List *list, TransactionId datum)
bool list_member_xid(const List *list, TransactionId 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_rel_relkind(Oid relid)
Oid get_rel_namespace(Oid relid)
char * get_namespace_name(Oid nspid)
Expr * make_orclause(List *orclauses)
void MemoryContextReset(MemoryContext context)
void * MemoryContextAllocZero(MemoryContext context, Size size)
void pfree(void *pointer)
void * palloc0(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, var_or_cell)
#define foreach_xid(var, lst)
#define foreach_ptr(type, var, lst)
List * GetRelationPublications(Oid relid)
Publication * GetPublicationByName(const char *pubname, bool missing_ok)
List * GetSchemaPublications(Oid schemaid)
Oid GetTopMostAncestorInPublication(Oid puboid, List *ancestors, int *ancestor_level)
Bitmapset * pub_form_cols_map(Relation relation, PublishGencolsType include_gencols_type)
bool check_and_fetch_column_list(Publication *pub, Oid relid, MemoryContext mcxt, Bitmapset **cols)
bool is_publishable_relation(Relation rel)
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 send_relation_and_attrs(Relation relation, TransactionId xid, LogicalDecodingContext *ctx, RelationSyncEntry *relentry)
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 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 check_and_init_gencol(PGOutputData *data, List *publications, RelationSyncEntry *entry)
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_insert(StringInfo out, TransactionId xid, Relation rel, TupleTableSlot *newslot, bool binary, Bitmapset *columns, PublishGencolsType include_gencols_type)
void logicalrep_write_origin(StringInfo out, const char *origin, XLogRecPtr origin_lsn)
void logicalrep_write_rel(StringInfo out, TransactionId xid, Relation rel, Bitmapset *columns, PublishGencolsType include_gencols_type)
void logicalrep_write_stream_abort(StringInfo out, TransactionId xid, TransactionId subxid, XLogRecPtr abort_lsn, TimestampTz abort_time, bool write_abort_info)
void logicalrep_write_message(StringInfo out, TransactionId xid, XLogRecPtr lsn, bool transactional, const char *prefix, Size sz, const char *message)
void logicalrep_write_prepare(StringInfo out, ReorderBufferTXN *txn, XLogRecPtr prepare_lsn)
void logicalrep_write_typ(StringInfo out, TransactionId xid, Oid typoid)
void logicalrep_write_delete(StringInfo out, TransactionId xid, Relation rel, TupleTableSlot *oldslot, bool binary, Bitmapset *columns, PublishGencolsType include_gencols_type)
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_begin_prepare(StringInfo out, ReorderBufferTXN *txn)
bool logicalrep_should_publish_column(Form_pg_attribute att, Bitmapset *columns, PublishGencolsType include_gencols_type)
void logicalrep_write_stream_start(StringInfo out, TransactionId xid, bool first_segment)
void logicalrep_write_update(StringInfo out, TransactionId xid, Relation rel, TupleTableSlot *oldslot, TupleTableSlot *newslot, bool binary, Bitmapset *columns, PublishGencolsType include_gencols_type)
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
Node * expand_generated_columns_in_expr(Node *node, Relation rel, int rt_index)
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
PublishGencolsType include_gencols_type
TupleTableSlot * new_slot
struct ReorderBufferChange::@110::@112 truncate
ReorderBufferChangeType action
union ReorderBufferChange::@110 data
struct ReorderBufferTXN * txn
struct ReorderBufferChange::@110::@111 tp
void * output_plugin_private
union ReorderBufferTXN::@116 xact_time
TupleDesc tts_tupleDescriptor
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)
#define SearchSysCacheExists2(cacheId, key1, key2)
#define InvalidTransactionId
#define FirstGenbkiObjectId
TupleTableSlot * execute_attr_map_slot(AttrMap *attrMap, TupleTableSlot *in_slot, TupleTableSlot *out_slot)
TupleDesc CreateTupleDescCopyConstr(TupleDesc tupdesc)
void FreeTupleDesc(TupleDesc tupdesc)
static FormData_pg_attribute * TupleDescAttr(TupleDesc tupdesc, int i)
static CompactAttribute * TupleDescCompactAttr(TupleDesc tupdesc, int 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