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  XL_ROUTINE(.page_read = read_local_xlog_page,
156  .segment_open = wal_segment_open,
157  .segment_close = wal_segment_close),
158  NULL, NULL, NULL);
159 
160  /*
161  * If caller needs us to determine the decoding start point, do so now.
162  * This might take a while.
163  */
164  if (find_startpoint)
166 
167  /* don't need the decoding context anymore */
168  FreeDecodingContext(ctx);
169 }
170 
171 /*
172  * SQL function for creating a new logical replication slot.
173  */
174 Datum
176 {
177  Name name = PG_GETARG_NAME(0);
179  bool temporary = PG_GETARG_BOOL(2);
180  Datum result;
181  TupleDesc tupdesc;
182  HeapTuple tuple;
183  Datum values[2];
184  bool nulls[2];
185 
186  if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
187  elog(ERROR, "return type must be a row type");
188 
190 
192 
194  NameStr(*plugin),
195  temporary,
197  true);
198 
199  values[0] = NameGetDatum(&MyReplicationSlot->data.name);
201 
202  memset(nulls, 0, sizeof(nulls));
203 
204  tuple = heap_form_tuple(tupdesc, values, nulls);
205  result = HeapTupleGetDatum(tuple);
206 
207  /* ok, slot is now fully created, mark it as persistent if needed */
208  if (!temporary)
211 
212  PG_RETURN_DATUM(result);
213 }
214 
215 
216 /*
217  * SQL function for dropping a replication slot.
218  */
219 Datum
221 {
222  Name name = PG_GETARG_NAME(0);
223 
225 
227 
228  ReplicationSlotDrop(NameStr(*name), true);
229 
230  PG_RETURN_VOID();
231 }
232 
233 /*
234  * pg_get_replication_slots - SQL SRF showing active replication slots.
235  */
236 Datum
238 {
239 #define PG_GET_REPLICATION_SLOTS_COLS 13
240  ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
241  TupleDesc tupdesc;
242  Tuplestorestate *tupstore;
243  MemoryContext per_query_ctx;
244  MemoryContext oldcontext;
245  int slotno;
246 
247  /* check to see if caller supports us returning a tuplestore */
248  if (rsinfo == NULL || !IsA(rsinfo, ReturnSetInfo))
249  ereport(ERROR,
250  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
251  errmsg("set-valued function called in context that cannot accept a set")));
252  if (!(rsinfo->allowedModes & SFRM_Materialize))
253  ereport(ERROR,
254  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
255  errmsg("materialize mode required, but it is not allowed in this context")));
256 
257  /* Build a tuple descriptor for our result type */
258  if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
259  elog(ERROR, "return type must be a row type");
260 
261  /*
262  * We don't require any special permission to see this function's data
263  * because nothing should be sensitive. The most critical being the slot
264  * name, which shouldn't contain anything particularly sensitive.
265  */
266 
267  per_query_ctx = rsinfo->econtext->ecxt_per_query_memory;
268  oldcontext = MemoryContextSwitchTo(per_query_ctx);
269 
270  tupstore = tuplestore_begin_heap(true, false, work_mem);
271  rsinfo->returnMode = SFRM_Materialize;
272  rsinfo->setResult = tupstore;
273  rsinfo->setDesc = tupdesc;
274 
275  MemoryContextSwitchTo(oldcontext);
276 
277  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
278  for (slotno = 0; slotno < max_replication_slots; slotno++)
279  {
281  ReplicationSlot slot_contents;
283  bool nulls[PG_GET_REPLICATION_SLOTS_COLS];
284  WALAvailability walstate;
285  XLogSegNo last_removed_seg;
286  int i;
287 
288  if (!slot->in_use)
289  continue;
290 
291  /* Copy slot contents while holding spinlock, then examine at leisure */
292  SpinLockAcquire(&slot->mutex);
293  slot_contents = *slot;
294  SpinLockRelease(&slot->mutex);
295 
296  memset(values, 0, sizeof(values));
297  memset(nulls, 0, sizeof(nulls));
298 
299  i = 0;
300  values[i++] = NameGetDatum(&slot_contents.data.name);
301 
302  if (slot_contents.data.database == InvalidOid)
303  nulls[i++] = true;
304  else
305  values[i++] = NameGetDatum(&slot_contents.data.plugin);
306 
307  if (slot_contents.data.database == InvalidOid)
308  values[i++] = CStringGetTextDatum("physical");
309  else
310  values[i++] = CStringGetTextDatum("logical");
311 
312  if (slot_contents.data.database == InvalidOid)
313  nulls[i++] = true;
314  else
315  values[i++] = ObjectIdGetDatum(slot_contents.data.database);
316 
317  values[i++] = BoolGetDatum(slot_contents.data.persistency == RS_TEMPORARY);
318  values[i++] = BoolGetDatum(slot_contents.active_pid != 0);
319 
320  if (slot_contents.active_pid != 0)
321  values[i++] = Int32GetDatum(slot_contents.active_pid);
322  else
323  nulls[i++] = true;
324 
325  if (slot_contents.data.xmin != InvalidTransactionId)
326  values[i++] = TransactionIdGetDatum(slot_contents.data.xmin);
327  else
328  nulls[i++] = true;
329 
330  if (slot_contents.data.catalog_xmin != InvalidTransactionId)
331  values[i++] = TransactionIdGetDatum(slot_contents.data.catalog_xmin);
332  else
333  nulls[i++] = true;
334 
335  if (slot_contents.data.restart_lsn != InvalidXLogRecPtr)
336  values[i++] = LSNGetDatum(slot_contents.data.restart_lsn);
337  else
338  nulls[i++] = true;
339 
340  if (slot_contents.data.confirmed_flush != InvalidXLogRecPtr)
341  values[i++] = LSNGetDatum(slot_contents.data.confirmed_flush);
342  else
343  nulls[i++] = true;
344 
345  /*
346  * If invalidated_at is valid and restart_lsn is invalid, we know for
347  * certain that the slot has been invalidated. Otherwise, test
348  * availability from restart_lsn.
349  */
350  if (XLogRecPtrIsInvalid(slot_contents.data.restart_lsn) &&
351  !XLogRecPtrIsInvalid(slot_contents.data.invalidated_at))
352  walstate = WALAVAIL_REMOVED;
353  else
354  walstate = GetWALAvailability(slot_contents.data.restart_lsn);
355 
356  switch (walstate)
357  {
359  nulls[i++] = true;
360  break;
361 
362  case WALAVAIL_RESERVED:
363  values[i++] = CStringGetTextDatum("reserved");
364  break;
365 
366  case WALAVAIL_EXTENDED:
367  values[i++] = CStringGetTextDatum("extended");
368  break;
369 
370  case WALAVAIL_UNRESERVED:
371  values[i++] = CStringGetTextDatum("unreserved");
372  break;
373 
374  case WALAVAIL_REMOVED:
375 
376  /*
377  * If we read the restart_lsn long enough ago, maybe that file
378  * has been removed by now. However, the walsender could have
379  * moved forward enough that it jumped to another file after
380  * we looked. If checkpointer signalled the process to
381  * termination, then it's definitely lost; but if a process is
382  * still alive, then "unreserved" seems more appropriate.
383  */
384  if (!XLogRecPtrIsInvalid(slot_contents.data.restart_lsn))
385  {
386  int pid;
387 
388  SpinLockAcquire(&slot->mutex);
389  pid = slot->active_pid;
390  SpinLockRelease(&slot->mutex);
391  if (pid != 0)
392  {
393  values[i++] = CStringGetTextDatum("unreserved");
394  break;
395  }
396  }
397  values[i++] = CStringGetTextDatum("lost");
398  break;
399  }
400 
401  if (max_slot_wal_keep_size_mb >= 0 &&
402  (walstate == WALAVAIL_RESERVED || walstate == WALAVAIL_EXTENDED) &&
403  ((last_removed_seg = XLogGetLastRemovedSegno()) != 0))
404  {
405  XLogRecPtr min_safe_lsn;
406 
407  XLogSegNoOffsetToRecPtr(last_removed_seg + 1, 0,
408  wal_segment_size, min_safe_lsn);
409  values[i++] = Int64GetDatum(min_safe_lsn);
410  }
411  else
412  nulls[i++] = true;
413 
415 
416  tuplestore_putvalues(tupstore, tupdesc, values, nulls);
417  }
418 
419  LWLockRelease(ReplicationSlotControlLock);
420 
421  tuplestore_donestoring(tupstore);
422 
423  return (Datum) 0;
424 }
425 
426 /*
427  * Helper function for advancing our physical replication slot forward.
428  *
429  * The LSN position to move to is compared simply to the slot's restart_lsn,
430  * knowing that any position older than that would be removed by successive
431  * checkpoints.
432  */
433 static XLogRecPtr
435 {
437  XLogRecPtr retlsn = startlsn;
438 
439  Assert(moveto != InvalidXLogRecPtr);
440 
441  if (startlsn < moveto)
442  {
446  retlsn = moveto;
447 
448  /*
449  * Dirty the slot so as it is written out at the next checkpoint. Note
450  * that the LSN position advanced may still be lost in the event of a
451  * crash, but this makes the data consistent after a clean shutdown.
452  */
454  }
455 
456  return retlsn;
457 }
458 
459 /*
460  * Helper function for advancing our logical replication slot forward.
461  *
462  * The slot's restart_lsn is used as start point for reading records, while
463  * confirmed_flush is used as base point for the decoding context.
464  *
465  * We cannot just do LogicalConfirmReceivedLocation to update confirmed_flush,
466  * because we need to digest WAL to advance restart_lsn allowing to recycle
467  * WAL and removal of old catalog tuples. As decoding is done in fast_forward
468  * mode, no changes are generated anyway.
469  */
470 static XLogRecPtr
472 {
474  ResourceOwner old_resowner = CurrentResourceOwner;
475  XLogRecPtr retlsn;
476 
477  Assert(moveto != InvalidXLogRecPtr);
478 
479  PG_TRY();
480  {
481  /*
482  * Create our decoding context in fast_forward mode, passing start_lsn
483  * as InvalidXLogRecPtr, so that we start processing from my slot's
484  * confirmed_flush.
485  */
487  NIL,
488  true, /* fast_forward */
489  XL_ROUTINE(.page_read = read_local_xlog_page,
490  .segment_open = wal_segment_open,
491  .segment_close = wal_segment_close),
492  NULL, NULL, NULL);
493 
494  /*
495  * Start reading at the slot's restart_lsn, which we know to point to
496  * a valid record.
497  */
499 
500  /* Initialize our return value in case we don't do anything */
502 
503  /* invalidate non-timetravel entries */
505 
506  /* Decode at least one record, until we run out of records */
507  while (ctx->reader->EndRecPtr < moveto)
508  {
509  char *errm = NULL;
510  XLogRecord *record;
511 
512  /*
513  * Read records. No changes are generated in fast_forward mode,
514  * but snapbuilder/slot statuses are updated properly.
515  */
516  record = XLogReadRecord(ctx->reader, &errm);
517  if (errm)
518  elog(ERROR, "%s", errm);
519 
520  /*
521  * Process the record. Storage-level changes are ignored in
522  * fast_forward mode, but other modules (such as snapbuilder)
523  * might still have critical updates to do.
524  */
525  if (record)
527 
528  /* Stop once the requested target has been reached */
529  if (moveto <= ctx->reader->EndRecPtr)
530  break;
531 
533  }
534 
535  /*
536  * Logical decoding could have clobbered CurrentResourceOwner during
537  * transaction management, so restore the executor's value. (This is
538  * a kluge, but it's not worth cleaning up right now.)
539  */
540  CurrentResourceOwner = old_resowner;
541 
542  if (ctx->reader->EndRecPtr != InvalidXLogRecPtr)
543  {
545 
546  /*
547  * If only the confirmed_flush LSN has changed the slot won't get
548  * marked as dirty by the above. Callers on the walsender
549  * interface are expected to keep track of their own progress and
550  * don't need it written out. But SQL-interface users cannot
551  * specify their own start positions and it's harder for them to
552  * keep track of their progress, so we should make more of an
553  * effort to save it for them.
554  *
555  * Dirty the slot so it is written out at the next checkpoint. The
556  * LSN position advanced to may still be lost on a crash but this
557  * makes the data consistent after a clean shutdown.
558  */
560  }
561 
563 
564  /* free context, call shutdown callback */
565  FreeDecodingContext(ctx);
566 
568  }
569  PG_CATCH();
570  {
571  /* clear all timetravel entries */
573 
574  PG_RE_THROW();
575  }
576  PG_END_TRY();
577 
578  return retlsn;
579 }
580 
581 /*
582  * SQL function for moving the position in a replication slot.
583  */
584 Datum
586 {
587  Name slotname = PG_GETARG_NAME(0);
588  XLogRecPtr moveto = PG_GETARG_LSN(1);
589  XLogRecPtr endlsn;
590  XLogRecPtr minlsn;
591  TupleDesc tupdesc;
592  Datum values[2];
593  bool nulls[2];
594  HeapTuple tuple;
595  Datum result;
596 
598 
600 
601  if (XLogRecPtrIsInvalid(moveto))
602  ereport(ERROR,
603  (errmsg("invalid target WAL LSN")));
604 
605  /* Build a tuple descriptor for our result type */
606  if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
607  elog(ERROR, "return type must be a row type");
608 
609  /*
610  * We can't move slot past what's been flushed/replayed so clamp the
611  * target position accordingly.
612  */
613  if (!RecoveryInProgress())
614  moveto = Min(moveto, GetFlushRecPtr());
615  else
616  moveto = Min(moveto, GetXLogReplayRecPtr(&ThisTimeLineID));
617 
618  /* Acquire the slot so we "own" it */
619  (void) ReplicationSlotAcquire(NameStr(*slotname), SAB_Error);
620 
621  /* A slot whose restart_lsn has never been reserved cannot be advanced */
623  ereport(ERROR,
624  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
625  errmsg("replication slot \"%s\" cannot be advanced",
626  NameStr(*slotname)),
627  errdetail("This slot has never previously reserved WAL, or has been invalidated.")));
628 
629  /*
630  * Check if the slot is not moving backwards. Physical slots rely simply
631  * on restart_lsn as a minimum point, while logical slots have confirmed
632  * consumption up to confirmed_flush, meaning that in both cases data
633  * older than that is not available anymore.
634  */
637  else
639 
640  if (moveto < minlsn)
641  ereport(ERROR,
642  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
643  errmsg("cannot advance replication slot to %X/%X, minimum is %X/%X",
644  (uint32) (moveto >> 32), (uint32) moveto,
645  (uint32) (minlsn >> 32), (uint32) minlsn)));
646 
647  /* Do the actual slot update, depending on the slot type */
649  endlsn = pg_logical_replication_slot_advance(moveto);
650  else
651  endlsn = pg_physical_replication_slot_advance(moveto);
652 
653  values[0] = NameGetDatum(&MyReplicationSlot->data.name);
654  nulls[0] = false;
655 
656  /*
657  * Recompute the minimum LSN and xmin across all slots to adjust with the
658  * advancing potentially done.
659  */
662 
664 
665  /* Return the reached position. */
666  values[1] = LSNGetDatum(endlsn);
667  nulls[1] = false;
668 
669  tuple = heap_form_tuple(tupdesc, values, nulls);
670  result = HeapTupleGetDatum(tuple);
671 
672  PG_RETURN_DATUM(result);
673 }
674 
675 /*
676  * Helper function of copying a replication slot.
677  */
678 static Datum
679 copy_replication_slot(FunctionCallInfo fcinfo, bool logical_slot)
680 {
681  Name src_name = PG_GETARG_NAME(0);
682  Name dst_name = PG_GETARG_NAME(1);
683  ReplicationSlot *src = NULL;
684  ReplicationSlot first_slot_contents;
685  ReplicationSlot second_slot_contents;
686  XLogRecPtr src_restart_lsn;
687  bool src_islogical;
688  bool temporary;
689  char *plugin;
690  Datum values[2];
691  bool nulls[2];
692  Datum result;
693  TupleDesc tupdesc;
694  HeapTuple tuple;
695 
696  if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
697  elog(ERROR, "return type must be a row type");
698 
700 
701  if (logical_slot)
703  else
705 
706  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
707 
708  /*
709  * We need to prevent the source slot's reserved WAL from being removed,
710  * but we don't want to lock that slot for very long, and it can advance
711  * in the meantime. So obtain the source slot's data, and create a new
712  * slot using its restart_lsn. Afterwards we lock the source slot again
713  * and verify that the data we copied (name, type) has not changed
714  * incompatibly. No inconvenient WAL removal can occur once the new slot
715  * is created -- but since WAL removal could have occurred before we
716  * managed to create the new slot, we advance the new slot's restart_lsn
717  * to the source slot's updated restart_lsn the second time we lock it.
718  */
719  for (int i = 0; i < max_replication_slots; i++)
720  {
722 
723  if (s->in_use && strcmp(NameStr(s->data.name), NameStr(*src_name)) == 0)
724  {
725  /* Copy the slot contents while holding spinlock */
726  SpinLockAcquire(&s->mutex);
727  first_slot_contents = *s;
728  SpinLockRelease(&s->mutex);
729  src = s;
730  break;
731  }
732  }
733 
734  LWLockRelease(ReplicationSlotControlLock);
735 
736  if (src == NULL)
737  ereport(ERROR,
738  (errcode(ERRCODE_UNDEFINED_OBJECT),
739  errmsg("replication slot \"%s\" does not exist", NameStr(*src_name))));
740 
741  src_islogical = SlotIsLogical(&first_slot_contents);
742  src_restart_lsn = first_slot_contents.data.restart_lsn;
743  temporary = (first_slot_contents.data.persistency == RS_TEMPORARY);
744  plugin = logical_slot ? NameStr(first_slot_contents.data.plugin) : NULL;
745 
746  /* Check type of replication slot */
747  if (src_islogical != logical_slot)
748  ereport(ERROR,
749  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
750  src_islogical ?
751  errmsg("cannot copy physical replication slot \"%s\" as a logical replication slot",
752  NameStr(*src_name)) :
753  errmsg("cannot copy logical replication slot \"%s\" as a physical replication slot",
754  NameStr(*src_name))));
755 
756  /* Copying non-reserved slot doesn't make sense */
757  if (XLogRecPtrIsInvalid(src_restart_lsn))
758  ereport(ERROR,
759  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
760  errmsg("cannot copy a replication slot that doesn't reserve WAL")));
761 
762  /* Overwrite params from optional arguments */
763  if (PG_NARGS() >= 3)
764  temporary = PG_GETARG_BOOL(2);
765  if (PG_NARGS() >= 4)
766  {
767  Assert(logical_slot);
768  plugin = NameStr(*(PG_GETARG_NAME(3)));
769  }
770 
771  /* Create new slot and acquire it */
772  if (logical_slot)
773  {
774  /*
775  * We must not try to read WAL, since we haven't reserved it yet --
776  * hence pass find_startpoint false. confirmed_flush will be set
777  * below, by copying from the source slot.
778  */
780  plugin,
781  temporary,
782  src_restart_lsn,
783  false);
784  }
785  else
787  true,
788  temporary,
789  src_restart_lsn);
790 
791  /*
792  * Update the destination slot to current values of the source slot;
793  * recheck that the source slot is still the one we saw previously.
794  */
795  {
796  TransactionId copy_effective_xmin;
797  TransactionId copy_effective_catalog_xmin;
798  TransactionId copy_xmin;
799  TransactionId copy_catalog_xmin;
800  XLogRecPtr copy_restart_lsn;
801  XLogRecPtr copy_confirmed_flush;
802  bool copy_islogical;
803  char *copy_name;
804 
805  /* Copy data of source slot again */
806  SpinLockAcquire(&src->mutex);
807  second_slot_contents = *src;
808  SpinLockRelease(&src->mutex);
809 
810  copy_effective_xmin = second_slot_contents.effective_xmin;
811  copy_effective_catalog_xmin = second_slot_contents.effective_catalog_xmin;
812 
813  copy_xmin = second_slot_contents.data.xmin;
814  copy_catalog_xmin = second_slot_contents.data.catalog_xmin;
815  copy_restart_lsn = second_slot_contents.data.restart_lsn;
816  copy_confirmed_flush = second_slot_contents.data.confirmed_flush;
817 
818  /* for existence check */
819  copy_name = NameStr(second_slot_contents.data.name);
820  copy_islogical = SlotIsLogical(&second_slot_contents);
821 
822  /*
823  * Check if the source slot still exists and is valid. We regard it as
824  * invalid if the type of replication slot or name has been changed,
825  * or the restart_lsn either is invalid or has gone backward. (The
826  * restart_lsn could go backwards if the source slot is dropped and
827  * copied from an older slot during installation.)
828  *
829  * Since erroring out will release and drop the destination slot we
830  * don't need to release it here.
831  */
832  if (copy_restart_lsn < src_restart_lsn ||
833  src_islogical != copy_islogical ||
834  strcmp(copy_name, NameStr(*src_name)) != 0)
835  ereport(ERROR,
836  (errmsg("could not copy replication slot \"%s\"",
837  NameStr(*src_name)),
838  errdetail("The source replication slot was modified incompatibly during the copy operation.")));
839 
840  /* The source slot must have a consistent snapshot */
841  if (src_islogical && XLogRecPtrIsInvalid(copy_confirmed_flush))
842  ereport(ERROR,
843  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
844  errmsg("cannot copy unfinished logical replication slot \"%s\"",
845  NameStr(*src_name)),
846  errhint("Retry when the source replication slot's confirmed_flush_lsn is valid.")));
847 
848  /* Install copied values again */
850  MyReplicationSlot->effective_xmin = copy_effective_xmin;
851  MyReplicationSlot->effective_catalog_xmin = copy_effective_catalog_xmin;
852 
853  MyReplicationSlot->data.xmin = copy_xmin;
854  MyReplicationSlot->data.catalog_xmin = copy_catalog_xmin;
855  MyReplicationSlot->data.restart_lsn = copy_restart_lsn;
856  MyReplicationSlot->data.confirmed_flush = copy_confirmed_flush;
858 
863 
864 #ifdef USE_ASSERT_CHECKING
865  /* Check that the restart_lsn is available */
866  {
867  XLogSegNo segno;
868 
869  XLByteToSeg(copy_restart_lsn, segno, wal_segment_size);
870  Assert(XLogGetLastRemovedSegno() < segno);
871  }
872 #endif
873  }
874 
875  /* target slot fully created, mark as persistent if needed */
876  if (logical_slot && !temporary)
878 
879  /* All done. Set up the return values */
880  values[0] = NameGetDatum(dst_name);
881  nulls[0] = false;
883  {
885  nulls[1] = false;
886  }
887  else
888  nulls[1] = true;
889 
890  tuple = heap_form_tuple(tupdesc, values, nulls);
891  result = HeapTupleGetDatum(tuple);
892 
894 
895  PG_RETURN_DATUM(result);
896 }
897 
898 /* The wrappers below are all to appease opr_sanity */
899 Datum
901 {
902  return copy_replication_slot(fcinfo, true);
903 }
904 
905 Datum
907 {
908  return copy_replication_slot(fcinfo, true);
909 }
910 
911 Datum
913 {
914  return copy_replication_slot(fcinfo, true);
915 }
916 
917 Datum
919 {
920  return copy_replication_slot(fcinfo, false);
921 }
922 
923 Datum
925 {
926  return copy_replication_slot(fcinfo, false);
927 }
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:471
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:1031
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
#define IsA(nodeptr, _type_)
Definition: nodes.h:580
#define NameGetDatum(X)
Definition: postgres.h:595
TypeFuncClass get_call_result_type(FunctionCallInfo fcinfo, Oid *resultTypeId, TupleDesc *resultTupleDesc)
Definition: funcapi.c:205
int errhint(const char *fmt,...)
Definition: elog.c:1071
void wal_segment_close(XLogReaderState *state)
Definition: xlogutils.c:813
static void create_physical_replication_slot(char *name, bool immediately_reserve, bool temporary, XLogRecPtr restart_lsn)
Definition: slotfuncs.c:46
int wal_segment_size
Definition: xlog.c:116
#define PG_GET_REPLICATION_SLOTS_COLS
uint32 TransactionId
Definition: c.h:513
Datum pg_get_replication_slots(PG_FUNCTION_ARGS)
Definition: slotfuncs.c:237
Oid GetUserId(void)
Definition: miscinit.c:448
ResourceOwner CurrentResourceOwner
Definition: resowner.c:142
Datum pg_create_logical_replication_slot(PG_FUNCTION_ARGS)
Definition: slotfuncs.c:175
WALAvailability
Definition: xlog.h:270
#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:222
ReplicationSlotPersistency persistency
Definition: slot.h:61
void wal_segment_open(XLogReaderState *state, XLogSegNo nextSegNo, TimeLineID *tli_p)
Definition: xlogutils.c:788
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:697
XLogRecPtr GetFlushRecPtr(void)
Definition: xlog.c:8423
HeapTuple heap_form_tuple(TupleDesc tupleDescriptor, Datum *values, bool *isnull)
Definition: heaptuple.c:1020
#define PG_GETARG_BOOL(n)
Definition: fmgr.h:274
Datum pg_copy_physical_replication_slot_a(PG_FUNCTION_ARGS)
Definition: slotfuncs.c:918
ReplicationSlotPersistentData data
Definition: slot.h:143
Datum pg_copy_logical_replication_slot_c(PG_FUNCTION_ARGS)
Definition: slotfuncs.c:912
bool RecoveryInProgress(void)
Definition: xlog.c:8072
#define OidIsValid(objectId)
Definition: c.h:644
void InvalidateSystemCaches(void)
Definition: inval.c:643
XLogRecPtr confirmed_flush
Definition: slot.h:91
XLogRecPtr EndRecPtr
Definition: xlogreader.h:176
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1812
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:3959
void ReplicationSlotReserveWal(void)
Definition: slot.c:1056
void DecodingContextFindStartpoint(LogicalDecodingContext *ctx)
Definition: logical.c:468
void ReplicationSlotsComputeRequiredLSN(void)
Definition: slot.c:804
XLogRecord * XLogReadRecord(XLogReaderState *state, char **errormsg)
Definition: xlogreader.c:268
#define ObjectIdGetDatum(X)
Definition: postgres.h:507
#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:900
static Datum copy_replication_slot(FunctionCallInfo fcinfo, bool logical_slot)
Definition: slotfuncs.c:679
XLogRecPtr GetXLogReplayRecPtr(TimeLineID *replayTLI)
Definition: xlog.c:11469
void ReplicationSlotPersist(void)
Definition: slot.c:732
TransactionId effective_xmin
Definition: slot.h:139
LogicalDecodingContext * CreateDecodingContext(XLogRecPtr start_lsn, List *output_plugin_options, bool fast_forward, XLogReaderRoutine *xl_routine, LogicalOutputPluginWriterPrepareWrite prepare_write, LogicalOutputPluginWriterWrite do_write, LogicalOutputPluginWriterUpdateProgress update_progress)
Definition: logical.c:376
Definition: c.h:609
Datum pg_copy_physical_replication_slot_b(PG_FUNCTION_ARGS)
Definition: slotfuncs.c:924
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:585
TransactionId catalog_xmin
Definition: slot.h:77
#define InvalidTransactionId
Definition: transam.h:31
void XLogBeginRead(XLogReaderState *state, XLogRecPtr RecPtr)
Definition: xlogreader.c:240
unsigned int uint32
Definition: c.h:367
void ReplicationSlotRelease(void)
Definition: slot.c:476
LogicalDecodingContext * CreateInitDecodingContext(char *plugin, List *output_plugin_options, bool need_full_snapshot, XLogRecPtr restart_lsn, XLogReaderRoutine *xl_routine, LogicalOutputPluginWriterPrepareWrite prepare_write, LogicalOutputPluginWriterWrite do_write, LogicalOutputPluginWriterUpdateProgress update_progress)
Definition: logical.c:230
TransactionId xmin
Definition: slot.h:69
#define SlotIsLogical(slot)
Definition: slot.h:165
Datum Int64GetDatum(int64 X)
Definition: fmgr.c:1701
static void check_permissions(void)
Definition: slotfuncs.c:29
#define XL_ROUTINE(...)
Definition: xlogreader.h:116
#define XLogRecPtrIsInvalid(r)
Definition: xlogdefs.h:29
bool in_use
Definition: slot.h:119
#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
#define PG_GETARG_LSN(n)
Definition: pg_lsn.h:24
Datum pg_copy_logical_replication_slot_b(PG_FUNCTION_ARGS)
Definition: slotfuncs.c:906
uintptr_t Datum
Definition: postgres.h:367
Datum pg_drop_replication_slot(PG_FUNCTION_ARGS)
Definition: slotfuncs.c:220
#define PG_RETURN_DATUM(x)
Definition: fmgr.h:352
TransactionId effective_catalog_xmin
Definition: slot.h:140
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:191
#define ereport(elevel,...)
Definition: elog.h:144
int allowedModes
Definition: execnodes.h:305
#define PG_RETURN_VOID()
Definition: fmgr.h:348
int max_slot_wal_keep_size_mb
Definition: xlog.c:110
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
#define XLogSegNoOffsetToRecPtr(segno, offset, wal_segsz_bytes, dest)
uint64 XLogRecPtr
Definition: xlogdefs.h:21
#define Assert(condition)
Definition: c.h:738
void FreeDecodingContext(LogicalDecodingContext *ctx)
Definition: logical.c:511
XLogRecPtr restart_lsn
Definition: slot.h:80
bool has_rolreplication(Oid roleid)
Definition: miscinit.c:629
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1208
#define PG_NARGS()
Definition: fmgr.h:203
#define PG_RE_THROW()
Definition: elog.h:336
#define HeapTupleGetDatum(tuple)
Definition: funcapi.h:220
void LogicalConfirmReceivedLocation(XLogRecPtr lsn)
Definition: logical.c:1012
MemoryContext ecxt_per_query_memory
Definition: execnodes.h:233
const char * name
Definition: encode.c:561
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:569
TupleDesc setDesc
Definition: execnodes.h:311
int errmsg(const char *fmt,...)
Definition: elog.c:824
XLogReaderState * reader
Definition: logical.h:41
int ReplicationSlotAcquire(const char *name, SlotAcquireBehavior behavior)
Definition: slot.c:367
pid_t active_pid
Definition: slot.h:122
WALAvailability GetWALAvailability(XLogRecPtr targetLSN)
Definition: xlog.c:9509
#define elog(elevel,...)
Definition: elog.h:214
int i
Definition: slot.h:42
#define NameStr(name)
Definition: c.h:615
#define CStringGetTextDatum(s)
Definition: builtins.h:87
#define PG_FUNCTION_ARGS
Definition: fmgr.h:193
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:99
ReplicationSlot replication_slots[1]
Definition: slot.h:176
static XLogRecPtr pg_physical_replication_slot_advance(XLogRecPtr moveto)
Definition: slotfuncs.c:434
XLogRecPtr invalidated_at
Definition: slot.h:83
slock_t mutex
Definition: slot.h:116
#define PG_TRY()
Definition: elog.h:295
int read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen, XLogRecPtr targetRecPtr, char *cur_page)
Definition: xlogutils.c:832
void CheckLogicalDecodingRequirements(void)
Definition: logical.c:75
void ReplicationSlotsComputeRequiredXmin(bool already_locked)
Definition: slot.c:754
#define PG_END_TRY()
Definition: elog.h:320
void ReplicationSlotMarkDirty(void)
Definition: slot.c:715
#define PG_GETARG_NAME(n)
Definition: fmgr.h:278
#define XLByteToSeg(xlrp, logSegNo, wal_segsz_bytes)