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 "access/xlogutils.h"
18 #include "funcapi.h"
19 #include "miscadmin.h"
20 #include "replication/decode.h"
21 #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  * When find_startpoint is false, the slot's confirmed_flush is not set; it's
123  * caller's responsibility to ensure it's set to something sensible.
124  */
125 static void
127  bool temporary, XLogRecPtr restart_lsn,
128  bool find_startpoint)
129 {
130  LogicalDecodingContext *ctx = NULL;
131 
133 
134  /*
135  * Acquire a logical decoding slot, this will check for conflicting names.
136  * Initially create persistent slot as ephemeral - that allows us to
137  * nicely handle errors during initialization because it'll get dropped if
138  * this transaction fails. We'll make it persistent at the end. Temporary
139  * slots can be created as temporary from beginning as they get dropped on
140  * error as well.
141  */
142  ReplicationSlotCreate(name, true,
143  temporary ? RS_TEMPORARY : RS_EPHEMERAL);
144 
145  /*
146  * Create logical decoding context to find start point or, if we don't
147  * need it, to 1) bump slot's restart_lsn and xmin 2) check plugin sanity.
148  *
149  * Note: when !find_startpoint this is still important, because it's at
150  * this point that the output plugin is validated.
151  */
152  ctx = CreateInitDecodingContext(plugin, NIL,
153  false, /* just catalogs is OK */
154  restart_lsn,
155  read_local_xlog_page, NULL, NULL,
156  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 }
168 
169 /*
170  * SQL function for creating a new logical replication slot.
171  */
172 Datum
174 {
175  Name name = PG_GETARG_NAME(0);
177  bool temporary = PG_GETARG_BOOL(2);
178  Datum result;
179  TupleDesc tupdesc;
180  HeapTuple tuple;
181  Datum values[2];
182  bool nulls[2];
183 
184  if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
185  elog(ERROR, "return type must be a row type");
186 
188 
190 
192  NameStr(*plugin),
193  temporary,
195  true);
196 
197  values[0] = NameGetDatum(&MyReplicationSlot->data.name);
199 
200  memset(nulls, 0, sizeof(nulls));
201 
202  tuple = heap_form_tuple(tupdesc, values, nulls);
203  result = HeapTupleGetDatum(tuple);
204 
205  /* ok, slot is now fully created, mark it as persistent if needed */
206  if (!temporary)
209 
210  PG_RETURN_DATUM(result);
211 }
212 
213 
214 /*
215  * SQL function for dropping a replication slot.
216  */
217 Datum
219 {
220  Name name = PG_GETARG_NAME(0);
221 
223 
225 
226  ReplicationSlotDrop(NameStr(*name), true);
227 
228  PG_RETURN_VOID();
229 }
230 
231 /*
232  * pg_get_replication_slots - SQL SRF showing active replication slots.
233  */
234 Datum
236 {
237 #define PG_GET_REPLICATION_SLOTS_COLS 11
238  ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
239  TupleDesc tupdesc;
240  Tuplestorestate *tupstore;
241  MemoryContext per_query_ctx;
242  MemoryContext oldcontext;
243  int slotno;
244 
245  /* check to see if caller supports us returning a tuplestore */
246  if (rsinfo == NULL || !IsA(rsinfo, ReturnSetInfo))
247  ereport(ERROR,
248  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
249  errmsg("set-valued function called in context that cannot accept a set")));
250  if (!(rsinfo->allowedModes & SFRM_Materialize))
251  ereport(ERROR,
252  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
253  errmsg("materialize mode required, but it is not allowed in this context")));
254 
255  /* Build a tuple descriptor for our result type */
256  if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
257  elog(ERROR, "return type must be a row type");
258 
259  /*
260  * We don't require any special permission to see this function's data
261  * because nothing should be sensitive. The most critical being the slot
262  * name, which shouldn't contain anything particularly sensitive.
263  */
264 
265  per_query_ctx = rsinfo->econtext->ecxt_per_query_memory;
266  oldcontext = MemoryContextSwitchTo(per_query_ctx);
267 
268  tupstore = tuplestore_begin_heap(true, false, work_mem);
269  rsinfo->returnMode = SFRM_Materialize;
270  rsinfo->setResult = tupstore;
271  rsinfo->setDesc = tupdesc;
272 
273  MemoryContextSwitchTo(oldcontext);
274 
275  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
276  for (slotno = 0; slotno < max_replication_slots; slotno++)
277  {
280  bool nulls[PG_GET_REPLICATION_SLOTS_COLS];
281 
282  ReplicationSlotPersistency persistency;
283  TransactionId xmin;
284  TransactionId catalog_xmin;
285  XLogRecPtr restart_lsn;
286  XLogRecPtr confirmed_flush_lsn;
287  pid_t active_pid;
288  Oid database;
289  NameData slot_name;
291  int i;
292 
293  if (!slot->in_use)
294  continue;
295 
296  SpinLockAcquire(&slot->mutex);
297 
298  xmin = slot->data.xmin;
299  catalog_xmin = slot->data.catalog_xmin;
300  database = slot->data.database;
301  restart_lsn = slot->data.restart_lsn;
302  confirmed_flush_lsn = slot->data.confirmed_flush;
303  namecpy(&slot_name, &slot->data.name);
304  namecpy(&plugin, &slot->data.plugin);
305  active_pid = slot->active_pid;
306  persistency = slot->data.persistency;
307 
308  SpinLockRelease(&slot->mutex);
309 
310  memset(nulls, 0, sizeof(nulls));
311 
312  i = 0;
313  values[i++] = NameGetDatum(&slot_name);
314 
315  if (database == InvalidOid)
316  nulls[i++] = true;
317  else
318  values[i++] = NameGetDatum(&plugin);
319 
320  if (database == InvalidOid)
321  values[i++] = CStringGetTextDatum("physical");
322  else
323  values[i++] = CStringGetTextDatum("logical");
324 
325  if (database == InvalidOid)
326  nulls[i++] = true;
327  else
328  values[i++] = database;
329 
330  values[i++] = BoolGetDatum(persistency == RS_TEMPORARY);
331  values[i++] = BoolGetDatum(active_pid != 0);
332 
333  if (active_pid != 0)
334  values[i++] = Int32GetDatum(active_pid);
335  else
336  nulls[i++] = true;
337 
338  if (xmin != InvalidTransactionId)
339  values[i++] = TransactionIdGetDatum(xmin);
340  else
341  nulls[i++] = true;
342 
343  if (catalog_xmin != InvalidTransactionId)
344  values[i++] = TransactionIdGetDatum(catalog_xmin);
345  else
346  nulls[i++] = true;
347 
348  if (restart_lsn != InvalidXLogRecPtr)
349  values[i++] = LSNGetDatum(restart_lsn);
350  else
351  nulls[i++] = true;
352 
353  if (confirmed_flush_lsn != InvalidXLogRecPtr)
354  values[i++] = LSNGetDatum(confirmed_flush_lsn);
355  else
356  nulls[i++] = true;
357 
358  tuplestore_putvalues(tupstore, tupdesc, values, nulls);
359  }
360  LWLockRelease(ReplicationSlotControlLock);
361 
362  tuplestore_donestoring(tupstore);
363 
364  return (Datum) 0;
365 }
366 
367 /*
368  * Helper function for advancing our physical replication slot forward.
369  *
370  * The LSN position to move to is compared simply to the slot's restart_lsn,
371  * knowing that any position older than that would be removed by successive
372  * checkpoints.
373  */
374 static XLogRecPtr
376 {
378  XLogRecPtr retlsn = startlsn;
379 
380  if (startlsn < moveto)
381  {
385  retlsn = moveto;
386 
387  /*
388  * Dirty the slot so as it is written out at the next checkpoint.
389  * Note that the LSN position advanced may still be lost in the
390  * event of a crash, but this makes the data consistent after a
391  * clean shutdown.
392  */
394  }
395 
396  return retlsn;
397 }
398 
399 /*
400  * Helper function for advancing our logical replication slot forward.
401  *
402  * The slot's restart_lsn is used as start point for reading records, while
403  * confirmed_flush is used as base point for the decoding context.
404  *
405  * We cannot just do LogicalConfirmReceivedLocation to update confirmed_flush,
406  * because we need to digest WAL to advance restart_lsn allowing to recycle
407  * WAL and removal of old catalog tuples. As decoding is done in fast_forward
408  * mode, no changes are generated anyway.
409  */
410 static XLogRecPtr
412 {
414  ResourceOwner old_resowner = CurrentResourceOwner;
415  XLogRecPtr retlsn;
416 
417  PG_TRY();
418  {
419  /*
420  * Create our decoding context in fast_forward mode, passing start_lsn
421  * as InvalidXLogRecPtr, so that we start processing from my slot's
422  * confirmed_flush.
423  */
425  NIL,
426  true, /* fast_forward */
428  NULL, NULL, NULL);
429 
430  /*
431  * Start reading at the slot's restart_lsn, which we know to point to
432  * a valid record.
433  */
435 
436  /* Initialize our return value in case we don't do anything */
438 
439  /* invalidate non-timetravel entries */
441 
442  /* Decode at least one record, until we run out of records */
443  while (ctx->reader->EndRecPtr < moveto)
444  {
445  char *errm = NULL;
446  XLogRecord *record;
447 
448  /*
449  * Read records. No changes are generated in fast_forward mode,
450  * but snapbuilder/slot statuses are updated properly.
451  */
452  record = XLogReadRecord(ctx->reader, &errm);
453  if (errm)
454  elog(ERROR, "%s", errm);
455 
456  /*
457  * Process the record. Storage-level changes are ignored in
458  * fast_forward mode, but other modules (such as snapbuilder)
459  * might still have critical updates to do.
460  */
461  if (record)
463 
464  /* Stop once the requested target has been reached */
465  if (moveto <= ctx->reader->EndRecPtr)
466  break;
467 
469  }
470 
471  /*
472  * Logical decoding could have clobbered CurrentResourceOwner during
473  * transaction management, so restore the executor's value. (This is
474  * a kluge, but it's not worth cleaning up right now.)
475  */
476  CurrentResourceOwner = old_resowner;
477 
478  if (ctx->reader->EndRecPtr != InvalidXLogRecPtr)
479  {
481 
482  /*
483  * If only the confirmed_flush LSN has changed the slot won't get
484  * marked as dirty by the above. Callers on the walsender
485  * interface are expected to keep track of their own progress and
486  * don't need it written out. But SQL-interface users cannot
487  * specify their own start positions and it's harder for them to
488  * keep track of their progress, so we should make more of an
489  * effort to save it for them.
490  *
491  * Dirty the slot so it is written out at the next checkpoint.
492  * The LSN position advanced to may still be lost on a crash
493  * but this makes the data consistent after a clean shutdown.
494  */
496  }
497 
499 
500  /* free context, call shutdown callback */
501  FreeDecodingContext(ctx);
502 
504  }
505  PG_CATCH();
506  {
507  /* clear all timetravel entries */
509 
510  PG_RE_THROW();
511  }
512  PG_END_TRY();
513 
514  return retlsn;
515 }
516 
517 /*
518  * SQL function for moving the position in a replication slot.
519  */
520 Datum
522 {
523  Name slotname = PG_GETARG_NAME(0);
524  XLogRecPtr moveto = PG_GETARG_LSN(1);
525  XLogRecPtr endlsn;
526  XLogRecPtr minlsn;
527  TupleDesc tupdesc;
528  Datum values[2];
529  bool nulls[2];
530  HeapTuple tuple;
531  Datum result;
532 
534 
536 
537  if (XLogRecPtrIsInvalid(moveto))
538  ereport(ERROR,
539  (errmsg("invalid target WAL LSN")));
540 
541  /* Build a tuple descriptor for our result type */
542  if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
543  elog(ERROR, "return type must be a row type");
544 
545  /*
546  * We can't move slot past what's been flushed/replayed so clamp the
547  * target position accordingly.
548  */
549  if (!RecoveryInProgress())
550  moveto = Min(moveto, GetFlushRecPtr());
551  else
552  moveto = Min(moveto, GetXLogReplayRecPtr(&ThisTimeLineID));
553 
554  /* Acquire the slot so we "own" it */
555  ReplicationSlotAcquire(NameStr(*slotname), true);
556 
557  /* A slot whose restart_lsn has never been reserved cannot be advanced */
559  ereport(ERROR,
560  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
561  errmsg("cannot advance replication slot that has not previously reserved WAL")));
562 
563  /*
564  * Check if the slot is not moving backwards. Physical slots rely simply
565  * on restart_lsn as a minimum point, while logical slots have confirmed
566  * consumption up to confirmed_flush, meaning that in both cases data
567  * older than that is not available anymore.
568  */
571  else
573 
574  if (moveto < minlsn)
575  ereport(ERROR,
576  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
577  errmsg("cannot advance replication slot to %X/%X, minimum is %X/%X",
578  (uint32) (moveto >> 32), (uint32) moveto,
579  (uint32) (minlsn >> 32), (uint32) minlsn)));
580 
581  /* Do the actual slot update, depending on the slot type */
583  endlsn = pg_logical_replication_slot_advance(moveto);
584  else
585  endlsn = pg_physical_replication_slot_advance(moveto);
586 
587  values[0] = NameGetDatum(&MyReplicationSlot->data.name);
588  nulls[0] = false;
589 
591 
592  /* Return the reached position. */
593  values[1] = LSNGetDatum(endlsn);
594  nulls[1] = false;
595 
596  tuple = heap_form_tuple(tupdesc, values, nulls);
597  result = HeapTupleGetDatum(tuple);
598 
599  PG_RETURN_DATUM(result);
600 }
601 
602 /*
603  * Helper function of copying a replication slot.
604  */
605 static Datum
606 copy_replication_slot(FunctionCallInfo fcinfo, bool logical_slot)
607 {
608  Name src_name = PG_GETARG_NAME(0);
609  Name dst_name = PG_GETARG_NAME(1);
610  ReplicationSlot *src = NULL;
611  XLogRecPtr src_restart_lsn;
612  bool src_islogical;
613  bool temporary;
614  char *plugin;
615  Datum values[2];
616  bool nulls[2];
617  Datum result;
618  TupleDesc tupdesc;
619  HeapTuple tuple;
620 
621  if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
622  elog(ERROR, "return type must be a row type");
623 
625 
626  if (logical_slot)
628  else
630 
631  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
632 
633  /*
634  * We need to prevent the source slot's reserved WAL from being removed,
635  * but we don't want to lock that slot for very long, and it can advance
636  * in the meantime. So obtain the source slot's data, and create a new
637  * slot using its restart_lsn. Afterwards we lock the source slot again
638  * and verify that the data we copied (name, type) has not changed
639  * incompatibly. No inconvenient WAL removal can occur once the new slot
640  * is created -- but since WAL removal could have occurred before we
641  * managed to create the new slot, we advance the new slot's restart_lsn
642  * to the source slot's updated restart_lsn the second time we lock it.
643  */
644  for (int i = 0; i < max_replication_slots; i++)
645  {
647 
648  if (s->in_use && strcmp(NameStr(s->data.name), NameStr(*src_name)) == 0)
649  {
650  SpinLockAcquire(&s->mutex);
651  src_islogical = SlotIsLogical(s);
652  src_restart_lsn = s->data.restart_lsn;
653  temporary = s->data.persistency == RS_TEMPORARY;
654  plugin = logical_slot ? pstrdup(NameStr(s->data.plugin)) : NULL;
655  SpinLockRelease(&s->mutex);
656 
657  src = s;
658  break;
659  }
660  }
661 
662  LWLockRelease(ReplicationSlotControlLock);
663 
664  if (src == NULL)
665  ereport(ERROR,
666  (errcode(ERRCODE_UNDEFINED_OBJECT),
667  errmsg("replication slot \"%s\" does not exist", NameStr(*src_name))));
668 
669  /* Check type of replication slot */
670  if (src_islogical != logical_slot)
671  ereport(ERROR,
672  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
673  src_islogical ?
674  errmsg("cannot copy physical replication slot \"%s\" as a logical replication slot",
675  NameStr(*src_name)) :
676  errmsg("cannot copy logical replication slot \"%s\" as a physical replication slot",
677  NameStr(*src_name))));
678 
679  /* Copying non-reserved slot doesn't make sense */
680  if (XLogRecPtrIsInvalid(src_restart_lsn))
681  {
682  Assert(!logical_slot);
683  ereport(ERROR,
684  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
685  errmsg("cannot copy a replication slot that doesn't reserve WAL")));
686  }
687 
688  /* Overwrite params from optional arguments */
689  if (PG_NARGS() >= 3)
690  temporary = PG_GETARG_BOOL(2);
691  if (PG_NARGS() >= 4)
692  {
693  Assert(logical_slot);
694  plugin = NameStr(*(PG_GETARG_NAME(3)));
695  }
696 
697  /* Create new slot and acquire it */
698  if (logical_slot)
699  {
700  /*
701  * We must not try to read WAL, since we haven't reserved it yet --
702  * hence pass find_startpoint false. confirmed_flush will be set
703  * below, by copying from the source slot.
704  */
706  plugin,
707  temporary,
708  src_restart_lsn,
709  false);
710  }
711  else
713  true,
714  temporary,
715  src_restart_lsn);
716 
717  /*
718  * Update the destination slot to current values of the source slot;
719  * recheck that the source slot is still the one we saw previously.
720  */
721  {
722  TransactionId copy_effective_xmin;
723  TransactionId copy_effective_catalog_xmin;
724  TransactionId copy_xmin;
725  TransactionId copy_catalog_xmin;
726  XLogRecPtr copy_restart_lsn;
727  XLogRecPtr copy_confirmed_flush;
728  bool copy_islogical;
729  char *copy_name;
730 
731  /* Copy data of source slot again */
732  SpinLockAcquire(&src->mutex);
733  copy_effective_xmin = src->effective_xmin;
734  copy_effective_catalog_xmin = src->effective_catalog_xmin;
735 
736  copy_xmin = src->data.xmin;
737  copy_catalog_xmin = src->data.catalog_xmin;
738  copy_restart_lsn = src->data.restart_lsn;
739  copy_confirmed_flush = src->data.confirmed_flush;
740 
741  /* for existence check */
742  copy_name = pstrdup(NameStr(src->data.name));
743  copy_islogical = SlotIsLogical(src);
744  SpinLockRelease(&src->mutex);
745 
746  /*
747  * Check if the source slot still exists and is valid. We regard it as
748  * invalid if the type of replication slot or name has been changed,
749  * or the restart_lsn either is invalid or has gone backward. (The
750  * restart_lsn could go backwards if the source slot is dropped and
751  * copied from an older slot during installation.)
752  *
753  * Since erroring out will release and drop the destination slot we
754  * don't need to release it here.
755  */
756  if (copy_restart_lsn < src_restart_lsn ||
757  src_islogical != copy_islogical ||
758  strcmp(copy_name, NameStr(*src_name)) != 0)
759  ereport(ERROR,
760  (errmsg("could not copy replication slot \"%s\"",
761  NameStr(*src_name)),
762  errdetail("The source replication slot was modified incompatibly during the copy operation.")));
763 
764  /* The source slot must have a consistent snapshot */
765  if (src_islogical && XLogRecPtrIsInvalid(copy_confirmed_flush))
766  ereport(ERROR,
767  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
768  errmsg("cannot copy unfinished logical replication slot \"%s\"",
769  NameStr(*src_name)),
770  errhint("Retry when the source replication slot's confirmed_flush_lsn is valid.")));
771 
772  /* Install copied values again */
774  MyReplicationSlot->effective_xmin = copy_effective_xmin;
775  MyReplicationSlot->effective_catalog_xmin = copy_effective_catalog_xmin;
776 
777  MyReplicationSlot->data.xmin = copy_xmin;
778  MyReplicationSlot->data.catalog_xmin = copy_catalog_xmin;
779  MyReplicationSlot->data.restart_lsn = copy_restart_lsn;
780  MyReplicationSlot->data.confirmed_flush = copy_confirmed_flush;
782 
787 
788 #ifdef USE_ASSERT_CHECKING
789  /* Check that the restart_lsn is available */
790  {
791  XLogSegNo segno;
792 
793  XLByteToSeg(copy_restart_lsn, segno, wal_segment_size);
794  Assert(XLogGetLastRemovedSegno() < segno);
795  }
796 #endif
797  }
798 
799  /* target slot fully created, mark as persistent if needed */
800  if (logical_slot && !temporary)
802 
803  /* All done. Set up the return values */
804  values[0] = NameGetDatum(dst_name);
805  nulls[0] = false;
807  {
809  nulls[1] = false;
810  }
811  else
812  nulls[1] = true;
813 
814  tuple = heap_form_tuple(tupdesc, values, nulls);
815  result = HeapTupleGetDatum(tuple);
816 
818 
819  PG_RETURN_DATUM(result);
820 }
821 
822 /* The wrappers below are all to appease opr_sanity */
823 Datum
825 {
826  return copy_replication_slot(fcinfo, true);
827 }
828 
829 Datum
831 {
832  return copy_replication_slot(fcinfo, true);
833 }
834 
835 Datum
837 {
838  return copy_replication_slot(fcinfo, true);
839 }
840 
841 Datum
843 {
844  return copy_replication_slot(fcinfo, false);
845 }
846 
847 Datum
849 {
850  return copy_replication_slot(fcinfo, false);
851 }
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:411
static void create_logical_replication_slot(char *name, char *plugin, bool temporary, XLogRecPtr restart_lsn, bool find_startpoint)
Definition: slotfuncs.c:126
#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:577
#define NameGetDatum(X)
Definition: postgres.h:595
TypeFuncClass get_call_result_type(FunctionCallInfo fcinfo, Oid *resultTypeId, TupleDesc *resultTupleDesc)
Definition: funcapi.c:205
int errhint(const char *fmt,...)
Definition: elog.c:1071
static void create_physical_replication_slot(char *name, bool immediately_reserve, bool temporary, XLogRecPtr restart_lsn)
Definition: slotfuncs.c:46
int wal_segment_size
Definition: xlog.c:114
#define PG_GET_REPLICATION_SLOTS_COLS
uint32 TransactionId
Definition: c.h:513
Datum pg_get_replication_slots(PG_FUNCTION_ARGS)
Definition: slotfuncs.c:235
Oid GetUserId(void)
Definition: miscinit.c:439
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:173
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:920
#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:610
#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:8393
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:842
ReplicationSlotPersistentData data
Definition: slot.h:132
Datum pg_copy_logical_replication_slot_c(PG_FUNCTION_ARGS)
Definition: slotfuncs.c:836
unsigned int Oid
Definition: postgres_ext.h:31
bool RecoveryInProgress(void)
Definition: xlog.c:8060
#define OidIsValid(objectId)
Definition: c.h:644
void InvalidateSystemCaches(void)
Definition: inval.c:643
XLogRecPtr confirmed_flush
Definition: slot.h:80
XLogRecPtr EndRecPtr
Definition: xlogreader.h:135
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1727
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:3921
void ReplicationSlotReserveWal(void)
Definition: slot.c:997
void DecodingContextFindStartpoint(LogicalDecodingContext *ctx)
Definition: logical.c:464
void ReplicationSlotsComputeRequiredLSN(void)
Definition: slot.c:748
XLogRecord * XLogReadRecord(XLogReaderState *state, char **errormsg)
Definition: xlogreader.c:261
#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:824
static Datum copy_replication_slot(FunctionCallInfo fcinfo, bool logical_slot)
Definition: slotfuncs.c:606
XLogRecPtr GetXLogReplayRecPtr(TimeLineID *replayTLI)
Definition: xlog.c:11338
void ReplicationSlotPersist(void)
Definition: slot.c:680
TransactionId effective_xmin
Definition: slot.h:128
Definition: c.h:609
Datum pg_copy_physical_replication_slot_b(PG_FUNCTION_ARGS)
Definition: slotfuncs.c:848
uint64 XLogSegNo
Definition: xlogdefs.h:41
int errdetail(const char *fmt,...)
Definition: elog.c:957
Datum pg_replication_slot_advance(PG_FUNCTION_ARGS)
Definition: slotfuncs.c:521
TransactionId catalog_xmin
Definition: slot.h:69
#define InvalidTransactionId
Definition: transam.h:31
void XLogBeginRead(XLogReaderState *state, XLogRecPtr RecPtr)
Definition: xlogreader.c:233
unsigned int uint32
Definition: c.h:367
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 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
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:229
#define PG_GETARG_LSN(n)
Definition: pg_lsn.h:24
Datum pg_copy_logical_replication_slot_b(PG_FUNCTION_ARGS)
Definition: slotfuncs.c:830
uintptr_t Datum
Definition: postgres.h:367
Datum pg_drop_replication_slot(PG_FUNCTION_ARGS)
Definition: slotfuncs.c:218
#define PG_RETURN_DATUM(x)
Definition: fmgr.h:347
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:372
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:189
#define ereport(elevel,...)
Definition: elog.h:144
int allowedModes
Definition: execnodes.h:305
#define PG_RETURN_VOID()
Definition: fmgr.h:343
SetFunctionReturnMode returnMode
Definition: execnodes.h:307
#define PG_CATCH()
Definition: elog.h:305
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
int max_replication_slots
Definition: slot.c:99
uint64 XLogRecPtr
Definition: xlogdefs.h:21
#define Assert(condition)
Definition: c.h:738
void FreeDecodingContext(LogicalDecodingContext *ctx)
Definition: logical.c:507
XLogRecPtr restart_lsn
Definition: slot.h:72
bool has_rolreplication(Oid roleid)
Definition: miscinit.c:620
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1123
#define PG_NARGS()
Definition: fmgr.h:198
#define PG_RE_THROW()
Definition: elog.h:336
#define HeapTupleGetDatum(tuple)
Definition: funcapi.h:220
void LogicalConfirmReceivedLocation(XLogRecPtr lsn)
Definition: logical.c:1000
MemoryContext ecxt_per_query_memory
Definition: execnodes.h:233
const char * name
Definition: encode.c:521
Tuplestorestate * setResult
Definition: execnodes.h:310
static Datum values[MAXATTR]
Definition: bootstrap.c:167
ExprContext * econtext
Definition: execnodes.h:303
#define Int32GetDatum(X)
Definition: postgres.h:479
void ReplicationSlotDrop(const char *name, bool nowait)
Definition: slot.c:517
TupleDesc setDesc
Definition: execnodes.h:311
ReplicationSlotPersistency
Definition: slot.h:32
int errmsg(const char *fmt,...)
Definition: elog.c:824
XLogReaderState * reader
Definition: logical.h:41
pid_t active_pid
Definition: slot.h:111
#define elog(elevel,...)
Definition: elog.h:214
int i
#define NameStr(name)
Definition: c.h:615
#define CStringGetTextDatum(s)
Definition: builtins.h:87
#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:375
slock_t mutex
Definition: slot.h:105
#define PG_TRY()
Definition: elog.h:295
int read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen, XLogRecPtr targetRecPtr, char *cur_page)
Definition: xlogutils.c:822
void CheckLogicalDecodingRequirements(void)
Definition: logical.c:75
void ReplicationSlotsComputeRequiredXmin(bool already_locked)
Definition: slot.c:702
#define PG_END_TRY()
Definition: elog.h:320
void ReplicationSlotMarkDirty(void)
Definition: slot.c:663
#define PG_GETARG_NAME(n)
Definition: fmgr.h:273
#define XLByteToSeg(xlrp, logSegNo, wal_segsz_bytes)