147 #include <sys/stat.h>
195 #define NAPTIME_PER_CYCLE 1000
336 #define is_skipping_changes() (unlikely(!XLogRecPtrIsInvalid(skip_xact_finish_lsn)))
426 char *originname,
Size szoriginname)
431 snprintf(originname, szoriginname,
"pg_%u_%u", suboid, relid);
436 snprintf(originname, szoriginname,
"pg_%u", suboid);
474 if (rel->
state != SUBREL_STATE_READY &&
475 rel->
state != SUBREL_STATE_UNKNOWN)
477 (
errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
478 errmsg(
"logical replication parallel apply worker for subscription \"%s\" will stop",
480 errdetail(
"Cannot handle streamed replication transactions using parallel apply workers until all tables have been synchronized.")));
482 return rel->
state == SUBREL_STATE_READY;
485 return (rel->
state == SUBREL_STATE_READY ||
486 (rel->
state == SUBREL_STATE_SYNCDONE &&
587 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
588 errmsg_internal(
"invalid transaction ID in streamed replication transaction")));
590 switch (apply_action)
637 elog(
ERROR,
"unexpected apply action: %d", (
int) apply_action);
741 int num_phys_attrs = desc->
natts;
755 defmap = (
int *)
palloc(num_phys_attrs *
sizeof(
int));
778 defmap[num_defaults] =
attnum;
783 for (
i = 0;
i < num_defaults;
i++)
804 for (
i = 0;
i < natts;
i++)
809 if (!att->attisdropped && remoteattnum >= 0)
813 Assert(remoteattnum < tupleData->ncols);
826 typioparam, att->atttypmod);
843 typioparam, att->atttypmod);
848 (
errcode(ERRCODE_INVALID_BINARY_REPRESENTATION),
849 errmsg(
"incorrect binary data format in logical replication column %d",
916 for (
i = 0;
i < natts;
i++)
921 if (remoteattnum < 0)
924 Assert(remoteattnum < tupleData->ncols);
941 typioparam, att->atttypmod);
958 typioparam, att->atttypmod);
963 (
errcode(ERRCODE_INVALID_BINARY_REPRESENTATION),
964 errmsg(
"incorrect binary data format in logical replication column %d",
1021 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
1022 errmsg_internal(
"incorrect commit LSN %X/%X in commit message (expected %X/%X)",
1046 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
1047 errmsg_internal(
"tablesync worker received a BEGIN PREPARE message")));
1113 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
1114 errmsg_internal(
"incorrect prepare LSN %X/%X in prepare message (expected %X/%X)",
1279 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
1285 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
1286 errmsg_internal(
"tablesync worker received a STREAM PREPARE message")));
1293 switch (apply_action)
1316 elog(
DEBUG1,
"finished processing the STREAM PREPARE command");
1374 elog(
DEBUG1,
"finished processing the STREAM PREPARE command");
1378 elog(
ERROR,
"unexpected apply action: %d", (
int) apply_action);
1415 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
1475 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
1489 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
1490 errmsg_internal(
"invalid transaction ID in streamed replication transaction")));
1500 switch (apply_action)
1586 elog(
ERROR,
"unexpected apply action: %d", (
int) apply_action);
1630 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
1635 switch (apply_action)
1672 elog(
DEBUG1,
"applied %u changes in the streaming chunk",
1702 elog(
ERROR,
"unexpected apply action: %d", (
int) apply_action);
1823 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
1830 xid = abort_data.
xid;
1831 subxid = abort_data.
subxid;
1832 toplevel_xact = (xid == subxid);
1838 switch (apply_action)
1848 elog(
DEBUG1,
"finished processing the STREAM ABORT command");
1951 elog(
DEBUG1,
"finished processing the STREAM ABORT command");
1955 elog(
ERROR,
"unexpected apply action: %d", (
int) apply_action);
1989 if (last_fileno != fileno || last_offset != offset)
1990 elog(
ERROR,
"unexpected message left in streaming transaction's changes file \"%s\"",
2003 char *buffer = NULL;
2024 elog(
DEBUG1,
"replaying changes from file \"%s\"", path);
2074 elog(
ERROR,
"incorrect length %d in streaming transaction's changes file \"%s\"",
2110 if (nchanges % 1000 == 0)
2111 elog(
DEBUG1,
"replayed %d changes from file \"%s\"",
2118 elog(
DEBUG1,
"replayed %d (all) changes from file \"%s\"",
2140 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
2148 switch (apply_action)
2164 elog(
DEBUG1,
"finished processing the STREAM COMMIT command");
2218 elog(
DEBUG1,
"finished processing the STREAM COMMIT command");
2222 elog(
ERROR,
"unexpected apply action: %d", (
int) apply_action);
2357 (
errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
2358 errmsg(
"user \"%s\" cannot replicate into relation with row-level security enabled: \"%s\"",
2496 (
errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
2497 errmsg(
"publisher did not send replica identity column "
2498 "expected by the logical replication target relation \"%s.%s\"",
2503 (
errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
2504 errmsg(
"logical replication target relation \"%s.%s\" has "
2505 "neither REPLICA IDENTITY index nor PRIMARY "
2506 "KEY and published relation does not have "
2507 "REPLICA IDENTITY FULL",
2591 if (!att->attisdropped && remoteattnum >= 0)
2604 has_oldtup ? &oldtup : &newtup);
2654 remoteslot, &localslot);
2685 "logical replication did not find row to be updated "
2686 "in replication target relation \"%s\"",
2805 remoteslot, &localslot);
2825 "logical replication did not find row to be deleted "
2826 "in replication target relation \"%s\"",
2861 (remoterel->
replident == REPLICA_IDENTITY_FULL));
2865 #ifdef USE_ASSERT_CHECKING
2877 remoteslot, *localslot);
2881 remoteslot, *localslot);
2922 Assert(remoteslot != NULL);
2925 remoteslot, estate);
2926 Assert(partrelinfo != NULL);
2944 if (remoteslot_part == NULL)
2955 remoteslot_part =
ExecCopySlot(remoteslot_part, remoteslot);
2999 remoteslot_part, &localslot);
3009 "logical replication did not find row to be updated "
3010 "in replication target relation's partition \"%s\"",
3028 if (!partrel->
rd_rel->relispartition ||
3048 localslot, remoteslot_part);
3071 remoteslot_part, remoteslot);
3075 remoteslot =
ExecCopySlot(remoteslot, remoteslot_part);
3085 Assert(partrelinfo_new != partrelinfo);
3106 if (remoteslot_part == NULL)
3130 elog(
ERROR,
"unrecognized CmdType: %d", (
int) operation);
3143 bool cascade =
false;
3144 bool restart_seqs =
false;
3166 foreach(lc, remote_relids)
3182 remote_rels =
lappend(remote_rels, rel);
3200 foreach(child, children)
3222 rels =
lappend(rels, childrel);
3223 part_rels =
lappend(part_rels, childrel);
3227 relids_logged =
lappend_oid(relids_logged, childrelid);
3248 foreach(lc, remote_rels)
3254 foreach(lc, part_rels)
3367 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
3368 errmsg(
"invalid logical replication message type \"??? (%d)\"",
action)));
3390 bool *have_pending_txes)
3421 *have_pending_txes =
true;
3478 bool ping_sent =
false;
3487 "ApplyMessageContext",
3495 "LogicalStreamingContext",
3517 bool endofstream =
false;
3540 (
errmsg(
"data stream from publisher has ended")));
3576 if (last_received < start_lsn)
3577 last_received = start_lsn;
3579 if (last_received < end_lsn)
3580 last_received = end_lsn;
3590 bool reply_requested;
3596 if (last_received < end_lsn)
3597 last_received = end_lsn;
3652 WAIT_EVENT_LOGICAL_APPLY_MAIN);
3676 bool requestReply =
false;
3693 (
errcode(ERRCODE_CONNECTION_FAILURE),
3694 errmsg(
"terminating logical replication worker due to timeout")));
3703 requestReply =
true;
3751 bool have_pending_txes;
3761 if (recvpos < last_recvpos)
3762 recvpos = last_recvpos;
3770 if (!have_pending_txes)
3771 flushpos = writepos = recvpos;
3773 if (writepos < last_writepos)
3774 writepos = last_writepos;
3776 if (flushpos < last_flushpos)
3777 flushpos = last_flushpos;
3783 writepos == last_writepos &&
3784 flushpos == last_flushpos &&
3807 elog(
DEBUG2,
"sending feedback (force %d) to recv %X/%X, write %X/%X, flush %X/%X",
3816 if (recvpos > last_recvpos)
3817 last_recvpos = recvpos;
3818 if (writepos > last_writepos)
3819 last_writepos = writepos;
3820 if (flushpos > last_flushpos)
3821 last_flushpos = flushpos;
3863 bool started_tx =
false;
3888 (
errmsg(
"logical replication worker for subscription \"%s\" will stop because the subscription was removed",
3902 (
errmsg(
"logical replication worker for subscription \"%s\" will stop because the subscription was disabled",
3933 (
errmsg(
"logical replication parallel apply worker for subscription \"%s\" will stop because of a parameter change",
3937 (
errmsg(
"logical replication worker for subscription \"%s\" will restart because of a parameter change",
3951 errmsg(
"logical replication parallel apply worker for subscription \"%s\" will stop because the subscription owner's superuser privileges have been revoked",
3955 errmsg(
"logical replication worker for subscription \"%s\" will restart because the subscription owner's superuser privileges have been revoked",
3964 elog(
ERROR,
"subscription %u changed unexpectedly",
4139 if (subxacts[
i - 1].xid == xid)
4237 elog(
DEBUG1,
"opening file \"%s\" for streamed changes", path);
4259 path, O_RDWR,
false);
4342 options->startpoint = *origin_startpos;
4346 options->proto.logical.proto_version =
4362 options->proto.logical.streaming_str =
"parallel";
4368 options->proto.logical.streaming_str =
"on";
4373 options->proto.logical.streaming_str = NULL;
4377 options->proto.logical.twophase =
false;
4440 char *slotname = NULL;
4445 bool must_use_password;
4456 (
errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
4457 errmsg(
"subscription has no replication slot set")));
4461 originname,
sizeof(originname));
4476 true, must_use_password,
4481 (
errcode(ERRCODE_CONNECTION_FAILURE),
4482 errmsg(
"apply worker for subscription \"%s\" could not connect to the publisher: %s",
4508 options.proto.logical.twophase =
true;
4522 (
errmsg_internal(
"logical replication apply worker for subscription \"%s\" two_phase is %s",
4571 (
errmsg(
"logical replication worker for subscription %u will not start because the subscription was removed during startup",
4587 (
errmsg(
"logical replication worker for subscription \"%s\" will not start because the subscription was disabled during startup",
4611 (
errmsg(
"logical replication table synchronization worker for subscription \"%s\", table \"%s\" has started",
4616 (
errmsg(
"logical replication apply worker for subscription \"%s\" has started",
4651 elog(
DEBUG1,
"connecting to publisher using connection string \"%s\"",
4714 errmsg(
"subscription \"%s\" has been disabled because of an error",
4762 errmsg(
"logical replication starts skipping transaction at LSN %X/%X",
4776 (
errmsg(
"logical replication completed skipping transaction at LSN %X/%X",
4798 bool started_tx =
false;
4838 if (subform->subskiplsn == myskiplsn)
4840 bool nulls[Natts_pg_subscription];
4841 bool replaces[Natts_pg_subscription];
4845 memset(nulls,
false,
sizeof(nulls));
4846 memset(replaces,
false,
sizeof(replaces));
4850 replaces[Anum_pg_subscription_subskiplsn - 1] =
true;
4856 if (myskiplsn != finish_lsn)
4859 errdetail(
"Remote transaction's finish WAL location (LSN) %X/%X did not match skip-LSN %X/%X.",
4882 if (errarg->
rel == NULL)
4885 errcontext(
"processing remote data for replication origin \"%s\" during message type \"%s\"",
4889 errcontext(
"processing remote data for replication origin \"%s\" during message type \"%s\" in transaction %u",
4894 errcontext(
"processing remote data for replication origin \"%s\" during message type \"%s\" in transaction %u, finished at %X/%X",
4905 errcontext(
"processing remote data for replication origin \"%s\" during message type \"%s\" for replication target relation \"%s.%s\" in transaction %u",
4912 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",
4923 errcontext(
"processing remote data for replication origin \"%s\" during message type \"%s\" for replication target relation \"%s.%s\" column \"%s\" in transaction %u",
4931 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",
4997 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 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)
static 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
void BackgroundWorkerUnblockSignals(void)
void BackgroundWorkerInitializeConnectionByOid(Oid dboid, Oid useroid, uint32 flags)
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 Assert(condition)
#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)
List * logicalrep_workers_find(Oid subid, bool only_running, bool acquire_lock)
void logicalrep_worker_attach(int slot)
void logicalrep_worker_wakeup(Oid subid, Oid relid)
LogicalRepWorker * MyLogicalRepWorker
void ApplyLauncherForgetWorkerStartTime(Oid subid)
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
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)
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)
MemoryContextSwitchTo(old_ctx)
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)
void TwoPhaseTransactionGid(Oid subid, TransactionId xid, char *gid_res, int szgid)
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