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 "replication/slotsync.h"
#include "utils/builtins.h"
#include "utils/guc.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   17
 

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, bool failover, 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)
 
Datum pg_sync_replication_slots (PG_FUNCTION_ARGS)
 

Macro Definition Documentation

◆ PG_GET_REPLICATION_SLOTS_COLS

#define PG_GET_REPLICATION_SLOTS_COLS   17

Function Documentation

◆ copy_replication_slot()

static Datum copy_replication_slot ( FunctionCallInfo  fcinfo,
bool  logical_slot 
)
static

Definition at line 685 of file slotfuncs.c.

686 {
687  Name src_name = PG_GETARG_NAME(0);
688  Name dst_name = PG_GETARG_NAME(1);
689  ReplicationSlot *src = NULL;
690  ReplicationSlot first_slot_contents;
691  ReplicationSlot second_slot_contents;
692  XLogRecPtr src_restart_lsn;
693  bool src_islogical;
694  bool temporary;
695  char *plugin;
696  Datum values[2];
697  bool nulls[2];
698  Datum result;
699  TupleDesc tupdesc;
700  HeapTuple tuple;
701 
702  if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
703  elog(ERROR, "return type must be a row type");
704 
706 
707  if (logical_slot)
709  else
711 
712  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
713 
714  /*
715  * We need to prevent the source slot's reserved WAL from being removed,
716  * but we don't want to lock that slot for very long, and it can advance
717  * in the meantime. So obtain the source slot's data, and create a new
718  * slot using its restart_lsn. Afterwards we lock the source slot again
719  * and verify that the data we copied (name, type) has not changed
720  * incompatibly. No inconvenient WAL removal can occur once the new slot
721  * is created -- but since WAL removal could have occurred before we
722  * managed to create the new slot, we advance the new slot's restart_lsn
723  * to the source slot's updated restart_lsn the second time we lock it.
724  */
725  for (int i = 0; i < max_replication_slots; i++)
726  {
728 
729  if (s->in_use && strcmp(NameStr(s->data.name), NameStr(*src_name)) == 0)
730  {
731  /* Copy the slot contents while holding spinlock */
732  SpinLockAcquire(&s->mutex);
733  first_slot_contents = *s;
734  SpinLockRelease(&s->mutex);
735  src = s;
736  break;
737  }
738  }
739 
740  LWLockRelease(ReplicationSlotControlLock);
741 
742  if (src == NULL)
743  ereport(ERROR,
744  (errcode(ERRCODE_UNDEFINED_OBJECT),
745  errmsg("replication slot \"%s\" does not exist", NameStr(*src_name))));
746 
747  src_islogical = SlotIsLogical(&first_slot_contents);
748  src_restart_lsn = first_slot_contents.data.restart_lsn;
749  temporary = (first_slot_contents.data.persistency == RS_TEMPORARY);
750  plugin = logical_slot ? NameStr(first_slot_contents.data.plugin) : NULL;
751 
752  /* Check type of replication slot */
753  if (src_islogical != logical_slot)
754  ereport(ERROR,
755  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
756  src_islogical ?
757  errmsg("cannot copy physical replication slot \"%s\" as a logical replication slot",
758  NameStr(*src_name)) :
759  errmsg("cannot copy logical replication slot \"%s\" as a physical replication slot",
760  NameStr(*src_name))));
761 
762  /* Copying non-reserved slot doesn't make sense */
763  if (XLogRecPtrIsInvalid(src_restart_lsn))
764  ereport(ERROR,
765  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
766  errmsg("cannot copy a replication slot that doesn't reserve WAL")));
767 
768  /* Overwrite params from optional arguments */
769  if (PG_NARGS() >= 3)
770  temporary = PG_GETARG_BOOL(2);
771  if (PG_NARGS() >= 4)
772  {
773  Assert(logical_slot);
774  plugin = NameStr(*(PG_GETARG_NAME(3)));
775  }
776 
777  /* Create new slot and acquire it */
778  if (logical_slot)
779  {
780  /*
781  * We must not try to read WAL, since we haven't reserved it yet --
782  * hence pass find_startpoint false. confirmed_flush will be set
783  * below, by copying from the source slot.
784  *
785  * To avoid potential issues with the slot synchronization where the
786  * restart_lsn of a replication slot can go backward, we set the
787  * failover option to false here. This situation occurs when a slot
788  * on the primary server is dropped and immediately replaced with a
789  * new slot of the same name, created by copying from another existing
790  * slot. However, the slot synchronization will only observe the
791  * restart_lsn of the same slot going backward.
792  */
794  plugin,
795  temporary,
796  false,
797  false,
798  src_restart_lsn,
799  false);
800  }
801  else
803  true,
804  temporary,
805  src_restart_lsn);
806 
807  /*
808  * Update the destination slot to current values of the source slot;
809  * recheck that the source slot is still the one we saw previously.
810  */
811  {
812  TransactionId copy_effective_xmin;
813  TransactionId copy_effective_catalog_xmin;
814  TransactionId copy_xmin;
815  TransactionId copy_catalog_xmin;
816  XLogRecPtr copy_restart_lsn;
817  XLogRecPtr copy_confirmed_flush;
818  bool copy_islogical;
819  char *copy_name;
820 
821  /* Copy data of source slot again */
822  SpinLockAcquire(&src->mutex);
823  second_slot_contents = *src;
824  SpinLockRelease(&src->mutex);
825 
826  copy_effective_xmin = second_slot_contents.effective_xmin;
827  copy_effective_catalog_xmin = second_slot_contents.effective_catalog_xmin;
828 
829  copy_xmin = second_slot_contents.data.xmin;
830  copy_catalog_xmin = second_slot_contents.data.catalog_xmin;
831  copy_restart_lsn = second_slot_contents.data.restart_lsn;
832  copy_confirmed_flush = second_slot_contents.data.confirmed_flush;
833 
834  /* for existence check */
835  copy_name = NameStr(second_slot_contents.data.name);
836  copy_islogical = SlotIsLogical(&second_slot_contents);
837 
838  /*
839  * Check if the source slot still exists and is valid. We regard it as
840  * invalid if the type of replication slot or name has been changed,
841  * or the restart_lsn either is invalid or has gone backward. (The
842  * restart_lsn could go backwards if the source slot is dropped and
843  * copied from an older slot during installation.)
844  *
845  * Since erroring out will release and drop the destination slot we
846  * don't need to release it here.
847  */
848  if (copy_restart_lsn < src_restart_lsn ||
849  src_islogical != copy_islogical ||
850  strcmp(copy_name, NameStr(*src_name)) != 0)
851  ereport(ERROR,
852  (errmsg("could not copy replication slot \"%s\"",
853  NameStr(*src_name)),
854  errdetail("The source replication slot was modified incompatibly during the copy operation.")));
855 
856  /* The source slot must have a consistent snapshot */
857  if (src_islogical && XLogRecPtrIsInvalid(copy_confirmed_flush))
858  ereport(ERROR,
859  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
860  errmsg("cannot copy unfinished logical replication slot \"%s\"",
861  NameStr(*src_name)),
862  errhint("Retry when the source replication slot's confirmed_flush_lsn is valid.")));
863 
864  /* Install copied values again */
866  MyReplicationSlot->effective_xmin = copy_effective_xmin;
867  MyReplicationSlot->effective_catalog_xmin = copy_effective_catalog_xmin;
868 
869  MyReplicationSlot->data.xmin = copy_xmin;
870  MyReplicationSlot->data.catalog_xmin = copy_catalog_xmin;
871  MyReplicationSlot->data.restart_lsn = copy_restart_lsn;
872  MyReplicationSlot->data.confirmed_flush = copy_confirmed_flush;
874 
879 
880 #ifdef USE_ASSERT_CHECKING
881  /* Check that the restart_lsn is available */
882  {
883  XLogSegNo segno;
884 
885  XLByteToSeg(copy_restart_lsn, segno, wal_segment_size);
886  Assert(XLogGetLastRemovedSegno() < segno);
887  }
888 #endif
889  }
890 
891  /* target slot fully created, mark as persistent if needed */
892  if (logical_slot && !temporary)
894 
895  /* All done. Set up the return values */
896  values[0] = NameGetDatum(dst_name);
897  nulls[0] = false;
899  {
901  nulls[1] = false;
902  }
903  else
904  nulls[1] = true;
905 
906  tuple = heap_form_tuple(tupdesc, values, nulls);
907  result = HeapTupleGetDatum(tuple);
908 
910 
911  PG_RETURN_DATUM(result);
912 }
static Datum values[MAXATTR]
Definition: bootstrap.c:156
#define NameStr(name)
Definition: c.h:733
uint32 TransactionId
Definition: c.h:639
int errdetail(const char *fmt,...)
Definition: elog.c:1206
int errhint(const char *fmt,...)
Definition: elog.c:1320
int errcode(int sqlerrcode)
Definition: elog.c:860
int errmsg(const char *fmt,...)
Definition: elog.c:1073
#define ERROR
Definition: elog.h:39
#define elog(elevel,...)
Definition: elog.h:224
#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, const Datum *values, const bool *isnull)
Definition: heaptuple.c:1117
int i
Definition: isn.c:73
Assert(fmt[strlen(fmt) - 1] !='\n')
void CheckLogicalDecodingRequirements(void)
Definition: logical.c:110
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1175
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1788
@ LW_SHARED
Definition: lwlock.h:117
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:946
void ReplicationSlotsComputeRequiredXmin(bool already_locked)
Definition: slot.c:985
void ReplicationSlotPersist(void)
Definition: slot.c:963
ReplicationSlot * MyReplicationSlot
Definition: slot.c:116
void ReplicationSlotSave(void)
Definition: slot.c:928
void CheckSlotPermissions(void)
Definition: slot.c:1320
void ReplicationSlotRelease(void)
Definition: slot.c:606
int max_replication_slots
Definition: slot.c:119
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:113
void ReplicationSlotsComputeRequiredLSN(void)
Definition: slot.c:1041
void CheckSlotRequirements(void)
Definition: slot.c:1298
@ RS_TEMPORARY
Definition: slot.h:37
#define SlotIsLogical(slot)
Definition: slot.h:207
static void create_logical_replication_slot(char *name, char *plugin, bool temporary, bool two_phase, bool failover, XLogRecPtr restart_lsn, bool find_startpoint)
Definition: slotfuncs.c:121
static void create_physical_replication_slot(char *name, bool immediately_reserve, bool temporary, XLogRecPtr restart_lsn)
Definition: slotfuncs.c:40
#define SpinLockRelease(lock)
Definition: spin.h:64
#define SpinLockAcquire(lock)
Definition: spin.h:62
ReplicationSlot replication_slots[1]
Definition: slot.h:218
TransactionId xmin
Definition: slot.h:82
TransactionId catalog_xmin
Definition: slot.h:90
XLogRecPtr restart_lsn
Definition: slot.h:93
XLogRecPtr confirmed_flush
Definition: slot.h:104
ReplicationSlotPersistency persistency
Definition: slot.h:74
TransactionId effective_catalog_xmin
Definition: slot.h:175
slock_t mutex
Definition: slot.h:151
bool in_use
Definition: slot.h:154
TransactionId effective_xmin
Definition: slot.h:174
ReplicationSlotPersistentData data
Definition: slot.h:178
Definition: c.h:728
XLogSegNo XLogGetLastRemovedSegno(void)
Definition: xlog.c:3693
int wal_segment_size
Definition: xlog.c:148
#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,
bool  failover,
XLogRecPtr  restart_lsn,
bool  find_startpoint 
)
static

Definition at line 121 of file slotfuncs.c.

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

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 40 of file slotfuncs.c.

42 {
44 
45  /* acquire replication slot, this will check for conflicting names */
47  temporary ? RS_TEMPORARY : RS_PERSISTENT, false,
48  false, false);
49 
50  if (immediately_reserve)
51  {
52  /* Reserve WAL as the user asked for it */
53  if (XLogRecPtrIsInvalid(restart_lsn))
55  else
56  MyReplicationSlot->data.restart_lsn = restart_lsn;
57 
58  /* Write this slot to disk */
61  }
62 }
void ReplicationSlotReserveWal(void)
Definition: slot.c:1337
@ 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 916 of file slotfuncs.c.

917 {
918  return copy_replication_slot(fcinfo, true);
919 }
static Datum copy_replication_slot(FunctionCallInfo fcinfo, bool logical_slot)
Definition: slotfuncs.c:685

References copy_replication_slot().

◆ pg_copy_logical_replication_slot_b()

Datum pg_copy_logical_replication_slot_b ( PG_FUNCTION_ARGS  )

Definition at line 922 of file slotfuncs.c.

923 {
924  return copy_replication_slot(fcinfo, true);
925 }

References copy_replication_slot().

◆ pg_copy_logical_replication_slot_c()

Datum pg_copy_logical_replication_slot_c ( PG_FUNCTION_ARGS  )

Definition at line 928 of file slotfuncs.c.

929 {
930  return copy_replication_slot(fcinfo, true);
931 }

References copy_replication_slot().

◆ pg_copy_physical_replication_slot_a()

Datum pg_copy_physical_replication_slot_a ( PG_FUNCTION_ARGS  )

Definition at line 934 of file slotfuncs.c.

935 {
936  return copy_replication_slot(fcinfo, false);
937 }

References copy_replication_slot().

◆ pg_copy_physical_replication_slot_b()

Datum pg_copy_physical_replication_slot_b ( PG_FUNCTION_ARGS  )

Definition at line 940 of file slotfuncs.c.

941 {
942  return copy_replication_slot(fcinfo, false);
943 }

References copy_replication_slot().

◆ pg_create_logical_replication_slot()

Datum pg_create_logical_replication_slot ( PG_FUNCTION_ARGS  )

Definition at line 173 of file slotfuncs.c.

174 {
175  Name name = PG_GETARG_NAME(0);
177  bool temporary = PG_GETARG_BOOL(2);
178  bool two_phase = PG_GETARG_BOOL(3);
179  bool failover = PG_GETARG_BOOL(4);
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,
196  two_phase,
197  failover,
199  true);
200 
203 
204  memset(nulls, 0, sizeof(nulls));
205 
206  tuple = heap_form_tuple(tupdesc, values, nulls);
207  result = HeapTupleGetDatum(tuple);
208 
209  /* ok, slot is now fully created, mark it as persistent if needed */
210  if (!temporary)
213 
214  PG_RETURN_DATUM(result);
215 }
#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 69 of file slotfuncs.c.

70 {
72  bool immediately_reserve = PG_GETARG_BOOL(1);
73  bool temporary = PG_GETARG_BOOL(2);
74  Datum values[2];
75  bool nulls[2];
76  TupleDesc tupdesc;
77  HeapTuple tuple;
78  Datum result;
79 
80  if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
81  elog(ERROR, "return type must be a row type");
82 
84 
86 
88  immediately_reserve,
89  temporary,
91 
93  nulls[0] = false;
94 
95  if (immediately_reserve)
96  {
98  nulls[1] = false;
99  }
100  else
101  nulls[1] = true;
102 
103  tuple = heap_form_tuple(tupdesc, values, nulls);
104  result = HeapTupleGetDatum(tuple);
105 
107 
108  PG_RETURN_DATUM(result);
109 }

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 222 of file slotfuncs.c.

223 {
224  Name name = PG_GETARG_NAME(0);
225 
227 
229 
231 
232  PG_RETURN_VOID();
233 }
#define PG_RETURN_VOID()
Definition: fmgr.h:349
void ReplicationSlotDrop(const char *name, bool nowait)
Definition: slot.c:720

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 240 of file slotfuncs.c.

241 {
242 #define PG_GET_REPLICATION_SLOTS_COLS 17
243  ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
244  XLogRecPtr currlsn;
245  int slotno;
246 
247  /*
248  * We don't require any special permission to see this function's data
249  * because nothing should be sensitive. The most critical being the slot
250  * name, which shouldn't contain anything particularly sensitive.
251  */
252 
253  InitMaterializedSRF(fcinfo, 0);
254 
255  currlsn = GetXLogWriteRecPtr();
256 
257  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
258  for (slotno = 0; slotno < max_replication_slots; slotno++)
259  {
261  ReplicationSlot slot_contents;
263  bool nulls[PG_GET_REPLICATION_SLOTS_COLS];
264  WALAvailability walstate;
265  int i;
266 
267  if (!slot->in_use)
268  continue;
269 
270  /* Copy slot contents while holding spinlock, then examine at leisure */
271  SpinLockAcquire(&slot->mutex);
272  slot_contents = *slot;
273  SpinLockRelease(&slot->mutex);
274 
275  memset(values, 0, sizeof(values));
276  memset(nulls, 0, sizeof(nulls));
277 
278  i = 0;
279  values[i++] = NameGetDatum(&slot_contents.data.name);
280 
281  if (slot_contents.data.database == InvalidOid)
282  nulls[i++] = true;
283  else
284  values[i++] = NameGetDatum(&slot_contents.data.plugin);
285 
286  if (slot_contents.data.database == InvalidOid)
287  values[i++] = CStringGetTextDatum("physical");
288  else
289  values[i++] = CStringGetTextDatum("logical");
290 
291  if (slot_contents.data.database == InvalidOid)
292  nulls[i++] = true;
293  else
294  values[i++] = ObjectIdGetDatum(slot_contents.data.database);
295 
296  values[i++] = BoolGetDatum(slot_contents.data.persistency == RS_TEMPORARY);
297  values[i++] = BoolGetDatum(slot_contents.active_pid != 0);
298 
299  if (slot_contents.active_pid != 0)
300  values[i++] = Int32GetDatum(slot_contents.active_pid);
301  else
302  nulls[i++] = true;
303 
304  if (slot_contents.data.xmin != InvalidTransactionId)
305  values[i++] = TransactionIdGetDatum(slot_contents.data.xmin);
306  else
307  nulls[i++] = true;
308 
309  if (slot_contents.data.catalog_xmin != InvalidTransactionId)
310  values[i++] = TransactionIdGetDatum(slot_contents.data.catalog_xmin);
311  else
312  nulls[i++] = true;
313 
314  if (slot_contents.data.restart_lsn != InvalidXLogRecPtr)
315  values[i++] = LSNGetDatum(slot_contents.data.restart_lsn);
316  else
317  nulls[i++] = true;
318 
319  if (slot_contents.data.confirmed_flush != InvalidXLogRecPtr)
320  values[i++] = LSNGetDatum(slot_contents.data.confirmed_flush);
321  else
322  nulls[i++] = true;
323 
324  /*
325  * If the slot has not been invalidated, test availability from
326  * restart_lsn.
327  */
328  if (slot_contents.data.invalidated != RS_INVAL_NONE)
329  walstate = WALAVAIL_REMOVED;
330  else
331  walstate = GetWALAvailability(slot_contents.data.restart_lsn);
332 
333  switch (walstate)
334  {
336  nulls[i++] = true;
337  break;
338 
339  case WALAVAIL_RESERVED:
340  values[i++] = CStringGetTextDatum("reserved");
341  break;
342 
343  case WALAVAIL_EXTENDED:
344  values[i++] = CStringGetTextDatum("extended");
345  break;
346 
347  case WALAVAIL_UNRESERVED:
348  values[i++] = CStringGetTextDatum("unreserved");
349  break;
350 
351  case WALAVAIL_REMOVED:
352 
353  /*
354  * If we read the restart_lsn long enough ago, maybe that file
355  * has been removed by now. However, the walsender could have
356  * moved forward enough that it jumped to another file after
357  * we looked. If checkpointer signalled the process to
358  * termination, then it's definitely lost; but if a process is
359  * still alive, then "unreserved" seems more appropriate.
360  *
361  * If we do change it, save the state for safe_wal_size below.
362  */
363  if (!XLogRecPtrIsInvalid(slot_contents.data.restart_lsn))
364  {
365  int pid;
366 
367  SpinLockAcquire(&slot->mutex);
368  pid = slot->active_pid;
369  slot_contents.data.restart_lsn = slot->data.restart_lsn;
370  SpinLockRelease(&slot->mutex);
371  if (pid != 0)
372  {
373  values[i++] = CStringGetTextDatum("unreserved");
374  walstate = WALAVAIL_UNRESERVED;
375  break;
376  }
377  }
378  values[i++] = CStringGetTextDatum("lost");
379  break;
380  }
381 
382  /*
383  * safe_wal_size is only computed for slots that have not been lost,
384  * and only if there's a configured maximum size.
385  */
386  if (walstate == WALAVAIL_REMOVED || max_slot_wal_keep_size_mb < 0)
387  nulls[i++] = true;
388  else
389  {
390  XLogSegNo targetSeg;
391  uint64 slotKeepSegs;
392  uint64 keepSegs;
393  XLogSegNo failSeg;
394  XLogRecPtr failLSN;
395 
396  XLByteToSeg(slot_contents.data.restart_lsn, targetSeg, wal_segment_size);
397 
398  /* determine how many segments can be kept by slots */
400  /* ditto for wal_keep_size */
402 
403  /* if currpos reaches failLSN, we lose our segment */
404  failSeg = targetSeg + Max(slotKeepSegs, keepSegs) + 1;
405  XLogSegNoOffsetToRecPtr(failSeg, 0, wal_segment_size, failLSN);
406 
407  values[i++] = Int64GetDatum(failLSN - currlsn);
408  }
409 
410  values[i++] = BoolGetDatum(slot_contents.data.two_phase);
411 
412  if (slot_contents.data.database == InvalidOid)
413  nulls[i++] = true;
414  else
415  {
416  ReplicationSlotInvalidationCause cause = slot_contents.data.invalidated;
417 
418  if (cause == RS_INVAL_NONE)
419  nulls[i++] = true;
420  else
422  }
423 
424  values[i++] = BoolGetDatum(slot_contents.data.failover);
425 
426  values[i++] = BoolGetDatum(slot_contents.data.synced);
427 
429 
430  tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc,
431  values, nulls);
432  }
433 
434  LWLockRelease(ReplicationSlotControlLock);
435 
436  return (Datum) 0;
437 }
#define CStringGetTextDatum(s)
Definition: builtins.h:97
#define Max(x, y)
Definition: c.h:985
Datum Int64GetDatum(int64 X)
Definition: fmgr.c:1807
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
const char *const SlotInvalidationCauses[]
Definition: slot.c:83
ReplicationSlotInvalidationCause
Definition: slot.h:48
@ RS_INVAL_NONE
Definition: slot.h:49
#define PG_GET_REPLICATION_SLOTS_COLS
ReplicationSlotInvalidationCause invalidated
Definition: slot.h:96
pid_t active_pid
Definition: slot.h:157
TupleDesc setDesc
Definition: execnodes.h:340
Tuplestorestate * setResult
Definition: execnodes.h:339
#define InvalidTransactionId
Definition: transam.h:31
void tuplestore_putvalues(Tuplestorestate *state, TupleDesc tdesc, const Datum *values, const bool *isnull)
Definition: tuplestore.c:750
int wal_keep_size_mb
Definition: xlog.c:121
WALAvailability GetWALAvailability(XLogRecPtr targetLSN)
Definition: xlog.c:7708
int max_slot_wal_keep_size_mb
Definition: xlog.c:140
XLogRecPtr GetXLogWriteRecPtr(void)
Definition: xlog.c:9287
WALAvailability
Definition: xlog.h:185
@ WALAVAIL_REMOVED
Definition: xlog.h:191
@ WALAVAIL_RESERVED
Definition: xlog.h:187
@ WALAVAIL_UNRESERVED
Definition: xlog.h:190
@ WALAVAIL_EXTENDED
Definition: xlog.h:188
@ WALAVAIL_INVALID_LSN
Definition: xlog.h:186
#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, ReplicationSlotPersistentData::failover, GetWALAvailability(), GetXLogWriteRecPtr(), i, ReplicationSlot::in_use, InitMaterializedSRF(), Int32GetDatum(), Int64GetDatum(), ReplicationSlotPersistentData::invalidated, 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_INVAL_NONE, RS_TEMPORARY, ReturnSetInfo::setDesc, ReturnSetInfo::setResult, SlotInvalidationCauses, SpinLockAcquire, SpinLockRelease, ReplicationSlotPersistentData::synced, 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 484 of file slotfuncs.c.

485 {
487  ResourceOwner old_resowner = CurrentResourceOwner;
488  XLogRecPtr retlsn;
489 
490  Assert(moveto != InvalidXLogRecPtr);
491 
492  PG_TRY();
493  {
494  /*
495  * Create our decoding context in fast_forward mode, passing start_lsn
496  * as InvalidXLogRecPtr, so that we start processing from my slot's
497  * confirmed_flush.
498  */
500  NIL,
501  true, /* fast_forward */
502  XL_ROUTINE(.page_read = read_local_xlog_page,
503  .segment_open = wal_segment_open,
504  .segment_close = wal_segment_close),
505  NULL, NULL, NULL);
506 
507  /*
508  * Start reading at the slot's restart_lsn, which we know to point to
509  * a valid record.
510  */
512 
513  /* invalidate non-timetravel entries */
515 
516  /* Decode records until we reach the requested target */
517  while (ctx->reader->EndRecPtr < moveto)
518  {
519  char *errm = NULL;
520  XLogRecord *record;
521 
522  /*
523  * Read records. No changes are generated in fast_forward mode,
524  * but snapbuilder/slot statuses are updated properly.
525  */
526  record = XLogReadRecord(ctx->reader, &errm);
527  if (errm)
528  elog(ERROR, "could not find record while advancing replication slot: %s",
529  errm);
530 
531  /*
532  * Process the record. Storage-level changes are ignored in
533  * fast_forward mode, but other modules (such as snapbuilder)
534  * might still have critical updates to do.
535  */
536  if (record)
538 
540  }
541 
542  /*
543  * Logical decoding could have clobbered CurrentResourceOwner during
544  * transaction management, so restore the executor's value. (This is
545  * a kluge, but it's not worth cleaning up right now.)
546  */
547  CurrentResourceOwner = old_resowner;
548 
549  if (ctx->reader->EndRecPtr != InvalidXLogRecPtr)
550  {
552 
553  /*
554  * If only the confirmed_flush LSN has changed the slot won't get
555  * marked as dirty by the above. Callers on the walsender
556  * interface are expected to keep track of their own progress and
557  * don't need it written out. But SQL-interface users cannot
558  * specify their own start positions and it's harder for them to
559  * keep track of their progress, so we should make more of an
560  * effort to save it for them.
561  *
562  * Dirty the slot so it is written out at the next checkpoint. The
563  * LSN position advanced to may still be lost on a crash but this
564  * makes the data consistent after a clean shutdown.
565  */
567  }
568 
570 
571  /* free context, call shutdown callback */
572  FreeDecodingContext(ctx);
573 
575  }
576  PG_CATCH();
577  {
578  /* clear all timetravel entries */
580 
581  PG_RE_THROW();
582  }
583  PG_END_TRY();
584 
585  return retlsn;
586 }
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:793
void LogicalConfirmReceivedLocation(XLogRecPtr lsn)
Definition: logical.c:1829
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:496
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:122
ResourceOwner CurrentResourceOwner
Definition: resowner.c:165
XLogReaderState * reader
Definition: logical.h:42
XLogRecPtr EndRecPtr
Definition: xlogreader.h:207
XLogRecord * XLogReadRecord(XLogReaderState *state, char **errormsg)
Definition: xlogreader.c:391
void XLogBeginRead(XLogReaderState *state, XLogRecPtr RecPtr)
Definition: xlogreader.c:233

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 447 of file slotfuncs.c.

448 {
450  XLogRecPtr retlsn = startlsn;
451 
452  Assert(moveto != InvalidXLogRecPtr);
453 
454  if (startlsn < moveto)
455  {
459  retlsn = moveto;
460 
461  /*
462  * Dirty the slot so as it is written out at the next checkpoint. Note
463  * that the LSN position advanced may still be lost in the event of a
464  * crash, but this makes the data consistent after a clean shutdown.
465  */
467  }
468 
469  return retlsn;
470 }

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 592 of file slotfuncs.c.

593 {
594  Name slotname = PG_GETARG_NAME(0);
595  XLogRecPtr moveto = PG_GETARG_LSN(1);
596  XLogRecPtr endlsn;
597  XLogRecPtr minlsn;
598  TupleDesc tupdesc;
599  Datum values[2];
600  bool nulls[2];
601  HeapTuple tuple;
602  Datum result;
603 
605 
607 
608  if (XLogRecPtrIsInvalid(moveto))
609  ereport(ERROR,
610  (errmsg("invalid target WAL LSN")));
611 
612  /* Build a tuple descriptor for our result type */
613  if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
614  elog(ERROR, "return type must be a row type");
615 
616  /*
617  * We can't move slot past what's been flushed/replayed so clamp the
618  * target position accordingly.
619  */
620  if (!RecoveryInProgress())
621  moveto = Min(moveto, GetFlushRecPtr(NULL));
622  else
623  moveto = Min(moveto, GetXLogReplayRecPtr(NULL));
624 
625  /* Acquire the slot so we "own" it */
626  ReplicationSlotAcquire(NameStr(*slotname), true);
627 
628  /* A slot whose restart_lsn has never been reserved cannot be advanced */
630  ereport(ERROR,
631  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
632  errmsg("replication slot \"%s\" cannot be advanced",
633  NameStr(*slotname)),
634  errdetail("This slot has never previously reserved WAL, or it has been invalidated.")));
635 
636  /*
637  * Check if the slot is not moving backwards. Physical slots rely simply
638  * on restart_lsn as a minimum point, while logical slots have confirmed
639  * consumption up to confirmed_flush, meaning that in both cases data
640  * older than that is not available anymore.
641  */
644  else
646 
647  if (moveto < minlsn)
648  ereport(ERROR,
649  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
650  errmsg("cannot advance replication slot to %X/%X, minimum is %X/%X",
651  LSN_FORMAT_ARGS(moveto), LSN_FORMAT_ARGS(minlsn))));
652 
653  /* Do the actual slot update, depending on the slot type */
655  endlsn = pg_logical_replication_slot_advance(moveto);
656  else
657  endlsn = pg_physical_replication_slot_advance(moveto);
658 
660  nulls[0] = false;
661 
662  /*
663  * Recompute the minimum LSN and xmin across all slots to adjust with the
664  * advancing potentially done.
665  */
668 
670 
671  /* Return the reached position. */
672  values[1] = LSNGetDatum(endlsn);
673  nulls[1] = false;
674 
675  tuple = heap_form_tuple(tupdesc, values, nulls);
676  result = HeapTupleGetDatum(tuple);
677 
678  PG_RETURN_DATUM(result);
679 }
#define Min(x, y)
Definition: c.h:991
#define OidIsValid(objectId)
Definition: c.h:762
#define PG_GETARG_LSN(n)
Definition: pg_lsn.h:33
void ReplicationSlotAcquire(const char *name, bool nowait)
Definition: slot.c:502
static XLogRecPtr pg_logical_replication_slot_advance(XLogRecPtr moveto)
Definition: slotfuncs.c:484
static XLogRecPtr pg_physical_replication_slot_advance(XLogRecPtr moveto)
Definition: slotfuncs.c:447
bool RecoveryInProgress(void)
Definition: xlog.c:6206
XLogRecPtr GetFlushRecPtr(TimeLineID *insertTLI)
Definition: xlog.c:6371
#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.

◆ pg_sync_replication_slots()

Datum pg_sync_replication_slots ( PG_FUNCTION_ARGS  )

Definition at line 950 of file slotfuncs.c.

951 {
953  char *err;
954  StringInfoData app_name;
955 
957 
958  if (!RecoveryInProgress())
959  ereport(ERROR,
960  errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
961  errmsg("replication slots can only be synchronized to a standby server"));
962 
964 
965  /* Load the libpq-specific functions */
966  load_file("libpqwalreceiver", false);
967 
969 
970  initStringInfo(&app_name);
971  if (cluster_name[0])
972  appendStringInfo(&app_name, "%s_slotsync", cluster_name);
973  else
974  appendStringInfoString(&app_name, "slotsync");
975 
976  /* Connect to the primary server. */
977  wrconn = walrcv_connect(PrimaryConnInfo, false, false, false,
978  app_name.data, &err);
979  pfree(app_name.data);
980 
981  if (!wrconn)
982  ereport(ERROR,
983  errcode(ERRCODE_CONNECTION_FAILURE),
984  errmsg("could not connect to the primary server: %s", err));
985 
987 
989 
990  PG_RETURN_VOID();
991 }
void load_file(const char *filename, bool restricted)
Definition: dfmgr.c:144
void err(int eval, const char *fmt,...)
Definition: err.c:43
char * cluster_name
Definition: guc_tables.c:538
void pfree(void *pointer)
Definition: mcxt.c:1405
void SyncReplicationSlots(WalReceiverConn *wrconn)
Definition: slotsync.c:1509
char * CheckAndGetDbnameFromConninfo(void)
Definition: slotsync.c:892
bool ValidateSlotSyncParams(int elevel)
Definition: slotsync.c:918
void appendStringInfo(StringInfo str, const char *fmt,...)
Definition: stringinfo.c:97
void appendStringInfoString(StringInfo str, const char *s)
Definition: stringinfo.c:182
void initStringInfo(StringInfo str)
Definition: stringinfo.c:59
static WalReceiverConn * wrconn
Definition: walreceiver.c:95
#define walrcv_connect(conninfo, replication, logical, must_use_password, appname, err)
Definition: walreceiver.h:432
#define walrcv_disconnect(conn)
Definition: walreceiver.h:464
char * PrimaryConnInfo
Definition: xlogrecovery.c:97

References appendStringInfo(), appendStringInfoString(), CheckAndGetDbnameFromConninfo(), CheckSlotPermissions(), cluster_name, StringInfoData::data, ereport, err(), errcode(), errmsg(), ERROR, initStringInfo(), load_file(), pfree(), PG_RETURN_VOID, PrimaryConnInfo, RecoveryInProgress(), SyncReplicationSlots(), ValidateSlotSyncParams(), walrcv_connect, walrcv_disconnect, and wrconn.