140 #include <sys/stat.h>
201 #include "utils/fmgroids.h"
213 #define NAPTIME_PER_CYCLE 1000
354 #define is_skipping_changes() (unlikely(!XLogRecPtrIsInvalid(skip_xact_finish_lsn)))
447 char *originname,
Size szoriginname)
452 snprintf(originname, szoriginname,
"pg_%u_%u", suboid, relid);
457 snprintf(originname, szoriginname,
"pg_%u", suboid);
495 if (rel->
state != SUBREL_STATE_READY &&
496 rel->
state != SUBREL_STATE_UNKNOWN)
498 (
errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
499 errmsg(
"logical replication parallel apply worker for subscription \"%s\" will stop",
501 errdetail(
"Cannot handle streamed replication transactions using parallel apply workers until all tables have been synchronized.")));
503 return rel->
state == SUBREL_STATE_READY;
506 return (rel->
state == SUBREL_STATE_READY ||
507 (rel->
state == SUBREL_STATE_SYNCDONE &&
608 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
609 errmsg_internal(
"invalid transaction ID in streamed replication transaction")));
611 switch (apply_action)
658 elog(
ERROR,
"unexpected apply action: %d", (
int) apply_action);
762 int num_phys_attrs = desc->
natts;
776 defmap = (
int *)
palloc(num_phys_attrs *
sizeof(
int));
799 defmap[num_defaults] =
attnum;
804 for (
i = 0;
i < num_defaults;
i++)
825 for (
i = 0;
i < natts;
i++)
830 if (!att->attisdropped && remoteattnum >= 0)
834 Assert(remoteattnum < tupleData->ncols);
847 typioparam, att->atttypmod);
864 typioparam, att->atttypmod);
869 (
errcode(ERRCODE_INVALID_BINARY_REPRESENTATION),
870 errmsg(
"incorrect binary data format in logical replication column %d",
937 for (
i = 0;
i < natts;
i++)
942 if (remoteattnum < 0)
945 Assert(remoteattnum < tupleData->ncols);
962 typioparam, att->atttypmod);
979 typioparam, att->atttypmod);
984 (
errcode(ERRCODE_INVALID_BINARY_REPRESENTATION),
985 errmsg(
"incorrect binary data format in logical replication column %d",
1042 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
1043 errmsg_internal(
"incorrect commit LSN %X/%X in commit message (expected %X/%X)",
1067 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
1068 errmsg_internal(
"tablesync worker received a BEGIN PREPARE message")));
1134 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
1135 errmsg_internal(
"incorrect prepare LSN %X/%X in prepare message (expected %X/%X)",
1300 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
1306 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
1307 errmsg_internal(
"tablesync worker received a STREAM PREPARE message")));
1314 switch (apply_action)
1337 elog(
DEBUG1,
"finished processing the STREAM PREPARE command");
1395 elog(
DEBUG1,
"finished processing the STREAM PREPARE command");
1399 elog(
ERROR,
"unexpected apply action: %d", (
int) apply_action);
1436 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
1496 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
1510 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
1511 errmsg_internal(
"invalid transaction ID in streamed replication transaction")));
1521 switch (apply_action)
1607 elog(
ERROR,
"unexpected apply action: %d", (
int) apply_action);
1651 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
1656 switch (apply_action)
1693 elog(
DEBUG1,
"applied %u changes in the streaming chunk",
1723 elog(
ERROR,
"unexpected apply action: %d", (
int) apply_action);
1844 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
1851 xid = abort_data.
xid;
1852 subxid = abort_data.
subxid;
1853 toplevel_xact = (xid == subxid);
1859 switch (apply_action)
1869 elog(
DEBUG1,
"finished processing the STREAM ABORT command");
1972 elog(
DEBUG1,
"finished processing the STREAM ABORT command");
1976 elog(
ERROR,
"unexpected apply action: %d", (
int) apply_action);
2010 if (last_fileno != fileno || last_offset != offset)
2011 elog(
ERROR,
"unexpected message left in streaming transaction's changes file \"%s\"",
2025 char *buffer = NULL;
2046 elog(
DEBUG1,
"replaying changes from file \"%s\"", path);
2096 elog(
ERROR,
"incorrect length %d in streaming transaction's changes file \"%s\"",
2133 if (nchanges % 1000 == 0)
2134 elog(
DEBUG1,
"replayed %d changes from file \"%s\"",
2141 elog(
DEBUG1,
"replayed %d (all) changes from file \"%s\"",
2163 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
2171 switch (apply_action)
2187 elog(
DEBUG1,
"finished processing the STREAM COMMIT command");
2241 elog(
DEBUG1,
"finished processing the STREAM COMMIT command");
2245 elog(
ERROR,
"unexpected apply action: %d", (
int) apply_action);
2380 (
errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
2381 errmsg(
"user \"%s\" cannot replicate into relation with row-level security enabled: \"%s\"",
2519 (
errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
2520 errmsg(
"publisher did not send replica identity column "
2521 "expected by the logical replication target relation \"%s.%s\"",
2526 (
errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
2527 errmsg(
"logical replication target relation \"%s.%s\" has "
2528 "neither REPLICA IDENTITY index nor PRIMARY "
2529 "KEY and published relation does not have "
2530 "REPLICA IDENTITY FULL",
2614 if (!att->attisdropped && remoteattnum >= 0)
2627 has_oldtup ? &oldtup : &newtup);
2677 remoteslot, &localslot);
2708 "logical replication did not find row to be updated "
2709 "in replication target relation \"%s\"",
2828 remoteslot, &localslot);
2848 "logical replication did not find row to be deleted "
2849 "in replication target relation \"%s\"",
2884 (remoterel->
replident == REPLICA_IDENTITY_FULL));
2888 #ifdef USE_ASSERT_CHECKING
2900 remoteslot, *localslot);
2904 remoteslot, *localslot);
2945 Assert(remoteslot != NULL);
2948 remoteslot, estate);
2949 Assert(partrelinfo != NULL);
2967 if (remoteslot_part == NULL)
2978 remoteslot_part =
ExecCopySlot(remoteslot_part, remoteslot);
3022 remoteslot_part, &localslot);
3032 "logical replication did not find row to be updated "
3033 "in replication target relation's partition \"%s\"",
3051 if (!partrel->
rd_rel->relispartition ||
3071 localslot, remoteslot_part);
3094 remoteslot_part, remoteslot);
3098 remoteslot =
ExecCopySlot(remoteslot, remoteslot_part);
3108 Assert(partrelinfo_new != partrelinfo);
3129 if (remoteslot_part == NULL)
3153 elog(
ERROR,
"unrecognized CmdType: %d", (
int) operation);
3166 bool cascade =
false;
3167 bool restart_seqs =
false;
3189 foreach(lc, remote_relids)
3205 remote_rels =
lappend(remote_rels, rel);
3223 foreach(child, children)
3245 rels =
lappend(rels, childrel);
3246 part_rels =
lappend(part_rels, childrel);
3250 relids_logged =
lappend_oid(relids_logged, childrelid);
3271 foreach(lc, remote_rels)
3277 foreach(lc, part_rels)
3390 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
3391 errmsg(
"invalid logical replication message type \"??? (%d)\"",
action)));
3413 bool *have_pending_txes)
3444 *have_pending_txes =
true;
3501 bool ping_sent =
false;
3510 "ApplyMessageContext",
3518 "LogicalStreamingContext",
3540 bool endofstream =
false;
3563 (
errmsg(
"data stream from publisher has ended")));
3602 if (last_received < start_lsn)
3603 last_received = start_lsn;
3605 if (last_received < end_lsn)
3606 last_received = end_lsn;
3616 bool reply_requested;
3622 if (last_received < end_lsn)
3623 last_received = end_lsn;
3678 WAIT_EVENT_LOGICAL_APPLY_MAIN);
3702 bool requestReply =
false;
3719 (
errcode(ERRCODE_CONNECTION_FAILURE),
3720 errmsg(
"terminating logical replication worker due to timeout")));
3729 requestReply =
true;
3777 bool have_pending_txes;
3787 if (recvpos < last_recvpos)
3788 recvpos = last_recvpos;
3796 if (!have_pending_txes)
3797 flushpos = writepos = recvpos;
3799 if (writepos < last_writepos)
3800 writepos = last_writepos;
3802 if (flushpos < last_flushpos)
3803 flushpos = last_flushpos;
3809 writepos == last_writepos &&
3810 flushpos == last_flushpos &&
3833 elog(
DEBUG2,
"sending feedback (force %d) to recv %X/%X, write %X/%X, flush %X/%X",
3842 if (recvpos > last_recvpos)
3843 last_recvpos = recvpos;
3844 if (writepos > last_writepos)
3845 last_writepos = writepos;
3846 if (flushpos > last_flushpos)
3847 last_flushpos = flushpos;
3889 bool started_tx =
false;
3914 (
errmsg(
"logical replication worker for subscription \"%s\" will stop because the subscription was removed",
3928 (
errmsg(
"logical replication worker for subscription \"%s\" will stop because the subscription was disabled",
3959 (
errmsg(
"logical replication parallel apply worker for subscription \"%s\" will stop because of a parameter change",
3963 (
errmsg(
"logical replication worker for subscription \"%s\" will restart because of a parameter change",
3972 elog(
ERROR,
"subscription %u changed unexpectedly",
4147 if (subxacts[
i - 1].xid == xid)
4245 elog(
DEBUG1,
"opening file \"%s\" for streamed changes", path);
4267 path, O_RDWR,
false);
4350 options->startpoint = *origin_startpos;
4354 options->proto.logical.proto_version =
4370 options->proto.logical.streaming_str =
"parallel";
4376 options->proto.logical.streaming_str =
"on";
4381 options->proto.logical.streaming_str = NULL;
4385 options->proto.logical.twophase =
false;
4416 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
4419 snprintf(gid, szgid,
"pg_gid_%u_%u", subid, xid);
4466 char *slotname = NULL;
4471 bool must_use_password;
4482 (
errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
4483 errmsg(
"subscription has no replication slot set")));
4487 originname,
sizeof(originname));
4509 (
errcode(ERRCODE_CONNECTION_FAILURE),
4510 errmsg(
"could not connect to the publisher: %s",
err)));
4535 options.proto.logical.twophase =
true;
4549 (
errmsg_internal(
"logical replication apply worker for subscription \"%s\" two_phase is %s",
4598 (
errmsg(
"logical replication worker for subscription %u will not start because the subscription was removed during startup",
4614 (
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)
elog(ERROR, "%s: %s", p2, msg)
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)