PostgreSQL Source Code  git master
slotfuncs.c File Reference
#include "postgres.h"
#include "access/htup_details.h"
#include "access/xlog_internal.h"
#include "access/xlogutils.h"
#include "funcapi.h"
#include "miscadmin.h"
#include "replication/decode.h"
#include "replication/logical.h"
#include "replication/slot.h"
#include "utils/builtins.h"
#include "utils/inval.h"
#include "utils/pg_lsn.h"
#include "utils/resowner.h"
Include dependency graph for slotfuncs.c:

Go to the source code of this file.

Macros

#define PG_GET_REPLICATION_SLOTS_COLS   13
 

Functions

static void check_permissions (void)
 
static void create_physical_replication_slot (char *name, bool immediately_reserve, bool temporary, XLogRecPtr restart_lsn)
 
Datum pg_create_physical_replication_slot (PG_FUNCTION_ARGS)
 
static void create_logical_replication_slot (char *name, char *plugin, bool temporary, XLogRecPtr restart_lsn, bool find_startpoint)
 
Datum pg_create_logical_replication_slot (PG_FUNCTION_ARGS)
 
Datum pg_drop_replication_slot (PG_FUNCTION_ARGS)
 
Datum pg_get_replication_slots (PG_FUNCTION_ARGS)
 
static XLogRecPtr pg_physical_replication_slot_advance (XLogRecPtr moveto)
 
static XLogRecPtr pg_logical_replication_slot_advance (XLogRecPtr moveto)
 
Datum pg_replication_slot_advance (PG_FUNCTION_ARGS)
 
static Datum copy_replication_slot (FunctionCallInfo fcinfo, bool logical_slot)
 
Datum pg_copy_logical_replication_slot_a (PG_FUNCTION_ARGS)
 
Datum pg_copy_logical_replication_slot_b (PG_FUNCTION_ARGS)
 
Datum pg_copy_logical_replication_slot_c (PG_FUNCTION_ARGS)
 
Datum pg_copy_physical_replication_slot_a (PG_FUNCTION_ARGS)
 
Datum pg_copy_physical_replication_slot_b (PG_FUNCTION_ARGS)
 

Macro Definition Documentation

◆ PG_GET_REPLICATION_SLOTS_COLS

#define PG_GET_REPLICATION_SLOTS_COLS   13

Function Documentation

◆ check_permissions()

static void check_permissions ( void  )
static

Definition at line 29 of file slotfuncs.c.

References ereport, errcode(), errmsg(), ERROR, GetUserId(), has_rolreplication(), and superuser().

Referenced by copy_replication_slot(), pg_create_logical_replication_slot(), pg_create_physical_replication_slot(), pg_drop_replication_slot(), and pg_replication_slot_advance().

30 {
32  ereport(ERROR,
33  (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
34  errmsg("must be superuser or replication role to use replication slots")));
35 }
Oid GetUserId(void)
Definition: miscinit.c:448
int errcode(int sqlerrcode)
Definition: elog.c:610
bool superuser(void)
Definition: superuser.c:46
#define ERROR
Definition: elog.h:43
#define ereport(elevel,...)
Definition: elog.h:144
bool has_rolreplication(Oid roleid)
Definition: miscinit.c:629
int errmsg(const char *fmt,...)
Definition: elog.c:824

◆ copy_replication_slot()

static Datum copy_replication_slot ( FunctionCallInfo  fcinfo,
bool  logical_slot 
)
static

Definition at line 651 of file slotfuncs.c.

References Assert, ReplicationSlotPersistentData::catalog_xmin, check_permissions(), CheckLogicalDecodingRequirements(), CheckSlotRequirements(), ReplicationSlotPersistentData::confirmed_flush, create_logical_replication_slot(), create_physical_replication_slot(), ReplicationSlot::data, ReplicationSlot::effective_catalog_xmin, ReplicationSlot::effective_xmin, elog, ereport, errcode(), errdetail(), errhint(), errmsg(), ERROR, get_call_result_type(), heap_form_tuple(), HeapTupleGetDatum, i, ReplicationSlot::in_use, LSNGetDatum, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_replication_slots, ReplicationSlot::mutex, MyReplicationSlot, ReplicationSlotPersistentData::name, NameGetDatum, NameStr, ReplicationSlotPersistentData::persistency, PG_GETARG_BOOL, PG_GETARG_NAME, PG_NARGS, PG_RETURN_DATUM, plugin, ReplicationSlotPersistentData::plugin, pstrdup(), ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, ReplicationSlotMarkDirty(), ReplicationSlotPersist(), ReplicationSlotRelease(), ReplicationSlotSave(), ReplicationSlotsComputeRequiredLSN(), ReplicationSlotsComputeRequiredXmin(), ReplicationSlotPersistentData::restart_lsn, RS_TEMPORARY, SlotIsLogical, SpinLockAcquire, SpinLockRelease, TYPEFUNC_COMPOSITE, values, wal_segment_size, XLByteToSeg, XLogGetLastRemovedSegno(), XLogRecPtrIsInvalid, and ReplicationSlotPersistentData::xmin.

Referenced by pg_copy_logical_replication_slot_a(), pg_copy_logical_replication_slot_b(), pg_copy_logical_replication_slot_c(), pg_copy_physical_replication_slot_a(), and pg_copy_physical_replication_slot_b().

652 {
653  Name src_name = PG_GETARG_NAME(0);
654  Name dst_name = PG_GETARG_NAME(1);
655  ReplicationSlot *src = NULL;
656  XLogRecPtr src_restart_lsn;
657  bool src_islogical;
658  bool temporary;
659  char *plugin;
660  Datum values[2];
661  bool nulls[2];
662  Datum result;
663  TupleDesc tupdesc;
664  HeapTuple tuple;
665 
666  if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
667  elog(ERROR, "return type must be a row type");
668 
670 
671  if (logical_slot)
673  else
675 
676  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
677 
678  /*
679  * We need to prevent the source slot's reserved WAL from being removed,
680  * but we don't want to lock that slot for very long, and it can advance
681  * in the meantime. So obtain the source slot's data, and create a new
682  * slot using its restart_lsn. Afterwards we lock the source slot again
683  * and verify that the data we copied (name, type) has not changed
684  * incompatibly. No inconvenient WAL removal can occur once the new slot
685  * is created -- but since WAL removal could have occurred before we
686  * managed to create the new slot, we advance the new slot's restart_lsn
687  * to the source slot's updated restart_lsn the second time we lock it.
688  */
689  for (int i = 0; i < max_replication_slots; i++)
690  {
692 
693  if (s->in_use && strcmp(NameStr(s->data.name), NameStr(*src_name)) == 0)
694  {
695  SpinLockAcquire(&s->mutex);
696  src_islogical = SlotIsLogical(s);
697  src_restart_lsn = s->data.restart_lsn;
698  temporary = s->data.persistency == RS_TEMPORARY;
699  plugin = logical_slot ? pstrdup(NameStr(s->data.plugin)) : NULL;
700  SpinLockRelease(&s->mutex);
701 
702  src = s;
703  break;
704  }
705  }
706 
707  LWLockRelease(ReplicationSlotControlLock);
708 
709  if (src == NULL)
710  ereport(ERROR,
711  (errcode(ERRCODE_UNDEFINED_OBJECT),
712  errmsg("replication slot \"%s\" does not exist", NameStr(*src_name))));
713 
714  /* Check type of replication slot */
715  if (src_islogical != logical_slot)
716  ereport(ERROR,
717  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
718  src_islogical ?
719  errmsg("cannot copy physical replication slot \"%s\" as a logical replication slot",
720  NameStr(*src_name)) :
721  errmsg("cannot copy logical replication slot \"%s\" as a physical replication slot",
722  NameStr(*src_name))));
723 
724  /* Copying non-reserved slot doesn't make sense */
725  if (XLogRecPtrIsInvalid(src_restart_lsn))
726  {
727  Assert(!logical_slot);
728  ereport(ERROR,
729  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
730  errmsg("cannot copy a replication slot that doesn't reserve WAL")));
731  }
732 
733  /* Overwrite params from optional arguments */
734  if (PG_NARGS() >= 3)
735  temporary = PG_GETARG_BOOL(2);
736  if (PG_NARGS() >= 4)
737  {
738  Assert(logical_slot);
739  plugin = NameStr(*(PG_GETARG_NAME(3)));
740  }
741 
742  /* Create new slot and acquire it */
743  if (logical_slot)
744  {
745  /*
746  * We must not try to read WAL, since we haven't reserved it yet --
747  * hence pass find_startpoint false. confirmed_flush will be set
748  * below, by copying from the source slot.
749  */
751  plugin,
752  temporary,
753  src_restart_lsn,
754  false);
755  }
756  else
758  true,
759  temporary,
760  src_restart_lsn);
761 
762  /*
763  * Update the destination slot to current values of the source slot;
764  * recheck that the source slot is still the one we saw previously.
765  */
766  {
767  TransactionId copy_effective_xmin;
768  TransactionId copy_effective_catalog_xmin;
769  TransactionId copy_xmin;
770  TransactionId copy_catalog_xmin;
771  XLogRecPtr copy_restart_lsn;
772  XLogRecPtr copy_confirmed_flush;
773  bool copy_islogical;
774  char *copy_name;
775 
776  /* Copy data of source slot again */
777  SpinLockAcquire(&src->mutex);
778  copy_effective_xmin = src->effective_xmin;
779  copy_effective_catalog_xmin = src->effective_catalog_xmin;
780 
781  copy_xmin = src->data.xmin;
782  copy_catalog_xmin = src->data.catalog_xmin;
783  copy_restart_lsn = src->data.restart_lsn;
784  copy_confirmed_flush = src->data.confirmed_flush;
785 
786  /* for existence check */
787  copy_name = pstrdup(NameStr(src->data.name));
788  copy_islogical = SlotIsLogical(src);
789  SpinLockRelease(&src->mutex);
790 
791  /*
792  * Check if the source slot still exists and is valid. We regard it as
793  * invalid if the type of replication slot or name has been changed,
794  * or the restart_lsn either is invalid or has gone backward. (The
795  * restart_lsn could go backwards if the source slot is dropped and
796  * copied from an older slot during installation.)
797  *
798  * Since erroring out will release and drop the destination slot we
799  * don't need to release it here.
800  */
801  if (copy_restart_lsn < src_restart_lsn ||
802  src_islogical != copy_islogical ||
803  strcmp(copy_name, NameStr(*src_name)) != 0)
804  ereport(ERROR,
805  (errmsg("could not copy replication slot \"%s\"",
806  NameStr(*src_name)),
807  errdetail("The source replication slot was modified incompatibly during the copy operation.")));
808 
809  /* The source slot must have a consistent snapshot */
810  if (src_islogical && XLogRecPtrIsInvalid(copy_confirmed_flush))
811  ereport(ERROR,
812  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
813  errmsg("cannot copy unfinished logical replication slot \"%s\"",
814  NameStr(*src_name)),
815  errhint("Retry when the source replication slot's confirmed_flush_lsn is valid.")));
816 
817  /* Install copied values again */
819  MyReplicationSlot->effective_xmin = copy_effective_xmin;
820  MyReplicationSlot->effective_catalog_xmin = copy_effective_catalog_xmin;
821 
822  MyReplicationSlot->data.xmin = copy_xmin;
823  MyReplicationSlot->data.catalog_xmin = copy_catalog_xmin;
824  MyReplicationSlot->data.restart_lsn = copy_restart_lsn;
825  MyReplicationSlot->data.confirmed_flush = copy_confirmed_flush;
827 
832 
833 #ifdef USE_ASSERT_CHECKING
834  /* Check that the restart_lsn is available */
835  {
836  XLogSegNo segno;
837 
838  XLByteToSeg(copy_restart_lsn, segno, wal_segment_size);
839  Assert(XLogGetLastRemovedSegno() < segno);
840  }
841 #endif
842  }
843 
844  /* target slot fully created, mark as persistent if needed */
845  if (logical_slot && !temporary)
847 
848  /* All done. Set up the return values */
849  values[0] = NameGetDatum(dst_name);
850  nulls[0] = false;
852  {
854  nulls[1] = false;
855  }
856  else
857  nulls[1] = true;
858 
859  tuple = heap_form_tuple(tupdesc, values, nulls);
860  result = HeapTupleGetDatum(tuple);
861 
863 
864  PG_RETURN_DATUM(result);
865 }
static void create_logical_replication_slot(char *name, char *plugin, bool temporary, XLogRecPtr restart_lsn, bool find_startpoint)
Definition: slotfuncs.c:126
static const char * plugin
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
void CheckSlotRequirements(void)
Definition: slot.c:988
#define NameGetDatum(X)
Definition: postgres.h:595
TypeFuncClass get_call_result_type(FunctionCallInfo fcinfo, Oid *resultTypeId, TupleDesc *resultTupleDesc)
Definition: funcapi.c:205
int errhint(const char *fmt,...)
Definition: elog.c:1071
static void create_physical_replication_slot(char *name, bool immediately_reserve, bool temporary, XLogRecPtr restart_lsn)
Definition: slotfuncs.c:46
int wal_segment_size
Definition: xlog.c:116
uint32 TransactionId
Definition: c.h:513
char * pstrdup(const char *in)
Definition: mcxt.c:1186
ReplicationSlotPersistency persistency
Definition: slot.h:61
int errcode(int sqlerrcode)
Definition: elog.c:610
#define LSNGetDatum(X)
Definition: pg_lsn.h:22
void ReplicationSlotSave(void)
Definition: slot.c:654
HeapTuple heap_form_tuple(TupleDesc tupleDescriptor, Datum *values, bool *isnull)
Definition: heaptuple.c:1020
#define PG_GETARG_BOOL(n)
Definition: fmgr.h:274
ReplicationSlotPersistentData data
Definition: slot.h:140
XLogRecPtr confirmed_flush
Definition: slot.h:88
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1812
#define SpinLockAcquire(lock)
Definition: spin.h:62
XLogSegNo XLogGetLastRemovedSegno(void)
Definition: xlog.c:3956
void ReplicationSlotsComputeRequiredLSN(void)
Definition: slot.c:761
#define ERROR
Definition: elog.h:43
void ReplicationSlotPersist(void)
Definition: slot.c:689
TransactionId effective_xmin
Definition: slot.h:136
Definition: c.h:609
uint64 XLogSegNo
Definition: xlogdefs.h:41
int errdetail(const char *fmt,...)
Definition: elog.c:957
TransactionId catalog_xmin
Definition: slot.h:77
void ReplicationSlotRelease(void)
Definition: slot.c:433
TransactionId xmin
Definition: slot.h:69
#define SlotIsLogical(slot)
Definition: slot.h:162
static void check_permissions(void)
Definition: slotfuncs.c:29
#define XLogRecPtrIsInvalid(r)
Definition: xlogdefs.h:29
bool in_use
Definition: slot.h:116
#define SpinLockRelease(lock)
Definition: spin.h:64
uintptr_t Datum
Definition: postgres.h:367
#define PG_RETURN_DATUM(x)
Definition: fmgr.h:352
TransactionId effective_catalog_xmin
Definition: slot.h:137
#define ereport(elevel,...)
Definition: elog.h:144
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
int max_replication_slots
Definition: slot.c:99
uint64 XLogRecPtr
Definition: xlogdefs.h:21
#define Assert(condition)
Definition: c.h:738
XLogRecPtr restart_lsn
Definition: slot.h:80
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1208
#define PG_NARGS()
Definition: fmgr.h:203
#define HeapTupleGetDatum(tuple)
Definition: funcapi.h:220
static Datum values[MAXATTR]
Definition: bootstrap.c:167
int errmsg(const char *fmt,...)
Definition: elog.c:824
#define elog(elevel,...)
Definition: elog.h:214
int i
#define NameStr(name)
Definition: c.h:615
ReplicationSlot replication_slots[1]
Definition: slot.h:173
slock_t mutex
Definition: slot.h:113
void CheckLogicalDecodingRequirements(void)
Definition: logical.c:75
void ReplicationSlotsComputeRequiredXmin(bool already_locked)
Definition: slot.c:711
void ReplicationSlotMarkDirty(void)
Definition: slot.c:672
#define PG_GETARG_NAME(n)
Definition: fmgr.h:278
#define XLByteToSeg(xlrp, logSegNo, wal_segsz_bytes)

◆ create_logical_replication_slot()

static void create_logical_replication_slot ( char *  name,
char *  plugin,
bool  temporary,
XLogRecPtr  restart_lsn,
bool  find_startpoint 
)
static

Definition at line 126 of file slotfuncs.c.

References Assert, CreateInitDecodingContext(), DecodingContextFindStartpoint(), FreeDecodingContext(), MyReplicationSlot, NIL, read_local_xlog_page(), ReplicationSlotCreate(), RS_EPHEMERAL, RS_TEMPORARY, wal_segment_close(), wal_segment_open(), and XL_ROUTINE.

Referenced by copy_replication_slot(), and pg_create_logical_replication_slot().

129 {
130  LogicalDecodingContext *ctx = NULL;
131 
133 
134  /*
135  * Acquire a logical decoding slot, this will check for conflicting names.
136  * Initially create persistent slot as ephemeral - that allows us to
137  * nicely handle errors during initialization because it'll get dropped if
138  * this transaction fails. We'll make it persistent at the end. Temporary
139  * slots can be created as temporary from beginning as they get dropped on
140  * error as well.
141  */
143  temporary ? RS_TEMPORARY : RS_EPHEMERAL);
144 
145  /*
146  * Create logical decoding context to find start point or, if we don't
147  * need it, to 1) bump slot's restart_lsn and xmin 2) check plugin sanity.
148  *
149  * Note: when !find_startpoint this is still important, because it's at
150  * this point that the output plugin is validated.
151  */
153  false, /* just catalogs is OK */
154  restart_lsn,
155  XL_ROUTINE(.page_read = read_local_xlog_page,
156  .segment_open = wal_segment_open,
157  .segment_close = wal_segment_close),
158  NULL, NULL, NULL);
159 
160  /*
161  * If caller needs us to determine the decoding start point, do so now.
162  * This might take a while.
163  */
164  if (find_startpoint)
166 
167  /* don't need the decoding context anymore */
168  FreeDecodingContext(ctx);
169 }
#define NIL
Definition: pg_list.h:65
static const char * plugin
void wal_segment_close(XLogReaderState *state)
Definition: xlogutils.c:813
void ReplicationSlotCreate(const char *name, bool db_specific, ReplicationSlotPersistency persistency)
Definition: slot.c:219
void wal_segment_open(XLogReaderState *state, XLogSegNo nextSegNo, TimeLineID *tli_p)
Definition: xlogutils.c:788
void DecodingContextFindStartpoint(LogicalDecodingContext *ctx)
Definition: logical.c:468
LogicalDecodingContext * CreateInitDecodingContext(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)
Definition: logical.c:230
#define XL_ROUTINE(...)
Definition: xlogreader.h:116
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
#define Assert(condition)
Definition: c.h:738
void FreeDecodingContext(LogicalDecodingContext *ctx)
Definition: logical.c:511
const char * name
Definition: encode.c:555
int read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen, XLogRecPtr targetRecPtr, char *cur_page)
Definition: xlogutils.c:832

◆ create_physical_replication_slot()

static void create_physical_replication_slot ( char *  name,
bool  immediately_reserve,
bool  temporary,
XLogRecPtr  restart_lsn 
)
static

Definition at line 46 of file slotfuncs.c.

References Assert, ReplicationSlot::data, MyReplicationSlot, ReplicationSlotCreate(), ReplicationSlotMarkDirty(), ReplicationSlotReserveWal(), ReplicationSlotSave(), ReplicationSlotPersistentData::restart_lsn, RS_PERSISTENT, RS_TEMPORARY, and XLogRecPtrIsInvalid.

Referenced by copy_replication_slot(), and pg_create_physical_replication_slot().

48 {
50 
51  /* acquire replication slot, this will check for conflicting names */
53  temporary ? RS_TEMPORARY : RS_PERSISTENT);
54 
55  if (immediately_reserve)
56  {
57  /* Reserve WAL as the user asked for it */
58  if (XLogRecPtrIsInvalid(restart_lsn))
60  else
61  MyReplicationSlot->data.restart_lsn = restart_lsn;
62 
63  /* Write this slot to disk */
66  }
67 }
void ReplicationSlotCreate(const char *name, bool db_specific, ReplicationSlotPersistency persistency)
Definition: slot.c:219
void ReplicationSlotSave(void)
Definition: slot.c:654
ReplicationSlotPersistentData data
Definition: slot.h:140
void ReplicationSlotReserveWal(void)
Definition: slot.c:1013
#define XLogRecPtrIsInvalid(r)
Definition: xlogdefs.h:29
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
#define Assert(condition)
Definition: c.h:738
XLogRecPtr restart_lsn
Definition: slot.h:80
const char * name
Definition: encode.c:555
void ReplicationSlotMarkDirty(void)
Definition: slot.c:672

◆ pg_copy_logical_replication_slot_a()

Datum pg_copy_logical_replication_slot_a ( PG_FUNCTION_ARGS  )

Definition at line 869 of file slotfuncs.c.

References copy_replication_slot().

870 {
871  return copy_replication_slot(fcinfo, true);
872 }
static Datum copy_replication_slot(FunctionCallInfo fcinfo, bool logical_slot)
Definition: slotfuncs.c:651

◆ pg_copy_logical_replication_slot_b()

Datum pg_copy_logical_replication_slot_b ( PG_FUNCTION_ARGS  )

Definition at line 875 of file slotfuncs.c.

References copy_replication_slot().

876 {
877  return copy_replication_slot(fcinfo, true);
878 }
static Datum copy_replication_slot(FunctionCallInfo fcinfo, bool logical_slot)
Definition: slotfuncs.c:651

◆ pg_copy_logical_replication_slot_c()

Datum pg_copy_logical_replication_slot_c ( PG_FUNCTION_ARGS  )

Definition at line 881 of file slotfuncs.c.

References copy_replication_slot().

882 {
883  return copy_replication_slot(fcinfo, true);
884 }
static Datum copy_replication_slot(FunctionCallInfo fcinfo, bool logical_slot)
Definition: slotfuncs.c:651

◆ pg_copy_physical_replication_slot_a()

Datum pg_copy_physical_replication_slot_a ( PG_FUNCTION_ARGS  )

Definition at line 887 of file slotfuncs.c.

References copy_replication_slot().

888 {
889  return copy_replication_slot(fcinfo, false);
890 }
static Datum copy_replication_slot(FunctionCallInfo fcinfo, bool logical_slot)
Definition: slotfuncs.c:651

◆ pg_copy_physical_replication_slot_b()

Datum pg_copy_physical_replication_slot_b ( PG_FUNCTION_ARGS  )

Definition at line 893 of file slotfuncs.c.

References copy_replication_slot().

894 {
895  return copy_replication_slot(fcinfo, false);
896 }
static Datum copy_replication_slot(FunctionCallInfo fcinfo, bool logical_slot)
Definition: slotfuncs.c:651

◆ pg_create_logical_replication_slot()

Datum pg_create_logical_replication_slot ( PG_FUNCTION_ARGS  )

Definition at line 175 of file slotfuncs.c.

References check_permissions(), CheckLogicalDecodingRequirements(), ReplicationSlotPersistentData::confirmed_flush, create_logical_replication_slot(), ReplicationSlot::data, elog, ERROR, get_call_result_type(), heap_form_tuple(), HeapTupleGetDatum, InvalidXLogRecPtr, LSNGetDatum, MyReplicationSlot, ReplicationSlotPersistentData::name, name, NameGetDatum, NameStr, PG_GETARG_BOOL, PG_GETARG_NAME, PG_RETURN_DATUM, plugin, ReplicationSlotPersist(), ReplicationSlotRelease(), TYPEFUNC_COMPOSITE, and values.

176 {
177  Name name = PG_GETARG_NAME(0);
179  bool temporary = PG_GETARG_BOOL(2);
180  Datum result;
181  TupleDesc tupdesc;
182  HeapTuple tuple;
183  Datum values[2];
184  bool nulls[2];
185 
186  if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
187  elog(ERROR, "return type must be a row type");
188 
190 
192 
194  NameStr(*plugin),
195  temporary,
197  true);
198 
199  values[0] = NameGetDatum(&MyReplicationSlot->data.name);
201 
202  memset(nulls, 0, sizeof(nulls));
203 
204  tuple = heap_form_tuple(tupdesc, values, nulls);
205  result = HeapTupleGetDatum(tuple);
206 
207  /* ok, slot is now fully created, mark it as persistent if needed */
208  if (!temporary)
211 
212  PG_RETURN_DATUM(result);
213 }
static void create_logical_replication_slot(char *name, char *plugin, bool temporary, XLogRecPtr restart_lsn, bool find_startpoint)
Definition: slotfuncs.c:126
static const char * plugin
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
#define NameGetDatum(X)
Definition: postgres.h:595
TypeFuncClass get_call_result_type(FunctionCallInfo fcinfo, Oid *resultTypeId, TupleDesc *resultTupleDesc)
Definition: funcapi.c:205
#define LSNGetDatum(X)
Definition: pg_lsn.h:22
HeapTuple heap_form_tuple(TupleDesc tupleDescriptor, Datum *values, bool *isnull)
Definition: heaptuple.c:1020
#define PG_GETARG_BOOL(n)
Definition: fmgr.h:274
ReplicationSlotPersistentData data
Definition: slot.h:140
XLogRecPtr confirmed_flush
Definition: slot.h:88
#define ERROR
Definition: elog.h:43
void ReplicationSlotPersist(void)
Definition: slot.c:689
Definition: c.h:609
void ReplicationSlotRelease(void)
Definition: slot.c:433
static void check_permissions(void)
Definition: slotfuncs.c:29
uintptr_t Datum
Definition: postgres.h:367
#define PG_RETURN_DATUM(x)
Definition: fmgr.h:352
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
#define HeapTupleGetDatum(tuple)
Definition: funcapi.h:220
const char * name
Definition: encode.c:555
static Datum values[MAXATTR]
Definition: bootstrap.c:167
#define elog(elevel,...)
Definition: elog.h:214
#define NameStr(name)
Definition: c.h:615
void CheckLogicalDecodingRequirements(void)
Definition: logical.c:75
#define PG_GETARG_NAME(n)
Definition: fmgr.h:278

◆ pg_create_physical_replication_slot()

Datum pg_create_physical_replication_slot ( PG_FUNCTION_ARGS  )

Definition at line 74 of file slotfuncs.c.

References check_permissions(), CheckSlotRequirements(), create_physical_replication_slot(), ReplicationSlot::data, elog, ERROR, get_call_result_type(), heap_form_tuple(), HeapTupleGetDatum, InvalidXLogRecPtr, LSNGetDatum, MyReplicationSlot, ReplicationSlotPersistentData::name, name, NameGetDatum, NameStr, PG_GETARG_BOOL, PG_GETARG_NAME, PG_RETURN_DATUM, ReplicationSlotRelease(), ReplicationSlotPersistentData::restart_lsn, TYPEFUNC_COMPOSITE, and values.

75 {
77  bool immediately_reserve = PG_GETARG_BOOL(1);
78  bool temporary = PG_GETARG_BOOL(2);
79  Datum values[2];
80  bool nulls[2];
81  TupleDesc tupdesc;
82  HeapTuple tuple;
83  Datum result;
84 
85  if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
86  elog(ERROR, "return type must be a row type");
87 
89 
91 
93  immediately_reserve,
94  temporary,
96 
97  values[0] = NameGetDatum(&MyReplicationSlot->data.name);
98  nulls[0] = false;
99 
100  if (immediately_reserve)
101  {
103  nulls[1] = false;
104  }
105  else
106  nulls[1] = true;
107 
108  tuple = heap_form_tuple(tupdesc, values, nulls);
109  result = HeapTupleGetDatum(tuple);
110 
112 
113  PG_RETURN_DATUM(result);
114 }
void CheckSlotRequirements(void)
Definition: slot.c:988
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
#define NameGetDatum(X)
Definition: postgres.h:595
TypeFuncClass get_call_result_type(FunctionCallInfo fcinfo, Oid *resultTypeId, TupleDesc *resultTupleDesc)
Definition: funcapi.c:205
static void create_physical_replication_slot(char *name, bool immediately_reserve, bool temporary, XLogRecPtr restart_lsn)
Definition: slotfuncs.c:46
#define LSNGetDatum(X)
Definition: pg_lsn.h:22
HeapTuple heap_form_tuple(TupleDesc tupleDescriptor, Datum *values, bool *isnull)
Definition: heaptuple.c:1020
#define PG_GETARG_BOOL(n)
Definition: fmgr.h:274
ReplicationSlotPersistentData data
Definition: slot.h:140
#define ERROR
Definition: elog.h:43
Definition: c.h:609
void ReplicationSlotRelease(void)
Definition: slot.c:433
static void check_permissions(void)
Definition: slotfuncs.c:29
uintptr_t Datum
Definition: postgres.h:367
#define PG_RETURN_DATUM(x)
Definition: fmgr.h:352
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
XLogRecPtr restart_lsn
Definition: slot.h:80
#define HeapTupleGetDatum(tuple)
Definition: funcapi.h:220
const char * name
Definition: encode.c:555
static Datum values[MAXATTR]
Definition: bootstrap.c:167
#define elog(elevel,...)
Definition: elog.h:214
#define NameStr(name)
Definition: c.h:615
#define PG_GETARG_NAME(n)
Definition: fmgr.h:278

◆ pg_drop_replication_slot()

Datum pg_drop_replication_slot ( PG_FUNCTION_ARGS  )

Definition at line 220 of file slotfuncs.c.

References check_permissions(), CheckSlotRequirements(), name, NameStr, PG_GETARG_NAME, PG_RETURN_VOID, and ReplicationSlotDrop().

221 {
222  Name name = PG_GETARG_NAME(0);
223 
225 
227 
228  ReplicationSlotDrop(NameStr(*name), true);
229 
230  PG_RETURN_VOID();
231 }
void CheckSlotRequirements(void)
Definition: slot.c:988
Definition: c.h:609
static void check_permissions(void)
Definition: slotfuncs.c:29
#define PG_RETURN_VOID()
Definition: fmgr.h:348
const char * name
Definition: encode.c:555
void ReplicationSlotDrop(const char *name, bool nowait)
Definition: slot.c:526
#define NameStr(name)
Definition: c.h:615
#define PG_GETARG_NAME(n)
Definition: fmgr.h:278

◆ pg_get_replication_slots()

Datum pg_get_replication_slots ( PG_FUNCTION_ARGS  )

Definition at line 237 of file slotfuncs.c.

References ReplicationSlot::active_pid, ReturnSetInfo::allowedModes, BoolGetDatum, ReplicationSlotPersistentData::catalog_xmin, ReplicationSlotPersistentData::confirmed_flush, CStringGetTextDatum, ReplicationSlot::data, ReplicationSlotPersistentData::database, ReturnSetInfo::econtext, ExprContext::ecxt_per_query_memory, elog, ereport, errcode(), errmsg(), ERROR, get_call_result_type(), GetWALAvailability(), i, ReplicationSlot::in_use, Int32GetDatum, Int64GetDatum(), InvalidOid, InvalidTransactionId, InvalidXLogRecPtr, IsA, LSNGetDatum, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_replication_slots, max_slot_wal_keep_size_mb, MemoryContextSwitchTo(), ReplicationSlot::mutex, ReplicationSlotPersistentData::name, namecpy(), NameGetDatum, ReplicationSlotPersistentData::persistency, PG_GET_REPLICATION_SLOTS_COLS, plugin, ReplicationSlotPersistentData::plugin, ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, ReplicationSlotPersistentData::restart_lsn, ReturnSetInfo::returnMode, RS_TEMPORARY, ReturnSetInfo::setDesc, ReturnSetInfo::setResult, SFRM_Materialize, SpinLockAcquire, SpinLockRelease, TransactionIdGetDatum, tuplestore_begin_heap(), tuplestore_donestoring, tuplestore_putvalues(), TYPEFUNC_COMPOSITE, values, wal_segment_size, WALAVAIL_INVALID_LSN, WALAVAIL_NORMAL, WALAVAIL_REMOVED, WALAVAIL_RESERVED, work_mem, XLogGetLastRemovedSegno(), XLogSegNoOffsetToRecPtr, and ReplicationSlotPersistentData::xmin.

238 {
239 #define PG_GET_REPLICATION_SLOTS_COLS 13
240  ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
241  TupleDesc tupdesc;
242  Tuplestorestate *tupstore;
243  MemoryContext per_query_ctx;
244  MemoryContext oldcontext;
245  int slotno;
246 
247  /* check to see if caller supports us returning a tuplestore */
248  if (rsinfo == NULL || !IsA(rsinfo, ReturnSetInfo))
249  ereport(ERROR,
250  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
251  errmsg("set-valued function called in context that cannot accept a set")));
252  if (!(rsinfo->allowedModes & SFRM_Materialize))
253  ereport(ERROR,
254  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
255  errmsg("materialize mode required, but it is not allowed in this context")));
256 
257  /* Build a tuple descriptor for our result type */
258  if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
259  elog(ERROR, "return type must be a row type");
260 
261  /*
262  * We don't require any special permission to see this function's data
263  * because nothing should be sensitive. The most critical being the slot
264  * name, which shouldn't contain anything particularly sensitive.
265  */
266 
267  per_query_ctx = rsinfo->econtext->ecxt_per_query_memory;
268  oldcontext = MemoryContextSwitchTo(per_query_ctx);
269 
270  tupstore = tuplestore_begin_heap(true, false, work_mem);
271  rsinfo->returnMode = SFRM_Materialize;
272  rsinfo->setResult = tupstore;
273  rsinfo->setDesc = tupdesc;
274 
275  MemoryContextSwitchTo(oldcontext);
276 
277  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
278  for (slotno = 0; slotno < max_replication_slots; slotno++)
279  {
282  bool nulls[PG_GET_REPLICATION_SLOTS_COLS];
283 
284  ReplicationSlotPersistency persistency;
285  TransactionId xmin;
286  TransactionId catalog_xmin;
287  XLogRecPtr restart_lsn;
288  XLogRecPtr confirmed_flush_lsn;
289  pid_t active_pid;
290  Oid database;
291  NameData slot_name;
293  WALAvailability walstate;
294  XLogSegNo last_removed_seg;
295  int i;
296 
297  if (!slot->in_use)
298  continue;
299 
300  SpinLockAcquire(&slot->mutex);
301 
302  xmin = slot->data.xmin;
303  catalog_xmin = slot->data.catalog_xmin;
304  database = slot->data.database;
305  restart_lsn = slot->data.restart_lsn;
306  confirmed_flush_lsn = slot->data.confirmed_flush;
307  namecpy(&slot_name, &slot->data.name);
308  namecpy(&plugin, &slot->data.plugin);
309  active_pid = slot->active_pid;
310  persistency = slot->data.persistency;
311 
312  SpinLockRelease(&slot->mutex);
313 
314  memset(nulls, 0, sizeof(nulls));
315 
316  i = 0;
317  values[i++] = NameGetDatum(&slot_name);
318 
319  if (database == InvalidOid)
320  nulls[i++] = true;
321  else
322  values[i++] = NameGetDatum(&plugin);
323 
324  if (database == InvalidOid)
325  values[i++] = CStringGetTextDatum("physical");
326  else
327  values[i++] = CStringGetTextDatum("logical");
328 
329  if (database == InvalidOid)
330  nulls[i++] = true;
331  else
332  values[i++] = database;
333 
334  values[i++] = BoolGetDatum(persistency == RS_TEMPORARY);
335  values[i++] = BoolGetDatum(active_pid != 0);
336 
337  if (active_pid != 0)
338  values[i++] = Int32GetDatum(active_pid);
339  else
340  nulls[i++] = true;
341 
342  if (xmin != InvalidTransactionId)
343  values[i++] = TransactionIdGetDatum(xmin);
344  else
345  nulls[i++] = true;
346 
347  if (catalog_xmin != InvalidTransactionId)
348  values[i++] = TransactionIdGetDatum(catalog_xmin);
349  else
350  nulls[i++] = true;
351 
352  if (restart_lsn != InvalidXLogRecPtr)
353  values[i++] = LSNGetDatum(restart_lsn);
354  else
355  nulls[i++] = true;
356 
357  if (confirmed_flush_lsn != InvalidXLogRecPtr)
358  values[i++] = LSNGetDatum(confirmed_flush_lsn);
359  else
360  nulls[i++] = true;
361 
362  walstate = GetWALAvailability(restart_lsn);
363 
364  switch (walstate)
365  {
367  nulls[i++] = true;
368  break;
369 
370  case WALAVAIL_NORMAL:
371  values[i++] = CStringGetTextDatum("normal");
372  break;
373 
374  case WALAVAIL_RESERVED:
375  values[i++] = CStringGetTextDatum("reserved");
376  break;
377 
378  case WALAVAIL_REMOVED:
379  values[i++] = CStringGetTextDatum("lost");
380  break;
381  }
382 
383  if (max_slot_wal_keep_size_mb >= 0 &&
384  (walstate == WALAVAIL_NORMAL || walstate == WALAVAIL_RESERVED) &&
385  ((last_removed_seg = XLogGetLastRemovedSegno()) != 0))
386  {
387  XLogRecPtr min_safe_lsn;
388 
389  XLogSegNoOffsetToRecPtr(last_removed_seg + 1, 0,
390  wal_segment_size, min_safe_lsn);
391  values[i++] = Int64GetDatum(min_safe_lsn);
392  }
393  else
394  nulls[i++] = true;
395 
396  tuplestore_putvalues(tupstore, tupdesc, values, nulls);
397  }
398  LWLockRelease(ReplicationSlotControlLock);
399 
400  tuplestore_donestoring(tupstore);
401 
402  return (Datum) 0;
403 }
void tuplestore_putvalues(Tuplestorestate *state, TupleDesc tdesc, Datum *values, bool *isnull)
Definition: tuplestore.c:750
static const char * plugin
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
#define IsA(nodeptr, _type_)
Definition: nodes.h:580
#define NameGetDatum(X)
Definition: postgres.h:595
TypeFuncClass get_call_result_type(FunctionCallInfo fcinfo, Oid *resultTypeId, TupleDesc *resultTupleDesc)
Definition: funcapi.c:205
int wal_segment_size
Definition: xlog.c:116
#define PG_GET_REPLICATION_SLOTS_COLS
uint32 TransactionId
Definition: c.h:513
int namecpy(Name n1, const NameData *n2)
Definition: name.c:233
WALAvailability
Definition: xlog.h:270
#define tuplestore_donestoring(state)
Definition: tuplestore.h:60
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
Definition: palloc.h:109
ReplicationSlotPersistency persistency
Definition: slot.h:61
int errcode(int sqlerrcode)
Definition: elog.c:610
#define LSNGetDatum(X)
Definition: pg_lsn.h:22
ReplicationSlotPersistentData data
Definition: slot.h:140
unsigned int Oid
Definition: postgres_ext.h:31
XLogRecPtr confirmed_flush
Definition: slot.h:88
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1812
#define SpinLockAcquire(lock)
Definition: spin.h:62
XLogSegNo XLogGetLastRemovedSegno(void)
Definition: xlog.c:3956
#define ERROR
Definition: elog.h:43
Definition: c.h:609
uint64 XLogSegNo
Definition: xlogdefs.h:41
TransactionId catalog_xmin
Definition: slot.h:77
#define InvalidTransactionId
Definition: transam.h:31
TransactionId xmin
Definition: slot.h:69
Datum Int64GetDatum(int64 X)
Definition: fmgr.c:1701
bool in_use
Definition: slot.h:116
#define SpinLockRelease(lock)
Definition: spin.h:64
Tuplestorestate * tuplestore_begin_heap(bool randomAccess, bool interXact, int maxKBytes)
Definition: tuplestore.c:318
#define TransactionIdGetDatum(X)
Definition: postgres.h:521
uintptr_t Datum
Definition: postgres.h:367
int work_mem
Definition: globals.c:121
#define BoolGetDatum(X)
Definition: postgres.h:402
#define InvalidOid
Definition: postgres_ext.h:36
#define ereport(elevel,...)
Definition: elog.h:144
int allowedModes
Definition: execnodes.h:305
int max_slot_wal_keep_size_mb
Definition: xlog.c:110
SetFunctionReturnMode returnMode
Definition: execnodes.h:307
int max_replication_slots
Definition: slot.c:99
#define XLogSegNoOffsetToRecPtr(segno, offset, wal_segsz_bytes, dest)
uint64 XLogRecPtr
Definition: xlogdefs.h:21
XLogRecPtr restart_lsn
Definition: slot.h:80
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1208
MemoryContext ecxt_per_query_memory
Definition: execnodes.h:233
Tuplestorestate * setResult
Definition: execnodes.h:310
static Datum values[MAXATTR]
Definition: bootstrap.c:167
ExprContext * econtext
Definition: execnodes.h:303
#define Int32GetDatum(X)
Definition: postgres.h:479
TupleDesc setDesc
Definition: execnodes.h:311
ReplicationSlotPersistency
Definition: slot.h:32
int errmsg(const char *fmt,...)
Definition: elog.c:824
pid_t active_pid
Definition: slot.h:119
WALAvailability GetWALAvailability(XLogRecPtr targetLSN)
Definition: xlog.c:9501
#define elog(elevel,...)
Definition: elog.h:214
int i
#define CStringGetTextDatum(s)
Definition: builtins.h:87
ReplicationSlot replication_slots[1]
Definition: slot.h:173
slock_t mutex
Definition: slot.h:113

◆ pg_logical_replication_slot_advance()

static XLogRecPtr pg_logical_replication_slot_advance ( XLogRecPtr  moveto)
static

Definition at line 450 of file slotfuncs.c.

References Assert, CHECK_FOR_INTERRUPTS, ReplicationSlotPersistentData::confirmed_flush, CreateDecodingContext(), CurrentResourceOwner, ReplicationSlot::data, elog, XLogReaderState::EndRecPtr, ERROR, FreeDecodingContext(), InvalidateSystemCaches(), InvalidXLogRecPtr, LogicalConfirmReceivedLocation(), LogicalDecodingProcessRecord(), MyReplicationSlot, NIL, PG_CATCH, PG_END_TRY, PG_RE_THROW, PG_TRY, read_local_xlog_page(), LogicalDecodingContext::reader, ReplicationSlotMarkDirty(), ReplicationSlotPersistentData::restart_lsn, wal_segment_close(), wal_segment_open(), XL_ROUTINE, XLogBeginRead(), and XLogReadRecord().

Referenced by pg_replication_slot_advance().

451 {
453  ResourceOwner old_resowner = CurrentResourceOwner;
454  XLogRecPtr retlsn;
455 
456  Assert(moveto != InvalidXLogRecPtr);
457 
458  PG_TRY();
459  {
460  /*
461  * Create our decoding context in fast_forward mode, passing start_lsn
462  * as InvalidXLogRecPtr, so that we start processing from my slot's
463  * confirmed_flush.
464  */
466  NIL,
467  true, /* fast_forward */
468  XL_ROUTINE(.page_read = read_local_xlog_page,
469  .segment_open = wal_segment_open,
470  .segment_close = wal_segment_close),
471  NULL, NULL, NULL);
472 
473  /*
474  * Start reading at the slot's restart_lsn, which we know to point to
475  * a valid record.
476  */
478 
479  /* Initialize our return value in case we don't do anything */
481 
482  /* invalidate non-timetravel entries */
484 
485  /* Decode at least one record, until we run out of records */
486  while (ctx->reader->EndRecPtr < moveto)
487  {
488  char *errm = NULL;
489  XLogRecord *record;
490 
491  /*
492  * Read records. No changes are generated in fast_forward mode,
493  * but snapbuilder/slot statuses are updated properly.
494  */
495  record = XLogReadRecord(ctx->reader, &errm);
496  if (errm)
497  elog(ERROR, "%s", errm);
498 
499  /*
500  * Process the record. Storage-level changes are ignored in
501  * fast_forward mode, but other modules (such as snapbuilder)
502  * might still have critical updates to do.
503  */
504  if (record)
506 
507  /* Stop once the requested target has been reached */
508  if (moveto <= ctx->reader->EndRecPtr)
509  break;
510 
512  }
513 
514  /*
515  * Logical decoding could have clobbered CurrentResourceOwner during
516  * transaction management, so restore the executor's value. (This is
517  * a kluge, but it's not worth cleaning up right now.)
518  */
519  CurrentResourceOwner = old_resowner;
520 
521  if (ctx->reader->EndRecPtr != InvalidXLogRecPtr)
522  {
524 
525  /*
526  * If only the confirmed_flush LSN has changed the slot won't get
527  * marked as dirty by the above. Callers on the walsender
528  * interface are expected to keep track of their own progress and
529  * don't need it written out. But SQL-interface users cannot
530  * specify their own start positions and it's harder for them to
531  * keep track of their progress, so we should make more of an
532  * effort to save it for them.
533  *
534  * Dirty the slot so it is written out at the next checkpoint. The
535  * LSN position advanced to may still be lost on a crash but this
536  * makes the data consistent after a clean shutdown.
537  */
539  }
540 
542 
543  /* free context, call shutdown callback */
544  FreeDecodingContext(ctx);
545 
547  }
548  PG_CATCH();
549  {
550  /* clear all timetravel entries */
552 
553  PG_RE_THROW();
554  }
555  PG_END_TRY();
556 
557  return retlsn;
558 }
#define NIL
Definition: pg_list.h:65
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
void wal_segment_close(XLogReaderState *state)
Definition: xlogutils.c:813
ResourceOwner CurrentResourceOwner
Definition: resowner.c:142
void wal_segment_open(XLogReaderState *state, XLogSegNo nextSegNo, TimeLineID *tli_p)
Definition: xlogutils.c:788
ReplicationSlotPersistentData data
Definition: slot.h:140
void InvalidateSystemCaches(void)
Definition: inval.c:643
XLogRecPtr confirmed_flush
Definition: slot.h:88
XLogRecPtr EndRecPtr
Definition: xlogreader.h:176
XLogRecord * XLogReadRecord(XLogReaderState *state, char **errormsg)
Definition: xlogreader.c:266
#define ERROR
Definition: elog.h:43
void LogicalDecodingProcessRecord(LogicalDecodingContext *ctx, XLogReaderState *record)
Definition: decode.c:94
LogicalDecodingContext * CreateDecodingContext(XLogRecPtr start_lsn, List *output_plugin_options, bool fast_forward, XLogReaderRoutine *xl_routine, LogicalOutputPluginWriterPrepareWrite prepare_write, LogicalOutputPluginWriterWrite do_write, LogicalOutputPluginWriterUpdateProgress update_progress)
Definition: logical.c:376
void XLogBeginRead(XLogReaderState *state, XLogRecPtr RecPtr)
Definition: xlogreader.c:238
#define XL_ROUTINE(...)
Definition: xlogreader.h:116
#define PG_CATCH()
Definition: elog.h:305
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
uint64 XLogRecPtr
Definition: xlogdefs.h:21
#define Assert(condition)
Definition: c.h:738
void FreeDecodingContext(LogicalDecodingContext *ctx)
Definition: logical.c:511
XLogRecPtr restart_lsn
Definition: slot.h:80
#define PG_RE_THROW()
Definition: elog.h:336
void LogicalConfirmReceivedLocation(XLogRecPtr lsn)
Definition: logical.c:1004
XLogReaderState * reader
Definition: logical.h:41
#define elog(elevel,...)
Definition: elog.h:214
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:99
#define PG_TRY()
Definition: elog.h:295
int read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen, XLogRecPtr targetRecPtr, char *cur_page)
Definition: xlogutils.c:832
#define PG_END_TRY()
Definition: elog.h:320
void ReplicationSlotMarkDirty(void)
Definition: slot.c:672

◆ pg_physical_replication_slot_advance()

static XLogRecPtr pg_physical_replication_slot_advance ( XLogRecPtr  moveto)
static

Definition at line 413 of file slotfuncs.c.

References Assert, ReplicationSlot::data, InvalidXLogRecPtr, ReplicationSlot::mutex, MyReplicationSlot, ReplicationSlotMarkDirty(), ReplicationSlotPersistentData::restart_lsn, SpinLockAcquire, and SpinLockRelease.

Referenced by pg_replication_slot_advance().

414 {
416  XLogRecPtr retlsn = startlsn;
417 
418  Assert(moveto != InvalidXLogRecPtr);
419 
420  if (startlsn < moveto)
421  {
425  retlsn = moveto;
426 
427  /*
428  * Dirty the slot so as it is written out at the next checkpoint. Note
429  * that the LSN position advanced may still be lost in the event of a
430  * crash, but this makes the data consistent after a clean shutdown.
431  */
433  }
434 
435  return retlsn;
436 }
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
ReplicationSlotPersistentData data
Definition: slot.h:140
#define SpinLockAcquire(lock)
Definition: spin.h:62
#define SpinLockRelease(lock)
Definition: spin.h:64
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
uint64 XLogRecPtr
Definition: xlogdefs.h:21
#define Assert(condition)
Definition: c.h:738
XLogRecPtr restart_lsn
Definition: slot.h:80
slock_t mutex
Definition: slot.h:113
void ReplicationSlotMarkDirty(void)
Definition: slot.c:672

◆ pg_replication_slot_advance()

Datum pg_replication_slot_advance ( PG_FUNCTION_ARGS  )

Definition at line 564 of file slotfuncs.c.

References Assert, check_permissions(), ReplicationSlotPersistentData::confirmed_flush, ReplicationSlot::data, ReplicationSlotPersistentData::database, elog, ereport, errcode(), errdetail(), errmsg(), ERROR, get_call_result_type(), GetFlushRecPtr(), GetXLogReplayRecPtr(), heap_form_tuple(), HeapTupleGetDatum, LSNGetDatum, Min, MyReplicationSlot, ReplicationSlotPersistentData::name, NameGetDatum, NameStr, OidIsValid, PG_GETARG_LSN, PG_GETARG_NAME, pg_logical_replication_slot_advance(), pg_physical_replication_slot_advance(), PG_RETURN_DATUM, RecoveryInProgress(), ReplicationSlotAcquire(), ReplicationSlotRelease(), ReplicationSlotPersistentData::restart_lsn, SAB_Error, ThisTimeLineID, TYPEFUNC_COMPOSITE, values, and XLogRecPtrIsInvalid.

565 {
566  Name slotname = PG_GETARG_NAME(0);
567  XLogRecPtr moveto = PG_GETARG_LSN(1);
568  XLogRecPtr endlsn;
569  XLogRecPtr minlsn;
570  TupleDesc tupdesc;
571  Datum values[2];
572  bool nulls[2];
573  HeapTuple tuple;
574  Datum result;
575 
577 
579 
580  if (XLogRecPtrIsInvalid(moveto))
581  ereport(ERROR,
582  (errmsg("invalid target WAL LSN")));
583 
584  /* Build a tuple descriptor for our result type */
585  if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
586  elog(ERROR, "return type must be a row type");
587 
588  /*
589  * We can't move slot past what's been flushed/replayed so clamp the
590  * target position accordingly.
591  */
592  if (!RecoveryInProgress())
593  moveto = Min(moveto, GetFlushRecPtr());
594  else
595  moveto = Min(moveto, GetXLogReplayRecPtr(&ThisTimeLineID));
596 
597  /* Acquire the slot so we "own" it */
598  (void) ReplicationSlotAcquire(NameStr(*slotname), SAB_Error);
599 
600  /* A slot whose restart_lsn has never been reserved cannot be advanced */
602  ereport(ERROR,
603  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
604  errmsg("replication slot \"%s\" cannot be advanced",
605  NameStr(*slotname)),
606  errdetail("This slot has never previously reserved WAL, or has been invalidated.")));
607 
608  /*
609  * Check if the slot is not moving backwards. Physical slots rely simply
610  * on restart_lsn as a minimum point, while logical slots have confirmed
611  * consumption up to confirmed_flush, meaning that in both cases data
612  * older than that is not available anymore.
613  */
616  else
618 
619  if (moveto < minlsn)
620  ereport(ERROR,
621  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
622  errmsg("cannot advance replication slot to %X/%X, minimum is %X/%X",
623  (uint32) (moveto >> 32), (uint32) moveto,
624  (uint32) (minlsn >> 32), (uint32) minlsn)));
625 
626  /* Do the actual slot update, depending on the slot type */
628  endlsn = pg_logical_replication_slot_advance(moveto);
629  else
630  endlsn = pg_physical_replication_slot_advance(moveto);
631 
632  values[0] = NameGetDatum(&MyReplicationSlot->data.name);
633  nulls[0] = false;
634 
636 
637  /* Return the reached position. */
638  values[1] = LSNGetDatum(endlsn);
639  nulls[1] = false;
640 
641  tuple = heap_form_tuple(tupdesc, values, nulls);
642  result = HeapTupleGetDatum(tuple);
643 
644  PG_RETURN_DATUM(result);
645 }
static XLogRecPtr pg_logical_replication_slot_advance(XLogRecPtr moveto)
Definition: slotfuncs.c:450
#define NameGetDatum(X)
Definition: postgres.h:595
TypeFuncClass get_call_result_type(FunctionCallInfo fcinfo, Oid *resultTypeId, TupleDesc *resultTupleDesc)
Definition: funcapi.c:205
#define Min(x, y)
Definition: c.h:920
int errcode(int sqlerrcode)
Definition: elog.c:610
#define LSNGetDatum(X)
Definition: pg_lsn.h:22
XLogRecPtr GetFlushRecPtr(void)
Definition: xlog.c:8420
HeapTuple heap_form_tuple(TupleDesc tupleDescriptor, Datum *values, bool *isnull)
Definition: heaptuple.c:1020
ReplicationSlotPersistentData data
Definition: slot.h:140
bool RecoveryInProgress(void)
Definition: xlog.c:8069
#define OidIsValid(objectId)
Definition: c.h:644
XLogRecPtr confirmed_flush
Definition: slot.h:88
#define ERROR
Definition: elog.h:43
XLogRecPtr GetXLogReplayRecPtr(TimeLineID *replayTLI)
Definition: xlog.c:11463
Definition: c.h:609
int errdetail(const char *fmt,...)
Definition: elog.c:957
unsigned int uint32
Definition: c.h:367
void ReplicationSlotRelease(void)
Definition: slot.c:433
static void check_permissions(void)
Definition: slotfuncs.c:29
#define XLogRecPtrIsInvalid(r)
Definition: xlogdefs.h:29
#define PG_GETARG_LSN(n)
Definition: pg_lsn.h:24
uintptr_t Datum
Definition: postgres.h:367
#define PG_RETURN_DATUM(x)
Definition: fmgr.h:352
TimeLineID ThisTimeLineID
Definition: xlog.c:191
#define ereport(elevel,...)
Definition: elog.h:144
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
uint64 XLogRecPtr
Definition: xlogdefs.h:21
#define Assert(condition)
Definition: c.h:738
XLogRecPtr restart_lsn
Definition: slot.h:80
#define HeapTupleGetDatum(tuple)
Definition: funcapi.h:220
static Datum values[MAXATTR]
Definition: bootstrap.c:167
int errmsg(const char *fmt,...)
Definition: elog.c:824
int ReplicationSlotAcquire(const char *name, SlotAcquireBehavior behavior)
Definition: slot.c:334
#define elog(elevel,...)
Definition: elog.h:214
Definition: slot.h:42
#define NameStr(name)
Definition: c.h:615
static XLogRecPtr pg_physical_replication_slot_advance(XLogRecPtr moveto)
Definition: slotfuncs.c:413
#define PG_GETARG_NAME(n)
Definition: fmgr.h:278