33 (
errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
34 errmsg(
"must be superuser or replication role to use replication slots")));
55 if (immediately_reserve)
86 elog(
ERROR,
"return type must be a row type");
100 if (immediately_reserve)
128 bool find_startpoint)
187 elog(
ERROR,
"return type must be a row type");
202 memset(nulls, 0,
sizeof(nulls));
239 #define PG_GET_REPLICATION_SLOTS_COLS 13 251 (
errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
252 errmsg(
"set-valued function called in context that cannot accept a set")));
255 (
errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
256 errmsg(
"materialize mode required, but it is not allowed in this context")));
260 elog(
ERROR,
"return type must be a row type");
295 slot_contents = *slot;
298 memset(values, 0,
sizeof(values));
299 memset(nulls, 0,
sizeof(nulls));
429 failSeg = targetSeg +
Max(slotKeepSegs, keepSegs) + 1;
462 if (startlsn < moveto)
547 if (moveto <= ctx->reader->EndRecPtr)
621 (
errmsg(
"invalid target WAL LSN")));
625 elog(
ERROR,
"return type must be a row type");
642 (
errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
643 errmsg(
"replication slot \"%s\" cannot be advanced",
645 errdetail(
"This slot has never previously reserved WAL, or has been invalidated.")));
660 (
errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
661 errmsg(
"cannot advance replication slot to %X/%X, minimum is %X/%X",
715 elog(
ERROR,
"return type must be a row type");
745 first_slot_contents = *s;
756 (
errcode(ERRCODE_UNDEFINED_OBJECT),
757 errmsg(
"replication slot \"%s\" does not exist",
NameStr(*src_name))));
765 if (src_islogical != logical_slot)
767 (
errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
769 errmsg(
"cannot copy physical replication slot \"%s\" as a logical replication slot",
771 errmsg(
"cannot copy logical replication slot \"%s\" as a physical replication slot",
777 (
errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
778 errmsg(
"cannot copy a replication slot that doesn't reserve WAL")));
825 second_slot_contents = *src;
831 copy_xmin = second_slot_contents.
data.
xmin;
850 if (copy_restart_lsn < src_restart_lsn ||
851 src_islogical != copy_islogical ||
852 strcmp(copy_name,
NameStr(*src_name)) != 0)
854 (
errmsg(
"could not copy replication slot \"%s\"",
856 errdetail(
"The source replication slot was modified incompatibly during the copy operation.")));
861 (
errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
862 errmsg(
"cannot copy unfinished logical replication slot \"%s\"",
864 errhint(
"Retry when the source replication slot's confirmed_flush_lsn is valid.")));
882 #ifdef USE_ASSERT_CHECKING 894 if (logical_slot && !temporary)
void tuplestore_putvalues(Tuplestorestate *state, TupleDesc tdesc, Datum *values, bool *isnull)
static XLogRecPtr pg_logical_replication_slot_advance(XLogRecPtr moveto)
static void create_logical_replication_slot(char *name, char *plugin, bool temporary, XLogRecPtr restart_lsn, bool find_startpoint)
static const char * plugin
ReplicationSlotCtlData * ReplicationSlotCtl
void CheckSlotRequirements(void)
#define InvalidXLogRecPtr
#define IsA(nodeptr, _type_)
TypeFuncClass get_call_result_type(FunctionCallInfo fcinfo, Oid *resultTypeId, TupleDesc *resultTupleDesc)
int errhint(const char *fmt,...)
void wal_segment_close(XLogReaderState *state)
static void create_physical_replication_slot(char *name, bool immediately_reserve, bool temporary, XLogRecPtr restart_lsn)
#define PG_GET_REPLICATION_SLOTS_COLS
Datum pg_get_replication_slots(PG_FUNCTION_ARGS)
ResourceOwner CurrentResourceOwner
Datum pg_create_logical_replication_slot(PG_FUNCTION_ARGS)
#define tuplestore_donestoring(state)
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
void ReplicationSlotCreate(const char *name, bool db_specific, ReplicationSlotPersistency persistency)
ReplicationSlotPersistency persistency
void wal_segment_open(XLogReaderState *state, XLogSegNo nextSegNo, TimeLineID *tli_p)
int errcode(int sqlerrcode)
void ReplicationSlotSave(void)
XLogRecPtr GetFlushRecPtr(void)
HeapTuple heap_form_tuple(TupleDesc tupleDescriptor, Datum *values, bool *isnull)
#define PG_GETARG_BOOL(n)
Datum pg_copy_physical_replication_slot_a(PG_FUNCTION_ARGS)
ReplicationSlotPersistentData data
Datum pg_copy_logical_replication_slot_c(PG_FUNCTION_ARGS)
bool RecoveryInProgress(void)
#define OidIsValid(objectId)
void InvalidateSystemCaches(void)
XLogRecPtr confirmed_flush
void LWLockRelease(LWLock *lock)
Datum pg_create_physical_replication_slot(PG_FUNCTION_ARGS)
#define SpinLockAcquire(lock)
XLogSegNo XLogGetLastRemovedSegno(void)
void ReplicationSlotReserveWal(void)
void DecodingContextFindStartpoint(LogicalDecodingContext *ctx)
void ReplicationSlotsComputeRequiredLSN(void)
XLogRecord * XLogReadRecord(XLogReaderState *state, char **errormsg)
#define ObjectIdGetDatum(X)
void LogicalDecodingProcessRecord(LogicalDecodingContext *ctx, XLogReaderState *record)
Datum pg_copy_logical_replication_slot_a(PG_FUNCTION_ARGS)
static Datum copy_replication_slot(FunctionCallInfo fcinfo, bool logical_slot)
XLogRecPtr GetXLogReplayRecPtr(TimeLineID *replayTLI)
void ReplicationSlotPersist(void)
TransactionId effective_xmin
LogicalDecodingContext * CreateDecodingContext(XLogRecPtr start_lsn, List *output_plugin_options, bool fast_forward, XLogReaderRoutine *xl_routine, LogicalOutputPluginWriterPrepareWrite prepare_write, LogicalOutputPluginWriterWrite do_write, LogicalOutputPluginWriterUpdateProgress update_progress)
Datum pg_copy_physical_replication_slot_b(PG_FUNCTION_ARGS)
int errdetail(const char *fmt,...)
Datum pg_replication_slot_advance(PG_FUNCTION_ARGS)
TransactionId catalog_xmin
#define InvalidTransactionId
void XLogBeginRead(XLogReaderState *state, XLogRecPtr RecPtr)
void ReplicationSlotRelease(void)
#define SlotIsLogical(slot)
Datum Int64GetDatum(int64 X)
static void check_permissions(void)
#define XLogRecPtrIsInvalid(r)
#define SpinLockRelease(lock)
Tuplestorestate * tuplestore_begin_heap(bool randomAccess, bool interXact, int maxKBytes)
#define TransactionIdGetDatum(X)
Datum pg_copy_logical_replication_slot_b(PG_FUNCTION_ARGS)
Datum pg_drop_replication_slot(PG_FUNCTION_ARGS)
#define PG_RETURN_DATUM(x)
TransactionId effective_catalog_xmin
TimeLineID ThisTimeLineID
#define XLogMBVarToSegs(mbvar, wal_segsz_bytes)
#define ereport(elevel,...)
int max_slot_wal_keep_size_mb
XLogRecPtr GetXLogWriteRecPtr(void)
SetFunctionReturnMode returnMode
ReplicationSlot * MyReplicationSlot
int max_replication_slots
#define XLogSegNoOffsetToRecPtr(segno, offset, wal_segsz_bytes, dest)
#define Assert(condition)
void FreeDecodingContext(LogicalDecodingContext *ctx)
bool has_rolreplication(Oid roleid)
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
#define HeapTupleGetDatum(tuple)
void LogicalConfirmReceivedLocation(XLogRecPtr lsn)
MemoryContext ecxt_per_query_memory
Tuplestorestate * setResult
static Datum values[MAXATTR]
void ReplicationSlotDrop(const char *name, bool nowait)
int errmsg(const char *fmt,...)
int ReplicationSlotAcquire(const char *name, SlotAcquireBehavior behavior)
WALAvailability GetWALAvailability(XLogRecPtr targetLSN)
#define CStringGetTextDatum(s)
#define CHECK_FOR_INTERRUPTS()
ReplicationSlot replication_slots[1]
static XLogRecPtr pg_physical_replication_slot_advance(XLogRecPtr moveto)
XLogRecPtr invalidated_at
int read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen, XLogRecPtr targetRecPtr, char *cur_page)
void CheckLogicalDecodingRequirements(void)
LogicalDecodingContext * CreateInitDecodingContext(const char *plugin, List *output_plugin_options, bool need_full_snapshot, XLogRecPtr restart_lsn, XLogReaderRoutine *xl_routine, LogicalOutputPluginWriterPrepareWrite prepare_write, LogicalOutputPluginWriterWrite do_write, LogicalOutputPluginWriterUpdateProgress update_progress)
void ReplicationSlotsComputeRequiredXmin(bool already_locked)
void ReplicationSlotMarkDirty(void)
#define PG_GETARG_NAME(n)
#define XLByteToSeg(xlrp, logSegNo, wal_segsz_bytes)