PostgreSQL Source Code  git master
slotfuncs.c
Go to the documentation of this file.
1 /*-------------------------------------------------------------------------
2  *
3  * slotfuncs.c
4  * Support functions for replication slots
5  *
6  * Copyright (c) 2012-2020, PostgreSQL Global Development Group
7  *
8  * IDENTIFICATION
9  * src/backend/replication/slotfuncs.c
10  *
11  *-------------------------------------------------------------------------
12  */
13 #include "postgres.h"
14 
15 #include "access/htup_details.h"
16 #include "access/xlog_internal.h"
17 #include "funcapi.h"
18 #include "miscadmin.h"
19 #include "replication/decode.h"
20 #include "replication/logical.h"
22 #include "replication/slot.h"
23 #include "utils/builtins.h"
24 #include "utils/inval.h"
25 #include "utils/pg_lsn.h"
26 #include "utils/resowner.h"
27 
28 static void
30 {
32  ereport(ERROR,
33  (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
34  (errmsg("must be superuser or replication role to use replication slots"))));
35 }
36 
37 /*
38  * Helper function for creating a new physical replication slot with
39  * given arguments. Note that this function doesn't release the created
40  * slot.
41  *
42  * If restart_lsn is a valid value, we use it without WAL reservation
43  * routine. So the caller must guarantee that WAL is available.
44  */
45 static void
46 create_physical_replication_slot(char *name, bool immediately_reserve,
47  bool temporary, XLogRecPtr restart_lsn)
48 {
50 
51  /* acquire replication slot, this will check for conflicting names */
52  ReplicationSlotCreate(name, false,
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 }
68 
69 /*
70  * SQL function for creating a new physical (streaming replication)
71  * replication slot.
72  */
73 Datum
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 }
115 
116 
117 /*
118  * Helper function for creating a new logical replication slot with
119  * given arguments. Note that this function doesn't release the created
120  * slot.
121  */
122 static void
124  bool temporary, XLogRecPtr restart_lsn)
125 {
126  LogicalDecodingContext *ctx = NULL;
127 
129 
130  /*
131  * Acquire a logical decoding slot, this will check for conflicting names.
132  * Initially create persistent slot as ephemeral - that allows us to
133  * nicely handle errors during initialization because it'll get dropped if
134  * this transaction fails. We'll make it persistent at the end. Temporary
135  * slots can be created as temporary from beginning as they get dropped on
136  * error as well.
137  */
138  ReplicationSlotCreate(name, true,
139  temporary ? RS_TEMPORARY : RS_EPHEMERAL);
140 
141  /*
142  * Create logical decoding context, to build the initial snapshot.
143  */
144  ctx = CreateInitDecodingContext(plugin, NIL,
145  false, /* do not build snapshot */
146  restart_lsn,
147  logical_read_local_xlog_page, NULL, NULL,
148  NULL);
149 
150  /* build initial snapshot, might take a while */
152 
153  /* don't need the decoding context anymore */
154  FreeDecodingContext(ctx);
155 }
156 
157 /*
158  * SQL function for creating a new logical replication slot.
159  */
160 Datum
162 {
163  Name name = PG_GETARG_NAME(0);
165  bool temporary = PG_GETARG_BOOL(2);
166  Datum result;
167  TupleDesc tupdesc;
168  HeapTuple tuple;
169  Datum values[2];
170  bool nulls[2];
171 
172  if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
173  elog(ERROR, "return type must be a row type");
174 
176 
178 
180  NameStr(*plugin),
181  temporary,
183 
184  values[0] = NameGetDatum(&MyReplicationSlot->data.name);
186 
187  memset(nulls, 0, sizeof(nulls));
188 
189  tuple = heap_form_tuple(tupdesc, values, nulls);
190  result = HeapTupleGetDatum(tuple);
191 
192  /* ok, slot is now fully created, mark it as persistent if needed */
193  if (!temporary)
196 
197  PG_RETURN_DATUM(result);
198 }
199 
200 
201 /*
202  * SQL function for dropping a replication slot.
203  */
204 Datum
206 {
207  Name name = PG_GETARG_NAME(0);
208 
210 
212 
213  ReplicationSlotDrop(NameStr(*name), true);
214 
215  PG_RETURN_VOID();
216 }
217 
218 /*
219  * pg_get_replication_slots - SQL SRF showing active replication slots.
220  */
221 Datum
223 {
224 #define PG_GET_REPLICATION_SLOTS_COLS 11
225  ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
226  TupleDesc tupdesc;
227  Tuplestorestate *tupstore;
228  MemoryContext per_query_ctx;
229  MemoryContext oldcontext;
230  int slotno;
231 
232  /* check to see if caller supports us returning a tuplestore */
233  if (rsinfo == NULL || !IsA(rsinfo, ReturnSetInfo))
234  ereport(ERROR,
235  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
236  errmsg("set-valued function called in context that cannot accept a set")));
237  if (!(rsinfo->allowedModes & SFRM_Materialize))
238  ereport(ERROR,
239  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
240  errmsg("materialize mode required, but it is not allowed in this context")));
241 
242  /* Build a tuple descriptor for our result type */
243  if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
244  elog(ERROR, "return type must be a row type");
245 
246  /*
247  * We don't require any special permission to see this function's data
248  * because nothing should be sensitive. The most critical being the slot
249  * name, which shouldn't contain anything particularly sensitive.
250  */
251 
252  per_query_ctx = rsinfo->econtext->ecxt_per_query_memory;
253  oldcontext = MemoryContextSwitchTo(per_query_ctx);
254 
255  tupstore = tuplestore_begin_heap(true, false, work_mem);
256  rsinfo->returnMode = SFRM_Materialize;
257  rsinfo->setResult = tupstore;
258  rsinfo->setDesc = tupdesc;
259 
260  MemoryContextSwitchTo(oldcontext);
261 
262  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
263  for (slotno = 0; slotno < max_replication_slots; slotno++)
264  {
267  bool nulls[PG_GET_REPLICATION_SLOTS_COLS];
268 
269  ReplicationSlotPersistency persistency;
270  TransactionId xmin;
271  TransactionId catalog_xmin;
272  XLogRecPtr restart_lsn;
273  XLogRecPtr confirmed_flush_lsn;
274  pid_t active_pid;
275  Oid database;
276  NameData slot_name;
278  int i;
279 
280  if (!slot->in_use)
281  continue;
282 
283  SpinLockAcquire(&slot->mutex);
284 
285  xmin = slot->data.xmin;
286  catalog_xmin = slot->data.catalog_xmin;
287  database = slot->data.database;
288  restart_lsn = slot->data.restart_lsn;
289  confirmed_flush_lsn = slot->data.confirmed_flush;
290  namecpy(&slot_name, &slot->data.name);
291  namecpy(&plugin, &slot->data.plugin);
292  active_pid = slot->active_pid;
293  persistency = slot->data.persistency;
294 
295  SpinLockRelease(&slot->mutex);
296 
297  memset(nulls, 0, sizeof(nulls));
298 
299  i = 0;
300  values[i++] = NameGetDatum(&slot_name);
301 
302  if (database == InvalidOid)
303  nulls[i++] = true;
304  else
305  values[i++] = NameGetDatum(&plugin);
306 
307  if (database == InvalidOid)
308  values[i++] = CStringGetTextDatum("physical");
309  else
310  values[i++] = CStringGetTextDatum("logical");
311 
312  if (database == InvalidOid)
313  nulls[i++] = true;
314  else
315  values[i++] = database;
316 
317  values[i++] = BoolGetDatum(persistency == RS_TEMPORARY);
318  values[i++] = BoolGetDatum(active_pid != 0);
319 
320  if (active_pid != 0)
321  values[i++] = Int32GetDatum(active_pid);
322  else
323  nulls[i++] = true;
324 
325  if (xmin != InvalidTransactionId)
326  values[i++] = TransactionIdGetDatum(xmin);
327  else
328  nulls[i++] = true;
329 
330  if (catalog_xmin != InvalidTransactionId)
331  values[i++] = TransactionIdGetDatum(catalog_xmin);
332  else
333  nulls[i++] = true;
334 
335  if (restart_lsn != InvalidXLogRecPtr)
336  values[i++] = LSNGetDatum(restart_lsn);
337  else
338  nulls[i++] = true;
339 
340  if (confirmed_flush_lsn != InvalidXLogRecPtr)
341  values[i++] = LSNGetDatum(confirmed_flush_lsn);
342  else
343  nulls[i++] = true;
344 
345  tuplestore_putvalues(tupstore, tupdesc, values, nulls);
346  }
347  LWLockRelease(ReplicationSlotControlLock);
348 
349  tuplestore_donestoring(tupstore);
350 
351  return (Datum) 0;
352 }
353 
354 /*
355  * Helper function for advancing our physical replication slot forward.
356  *
357  * The LSN position to move to is compared simply to the slot's restart_lsn,
358  * knowing that any position older than that would be removed by successive
359  * checkpoints.
360  */
361 static XLogRecPtr
363 {
365  XLogRecPtr retlsn = startlsn;
366 
367  if (startlsn < moveto)
368  {
372  retlsn = moveto;
373  }
374 
375  return retlsn;
376 }
377 
378 /*
379  * Helper function for advancing our logical replication slot forward.
380  *
381  * The slot's restart_lsn is used as start point for reading records, while
382  * confirmed_flush is used as base point for the decoding context.
383  *
384  * We cannot just do LogicalConfirmReceivedLocation to update confirmed_flush,
385  * because we need to digest WAL to advance restart_lsn allowing to recycle
386  * WAL and removal of old catalog tuples. As decoding is done in fast_forward
387  * mode, no changes are generated anyway.
388  */
389 static XLogRecPtr
391 {
393  ResourceOwner old_resowner = CurrentResourceOwner;
394  XLogRecPtr startlsn;
395  XLogRecPtr retlsn;
396 
397  PG_TRY();
398  {
399  /*
400  * Create our decoding context in fast_forward mode, passing start_lsn
401  * as InvalidXLogRecPtr, so that we start processing from my slot's
402  * confirmed_flush.
403  */
405  NIL,
406  true, /* fast_forward */
408  NULL, NULL, NULL);
409 
410  /*
411  * Start reading at the slot's restart_lsn, which we know to point to
412  * a valid record.
413  */
414  startlsn = MyReplicationSlot->data.restart_lsn;
415 
416  /* Initialize our return value in case we don't do anything */
418 
419  /* invalidate non-timetravel entries */
421 
422  /* Decode at least one record, until we run out of records */
423  while ((!XLogRecPtrIsInvalid(startlsn) &&
424  startlsn < moveto) ||
426  ctx->reader->EndRecPtr < moveto))
427  {
428  char *errm = NULL;
429  XLogRecord *record;
430 
431  /*
432  * Read records. No changes are generated in fast_forward mode,
433  * but snapbuilder/slot statuses are updated properly.
434  */
435  record = XLogReadRecord(ctx->reader, startlsn, &errm);
436  if (errm)
437  elog(ERROR, "%s", errm);
438 
439  /* Read sequentially from now on */
440  startlsn = InvalidXLogRecPtr;
441 
442  /*
443  * Process the record. Storage-level changes are ignored in
444  * fast_forward mode, but other modules (such as snapbuilder)
445  * might still have critical updates to do.
446  */
447  if (record)
449 
450  /* Stop once the requested target has been reached */
451  if (moveto <= ctx->reader->EndRecPtr)
452  break;
453 
455  }
456 
457  /*
458  * Logical decoding could have clobbered CurrentResourceOwner during
459  * transaction management, so restore the executor's value. (This is
460  * a kluge, but it's not worth cleaning up right now.)
461  */
462  CurrentResourceOwner = old_resowner;
463 
464  if (ctx->reader->EndRecPtr != InvalidXLogRecPtr)
465  {
467 
468  /*
469  * If only the confirmed_flush LSN has changed the slot won't get
470  * marked as dirty by the above. Callers on the walsender
471  * interface are expected to keep track of their own progress and
472  * don't need it written out. But SQL-interface users cannot
473  * specify their own start positions and it's harder for them to
474  * keep track of their progress, so we should make more of an
475  * effort to save it for them.
476  *
477  * Dirty the slot so it's written out at the next checkpoint.
478  * We'll still lose its position on crash, as documented, but it's
479  * better than always losing the position even on clean restart.
480  */
482  }
483 
485 
486  /* free context, call shutdown callback */
487  FreeDecodingContext(ctx);
488 
490  }
491  PG_CATCH();
492  {
493  /* clear all timetravel entries */
495 
496  PG_RE_THROW();
497  }
498  PG_END_TRY();
499 
500  return retlsn;
501 }
502 
503 /*
504  * SQL function for moving the position in a replication slot.
505  */
506 Datum
508 {
509  Name slotname = PG_GETARG_NAME(0);
510  XLogRecPtr moveto = PG_GETARG_LSN(1);
511  XLogRecPtr endlsn;
512  XLogRecPtr minlsn;
513  TupleDesc tupdesc;
514  Datum values[2];
515  bool nulls[2];
516  HeapTuple tuple;
517  Datum result;
518 
520 
522 
523  if (XLogRecPtrIsInvalid(moveto))
524  ereport(ERROR,
525  (errmsg("invalid target WAL LSN")));
526 
527  /* Build a tuple descriptor for our result type */
528  if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
529  elog(ERROR, "return type must be a row type");
530 
531  /*
532  * We can't move slot past what's been flushed/replayed so clamp the
533  * target position accordingly.
534  */
535  if (!RecoveryInProgress())
536  moveto = Min(moveto, GetFlushRecPtr());
537  else
538  moveto = Min(moveto, GetXLogReplayRecPtr(&ThisTimeLineID));
539 
540  /* Acquire the slot so we "own" it */
541  ReplicationSlotAcquire(NameStr(*slotname), true);
542 
543  /* A slot whose restart_lsn has never been reserved cannot be advanced */
545  ereport(ERROR,
546  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
547  errmsg("cannot advance replication slot that has not previously reserved WAL")));
548 
549  /*
550  * Check if the slot is not moving backwards. Physical slots rely simply
551  * on restart_lsn as a minimum point, while logical slots have confirmed
552  * consumption up to confirmed_flush, meaning that in both cases data
553  * older than that is not available anymore.
554  */
557  else
559 
560  if (moveto < minlsn)
561  ereport(ERROR,
562  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
563  errmsg("cannot advance replication slot to %X/%X, minimum is %X/%X",
564  (uint32) (moveto >> 32), (uint32) moveto,
565  (uint32) (minlsn >> 32), (uint32) minlsn)));
566 
567  /* Do the actual slot update, depending on the slot type */
569  endlsn = pg_logical_replication_slot_advance(moveto);
570  else
571  endlsn = pg_physical_replication_slot_advance(moveto);
572 
573  values[0] = NameGetDatum(&MyReplicationSlot->data.name);
574  nulls[0] = false;
575 
576  /* Update the on disk state when lsn was updated. */
577  if (XLogRecPtrIsInvalid(endlsn))
578  {
583  }
584 
586 
587  /* Return the reached position. */
588  values[1] = LSNGetDatum(endlsn);
589  nulls[1] = false;
590 
591  tuple = heap_form_tuple(tupdesc, values, nulls);
592  result = HeapTupleGetDatum(tuple);
593 
594  PG_RETURN_DATUM(result);
595 }
596 
597 /*
598  * Helper function of copying a replication slot.
599  */
600 static Datum
601 copy_replication_slot(FunctionCallInfo fcinfo, bool logical_slot)
602 {
603  Name src_name = PG_GETARG_NAME(0);
604  Name dst_name = PG_GETARG_NAME(1);
605  ReplicationSlot *src = NULL;
606  XLogRecPtr src_restart_lsn;
607  bool src_islogical;
608  bool temporary;
609  char *plugin;
610  Datum values[2];
611  bool nulls[2];
612  Datum result;
613  TupleDesc tupdesc;
614  HeapTuple tuple;
615 
616  if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
617  elog(ERROR, "return type must be a row type");
618 
620 
621  if (logical_slot)
623  else
625 
626  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
627 
628  /*
629  * We need to prevent the source slot's reserved WAL from being removed,
630  * but we don't want to lock that slot for very long, and it can advance
631  * in the meantime. So obtain the source slot's data, and create a new
632  * slot using its restart_lsn. Afterwards we lock the source slot again
633  * and verify that the data we copied (name, type) has not changed
634  * incompatibly. No inconvenient WAL removal can occur once the new slot
635  * is created -- but since WAL removal could have occurred before we
636  * managed to create the new slot, we advance the new slot's restart_lsn
637  * to the source slot's updated restart_lsn the second time we lock it.
638  */
639  for (int i = 0; i < max_replication_slots; i++)
640  {
642 
643  if (s->in_use && strcmp(NameStr(s->data.name), NameStr(*src_name)) == 0)
644  {
645  SpinLockAcquire(&s->mutex);
646  src_islogical = SlotIsLogical(s);
647  src_restart_lsn = s->data.restart_lsn;
648  temporary = s->data.persistency == RS_TEMPORARY;
649  plugin = logical_slot ? pstrdup(NameStr(s->data.plugin)) : NULL;
650  SpinLockRelease(&s->mutex);
651 
652  src = s;
653  break;
654  }
655  }
656 
657  LWLockRelease(ReplicationSlotControlLock);
658 
659  if (src == NULL)
660  ereport(ERROR,
661  (errcode(ERRCODE_UNDEFINED_OBJECT),
662  errmsg("replication slot \"%s\" does not exist", NameStr(*src_name))));
663 
664  /* Check type of replication slot */
665  if (src_islogical != logical_slot)
666  ereport(ERROR,
667  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
668  src_islogical ?
669  errmsg("cannot copy physical replication slot \"%s\" as a logical replication slot",
670  NameStr(*src_name)) :
671  errmsg("cannot copy logical replication slot \"%s\" as a physical replication slot",
672  NameStr(*src_name))));
673 
674  /* Copying non-reserved slot doesn't make sense */
675  if (XLogRecPtrIsInvalid(src_restart_lsn))
676  {
677  Assert(!logical_slot);
678  ereport(ERROR,
679  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
680  (errmsg("cannot copy a replication slot that doesn't reserve WAL"))));
681  }
682 
683  /* Overwrite params from optional arguments */
684  if (PG_NARGS() >= 3)
685  temporary = PG_GETARG_BOOL(2);
686  if (PG_NARGS() >= 4)
687  {
688  Assert(logical_slot);
689  plugin = NameStr(*(PG_GETARG_NAME(3)));
690  }
691 
692  /* Create new slot and acquire it */
693  if (logical_slot)
695  plugin,
696  temporary,
697  src_restart_lsn);
698  else
700  true,
701  temporary,
702  src_restart_lsn);
703 
704  /*
705  * Update the destination slot to current values of the source slot;
706  * recheck that the source slot is still the one we saw previously.
707  */
708  {
709  TransactionId copy_effective_xmin;
710  TransactionId copy_effective_catalog_xmin;
711  TransactionId copy_xmin;
712  TransactionId copy_catalog_xmin;
713  XLogRecPtr copy_restart_lsn;
714  bool copy_islogical;
715  char *copy_name;
716 
717  /* Copy data of source slot again */
718  SpinLockAcquire(&src->mutex);
719  copy_effective_xmin = src->effective_xmin;
720  copy_effective_catalog_xmin = src->effective_catalog_xmin;
721 
722  copy_xmin = src->data.xmin;
723  copy_catalog_xmin = src->data.catalog_xmin;
724  copy_restart_lsn = src->data.restart_lsn;
725 
726  /* for existence check */
727  copy_name = pstrdup(NameStr(src->data.name));
728  copy_islogical = SlotIsLogical(src);
729  SpinLockRelease(&src->mutex);
730 
731  /*
732  * Check if the source slot still exists and is valid. We regard it as
733  * invalid if the type of replication slot or name has been changed,
734  * or the restart_lsn either is invalid or has gone backward. (The
735  * restart_lsn could go backwards if the source slot is dropped and
736  * copied from an older slot during installation.)
737  *
738  * Since erroring out will release and drop the destination slot we
739  * don't need to release it here.
740  */
741  if (copy_restart_lsn < src_restart_lsn ||
742  src_islogical != copy_islogical ||
743  strcmp(copy_name, NameStr(*src_name)) != 0)
744  ereport(ERROR,
745  (errmsg("could not copy replication slot \"%s\"",
746  NameStr(*src_name)),
747  errdetail("The source replication slot was modified incompatibly during the copy operation.")));
748 
749  /* Install copied values again */
751  MyReplicationSlot->effective_xmin = copy_effective_xmin;
752  MyReplicationSlot->effective_catalog_xmin = copy_effective_catalog_xmin;
753 
754  MyReplicationSlot->data.xmin = copy_xmin;
755  MyReplicationSlot->data.catalog_xmin = copy_catalog_xmin;
756  MyReplicationSlot->data.restart_lsn = copy_restart_lsn;
758 
763 
764 #ifdef USE_ASSERT_CHECKING
765  /* Check that the restart_lsn is available */
766  {
767  XLogSegNo segno;
768 
769  XLByteToSeg(copy_restart_lsn, segno, wal_segment_size);
770  Assert(XLogGetLastRemovedSegno() < segno);
771  }
772 #endif
773  }
774 
775  /* target slot fully created, mark as persistent if needed */
776  if (logical_slot && !temporary)
778 
779  /* All done. Set up the return values */
780  values[0] = NameGetDatum(dst_name);
781  nulls[0] = false;
783  {
785  nulls[1] = false;
786  }
787  else
788  nulls[1] = true;
789 
790  tuple = heap_form_tuple(tupdesc, values, nulls);
791  result = HeapTupleGetDatum(tuple);
792 
794 
795  PG_RETURN_DATUM(result);
796 }
797 
798 /* The wrappers below are all to appease opr_sanity */
799 Datum
801 {
802  return copy_replication_slot(fcinfo, true);
803 }
804 
805 Datum
807 {
808  return copy_replication_slot(fcinfo, true);
809 }
810 
811 Datum
813 {
814  return copy_replication_slot(fcinfo, true);
815 }
816 
817 Datum
819 {
820  return copy_replication_slot(fcinfo, false);
821 }
822 
823 Datum
825 {
826  return copy_replication_slot(fcinfo, false);
827 }
void tuplestore_putvalues(Tuplestorestate *state, TupleDesc tdesc, Datum *values, bool *isnull)
Definition: tuplestore.c:750
static XLogRecPtr pg_logical_replication_slot_advance(XLogRecPtr moveto)
Definition: slotfuncs.c:390
#define NIL
Definition: pg_list.h:65
static const char * plugin
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
void CheckSlotRequirements(void)
Definition: slot.c:972
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
#define IsA(nodeptr, _type_)
Definition: nodes.h:576
#define NameGetDatum(X)
Definition: postgres.h:595
TypeFuncClass get_call_result_type(FunctionCallInfo fcinfo, Oid *resultTypeId, TupleDesc *resultTupleDesc)
Definition: funcapi.c:196
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:112
#define PG_GET_REPLICATION_SLOTS_COLS
uint32 TransactionId
Definition: c.h:514
Datum pg_get_replication_slots(PG_FUNCTION_ARGS)
Definition: slotfuncs.c:222
Oid GetUserId(void)
Definition: miscinit.c:380
int namecpy(Name n1, const NameData *n2)
Definition: name.c:233
ResourceOwner CurrentResourceOwner
Definition: resowner.c:142
Datum pg_create_logical_replication_slot(PG_FUNCTION_ARGS)
Definition: slotfuncs.c:161
int logical_read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen, XLogRecPtr targetRecPtr, char *cur_page)
Definition: logicalfuncs.c:110
char * pstrdup(const char *in)
Definition: mcxt.c:1186
void ReplicationSlotAcquire(const char *name, bool nowait)
Definition: slot.c:330
#define Min(x, y)
Definition: c.h:911
#define tuplestore_donestoring(state)
Definition: tuplestore.h:60
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
Definition: palloc.h:109
void ReplicationSlotCreate(const char *name, bool db_specific, ReplicationSlotPersistency persistency)
Definition: slot.c:221
ReplicationSlotPersistency persistency
Definition: slot.h:53
int errcode(int sqlerrcode)
Definition: elog.c:608
#define LSNGetDatum(X)
Definition: pg_lsn.h:22
bool superuser(void)
Definition: superuser.c:46
void ReplicationSlotSave(void)
Definition: slot.c:645
XLogRecPtr GetFlushRecPtr(void)
Definition: xlog.c:8262
HeapTuple heap_form_tuple(TupleDesc tupleDescriptor, Datum *values, bool *isnull)
Definition: heaptuple.c:1020
#define PG_GETARG_BOOL(n)
Definition: fmgr.h:269
Datum pg_copy_physical_replication_slot_a(PG_FUNCTION_ARGS)
Definition: slotfuncs.c:818
ReplicationSlotPersistentData data
Definition: slot.h:132
Datum pg_copy_logical_replication_slot_c(PG_FUNCTION_ARGS)
Definition: slotfuncs.c:812
unsigned int Oid
Definition: postgres_ext.h:31
bool RecoveryInProgress(void)
Definition: xlog.c:7930
#define OidIsValid(objectId)
Definition: c.h:645
void InvalidateSystemCaches(void)
Definition: inval.c:643
XLogRecord * XLogReadRecord(XLogReaderState *state, XLogRecPtr RecPtr, char **errormsg)
Definition: xlogreader.c:238
XLogRecPtr confirmed_flush
Definition: slot.h:80
XLogRecPtr EndRecPtr
Definition: xlogreader.h:132
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1726
Datum pg_create_physical_replication_slot(PG_FUNCTION_ARGS)
Definition: slotfuncs.c:74
#define SpinLockAcquire(lock)
Definition: spin.h:62
XLogSegNo XLogGetLastRemovedSegno(void)
Definition: xlog.c:3890
void ReplicationSlotReserveWal(void)
Definition: slot.c:997
void DecodingContextFindStartpoint(LogicalDecodingContext *ctx)
Definition: logical.c:462
void ReplicationSlotsComputeRequiredLSN(void)
Definition: slot.c:748
#define ERROR
Definition: elog.h:43
void LogicalDecodingProcessRecord(LogicalDecodingContext *ctx, XLogReaderState *record)
Definition: decode.c:94
Datum pg_copy_logical_replication_slot_a(PG_FUNCTION_ARGS)
Definition: slotfuncs.c:800
static Datum copy_replication_slot(FunctionCallInfo fcinfo, bool logical_slot)
Definition: slotfuncs.c:601
XLogRecPtr GetXLogReplayRecPtr(TimeLineID *replayTLI)
Definition: xlog.c:11193
void ReplicationSlotPersist(void)
Definition: slot.c:680
TransactionId effective_xmin
Definition: slot.h:128
Definition: c.h:610
Datum pg_copy_physical_replication_slot_b(PG_FUNCTION_ARGS)
Definition: slotfuncs.c:824
uint64 XLogSegNo
Definition: xlogdefs.h:41
int errdetail(const char *fmt,...)
Definition: elog.c:955
Datum pg_replication_slot_advance(PG_FUNCTION_ARGS)
Definition: slotfuncs.c:507
TransactionId catalog_xmin
Definition: slot.h:69
#define InvalidTransactionId
Definition: transam.h:31
unsigned int uint32
Definition: c.h:359
void ReplicationSlotRelease(void)
Definition: slot.c:424
TransactionId xmin
Definition: slot.h:61
#define SlotIsLogical(slot)
Definition: slot.h:154
static void check_permissions(void)
Definition: slotfuncs.c:29
#define ereport(elevel, rest)
Definition: elog.h:141
#define XLogRecPtrIsInvalid(r)
Definition: xlogdefs.h:29
bool in_use
Definition: slot.h:108
#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
static void create_logical_replication_slot(char *name, char *plugin, bool temporary, XLogRecPtr restart_lsn)
Definition: slotfuncs.c:123
LogicalDecodingContext * CreateInitDecodingContext(char *plugin, List *output_plugin_options, bool need_full_snapshot, XLogRecPtr restart_lsn, XLogPageReadCB read_page, LogicalOutputPluginWriterPrepareWrite prepare_write, LogicalOutputPluginWriterWrite do_write, LogicalOutputPluginWriterUpdateProgress update_progress)
Definition: logical.c:227
#define PG_GETARG_LSN(n)
Definition: pg_lsn.h:24
Datum pg_copy_logical_replication_slot_b(PG_FUNCTION_ARGS)
Definition: slotfuncs.c:806
uintptr_t Datum
Definition: postgres.h:367
Datum pg_drop_replication_slot(PG_FUNCTION_ARGS)
Definition: slotfuncs.c:205
#define PG_RETURN_DATUM(x)
Definition: fmgr.h:343
TransactionId effective_catalog_xmin
Definition: slot.h:129
LogicalDecodingContext * CreateDecodingContext(XLogRecPtr start_lsn, List *output_plugin_options, bool fast_forward, XLogPageReadCB read_page, LogicalOutputPluginWriterPrepareWrite prepare_write, LogicalOutputPluginWriterWrite do_write, LogicalOutputPluginWriterUpdateProgress update_progress)
Definition: logical.c:370
int work_mem
Definition: globals.c:121
#define BoolGetDatum(X)
Definition: postgres.h:402
#define InvalidOid
Definition: postgres_ext.h:36
TimeLineID ThisTimeLineID
Definition: xlog.c:187
int allowedModes
Definition: execnodes.h:302
#define PG_RETURN_VOID()
Definition: fmgr.h:339
SetFunctionReturnMode returnMode
Definition: execnodes.h:304
#define PG_CATCH()
Definition: elog.h:332
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:739
void FreeDecodingContext(LogicalDecodingContext *ctx)
Definition: logical.c:508
XLogRecPtr restart_lsn
Definition: slot.h:72
bool has_rolreplication(Oid roleid)
Definition: miscinit.c:561
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1122
#define PG_NARGS()
Definition: fmgr.h:198
#define PG_RE_THROW()
Definition: elog.h:363
#define HeapTupleGetDatum(tuple)
Definition: funcapi.h:220
void LogicalConfirmReceivedLocation(XLogRecPtr lsn)
Definition: logical.c:1001
MemoryContext ecxt_per_query_memory
Definition: execnodes.h:230
const char * name
Definition: encode.c:521
Tuplestorestate * setResult
Definition: execnodes.h:307
static Datum values[MAXATTR]
Definition: bootstrap.c:167
ExprContext * econtext
Definition: execnodes.h:300
#define Int32GetDatum(X)
Definition: postgres.h:479
void ReplicationSlotDrop(const char *name, bool nowait)
Definition: slot.c:517
TupleDesc setDesc
Definition: execnodes.h:308
ReplicationSlotPersistency
Definition: slot.h:32
int errmsg(const char *fmt,...)
Definition: elog.c:822
XLogReaderState * reader
Definition: logical.h:41
pid_t active_pid
Definition: slot.h:111
#define elog(elevel,...)
Definition: elog.h:228
int i
#define NameStr(name)
Definition: c.h:616
#define CStringGetTextDatum(s)
Definition: builtins.h:83
#define PG_FUNCTION_ARGS
Definition: fmgr.h:188
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:99
ReplicationSlot replication_slots[1]
Definition: slot.h:165
static XLogRecPtr pg_physical_replication_slot_advance(XLogRecPtr moveto)
Definition: slotfuncs.c:362
slock_t mutex
Definition: slot.h:105
#define PG_TRY()
Definition: elog.h:322
void CheckLogicalDecodingRequirements(void)
Definition: logical.c:75
void ReplicationSlotsComputeRequiredXmin(bool already_locked)
Definition: slot.c:702
#define PG_END_TRY()
Definition: elog.h:347
void ReplicationSlotMarkDirty(void)
Definition: slot.c:663
#define PG_GETARG_NAME(n)
Definition: fmgr.h:273
#define XLByteToSeg(xlrp, logSegNo, wal_segsz_bytes)