147 #include <sys/stat.h>
195 #define NAPTIME_PER_CYCLE 1000
336 #define is_skipping_changes() (unlikely(!XLogRecPtrIsInvalid(skip_xact_finish_lsn)))
429 char *originname,
Size szoriginname)
434 snprintf(originname, szoriginname,
"pg_%u_%u", suboid, relid);
439 snprintf(originname, szoriginname,
"pg_%u", suboid);
477 if (rel->
state != SUBREL_STATE_READY &&
478 rel->
state != SUBREL_STATE_UNKNOWN)
480 (
errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
481 errmsg(
"logical replication parallel apply worker for subscription \"%s\" will stop",
483 errdetail(
"Cannot handle streamed replication transactions using parallel apply workers until all tables have been synchronized.")));
485 return rel->
state == SUBREL_STATE_READY;
488 return (rel->
state == SUBREL_STATE_READY ||
489 (rel->
state == SUBREL_STATE_SYNCDONE &&
590 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
591 errmsg_internal(
"invalid transaction ID in streamed replication transaction")));
593 switch (apply_action)
640 elog(
ERROR,
"unexpected apply action: %d", (
int) apply_action);
744 int num_phys_attrs = desc->
natts;
758 defmap = (
int *)
palloc(num_phys_attrs *
sizeof(
int));
781 defmap[num_defaults] =
attnum;
786 for (
i = 0;
i < num_defaults;
i++)
807 for (
i = 0;
i < natts;
i++)
812 if (!att->attisdropped && remoteattnum >= 0)
816 Assert(remoteattnum < tupleData->ncols);
829 typioparam, att->atttypmod);
846 typioparam, att->atttypmod);
851 (
errcode(ERRCODE_INVALID_BINARY_REPRESENTATION),
852 errmsg(
"incorrect binary data format in logical replication column %d",
919 for (
i = 0;
i < natts;
i++)
924 if (remoteattnum < 0)
927 Assert(remoteattnum < tupleData->ncols);
944 typioparam, att->atttypmod);
961 typioparam, att->atttypmod);
966 (
errcode(ERRCODE_INVALID_BINARY_REPRESENTATION),
967 errmsg(
"incorrect binary data format in logical replication column %d",
1024 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
1025 errmsg_internal(
"incorrect commit LSN %X/%X in commit message (expected %X/%X)",
1049 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
1050 errmsg_internal(
"tablesync worker received a BEGIN PREPARE message")));
1116 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
1117 errmsg_internal(
"incorrect prepare LSN %X/%X in prepare message (expected %X/%X)",
1282 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
1288 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
1289 errmsg_internal(
"tablesync worker received a STREAM PREPARE message")));
1296 switch (apply_action)
1319 elog(
DEBUG1,
"finished processing the STREAM PREPARE command");
1377 elog(
DEBUG1,
"finished processing the STREAM PREPARE command");
1381 elog(
ERROR,
"unexpected apply action: %d", (
int) apply_action);
1418 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
1478 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
1492 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
1493 errmsg_internal(
"invalid transaction ID in streamed replication transaction")));
1503 switch (apply_action)
1589 elog(
ERROR,
"unexpected apply action: %d", (
int) apply_action);
1633 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
1638 switch (apply_action)
1675 elog(
DEBUG1,
"applied %u changes in the streaming chunk",
1705 elog(
ERROR,
"unexpected apply action: %d", (
int) apply_action);
1826 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
1833 xid = abort_data.
xid;
1834 subxid = abort_data.
subxid;
1835 toplevel_xact = (xid == subxid);
1841 switch (apply_action)
1851 elog(
DEBUG1,
"finished processing the STREAM ABORT command");
1954 elog(
DEBUG1,
"finished processing the STREAM ABORT command");
1958 elog(
ERROR,
"unexpected apply action: %d", (
int) apply_action);
1992 if (last_fileno != fileno || last_offset != offset)
1993 elog(
ERROR,
"unexpected message left in streaming transaction's changes file \"%s\"",
2006 char *buffer = NULL;
2027 elog(
DEBUG1,
"replaying changes from file \"%s\"", path);
2077 elog(
ERROR,
"incorrect length %d in streaming transaction's changes file \"%s\"",
2113 if (nchanges % 1000 == 0)
2114 elog(
DEBUG1,
"replayed %d changes from file \"%s\"",
2121 elog(
DEBUG1,
"replayed %d (all) changes from file \"%s\"",
2143 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
2151 switch (apply_action)
2167 elog(
DEBUG1,
"finished processing the STREAM COMMIT command");
2221 elog(
DEBUG1,
"finished processing the STREAM COMMIT command");
2225 elog(
ERROR,
"unexpected apply action: %d", (
int) apply_action);
2360 (
errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
2361 errmsg(
"user \"%s\" cannot replicate into relation with row-level security enabled: \"%s\"",
2499 (
errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
2500 errmsg(
"publisher did not send replica identity column "
2501 "expected by the logical replication target relation \"%s.%s\"",
2506 (
errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
2507 errmsg(
"logical replication target relation \"%s.%s\" has "
2508 "neither REPLICA IDENTITY index nor PRIMARY "
2509 "KEY and published relation does not have "
2510 "REPLICA IDENTITY FULL",
2594 if (!att->attisdropped && remoteattnum >= 0)
2607 has_oldtup ? &oldtup : &newtup);
2657 remoteslot, &localslot);
2688 "logical replication did not find row to be updated "
2689 "in replication target relation \"%s\"",
2808 remoteslot, &localslot);
2828 "logical replication did not find row to be deleted "
2829 "in replication target relation \"%s\"",
2864 (remoterel->
replident == REPLICA_IDENTITY_FULL));
2868 #ifdef USE_ASSERT_CHECKING
2880 remoteslot, *localslot);
2884 remoteslot, *localslot);
2925 Assert(remoteslot != NULL);
2928 remoteslot, estate);
2929 Assert(partrelinfo != NULL);
2947 if (remoteslot_part == NULL)
2958 remoteslot_part =
ExecCopySlot(remoteslot_part, remoteslot);
3002 remoteslot_part, &localslot);
3012 "logical replication did not find row to be updated "
3013 "in replication target relation's partition \"%s\"",
3031 if (!partrel->
rd_rel->relispartition ||
3051 localslot, remoteslot_part);
3074 remoteslot_part, remoteslot);
3078 remoteslot =
ExecCopySlot(remoteslot, remoteslot_part);
3088 Assert(partrelinfo_new != partrelinfo);
3109 if (remoteslot_part == NULL)
3133 elog(
ERROR,
"unrecognized CmdType: %d", (
int) operation);
3146 bool cascade =
false;
3147 bool restart_seqs =
false;
3169 foreach(lc, remote_relids)
3185 remote_rels =
lappend(remote_rels, rel);
3203 foreach(child, children)
3225 rels =
lappend(rels, childrel);
3226 part_rels =
lappend(part_rels, childrel);
3230 relids_logged =
lappend_oid(relids_logged, childrelid);
3251 foreach(lc, remote_rels)
3257 foreach(lc, part_rels)
3370 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
3371 errmsg(
"invalid logical replication message type \"??? (%d)\"",
action)));
3393 bool *have_pending_txes)
3424 *have_pending_txes =
true;
3481 bool ping_sent =
false;
3490 "ApplyMessageContext",
3498 "LogicalStreamingContext",
3520 bool endofstream =
false;
3543 (
errmsg(
"data stream from publisher has ended")));
3579 if (last_received < start_lsn)
3580 last_received = start_lsn;
3582 if (last_received < end_lsn)
3583 last_received = end_lsn;
3593 bool reply_requested;
3599 if (last_received < end_lsn)
3600 last_received = end_lsn;
3655 WAIT_EVENT_LOGICAL_APPLY_MAIN);
3679 bool requestReply =
false;
3696 (
errcode(ERRCODE_CONNECTION_FAILURE),
3697 errmsg(
"terminating logical replication worker due to timeout")));
3706 requestReply =
true;
3754 bool have_pending_txes;
3764 if (recvpos < last_recvpos)
3765 recvpos = last_recvpos;
3773 if (!have_pending_txes)
3774 flushpos = writepos = recvpos;
3776 if (writepos < last_writepos)
3777 writepos = last_writepos;
3779 if (flushpos < last_flushpos)
3780 flushpos = last_flushpos;
3786 writepos == last_writepos &&
3787 flushpos == last_flushpos &&
3810 elog(
DEBUG2,
"sending feedback (force %d) to recv %X/%X, write %X/%X, flush %X/%X",
3819 if (recvpos > last_recvpos)
3820 last_recvpos = recvpos;
3821 if (writepos > last_writepos)
3822 last_writepos = writepos;
3823 if (flushpos > last_flushpos)
3824 last_flushpos = flushpos;
3866 bool started_tx =
false;
3891 (
errmsg(
"logical replication worker for subscription \"%s\" will stop because the subscription was removed",
3905 (
errmsg(
"logical replication worker for subscription \"%s\" will stop because the subscription was disabled",
3936 (
errmsg(
"logical replication parallel apply worker for subscription \"%s\" will stop because of a parameter change",
3940 (
errmsg(
"logical replication worker for subscription \"%s\" will restart because of a parameter change",
3954 errmsg(
"logical replication parallel apply worker for subscription \"%s\" will stop because the subscription owner's superuser privileges have been revoked",
3958 errmsg(
"logical replication worker for subscription \"%s\" will restart because the subscription owner's superuser privileges have been revoked",
3967 elog(
ERROR,
"subscription %u changed unexpectedly",
4142 if (subxacts[
i - 1].xid == xid)
4240 elog(
DEBUG1,
"opening file \"%s\" for streamed changes", path);
4262 path, O_RDWR,
false);
4345 options->startpoint = *origin_startpos;
4349 options->proto.logical.proto_version =
4365 options->proto.logical.streaming_str =
"parallel";
4371 options->proto.logical.streaming_str =
"on";
4376 options->proto.logical.streaming_str = NULL;
4380 options->proto.logical.twophase =
false;
4411 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
4414 snprintf(gid, szgid,
"pg_gid_%u_%u", subid, xid);
4461 char *slotname = NULL;
4466 bool must_use_password;
4477 (
errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
4478 errmsg(
"subscription has no replication slot set")));
4482 originname,
sizeof(originname));
4497 true, must_use_password,
4502 (
errcode(ERRCODE_CONNECTION_FAILURE),
4503 errmsg(
"could not connect to the publisher: %s",
err)));
4528 options.proto.logical.twophase =
true;
4542 (
errmsg_internal(
"logical replication apply worker for subscription \"%s\" two_phase is %s",
4591 (
errmsg(
"logical replication worker for subscription %u will not start because the subscription was removed during startup",
4607 (
errmsg(
"logical replication worker for subscription \"%s\" will not start because the subscription was disabled during startup",
4631 (
errmsg(
"logical replication table synchronization worker for subscription \"%s\", table \"%s\" has started",
4636 (
errmsg(
"logical replication apply worker for subscription \"%s\" has started",
4671 elog(
DEBUG1,
"connecting to publisher using connection string \"%s\"",
4734 errmsg(
"subscription \"%s\" has been disabled because of an error",
4782 errmsg(
"logical replication starts skipping transaction at LSN %X/%X",
4796 (
errmsg(
"logical replication completed skipping transaction at LSN %X/%X",
4818 bool started_tx =
false;
4858 if (subform->subskiplsn == myskiplsn)
4860 bool nulls[Natts_pg_subscription];
4861 bool replaces[Natts_pg_subscription];
4865 memset(nulls,
false,
sizeof(nulls));
4866 memset(replaces,
false,
sizeof(replaces));
4870 replaces[Anum_pg_subscription_subskiplsn - 1] =
true;
4876 if (myskiplsn != finish_lsn)
4879 errdetail(
"Remote transaction's finish WAL location (LSN) %X/%X did not match skip-LSN %X/%X.",
4902 if (errarg->
rel == NULL)
4905 errcontext(
"processing remote data for replication origin \"%s\" during message type \"%s\"",
4909 errcontext(
"processing remote data for replication origin \"%s\" during message type \"%s\" in transaction %u",
4914 errcontext(
"processing remote data for replication origin \"%s\" during message type \"%s\" in transaction %u, finished at %X/%X",
4925 errcontext(
"processing remote data for replication origin \"%s\" during message type \"%s\" for replication target relation \"%s.%s\" in transaction %u",
4932 errcontext(
"processing remote data for replication origin \"%s\" during message type \"%s\" for replication target relation \"%s.%s\" in transaction %u, finished at %X/%X",
4943 errcontext(
"processing remote data for replication origin \"%s\" during message type \"%s\" for replication target relation \"%s.%s\" column \"%s\" in transaction %u",
4951 errcontext(
"processing remote data for replication origin \"%s\" during message type \"%s\" for replication target relation \"%s.%s\" column \"%s\" in transaction %u, finished at %X/%X",
5017 foreach(lc2, workers)
void aclcheck_error(AclResult aclerr, ObjectType objtype, const char *objectname)
AclResult pg_class_aclcheck(Oid table_oid, Oid roleid, AclMode mode)
void pa_set_xact_state(ParallelApplyWorkerShared *wshared, ParallelTransState xact_state)
void pa_unlock_stream(TransactionId xid, LOCKMODE lockmode)
ParallelApplyWorkerInfo * pa_find_worker(TransactionId xid)
void pa_stream_abort(LogicalRepStreamAbortData *abort_data)
void pa_lock_stream(TransactionId xid, LOCKMODE lockmode)
void pa_set_fileset_state(ParallelApplyWorkerShared *wshared, PartialFileSetState fileset_state)
void pa_reset_subtrans(void)
void pa_lock_transaction(TransactionId xid, LOCKMODE lockmode)
ParallelApplyWorkerShared * MyParallelShared
void pa_start_subtrans(TransactionId current_xid, TransactionId top_xid)
void pa_switch_to_partial_serialize(ParallelApplyWorkerInfo *winfo, bool stream_locked)
void pa_xact_finish(ParallelApplyWorkerInfo *winfo, XLogRecPtr remote_lsn)
bool pa_send_data(ParallelApplyWorkerInfo *winfo, Size nbytes, const void *data)
void pa_allocate_worker(TransactionId xid)
void pa_set_stream_apply_worker(ParallelApplyWorkerInfo *winfo)
void pa_unlock_transaction(TransactionId xid, LOCKMODE lockmode)
void pa_decr_and_wait_stream_block(void)
static uint32 pg_atomic_add_fetch_u32(volatile pg_atomic_uint32 *ptr, int32 add_)
static void check_relation_updatable(LogicalRepRelMapEntry *rel)
static void subxact_filename(char *path, Oid subid, TransactionId xid)
static void begin_replication_step(void)
static void end_replication_step(void)
static void cleanup_subxact_info(void)
void set_stream_options(WalRcvStreamOptions *options, char *slotname, XLogRecPtr *origin_startpos)
static void apply_handle_stream_prepare(StringInfo s)
static void apply_handle_insert_internal(ApplyExecutionData *edata, ResultRelInfo *relinfo, TupleTableSlot *remoteslot)
static void subxact_info_add(TransactionId xid)
static ApplyExecutionData * create_edata_for_relation(LogicalRepRelMapEntry *rel)
void stream_cleanup_files(Oid subid, TransactionId xid)
MemoryContext ApplyMessageContext
static bool should_apply_changes_for_rel(LogicalRepRelMapEntry *rel)
static void apply_handle_type(StringInfo s)
static void apply_handle_truncate(StringInfo s)
static void UpdateWorkerStats(XLogRecPtr last_lsn, TimestampTz send_time, bool reply)
static void TwoPhaseTransactionGid(Oid subid, TransactionId xid, char *gid, int szgid)
static void subscription_change_cb(Datum arg, int cacheid, uint32 hashvalue)
static TransApplyAction get_transaction_apply_action(TransactionId xid, ParallelApplyWorkerInfo **winfo)
@ TRANS_LEADER_SEND_TO_PARALLEL
@ TRANS_LEADER_PARTIAL_SERIALIZE
static bool handle_streamed_transaction(LogicalRepMsgType action, StringInfo s)
static void stream_open_and_write_change(TransactionId xid, char action, StringInfo s)
struct ApplyExecutionData ApplyExecutionData
static void changes_filename(char *path, Oid subid, TransactionId xid)
bool InitializingApplyWorker
static void apply_worker_exit(void)
static BufFile * stream_fd
static void apply_handle_update(StringInfo s)
void stream_stop_internal(TransactionId xid)
static void apply_handle_stream_commit(StringInfo s)
void start_apply(XLogRecPtr origin_startpos)
static void stop_skipping_changes(void)
struct ApplySubXactData ApplySubXactData
#define NAPTIME_PER_CYCLE
static bool FindReplTupleInLocalRel(ApplyExecutionData *edata, Relation localrel, LogicalRepRelation *remoterel, Oid localidxoid, TupleTableSlot *remoteslot, TupleTableSlot **localslot)
static void get_flush_position(XLogRecPtr *write, XLogRecPtr *flush, bool *have_pending_txes)
static uint32 parallel_stream_nchanges
static void apply_handle_commit_prepared(StringInfo s)
static void LogicalRepApplyLoop(XLogRecPtr last_received)
void LogicalRepWorkersWakeupAtCommit(Oid subid)
bool IsLogicalWorker(void)
static ApplySubXactData subxact_data
static void apply_handle_tuple_routing(ApplyExecutionData *edata, TupleTableSlot *remoteslot, LogicalRepTupleData *newtup, CmdType operation)
ApplyErrorCallbackArg apply_error_callback_arg
bool in_remote_transaction
static XLogRecPtr skip_xact_finish_lsn
static void stream_open_file(Oid subid, TransactionId xid, bool first_segment)
static void apply_handle_delete(StringInfo s)
void apply_dispatch(StringInfo s)
#define is_skipping_changes()
static void stream_write_change(char action, StringInfo s)
static void clear_subscription_skip_lsn(XLogRecPtr finish_lsn)
static void apply_handle_update_internal(ApplyExecutionData *edata, ResultRelInfo *relinfo, TupleTableSlot *remoteslot, LogicalRepTupleData *newtup, Oid localindexoid)
static void ensure_last_message(FileSet *stream_fileset, TransactionId xid, int fileno, off_t offset)
static void apply_handle_begin(StringInfo s)
void DisableSubscriptionAndExit(void)
static dlist_head lsn_mapping
bool IsLogicalParallelApplyWorker(void)
void AtEOXact_LogicalRepWorkers(bool isCommit)
static void slot_store_data(TupleTableSlot *slot, LogicalRepRelMapEntry *rel, LogicalRepTupleData *tupleData)
void ReplicationOriginNameForLogicalRep(Oid suboid, Oid relid, char *originname, Size szoriginname)
static void finish_edata(ApplyExecutionData *edata)
static void slot_modify_data(TupleTableSlot *slot, TupleTableSlot *srcslot, LogicalRepRelMapEntry *rel, LogicalRepTupleData *tupleData)
static void set_apply_error_context_xact(TransactionId xid, XLogRecPtr lsn)
ErrorContextCallback * apply_error_context_stack
static void stream_abort_internal(TransactionId xid, TransactionId subxid)
static void apply_handle_commit(StringInfo s)
void stream_start_internal(TransactionId xid, bool first_segment)
static List * on_commit_wakeup_workers_subids
static void apply_handle_stream_abort(StringInfo s)
static void apply_handle_relation(StringInfo s)
void set_apply_error_context_origin(char *originname)
struct ApplyErrorCallbackArg ApplyErrorCallbackArg
MemoryContext ApplyContext
static void subxact_info_write(Oid subid, TransactionId xid)
static void TargetPrivilegesCheck(Relation rel, AclMode mode)
static void apply_handle_prepare(StringInfo s)
static void apply_handle_rollback_prepared(StringInfo s)
static void run_apply_worker()
void SetupApplyOrSyncWorker(int worker_slot)
static void apply_handle_stream_stop(StringInfo s)
static void apply_handle_origin(StringInfo s)
static void send_feedback(XLogRecPtr recvpos, bool force, bool requestReply)
WalReceiverConn * LogRepWorkerWalRcvConn
static XLogRecPtr remote_final_lsn
static bool MySubscriptionValid
void apply_error_callback(void *arg)
void store_flush_position(XLogRecPtr remote_lsn, XLogRecPtr local_lsn)
static MemoryContext LogicalStreamingContext
void maybe_reread_subscription(void)
static void apply_handle_commit_internal(LogicalRepCommitData *commit_data)
void InitializeLogRepWorker(void)
static bool in_streamed_transaction
struct SubXactInfo SubXactInfo
static void apply_handle_begin_prepare(StringInfo s)
struct FlushPosition FlushPosition
void ApplyWorkerMain(Datum main_arg)
void apply_spooled_messages(FileSet *stream_fileset, TransactionId xid, XLogRecPtr lsn)
static void apply_handle_stream_start(StringInfo s)
static void maybe_start_skipping_changes(XLogRecPtr finish_lsn)
Subscription * MySubscription
static void apply_handle_prepare_internal(LogicalRepPreparedTxnData *prepare_data)
static void stream_close_file(void)
static TransactionId stream_xid
static void apply_handle_insert(StringInfo s)
static void slot_fill_defaults(LogicalRepRelMapEntry *rel, EState *estate, TupleTableSlot *slot)
static void subxact_info_read(Oid subid, TransactionId xid)
static void apply_handle_delete_internal(ApplyExecutionData *edata, ResultRelInfo *relinfo, TupleTableSlot *remoteslot, Oid localindexoid)
static void reset_apply_error_context_info(void)
bool TimestampDifferenceExceeds(TimestampTz start_time, TimestampTz stop_time, int msec)
TimestampTz GetCurrentTimestamp(void)
Datum now(PG_FUNCTION_ARGS)
void pgstat_report_activity(BackendState state, const char *cmd_str)
@ STATE_IDLEINTRANSACTION
Bitmapset * bms_add_member(Bitmapset *a, int x)
static Datum values[MAXATTR]
void BufFileReadExact(BufFile *file, void *ptr, size_t size)
BufFile * BufFileOpenFileSet(FileSet *fileset, const char *name, int mode, bool missing_ok)
void BufFileTell(BufFile *file, int *fileno, off_t *offset)
void BufFileWrite(BufFile *file, const void *ptr, size_t size)
size_t BufFileReadMaybeEOF(BufFile *file, void *ptr, size_t size, bool eofOK)
void BufFileTruncateFileSet(BufFile *file, int fileno, off_t offset)
int BufFileSeek(BufFile *file, int fileno, off_t offset, int whence)
BufFile * BufFileCreateFileSet(FileSet *fileset, const char *name)
void BufFileClose(BufFile *file)
void BufFileDeleteFileSet(FileSet *fileset, const char *name, bool missing_ok)
#define OidIsValid(objectId)
void load_file(const char *filename, bool restricted)
int errmsg_internal(const char *fmt,...)
void EmitErrorReport(void)
int errdetail(const char *fmt,...)
ErrorContextCallback * error_context_stack
void FlushErrorState(void)
int errcode(int sqlerrcode)
int errmsg(const char *fmt,...)
#define ereport(elevel,...)
bool equal(const void *a, const void *b)
void err(int eval, const char *fmt,...)
ExprState * ExecInitExpr(Expr *node, PlanState *parent)
void ExecCloseIndices(ResultRelInfo *resultRelInfo)
void ExecOpenIndices(ResultRelInfo *resultRelInfo, bool speculative)
bool ExecPartitionCheck(ResultRelInfo *resultRelInfo, TupleTableSlot *slot, EState *estate, bool emitError)
void EvalPlanQualInit(EPQState *epqstate, EState *parentestate, Plan *subplan, List *auxrowmarks, int epqParam, List *resultRelations)
void InitResultRelInfo(ResultRelInfo *resultRelInfo, Relation resultRelationDesc, Index resultRelationIndex, ResultRelInfo *partition_root_rri, int instrument_options)
void EvalPlanQualEnd(EPQState *epqstate)
ResultRelInfo * ExecFindPartition(ModifyTableState *mtstate, ResultRelInfo *rootResultRelInfo, PartitionTupleRouting *proute, TupleTableSlot *slot, EState *estate)
PartitionTupleRouting * ExecSetupPartitionTupleRouting(EState *estate, Relation rel)
void ExecCleanupTupleRouting(ModifyTableState *mtstate, PartitionTupleRouting *proute)
bool RelationFindReplTupleSeq(Relation rel, LockTupleMode lockmode, TupleTableSlot *searchslot, TupleTableSlot *outslot)
bool RelationFindReplTupleByIndex(Relation rel, Oid idxoid, LockTupleMode lockmode, TupleTableSlot *searchslot, TupleTableSlot *outslot)
void ExecSimpleRelationDelete(ResultRelInfo *resultRelInfo, EState *estate, EPQState *epqstate, TupleTableSlot *searchslot)
void CheckSubscriptionRelkind(char relkind, const char *nspname, const char *relname)
void ExecSimpleRelationUpdate(ResultRelInfo *resultRelInfo, EState *estate, EPQState *epqstate, TupleTableSlot *searchslot, TupleTableSlot *slot)
void ExecSimpleRelationInsert(ResultRelInfo *resultRelInfo, EState *estate, TupleTableSlot *slot)
void ExecResetTupleTable(List *tupleTable, bool shouldFree)
const TupleTableSlotOps TTSOpsVirtual
TupleTableSlot * ExecStoreVirtualTuple(TupleTableSlot *slot)
TupleTableSlot * ExecInitExtraTupleSlot(EState *estate, TupleDesc tupledesc, const TupleTableSlotOps *tts_ops)
void ExecInitRangeTable(EState *estate, List *rangeTable, List *permInfos)
EState * CreateExecutorState(void)
TupleConversionMap * ExecGetRootToChildMap(ResultRelInfo *resultRelInfo, EState *estate)
void FreeExecutorState(EState *estate)
#define GetPerTupleExprContext(estate)
#define GetPerTupleMemoryContext(estate)
#define EvalPlanQualSetSlot(epqstate, slot)
static Datum ExecEvalExpr(ExprState *state, ExprContext *econtext, bool *isNull)
void FileSetInit(FileSet *fileset)
Datum OidReceiveFunctionCall(Oid functionId, StringInfo buf, Oid typioparam, int32 typmod)
Datum OidInputFunctionCall(Oid functionId, char *str, Oid typioparam, int32 typmod)
void SetConfigOption(const char *name, const char *value, GucContext context, GucSource source)
void ProcessConfigFile(GucContext context)
HeapTuple heap_modify_tuple(HeapTuple tuple, TupleDesc tupleDesc, const Datum *replValues, const bool *replIsnull, const bool *doReplace)
void heap_freetuple(HeapTuple htup)
#define HeapTupleIsValid(tuple)
static void dlist_delete(dlist_node *node)
#define dlist_tail_element(type, membername, lhead)
#define dlist_foreach_modify(iter, lhead)
static bool dlist_is_empty(const dlist_head *head)
static void dlist_push_tail(dlist_head *head, dlist_node *node)
#define DLIST_STATIC_INIT(name)
#define dlist_container(type, membername, ptr)
IndexInfo * BuildIndexInfo(Relation index)
void index_close(Relation relation, LOCKMODE lockmode)
Relation index_open(Oid relationId, LOCKMODE lockmode)
void CatalogTupleUpdate(Relation heapRel, ItemPointer otid, HeapTuple tup)
volatile sig_atomic_t ConfigReloadPending
void SignalHandlerForConfigReload(SIGNAL_ARGS)
void AcceptInvalidationMessages(void)
void CacheRegisterSyscacheCallback(int cacheid, SyscacheCallbackFunction func, Datum arg)
int WaitLatchOrSocket(Latch *latch, int wakeEvents, pgsocket sock, long timeout, uint32 wait_event_info)
void ResetLatch(Latch *latch)
#define WL_SOCKET_READABLE
#define WL_EXIT_ON_PM_DEATH
void logicalrep_worker_wakeup_ptr(LogicalRepWorker *worker)
void logicalrep_worker_attach(int slot)
List * logicalrep_workers_find(Oid subid, bool only_running)
void logicalrep_worker_wakeup(Oid subid, Oid relid)
LogicalRepWorker * MyLogicalRepWorker
void ApplyLauncherForgetWorkerStartTime(Oid subid)
Assert(fmt[strlen(fmt) - 1] !='\n')
List * lappend(List *list, void *datum)
List * lappend_oid(List *list, Oid datum)
bool list_member_oid(const List *list, Oid datum)
List * list_append_unique_oid(List *list, Oid datum)
void LockSharedObject(Oid classid, Oid objid, uint16 objsubid, LOCKMODE lockmode)
#define AccessExclusiveLock
#define LOGICALREP_PROTO_STREAM_PARALLEL_VERSION_NUM
#define LOGICALREP_PROTO_STREAM_VERSION_NUM
#define LOGICALREP_PROTO_TWOPHASE_VERSION_NUM
#define LOGICALREP_COLUMN_UNCHANGED
@ LOGICAL_REP_MSG_TRUNCATE
@ LOGICAL_REP_MSG_STREAM_STOP
@ LOGICAL_REP_MSG_STREAM_PREPARE
@ LOGICAL_REP_MSG_STREAM_ABORT
@ LOGICAL_REP_MSG_BEGIN_PREPARE
@ LOGICAL_REP_MSG_STREAM_START
@ LOGICAL_REP_MSG_PREPARE
@ LOGICAL_REP_MSG_RELATION
@ LOGICAL_REP_MSG_MESSAGE
@ LOGICAL_REP_MSG_ROLLBACK_PREPARED
@ LOGICAL_REP_MSG_COMMIT_PREPARED
@ LOGICAL_REP_MSG_STREAM_COMMIT
#define LOGICALREP_PROTO_VERSION_NUM
#define LOGICALREP_COLUMN_BINARY
#define LOGICALREP_COLUMN_TEXT
char * get_namespace_name(Oid nspid)
char * get_rel_name(Oid relid)
void getTypeInputInfo(Oid type, Oid *typInput, Oid *typIOParam)
void getTypeBinaryInputInfo(Oid type, Oid *typReceive, Oid *typIOParam)
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
void LWLockRelease(LWLock *lock)
void MemoryContextReset(MemoryContext context)
MemoryContext TopTransactionContext
char * pstrdup(const char *in)
void pfree(void *pointer)
MemoryContext TopMemoryContext
void * palloc0(Size size)
void * repalloc(void *pointer, Size size)
char * MemoryContextStrdup(MemoryContext context, const char *string)
#define AllocSetContextCreate
#define ALLOCSET_DEFAULT_SIZES
#define RESUME_INTERRUPTS()
#define CHECK_FOR_INTERRUPTS()
#define HOLD_INTERRUPTS()
char * GetUserNameFromId(Oid roleid, bool noerr)
ObjectType get_relkind_objtype(char relkind)
TimestampTz replorigin_session_origin_timestamp
RepOriginId replorigin_by_name(const char *roname, bool missing_ok)
RepOriginId replorigin_create(const char *roname)
RepOriginId replorigin_session_origin
void replorigin_session_setup(RepOriginId node, int acquired_by)
XLogRecPtr replorigin_session_get_progress(bool flush)
XLogRecPtr replorigin_session_origin_lsn
#define InvalidRepOriginId
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
RTEPermissionInfo * addRTEPermissionInfo(List **rteperminfos, RangeTblEntry *rte)
FormData_pg_attribute * Form_pg_attribute
static PgChecksumMode mode
static int server_version
List * find_all_inheritors(Oid parentrelId, LOCKMODE lockmode, List **numparents)
static void * list_nth(const List *list, int n)
static Datum LSNGetDatum(XLogRecPtr X)
void FreeSubscription(Subscription *sub)
void DisableSubscription(Oid subid)
Subscription * GetSubscription(Oid subid, bool missing_ok)
#define LOGICALREP_STREAM_OFF
#define LOGICALREP_STREAM_PARALLEL
#define LOGICALREP_TWOPHASE_STATE_DISABLED
#define LOGICALREP_TWOPHASE_STATE_PENDING
#define LOGICALREP_TWOPHASE_STATE_ENABLED
FormData_pg_subscription * Form_pg_subscription
long pgstat_report_stat(bool force)
void pgstat_report_subscription_error(Oid subid, bool is_apply_error)
Expr * expression_planner(Expr *expr)
pqsigfunc pqsignal(int signo, pqsigfunc func)
static Datum ObjectIdGetDatum(Oid X)
static int32 DatumGetInt32(Datum X)
void BackgroundWorkerUnblockSignals(void)
void BackgroundWorkerInitializeConnectionByOid(Oid dboid, Oid useroid, uint32 flags)
static int fd(const char *x, int i)
void logicalrep_read_commit(StringInfo in, LogicalRepCommitData *commit_data)
LogicalRepRelId logicalrep_read_delete(StringInfo in, LogicalRepTupleData *oldtup)
void logicalrep_read_rollback_prepared(StringInfo in, LogicalRepRollbackPreparedTxnData *rollback_data)
const char * logicalrep_message_type(LogicalRepMsgType action)
void logicalrep_read_begin_prepare(StringInfo in, LogicalRepPreparedTxnData *begin_data)
List * logicalrep_read_truncate(StringInfo in, bool *cascade, bool *restart_seqs)
void logicalrep_read_typ(StringInfo in, LogicalRepTyp *ltyp)
LogicalRepRelId logicalrep_read_update(StringInfo in, bool *has_oldtuple, LogicalRepTupleData *oldtup, LogicalRepTupleData *newtup)
void logicalrep_read_stream_abort(StringInfo in, LogicalRepStreamAbortData *abort_data, bool read_abort_info)
void logicalrep_read_begin(StringInfo in, LogicalRepBeginData *begin_data)
void logicalrep_read_commit_prepared(StringInfo in, LogicalRepCommitPreparedTxnData *prepare_data)
LogicalRepRelation * logicalrep_read_rel(StringInfo in)
void logicalrep_read_stream_prepare(StringInfo in, LogicalRepPreparedTxnData *prepare_data)
TransactionId logicalrep_read_stream_commit(StringInfo in, LogicalRepCommitData *commit_data)
LogicalRepRelId logicalrep_read_insert(StringInfo in, LogicalRepTupleData *newtup)
void logicalrep_read_prepare(StringInfo in, LogicalRepPreparedTxnData *prepare_data)
TransactionId logicalrep_read_stream_start(StringInfo in, bool *first_segment)
static color newsub(struct colormap *cm, color co)
#define RelationGetRelid(relation)
#define RelationIsLogicallyLogged(relation)
#define RelationGetDescr(relation)
#define RelationGetRelationName(relation)
#define RELATION_IS_OTHER_TEMP(relation)
#define RelationGetNamespace(relation)
ResourceOwner TopTransactionResourceOwner
ResourceOwner CurrentResourceOwner
Node * build_column_default(Relation rel, int attrno)
int check_enable_rls(Oid relid, Oid checkAsUser, bool noError)
Snapshot GetTransactionSnapshot(void)
void PushActiveSnapshot(Snapshot snapshot)
void PopActiveSnapshot(void)
void logicalrep_partmap_reset_relmap(LogicalRepRelation *remoterel)
bool IsIndexUsableForReplicaIdentityFull(IndexInfo *indexInfo, AttrMap *attrmap)
LogicalRepRelMapEntry * logicalrep_partition_open(LogicalRepRelMapEntry *root, Relation partrel, AttrMap *map)
Oid GetRelationIdentityOrPK(Relation rel)
void logicalrep_relmap_update(LogicalRepRelation *remoterel)
LogicalRepRelMapEntry * logicalrep_rel_open(LogicalRepRelId remoteid, LOCKMODE lockmode)
void logicalrep_rel_close(LogicalRepRelMapEntry *rel, LOCKMODE lockmode)
StringInfo makeStringInfo(void)
void resetStringInfo(StringInfo str)
static void initReadOnlyStringInfo(StringInfo str, char *data, int len)
LogicalRepMsgType command
LogicalRepRelMapEntry * rel
ResultRelInfo * targetRelInfo
PartitionTupleRouting * proute
ModifyTableState * mtstate
LogicalRepRelMapEntry * targetRel
TransactionId subxact_last
List * es_opened_result_relations
struct ErrorContextCallback * previous
void(* callback)(void *arg)
LogicalRepRelation remoterel
XLogRecPtr prepare_end_lsn
XLogRecPtr rollback_end_lsn
TimestampTz rollback_time
StringInfoData * colvalues
TimestampTz last_recv_time
LogicalRepWorkerType type
TimestampTz last_send_time
ResultRelInfo * resultRelInfo
ParallelApplyWorkerShared * shared
pg_atomic_uint32 pending_stream_count
XLogRecPtr last_commit_end
TupleTableSlot * ri_PartitionTupleSlot
TupleDesc tts_tupleDescriptor
#define FirstLowInvalidHeapAttributeNumber
#define SearchSysCacheCopy1(cacheId, key1)
void table_close(Relation relation, LOCKMODE lockmode)
Relation table_open(Oid relationId, LOCKMODE lockmode)
TupleTableSlot * table_slot_create(Relation relation, List **reglist)
void ExecuteTruncateGuts(List *explicit_rels, List *relids, List *relids_logged, DropBehavior behavior, bool restart_seqs, bool run_as_table_owner)
bool AllTablesyncsReady(void)
void invalidate_syncing_table_states(Datum arg, int cacheid, uint32 hashvalue)
void process_syncing_tables(XLogRecPtr current_lsn)
void UpdateTwoPhaseState(Oid suboid, char new_state)
#define InvalidTransactionId
#define TransactionIdIsValid(xid)
void AfterTriggerEndQuery(EState *estate)
void AfterTriggerBeginQuery(void)
TupleTableSlot * execute_attr_map_slot(AttrMap *attrMap, TupleTableSlot *in_slot, TupleTableSlot *out_slot)
TupleConversionMap * convert_tuples_by_name(TupleDesc indesc, TupleDesc outdesc)
#define TupleDescAttr(tupdesc, i)
static TupleTableSlot * ExecClearTuple(TupleTableSlot *slot)
static TupleTableSlot * ExecCopySlot(TupleTableSlot *dstslot, TupleTableSlot *srcslot)
static void slot_getallattrs(TupleTableSlot *slot)
bool LookupGXact(const char *gid, XLogRecPtr prepare_end_lsn, TimestampTz origin_prepare_timestamp)
void FinishPreparedTransaction(const char *gid, bool isCommit)
void SwitchToUntrustedUser(Oid userid, UserContext *context)
void RestoreUserContext(UserContext *context)
#define TimestampTzPlusMilliseconds(tz, ms)
static StringInfoData reply_message
int wal_receiver_status_interval
#define walrcv_startstreaming(conn, options)
#define walrcv_connect(conninfo, replication, logical, must_use_password, appname, err)
#define walrcv_send(conn, buffer, nbytes)
#define walrcv_server_version(conn)
#define walrcv_endstreaming(conn, next_tli)
#define walrcv_identify_system(conn, primary_tli)
#define walrcv_receive(conn, buffer, wait_fd)
@ PARALLEL_TRANS_FINISHED
static bool am_parallel_apply_worker(void)
@ WORKERTYPE_PARALLEL_APPLY
static bool am_tablesync_worker(void)
static bool am_leader_apply_worker(void)
bool IsTransactionOrTransactionBlock(void)
bool PrepareTransactionBlock(const char *gid)
bool IsTransactionState(void)
void CommandCounterIncrement(void)
void StartTransactionCommand(void)
void SetCurrentStatementStartTimestamp(void)
bool IsTransactionBlock(void)
void BeginTransactionBlock(void)
void CommitTransactionCommand(void)
bool EndTransactionBlock(bool chain)
void AbortOutOfAnyTransaction(void)
CommandId GetCurrentCommandId(bool used)
XLogRecPtr GetFlushRecPtr(TimeLineID *insertTLI)
XLogRecPtr XactLastCommitEnd
#define LSN_FORMAT_ARGS(lsn)
#define XLogRecPtrIsInvalid(r)
#define InvalidXLogRecPtr