PostgreSQL Source Code  git master
slotfuncs.c File Reference
#include "postgres.h"
#include "access/htup_details.h"
#include "access/xlog_internal.h"
#include "access/xlogrecovery.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   14
 

Functions

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, bool two_phase, 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   14

Function Documentation

◆ copy_replication_slot()

static Datum copy_replication_slot ( FunctionCallInfo  fcinfo,
bool  logical_slot 
)
static

Definition at line 668 of file slotfuncs.c.

669 {
670  Name src_name = PG_GETARG_NAME(0);
671  Name dst_name = PG_GETARG_NAME(1);
672  ReplicationSlot *src = NULL;
673  ReplicationSlot first_slot_contents;
674  ReplicationSlot second_slot_contents;
675  XLogRecPtr src_restart_lsn;
676  bool src_islogical;
677  bool temporary;
678  char *plugin;
679  Datum values[2];
680  bool nulls[2];
681  Datum result;
682  TupleDesc tupdesc;
683  HeapTuple tuple;
684 
685  if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
686  elog(ERROR, "return type must be a row type");
687 
689 
690  if (logical_slot)
692  else
694 
695  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
696 
697  /*
698  * We need to prevent the source slot's reserved WAL from being removed,
699  * but we don't want to lock that slot for very long, and it can advance
700  * in the meantime. So obtain the source slot's data, and create a new
701  * slot using its restart_lsn. Afterwards we lock the source slot again
702  * and verify that the data we copied (name, type) has not changed
703  * incompatibly. No inconvenient WAL removal can occur once the new slot
704  * is created -- but since WAL removal could have occurred before we
705  * managed to create the new slot, we advance the new slot's restart_lsn
706  * to the source slot's updated restart_lsn the second time we lock it.
707  */
708  for (int i = 0; i < max_replication_slots; i++)
709  {
711 
712  if (s->in_use && strcmp(NameStr(s->data.name), NameStr(*src_name)) == 0)
713  {
714  /* Copy the slot contents while holding spinlock */
715  SpinLockAcquire(&s->mutex);
716  first_slot_contents = *s;
717  SpinLockRelease(&s->mutex);
718  src = s;
719  break;
720  }
721  }
722 
723  LWLockRelease(ReplicationSlotControlLock);
724 
725  if (src == NULL)
726  ereport(ERROR,
727  (errcode(ERRCODE_UNDEFINED_OBJECT),
728  errmsg("replication slot \"%s\" does not exist", NameStr(*src_name))));
729 
730  src_islogical = SlotIsLogical(&first_slot_contents);
731  src_restart_lsn = first_slot_contents.data.restart_lsn;
732  temporary = (first_slot_contents.data.persistency == RS_TEMPORARY);
733  plugin = logical_slot ? NameStr(first_slot_contents.data.plugin) : NULL;
734 
735  /* Check type of replication slot */
736  if (src_islogical != logical_slot)
737  ereport(ERROR,
738  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
739  src_islogical ?
740  errmsg("cannot copy physical replication slot \"%s\" as a logical replication slot",
741  NameStr(*src_name)) :
742  errmsg("cannot copy logical replication slot \"%s\" as a physical replication slot",
743  NameStr(*src_name))));
744 
745  /* Copying non-reserved slot doesn't make sense */
746  if (XLogRecPtrIsInvalid(src_restart_lsn))
747  ereport(ERROR,
748  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
749  errmsg("cannot copy a replication slot that doesn't reserve WAL")));
750 
751  /* Overwrite params from optional arguments */
752  if (PG_NARGS() >= 3)
753  temporary = PG_GETARG_BOOL(2);
754  if (PG_NARGS() >= 4)
755  {
756  Assert(logical_slot);
757  plugin = NameStr(*(PG_GETARG_NAME(3)));
758  }
759 
760  /* Create new slot and acquire it */
761  if (logical_slot)
762  {
763  /*
764  * We must not try to read WAL, since we haven't reserved it yet --
765  * hence pass find_startpoint false. confirmed_flush will be set
766  * below, by copying from the source slot.
767  */
769  plugin,
770  temporary,
771  false,
772  src_restart_lsn,
773  false);
774  }
775  else
777  true,
778  temporary,
779  src_restart_lsn);
780 
781  /*
782  * Update the destination slot to current values of the source slot;
783  * recheck that the source slot is still the one we saw previously.
784  */
785  {
786  TransactionId copy_effective_xmin;
787  TransactionId copy_effective_catalog_xmin;
788  TransactionId copy_xmin;
789  TransactionId copy_catalog_xmin;
790  XLogRecPtr copy_restart_lsn;
791  XLogRecPtr copy_confirmed_flush;
792  bool copy_islogical;
793  char *copy_name;
794 
795  /* Copy data of source slot again */
796  SpinLockAcquire(&src->mutex);
797  second_slot_contents = *src;
798  SpinLockRelease(&src->mutex);
799 
800  copy_effective_xmin = second_slot_contents.effective_xmin;
801  copy_effective_catalog_xmin = second_slot_contents.effective_catalog_xmin;
802 
803  copy_xmin = second_slot_contents.data.xmin;
804  copy_catalog_xmin = second_slot_contents.data.catalog_xmin;
805  copy_restart_lsn = second_slot_contents.data.restart_lsn;
806  copy_confirmed_flush = second_slot_contents.data.confirmed_flush;
807 
808  /* for existence check */
809  copy_name = NameStr(second_slot_contents.data.name);
810  copy_islogical = SlotIsLogical(&second_slot_contents);
811 
812  /*
813  * Check if the source slot still exists and is valid. We regard it as
814  * invalid if the type of replication slot or name has been changed,
815  * or the restart_lsn either is invalid or has gone backward. (The
816  * restart_lsn could go backwards if the source slot is dropped and
817  * copied from an older slot during installation.)
818  *
819  * Since erroring out will release and drop the destination slot we
820  * don't need to release it here.
821  */
822  if (copy_restart_lsn < src_restart_lsn ||
823  src_islogical != copy_islogical ||
824  strcmp(copy_name, NameStr(*src_name)) != 0)
825  ereport(ERROR,
826  (errmsg("could not copy replication slot \"%s\"",
827  NameStr(*src_name)),
828  errdetail("The source replication slot was modified incompatibly during the copy operation.")));
829 
830  /* The source slot must have a consistent snapshot */
831  if (src_islogical && XLogRecPtrIsInvalid(copy_confirmed_flush))
832  ereport(ERROR,
833  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
834  errmsg("cannot copy unfinished logical replication slot \"%s\"",
835  NameStr(*src_name)),
836  errhint("Retry when the source replication slot's confirmed_flush_lsn is valid.")));
837 
838  /* Install copied values again */
840  MyReplicationSlot->effective_xmin = copy_effective_xmin;
841  MyReplicationSlot->effective_catalog_xmin = copy_effective_catalog_xmin;
842 
843  MyReplicationSlot->data.xmin = copy_xmin;
844  MyReplicationSlot->data.catalog_xmin = copy_catalog_xmin;
845  MyReplicationSlot->data.restart_lsn = copy_restart_lsn;
846  MyReplicationSlot->data.confirmed_flush = copy_confirmed_flush;
848 
853 
854 #ifdef USE_ASSERT_CHECKING
855  /* Check that the restart_lsn is available */
856  {
857  XLogSegNo segno;
858 
859  XLByteToSeg(copy_restart_lsn, segno, wal_segment_size);
860  Assert(XLogGetLastRemovedSegno() < segno);
861  }
862 #endif
863  }
864 
865  /* target slot fully created, mark as persistent if needed */
866  if (logical_slot && !temporary)
868 
869  /* All done. Set up the return values */
870  values[0] = NameGetDatum(dst_name);
871  nulls[0] = false;
873  {
875  nulls[1] = false;
876  }
877  else
878  nulls[1] = true;
879 
880  tuple = heap_form_tuple(tupdesc, values, nulls);
881  result = HeapTupleGetDatum(tuple);
882 
884 
885  PG_RETURN_DATUM(result);
886 }
static Datum values[MAXATTR]
Definition: bootstrap.c:156
#define NameStr(name)
Definition: c.h:730
uint32 TransactionId
Definition: c.h:636
int errdetail(const char *fmt,...)
Definition: elog.c:1202
int errhint(const char *fmt,...)
Definition: elog.c:1316
int errcode(int sqlerrcode)
Definition: elog.c:858
int errmsg(const char *fmt,...)
Definition: elog.c:1069
#define ERROR
Definition: elog.h:39
#define ereport(elevel,...)
Definition: elog.h:149
#define PG_NARGS()
Definition: fmgr.h:203
#define PG_GETARG_NAME(n)
Definition: fmgr.h:278
#define PG_GETARG_BOOL(n)
Definition: fmgr.h:274
#define PG_RETURN_DATUM(x)
Definition: fmgr.h:353
TypeFuncClass get_call_result_type(FunctionCallInfo fcinfo, Oid *resultTypeId, TupleDesc *resultTupleDesc)
Definition: funcapi.c:276
@ TYPEFUNC_COMPOSITE
Definition: funcapi.h:149
static Datum HeapTupleGetDatum(const HeapTupleData *tuple)
Definition: funcapi.h:230
HeapTuple heap_form_tuple(TupleDesc tupleDescriptor, Datum *values, bool *isnull)
Definition: heaptuple.c:1020
int i
Definition: isn.c:73
Assert(fmt[strlen(fmt) - 1] !='\n')
void CheckLogicalDecodingRequirements(void)
Definition: logical.c:108
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1195
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1803
@ LW_SHARED
Definition: lwlock.h:116
static Datum LSNGetDatum(XLogRecPtr X)
Definition: pg_lsn.h:28
static const char * plugin
uintptr_t Datum
Definition: postgres.h:64
static Datum NameGetDatum(const NameData *X)
Definition: postgres.h:373
void ReplicationSlotMarkDirty(void)
Definition: slot.c:796
void ReplicationSlotsComputeRequiredXmin(bool already_locked)
Definition: slot.c:835
void ReplicationSlotPersist(void)
Definition: slot.c:813
ReplicationSlot * MyReplicationSlot
Definition: slot.c:98
void ReplicationSlotSave(void)
Definition: slot.c:778
void CheckSlotPermissions(void)
Definition: slot.c:1141
void ReplicationSlotRelease(void)
Definition: slot.c:547
int max_replication_slots
Definition: slot.c:101
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:95
void ReplicationSlotsComputeRequiredLSN(void)
Definition: slot.c:892
void CheckSlotRequirements(void)
Definition: slot.c:1119
@ RS_TEMPORARY
Definition: slot.h:37
#define SlotIsLogical(slot)
Definition: slot.h:169
static void create_logical_replication_slot(char *name, char *plugin, bool temporary, bool two_phase, XLogRecPtr restart_lsn, bool find_startpoint)
Definition: slotfuncs.c:118
static void create_physical_replication_slot(char *name, bool immediately_reserve, bool temporary, XLogRecPtr restart_lsn)
Definition: slotfuncs.c:38
#define SpinLockRelease(lock)
Definition: spin.h:64
#define SpinLockAcquire(lock)
Definition: spin.h:62
ReplicationSlot replication_slots[1]
Definition: slot.h:180
TransactionId xmin
Definition: slot.h:62
TransactionId catalog_xmin
Definition: slot.h:70
XLogRecPtr restart_lsn
Definition: slot.h:73
XLogRecPtr confirmed_flush
Definition: slot.h:84
ReplicationSlotPersistency persistency
Definition: slot.h:54
TransactionId effective_catalog_xmin
Definition: slot.h:144
slock_t mutex
Definition: slot.h:120
bool in_use
Definition: slot.h:123
TransactionId effective_xmin
Definition: slot.h:143
ReplicationSlotPersistentData data
Definition: slot.h:147
Definition: c.h:725
XLogSegNo XLogGetLastRemovedSegno(void)
Definition: xlog.c:3466
int wal_segment_size
Definition: xlog.c:146
#define XLByteToSeg(xlrp, logSegNo, wal_segsz_bytes)
#define XLogRecPtrIsInvalid(r)
Definition: xlogdefs.h:29
uint64 XLogRecPtr
Definition: xlogdefs.h:21
uint64 XLogSegNo
Definition: xlogdefs.h:48

References Assert(), ReplicationSlotPersistentData::catalog_xmin, CheckLogicalDecodingRequirements(), CheckSlotPermissions(), 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, 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().

◆ create_logical_replication_slot()

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

Definition at line 118 of file slotfuncs.c.

122 {
123  LogicalDecodingContext *ctx = NULL;
124 
126 
127  /*
128  * Acquire a logical decoding slot, this will check for conflicting names.
129  * Initially create persistent slot as ephemeral - that allows us to
130  * nicely handle errors during initialization because it'll get dropped if
131  * this transaction fails. We'll make it persistent at the end. Temporary
132  * slots can be created as temporary from beginning as they get dropped on
133  * error as well.
134  */
136  temporary ? RS_TEMPORARY : RS_EPHEMERAL, two_phase);
137 
138  /*
139  * Create logical decoding context to find start point or, if we don't
140  * need it, to 1) bump slot's restart_lsn and xmin 2) check plugin sanity.
141  *
142  * Note: when !find_startpoint this is still important, because it's at
143  * this point that the output plugin is validated.
144  */
146  false, /* just catalogs is OK */
147  restart_lsn,
148  XL_ROUTINE(.page_read = read_local_xlog_page,
149  .segment_open = wal_segment_open,
150  .segment_close = wal_segment_close),
151  NULL, NULL, NULL);
152 
153  /*
154  * If caller needs us to determine the decoding start point, do so now.
155  * This might take a while.
156  */
157  if (find_startpoint)
159 
160  /* don't need the decoding context anymore */
161  FreeDecodingContext(ctx);
162 }
const char * name
Definition: encode.c:571
void FreeDecodingContext(LogicalDecodingContext *ctx)
Definition: logical.c:647
void DecodingContextFindStartpoint(LogicalDecodingContext *ctx)
Definition: logical.c:603
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)
Definition: logical.c:330
#define NIL
Definition: pg_list.h:68
static bool two_phase
void ReplicationSlotCreate(const char *name, bool db_specific, ReplicationSlotPersistency persistency, bool two_phase)
Definition: slot.c:252
@ RS_EPHEMERAL
Definition: slot.h:36
#define XL_ROUTINE(...)
Definition: xlogreader.h:117
void wal_segment_close(XLogReaderState *state)
Definition: xlogutils.c:859
void wal_segment_open(XLogReaderState *state, XLogSegNo nextSegNo, TimeLineID *tli_p)
Definition: xlogutils.c:834
int read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen, XLogRecPtr targetRecPtr, char *cur_page)
Definition: xlogutils.c:878

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

Referenced by copy_replication_slot(), and pg_create_logical_replication_slot().

◆ create_physical_replication_slot()

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

Definition at line 38 of file slotfuncs.c.

40 {
42 
43  /* acquire replication slot, this will check for conflicting names */
45  temporary ? RS_TEMPORARY : RS_PERSISTENT, false);
46 
47  if (immediately_reserve)
48  {
49  /* Reserve WAL as the user asked for it */
50  if (XLogRecPtrIsInvalid(restart_lsn))
52  else
53  MyReplicationSlot->data.restart_lsn = restart_lsn;
54 
55  /* Write this slot to disk */
58  }
59 }
void ReplicationSlotReserveWal(void)
Definition: slot.c:1158
@ RS_PERSISTENT
Definition: slot.h:35

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

Referenced by copy_replication_slot(), and pg_create_physical_replication_slot().

◆ pg_copy_logical_replication_slot_a()

Datum pg_copy_logical_replication_slot_a ( PG_FUNCTION_ARGS  )

Definition at line 890 of file slotfuncs.c.

891 {
892  return copy_replication_slot(fcinfo, true);
893 }
static Datum copy_replication_slot(FunctionCallInfo fcinfo, bool logical_slot)
Definition: slotfuncs.c:668

References copy_replication_slot().

◆ pg_copy_logical_replication_slot_b()

Datum pg_copy_logical_replication_slot_b ( PG_FUNCTION_ARGS  )

Definition at line 896 of file slotfuncs.c.

897 {
898  return copy_replication_slot(fcinfo, true);
899 }

References copy_replication_slot().

◆ pg_copy_logical_replication_slot_c()

Datum pg_copy_logical_replication_slot_c ( PG_FUNCTION_ARGS  )

Definition at line 902 of file slotfuncs.c.

903 {
904  return copy_replication_slot(fcinfo, true);
905 }

References copy_replication_slot().

◆ pg_copy_physical_replication_slot_a()

Datum pg_copy_physical_replication_slot_a ( PG_FUNCTION_ARGS  )

Definition at line 908 of file slotfuncs.c.

909 {
910  return copy_replication_slot(fcinfo, false);
911 }

References copy_replication_slot().

◆ pg_copy_physical_replication_slot_b()

Datum pg_copy_physical_replication_slot_b ( PG_FUNCTION_ARGS  )

Definition at line 914 of file slotfuncs.c.

915 {
916  return copy_replication_slot(fcinfo, false);
917 }

References copy_replication_slot().

◆ pg_create_logical_replication_slot()

Datum pg_create_logical_replication_slot ( PG_FUNCTION_ARGS  )

Definition at line 168 of file slotfuncs.c.

169 {
170  Name name = PG_GETARG_NAME(0);
172  bool temporary = PG_GETARG_BOOL(2);
173  bool two_phase = PG_GETARG_BOOL(3);
174  Datum result;
175  TupleDesc tupdesc;
176  HeapTuple tuple;
177  Datum values[2];
178  bool nulls[2];
179 
180  if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
181  elog(ERROR, "return type must be a row type");
182 
184 
186 
188  NameStr(*plugin),
189  temporary,
190  two_phase,
192  true);
193 
196 
197  memset(nulls, 0, sizeof(nulls));
198 
199  tuple = heap_form_tuple(tupdesc, values, nulls);
200  result = HeapTupleGetDatum(tuple);
201 
202  /* ok, slot is now fully created, mark it as persistent if needed */
203  if (!temporary)
206 
207  PG_RETURN_DATUM(result);
208 }
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28

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

◆ pg_create_physical_replication_slot()

Datum pg_create_physical_replication_slot ( PG_FUNCTION_ARGS  )

Definition at line 66 of file slotfuncs.c.

67 {
69  bool immediately_reserve = PG_GETARG_BOOL(1);
70  bool temporary = PG_GETARG_BOOL(2);
71  Datum values[2];
72  bool nulls[2];
73  TupleDesc tupdesc;
74  HeapTuple tuple;
75  Datum result;
76 
77  if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
78  elog(ERROR, "return type must be a row type");
79 
81 
83 
85  immediately_reserve,
86  temporary,
88 
90  nulls[0] = false;
91 
92  if (immediately_reserve)
93  {
95  nulls[1] = false;
96  }
97  else
98  nulls[1] = true;
99 
100  tuple = heap_form_tuple(tupdesc, values, nulls);
101  result = HeapTupleGetDatum(tuple);
102 
104 
105  PG_RETURN_DATUM(result);
106 }

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

◆ pg_drop_replication_slot()

Datum pg_drop_replication_slot ( PG_FUNCTION_ARGS  )

Definition at line 215 of file slotfuncs.c.

216 {
217  Name name = PG_GETARG_NAME(0);
218 
220 
222 
224 
225  PG_RETURN_VOID();
226 }
#define PG_RETURN_VOID()
Definition: fmgr.h:349
void ReplicationSlotDrop(const char *name, bool nowait)
Definition: slot.c:641

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

◆ pg_get_replication_slots()

Datum pg_get_replication_slots ( PG_FUNCTION_ARGS  )

Definition at line 233 of file slotfuncs.c.

234 {
235 #define PG_GET_REPLICATION_SLOTS_COLS 14
236  ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
237  XLogRecPtr currlsn;
238  int slotno;
239 
240  /*
241  * We don't require any special permission to see this function's data
242  * because nothing should be sensitive. The most critical being the slot
243  * name, which shouldn't contain anything particularly sensitive.
244  */
245 
246  InitMaterializedSRF(fcinfo, 0);
247 
248  currlsn = GetXLogWriteRecPtr();
249 
250  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
251  for (slotno = 0; slotno < max_replication_slots; slotno++)
252  {
254  ReplicationSlot slot_contents;
256  bool nulls[PG_GET_REPLICATION_SLOTS_COLS];
257  WALAvailability walstate;
258  int i;
259 
260  if (!slot->in_use)
261  continue;
262 
263  /* Copy slot contents while holding spinlock, then examine at leisure */
264  SpinLockAcquire(&slot->mutex);
265  slot_contents = *slot;
266  SpinLockRelease(&slot->mutex);
267 
268  memset(values, 0, sizeof(values));
269  memset(nulls, 0, sizeof(nulls));
270 
271  i = 0;
272  values[i++] = NameGetDatum(&slot_contents.data.name);
273 
274  if (slot_contents.data.database == InvalidOid)
275  nulls[i++] = true;
276  else
277  values[i++] = NameGetDatum(&slot_contents.data.plugin);
278 
279  if (slot_contents.data.database == InvalidOid)
280  values[i++] = CStringGetTextDatum("physical");
281  else
282  values[i++] = CStringGetTextDatum("logical");
283 
284  if (slot_contents.data.database == InvalidOid)
285  nulls[i++] = true;
286  else
287  values[i++] = ObjectIdGetDatum(slot_contents.data.database);
288 
289  values[i++] = BoolGetDatum(slot_contents.data.persistency == RS_TEMPORARY);
290  values[i++] = BoolGetDatum(slot_contents.active_pid != 0);
291 
292  if (slot_contents.active_pid != 0)
293  values[i++] = Int32GetDatum(slot_contents.active_pid);
294  else
295  nulls[i++] = true;
296 
297  if (slot_contents.data.xmin != InvalidTransactionId)
298  values[i++] = TransactionIdGetDatum(slot_contents.data.xmin);
299  else
300  nulls[i++] = true;
301 
302  if (slot_contents.data.catalog_xmin != InvalidTransactionId)
303  values[i++] = TransactionIdGetDatum(slot_contents.data.catalog_xmin);
304  else
305  nulls[i++] = true;
306 
307  if (slot_contents.data.restart_lsn != InvalidXLogRecPtr)
308  values[i++] = LSNGetDatum(slot_contents.data.restart_lsn);
309  else
310  nulls[i++] = true;
311 
312  if (slot_contents.data.confirmed_flush != InvalidXLogRecPtr)
313  values[i++] = LSNGetDatum(slot_contents.data.confirmed_flush);
314  else
315  nulls[i++] = true;
316 
317  /*
318  * If invalidated_at is valid and restart_lsn is invalid, we know for
319  * certain that the slot has been invalidated. Otherwise, test
320  * availability from restart_lsn.
321  */
322  if (XLogRecPtrIsInvalid(slot_contents.data.restart_lsn) &&
323  !XLogRecPtrIsInvalid(slot_contents.data.invalidated_at))
324  walstate = WALAVAIL_REMOVED;
325  else
326  walstate = GetWALAvailability(slot_contents.data.restart_lsn);
327 
328  switch (walstate)
329  {
331  nulls[i++] = true;
332  break;
333 
334  case WALAVAIL_RESERVED:
335  values[i++] = CStringGetTextDatum("reserved");
336  break;
337 
338  case WALAVAIL_EXTENDED:
339  values[i++] = CStringGetTextDatum("extended");
340  break;
341 
342  case WALAVAIL_UNRESERVED:
343  values[i++] = CStringGetTextDatum("unreserved");
344  break;
345 
346  case WALAVAIL_REMOVED:
347 
348  /*
349  * If we read the restart_lsn long enough ago, maybe that file
350  * has been removed by now. However, the walsender could have
351  * moved forward enough that it jumped to another file after
352  * we looked. If checkpointer signalled the process to
353  * termination, then it's definitely lost; but if a process is
354  * still alive, then "unreserved" seems more appropriate.
355  *
356  * If we do change it, save the state for safe_wal_size below.
357  */
358  if (!XLogRecPtrIsInvalid(slot_contents.data.restart_lsn))
359  {
360  int pid;
361 
362  SpinLockAcquire(&slot->mutex);
363  pid = slot->active_pid;
364  slot_contents.data.restart_lsn = slot->data.restart_lsn;
365  SpinLockRelease(&slot->mutex);
366  if (pid != 0)
367  {
368  values[i++] = CStringGetTextDatum("unreserved");
369  walstate = WALAVAIL_UNRESERVED;
370  break;
371  }
372  }
373  values[i++] = CStringGetTextDatum("lost");
374  break;
375  }
376 
377  /*
378  * safe_wal_size is only computed for slots that have not been lost,
379  * and only if there's a configured maximum size.
380  */
381  if (walstate == WALAVAIL_REMOVED || max_slot_wal_keep_size_mb < 0)
382  nulls[i++] = true;
383  else
384  {
385  XLogSegNo targetSeg;
386  uint64 slotKeepSegs;
387  uint64 keepSegs;
388  XLogSegNo failSeg;
389  XLogRecPtr failLSN;
390 
391  XLByteToSeg(slot_contents.data.restart_lsn, targetSeg, wal_segment_size);
392 
393  /* determine how many segments can be kept by slots */
395  /* ditto for wal_keep_size */
397 
398  /* if currpos reaches failLSN, we lose our segment */
399  failSeg = targetSeg + Max(slotKeepSegs, keepSegs) + 1;
400  XLogSegNoOffsetToRecPtr(failSeg, 0, wal_segment_size, failLSN);
401 
402  values[i++] = Int64GetDatum(failLSN - currlsn);
403  }
404 
405  values[i++] = BoolGetDatum(slot_contents.data.two_phase);
406 
408 
409  tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc,
410  values, nulls);
411  }
412 
413  LWLockRelease(ReplicationSlotControlLock);
414 
415  return (Datum) 0;
416 }
#define CStringGetTextDatum(s)
Definition: builtins.h:94
#define Max(x, y)
Definition: c.h:982
Datum Int64GetDatum(int64 X)
Definition: fmgr.c:1779
void InitMaterializedSRF(FunctionCallInfo fcinfo, bits32 flags)
Definition: funcapi.c:76
static Datum TransactionIdGetDatum(TransactionId X)
Definition: postgres.h:272
static Datum BoolGetDatum(bool X)
Definition: postgres.h:102
static Datum ObjectIdGetDatum(Oid X)
Definition: postgres.h:252
static Datum Int32GetDatum(int32 X)
Definition: postgres.h:212
#define InvalidOid
Definition: postgres_ext.h:36
#define PG_GET_REPLICATION_SLOTS_COLS
XLogRecPtr invalidated_at
Definition: slot.h:76
pid_t active_pid
Definition: slot.h:126
TupleDesc setDesc
Definition: execnodes.h:334
Tuplestorestate * setResult
Definition: execnodes.h:333
#define InvalidTransactionId
Definition: transam.h:31
void tuplestore_putvalues(Tuplestorestate *state, TupleDesc tdesc, Datum *values, bool *isnull)
Definition: tuplestore.c:750
int wal_keep_size_mb
Definition: xlog.c:119
WALAvailability GetWALAvailability(XLogRecPtr targetLSN)
Definition: xlog.c:7347
int max_slot_wal_keep_size_mb
Definition: xlog.c:138
XLogRecPtr GetXLogWriteRecPtr(void)
Definition: xlog.c:8873
WALAvailability
Definition: xlog.h:182
@ WALAVAIL_REMOVED
Definition: xlog.h:188
@ WALAVAIL_RESERVED
Definition: xlog.h:184
@ WALAVAIL_UNRESERVED
Definition: xlog.h:187
@ WALAVAIL_EXTENDED
Definition: xlog.h:185
@ WALAVAIL_INVALID_LSN
Definition: xlog.h:183
#define XLogSegNoOffsetToRecPtr(segno, offset, wal_segsz_bytes, dest)
#define XLogMBVarToSegs(mbvar, wal_segsz_bytes)

References ReplicationSlot::active_pid, Assert(), BoolGetDatum(), ReplicationSlotPersistentData::catalog_xmin, ReplicationSlotPersistentData::confirmed_flush, CStringGetTextDatum, ReplicationSlot::data, ReplicationSlotPersistentData::database, GetWALAvailability(), GetXLogWriteRecPtr(), i, ReplicationSlot::in_use, InitMaterializedSRF(), Int32GetDatum(), Int64GetDatum(), ReplicationSlotPersistentData::invalidated_at, InvalidOid, InvalidTransactionId, InvalidXLogRecPtr, LSNGetDatum(), LW_SHARED, LWLockAcquire(), LWLockRelease(), Max, max_replication_slots, max_slot_wal_keep_size_mb, ReplicationSlot::mutex, ReplicationSlotPersistentData::name, NameGetDatum(), ObjectIdGetDatum(), ReplicationSlotPersistentData::persistency, PG_GET_REPLICATION_SLOTS_COLS, ReplicationSlotPersistentData::plugin, ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, ReplicationSlotPersistentData::restart_lsn, RS_TEMPORARY, ReturnSetInfo::setDesc, ReturnSetInfo::setResult, SpinLockAcquire, SpinLockRelease, TransactionIdGetDatum(), tuplestore_putvalues(), ReplicationSlotPersistentData::two_phase, values, wal_keep_size_mb, wal_segment_size, WALAVAIL_EXTENDED, WALAVAIL_INVALID_LSN, WALAVAIL_REMOVED, WALAVAIL_RESERVED, WALAVAIL_UNRESERVED, XLByteToSeg, XLogMBVarToSegs, XLogRecPtrIsInvalid, XLogSegNoOffsetToRecPtr, and ReplicationSlotPersistentData::xmin.

◆ pg_logical_replication_slot_advance()

static XLogRecPtr pg_logical_replication_slot_advance ( XLogRecPtr  moveto)
static

Definition at line 463 of file slotfuncs.c.

464 {
466  ResourceOwner old_resowner = CurrentResourceOwner;
467  XLogRecPtr retlsn;
468 
469  Assert(moveto != InvalidXLogRecPtr);
470 
471  PG_TRY();
472  {
473  /*
474  * Create our decoding context in fast_forward mode, passing start_lsn
475  * as InvalidXLogRecPtr, so that we start processing from my slot's
476  * confirmed_flush.
477  */
479  NIL,
480  true, /* fast_forward */
481  XL_ROUTINE(.page_read = read_local_xlog_page,
482  .segment_open = wal_segment_open,
483  .segment_close = wal_segment_close),
484  NULL, NULL, NULL);
485 
486  /*
487  * Start reading at the slot's restart_lsn, which we know to point to
488  * a valid record.
489  */
491 
492  /* invalidate non-timetravel entries */
494 
495  /* Decode at least one record, until we run out of records */
496  while (ctx->reader->EndRecPtr < moveto)
497  {
498  char *errm = NULL;
499  XLogRecord *record;
500 
501  /*
502  * Read records. No changes are generated in fast_forward mode,
503  * but snapbuilder/slot statuses are updated properly.
504  */
505  record = XLogReadRecord(ctx->reader, &errm);
506  if (errm)
507  elog(ERROR, "could not find record while advancing replication slot: %s",
508  errm);
509 
510  /*
511  * Process the record. Storage-level changes are ignored in
512  * fast_forward mode, but other modules (such as snapbuilder)
513  * might still have critical updates to do.
514  */
515  if (record)
517 
518  /* Stop once the requested target has been reached */
519  if (moveto <= ctx->reader->EndRecPtr)
520  break;
521 
523  }
524 
525  /*
526  * Logical decoding could have clobbered CurrentResourceOwner during
527  * transaction management, so restore the executor's value. (This is
528  * a kluge, but it's not worth cleaning up right now.)
529  */
530  CurrentResourceOwner = old_resowner;
531 
532  if (ctx->reader->EndRecPtr != InvalidXLogRecPtr)
533  {
535 
536  /*
537  * If only the confirmed_flush LSN has changed the slot won't get
538  * marked as dirty by the above. Callers on the walsender
539  * interface are expected to keep track of their own progress and
540  * don't need it written out. But SQL-interface users cannot
541  * specify their own start positions and it's harder for them to
542  * keep track of their progress, so we should make more of an
543  * effort to save it for them.
544  *
545  * Dirty the slot so it is written out at the next checkpoint. The
546  * LSN position advanced to may still be lost on a crash but this
547  * makes the data consistent after a clean shutdown.
548  */
550  }
551 
553 
554  /* free context, call shutdown callback */
555  FreeDecodingContext(ctx);
556 
558  }
559  PG_CATCH();
560  {
561  /* clear all timetravel entries */
563 
564  PG_RE_THROW();
565  }
566  PG_END_TRY();
567 
568  return retlsn;
569 }
void LogicalDecodingProcessRecord(LogicalDecodingContext *ctx, XLogReaderState *record)
Definition: decode.c:91
#define PG_RE_THROW()
Definition: elog.h:411
#define PG_TRY(...)
Definition: elog.h:370
#define PG_END_TRY(...)
Definition: elog.h:395
#define PG_CATCH(...)
Definition: elog.h:380
void InvalidateSystemCaches(void)
Definition: inval.c:702
void LogicalConfirmReceivedLocation(XLogRecPtr lsn)
Definition: logical.c:1788
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:490
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:121
ResourceOwner CurrentResourceOwner
Definition: resowner.c:146
XLogReaderState * reader
Definition: logical.h:42
XLogRecPtr EndRecPtr
Definition: xlogreader.h:207
XLogRecord * XLogReadRecord(XLogReaderState *state, char **errormsg)
Definition: xlogreader.c:422
void XLogBeginRead(XLogReaderState *state, XLogRecPtr RecPtr)
Definition: xlogreader.c:264

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().

◆ pg_physical_replication_slot_advance()

static XLogRecPtr pg_physical_replication_slot_advance ( XLogRecPtr  moveto)
static

Definition at line 426 of file slotfuncs.c.

427 {
429  XLogRecPtr retlsn = startlsn;
430 
431  Assert(moveto != InvalidXLogRecPtr);
432 
433  if (startlsn < moveto)
434  {
438  retlsn = moveto;
439 
440  /*
441  * Dirty the slot so as it is written out at the next checkpoint. Note
442  * that the LSN position advanced may still be lost in the event of a
443  * crash, but this makes the data consistent after a clean shutdown.
444  */
446  }
447 
448  return retlsn;
449 }

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

Referenced by pg_replication_slot_advance().

◆ pg_replication_slot_advance()

Datum pg_replication_slot_advance ( PG_FUNCTION_ARGS  )

Definition at line 575 of file slotfuncs.c.

576 {
577  Name slotname = PG_GETARG_NAME(0);
578  XLogRecPtr moveto = PG_GETARG_LSN(1);
579  XLogRecPtr endlsn;
580  XLogRecPtr minlsn;
581  TupleDesc tupdesc;
582  Datum values[2];
583  bool nulls[2];
584  HeapTuple tuple;
585  Datum result;
586 
588 
590 
591  if (XLogRecPtrIsInvalid(moveto))
592  ereport(ERROR,
593  (errmsg("invalid target WAL LSN")));
594 
595  /* Build a tuple descriptor for our result type */
596  if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
597  elog(ERROR, "return type must be a row type");
598 
599  /*
600  * We can't move slot past what's been flushed/replayed so clamp the
601  * target position accordingly.
602  */
603  if (!RecoveryInProgress())
604  moveto = Min(moveto, GetFlushRecPtr(NULL));
605  else
606  moveto = Min(moveto, GetXLogReplayRecPtr(NULL));
607 
608  /* Acquire the slot so we "own" it */
609  ReplicationSlotAcquire(NameStr(*slotname), true);
610 
611  /* A slot whose restart_lsn has never been reserved cannot be advanced */
613  ereport(ERROR,
614  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
615  errmsg("replication slot \"%s\" cannot be advanced",
616  NameStr(*slotname)),
617  errdetail("This slot has never previously reserved WAL, or it has been invalidated.")));
618 
619  /*
620  * Check if the slot is not moving backwards. Physical slots rely simply
621  * on restart_lsn as a minimum point, while logical slots have confirmed
622  * consumption up to confirmed_flush, meaning that in both cases data
623  * older than that is not available anymore.
624  */
627  else
629 
630  if (moveto < minlsn)
631  ereport(ERROR,
632  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
633  errmsg("cannot advance replication slot to %X/%X, minimum is %X/%X",
634  LSN_FORMAT_ARGS(moveto), LSN_FORMAT_ARGS(minlsn))));
635 
636  /* Do the actual slot update, depending on the slot type */
638  endlsn = pg_logical_replication_slot_advance(moveto);
639  else
640  endlsn = pg_physical_replication_slot_advance(moveto);
641 
643  nulls[0] = false;
644 
645  /*
646  * Recompute the minimum LSN and xmin across all slots to adjust with the
647  * advancing potentially done.
648  */
651 
653 
654  /* Return the reached position. */
655  values[1] = LSNGetDatum(endlsn);
656  nulls[1] = false;
657 
658  tuple = heap_form_tuple(tupdesc, values, nulls);
659  result = HeapTupleGetDatum(tuple);
660 
661  PG_RETURN_DATUM(result);
662 }
#define Min(x, y)
Definition: c.h:988
#define OidIsValid(objectId)
Definition: c.h:759
#define PG_GETARG_LSN(n)
Definition: pg_lsn.h:33
void ReplicationSlotAcquire(const char *name, bool nowait)
Definition: slot.c:450
static XLogRecPtr pg_logical_replication_slot_advance(XLogRecPtr moveto)
Definition: slotfuncs.c:463
static XLogRecPtr pg_physical_replication_slot_advance(XLogRecPtr moveto)
Definition: slotfuncs.c:426
bool RecoveryInProgress(void)
Definition: xlog.c:5908
XLogRecPtr GetFlushRecPtr(TimeLineID *insertTLI)
Definition: xlog.c:6073
#define LSN_FORMAT_ARGS(lsn)
Definition: xlogdefs.h:43
XLogRecPtr GetXLogReplayRecPtr(TimeLineID *replayTLI)

References Assert(), CheckSlotPermissions(), ReplicationSlotPersistentData::confirmed_flush, ReplicationSlot::data, ReplicationSlotPersistentData::database, elog(), ereport, errcode(), errdetail(), errmsg(), ERROR, get_call_result_type(), GetFlushRecPtr(), GetXLogReplayRecPtr(), heap_form_tuple(), HeapTupleGetDatum(), LSN_FORMAT_ARGS, 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(), ReplicationSlotsComputeRequiredLSN(), ReplicationSlotsComputeRequiredXmin(), ReplicationSlotPersistentData::restart_lsn, TYPEFUNC_COMPOSITE, values, and XLogRecPtrIsInvalid.