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-2022, 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/xlogrecovery.h"
18 #include "access/xlogutils.h"
19 #include "funcapi.h"
20 #include "miscadmin.h"
21 #include "replication/decode.h"
22 #include "replication/logical.h"
23 #include "replication/slot.h"
24 #include "utils/builtins.h"
25 #include "utils/inval.h"
26 #include "utils/pg_lsn.h"
27 #include "utils/resowner.h"
28 
29 /*
30  * Helper function for creating a new physical replication slot with
31  * given arguments. Note that this function doesn't release the created
32  * slot.
33  *
34  * If restart_lsn is a valid value, we use it without WAL reservation
35  * routine. So the caller must guarantee that WAL is available.
36  */
37 static void
38 create_physical_replication_slot(char *name, bool immediately_reserve,
39  bool temporary, XLogRecPtr restart_lsn)
40 {
42 
43  /* acquire replication slot, this will check for conflicting names */
45  temporary ? RS_TEMPORARY : RS_PERSISTENT, false);
46 
47  if (immediately_reserve)
48  {
49  /* Reserve WAL as the user asked for it */
50  if (XLogRecPtrIsInvalid(restart_lsn))
52  else
53  MyReplicationSlot->data.restart_lsn = restart_lsn;
54 
55  /* Write this slot to disk */
58  }
59 }
60 
61 /*
62  * SQL function for creating a new physical (streaming replication)
63  * replication slot.
64  */
65 Datum
67 {
69  bool immediately_reserve = PG_GETARG_BOOL(1);
70  bool temporary = PG_GETARG_BOOL(2);
71  Datum values[2];
72  bool nulls[2];
73  TupleDesc tupdesc;
74  HeapTuple tuple;
75  Datum result;
76 
77  if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
78  elog(ERROR, "return type must be a row type");
79 
81 
83 
85  immediately_reserve,
86  temporary,
88 
90  nulls[0] = false;
91 
92  if (immediately_reserve)
93  {
95  nulls[1] = false;
96  }
97  else
98  nulls[1] = true;
99 
100  tuple = heap_form_tuple(tupdesc, values, nulls);
101  result = HeapTupleGetDatum(tuple);
102 
104 
105  PG_RETURN_DATUM(result);
106 }
107 
108 
109 /*
110  * Helper function for creating a new logical replication slot with
111  * given arguments. Note that this function doesn't release the created
112  * slot.
113  *
114  * When find_startpoint is false, the slot's confirmed_flush is not set; it's
115  * caller's responsibility to ensure it's set to something sensible.
116  */
117 static void
119  bool temporary, bool two_phase,
120  XLogRecPtr restart_lsn,
121  bool find_startpoint)
122 {
123  LogicalDecodingContext *ctx = NULL;
124 
126 
127  /*
128  * Acquire a logical decoding slot, this will check for conflicting names.
129  * Initially create persistent slot as ephemeral - that allows us to
130  * nicely handle errors during initialization because it'll get dropped if
131  * this transaction fails. We'll make it persistent at the end. Temporary
132  * slots can be created as temporary from beginning as they get dropped on
133  * error as well.
134  */
136  temporary ? RS_TEMPORARY : RS_EPHEMERAL, two_phase);
137 
138  /*
139  * Create logical decoding context to find start point or, if we don't
140  * need it, to 1) bump slot's restart_lsn and xmin 2) check plugin sanity.
141  *
142  * Note: when !find_startpoint this is still important, because it's at
143  * this point that the output plugin is validated.
144  */
146  false, /* just catalogs is OK */
147  restart_lsn,
148  XL_ROUTINE(.page_read = read_local_xlog_page,
149  .segment_open = wal_segment_open,
150  .segment_close = wal_segment_close),
151  NULL, NULL, NULL);
152 
153  /*
154  * If caller needs us to determine the decoding start point, do so now.
155  * This might take a while.
156  */
157  if (find_startpoint)
159 
160  /* don't need the decoding context anymore */
161  FreeDecodingContext(ctx);
162 }
163 
164 /*
165  * SQL function for creating a new logical replication slot.
166  */
167 Datum
169 {
170  Name name = PG_GETARG_NAME(0);
172  bool temporary = PG_GETARG_BOOL(2);
173  bool two_phase = PG_GETARG_BOOL(3);
174  Datum result;
175  TupleDesc tupdesc;
176  HeapTuple tuple;
177  Datum values[2];
178  bool nulls[2];
179 
180  if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
181  elog(ERROR, "return type must be a row type");
182 
184 
186 
188  NameStr(*plugin),
189  temporary,
190  two_phase,
192  true);
193 
196 
197  memset(nulls, 0, sizeof(nulls));
198 
199  tuple = heap_form_tuple(tupdesc, values, nulls);
200  result = HeapTupleGetDatum(tuple);
201 
202  /* ok, slot is now fully created, mark it as persistent if needed */
203  if (!temporary)
206 
207  PG_RETURN_DATUM(result);
208 }
209 
210 
211 /*
212  * SQL function for dropping a replication slot.
213  */
214 Datum
216 {
217  Name name = PG_GETARG_NAME(0);
218 
220 
222 
224 
225  PG_RETURN_VOID();
226 }
227 
228 /*
229  * pg_get_replication_slots - SQL SRF showing active replication slots.
230  */
231 Datum
233 {
234 #define PG_GET_REPLICATION_SLOTS_COLS 14
235  ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
236  XLogRecPtr currlsn;
237  int slotno;
238 
239  /*
240  * We don't require any special permission to see this function's data
241  * because nothing should be sensitive. The most critical being the slot
242  * name, which shouldn't contain anything particularly sensitive.
243  */
244 
245  SetSingleFuncCall(fcinfo, 0);
246 
247  currlsn = GetXLogWriteRecPtr();
248 
249  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
250  for (slotno = 0; slotno < max_replication_slots; slotno++)
251  {
253  ReplicationSlot slot_contents;
255  bool nulls[PG_GET_REPLICATION_SLOTS_COLS];
256  WALAvailability walstate;
257  int i;
258 
259  if (!slot->in_use)
260  continue;
261 
262  /* Copy slot contents while holding spinlock, then examine at leisure */
263  SpinLockAcquire(&slot->mutex);
264  slot_contents = *slot;
265  SpinLockRelease(&slot->mutex);
266 
267  memset(values, 0, sizeof(values));
268  memset(nulls, 0, sizeof(nulls));
269 
270  i = 0;
271  values[i++] = NameGetDatum(&slot_contents.data.name);
272 
273  if (slot_contents.data.database == InvalidOid)
274  nulls[i++] = true;
275  else
276  values[i++] = NameGetDatum(&slot_contents.data.plugin);
277 
278  if (slot_contents.data.database == InvalidOid)
279  values[i++] = CStringGetTextDatum("physical");
280  else
281  values[i++] = CStringGetTextDatum("logical");
282 
283  if (slot_contents.data.database == InvalidOid)
284  nulls[i++] = true;
285  else
286  values[i++] = ObjectIdGetDatum(slot_contents.data.database);
287 
288  values[i++] = BoolGetDatum(slot_contents.data.persistency == RS_TEMPORARY);
289  values[i++] = BoolGetDatum(slot_contents.active_pid != 0);
290 
291  if (slot_contents.active_pid != 0)
292  values[i++] = Int32GetDatum(slot_contents.active_pid);
293  else
294  nulls[i++] = true;
295 
296  if (slot_contents.data.xmin != InvalidTransactionId)
297  values[i++] = TransactionIdGetDatum(slot_contents.data.xmin);
298  else
299  nulls[i++] = true;
300 
301  if (slot_contents.data.catalog_xmin != InvalidTransactionId)
302  values[i++] = TransactionIdGetDatum(slot_contents.data.catalog_xmin);
303  else
304  nulls[i++] = true;
305 
306  if (slot_contents.data.restart_lsn != InvalidXLogRecPtr)
307  values[i++] = LSNGetDatum(slot_contents.data.restart_lsn);
308  else
309  nulls[i++] = true;
310 
311  if (slot_contents.data.confirmed_flush != InvalidXLogRecPtr)
312  values[i++] = LSNGetDatum(slot_contents.data.confirmed_flush);
313  else
314  nulls[i++] = true;
315 
316  /*
317  * If invalidated_at is valid and restart_lsn is invalid, we know for
318  * certain that the slot has been invalidated. Otherwise, test
319  * availability from restart_lsn.
320  */
321  if (XLogRecPtrIsInvalid(slot_contents.data.restart_lsn) &&
322  !XLogRecPtrIsInvalid(slot_contents.data.invalidated_at))
323  walstate = WALAVAIL_REMOVED;
324  else
325  walstate = GetWALAvailability(slot_contents.data.restart_lsn);
326 
327  switch (walstate)
328  {
330  nulls[i++] = true;
331  break;
332 
333  case WALAVAIL_RESERVED:
334  values[i++] = CStringGetTextDatum("reserved");
335  break;
336 
337  case WALAVAIL_EXTENDED:
338  values[i++] = CStringGetTextDatum("extended");
339  break;
340 
341  case WALAVAIL_UNRESERVED:
342  values[i++] = CStringGetTextDatum("unreserved");
343  break;
344 
345  case WALAVAIL_REMOVED:
346 
347  /*
348  * If we read the restart_lsn long enough ago, maybe that file
349  * has been removed by now. However, the walsender could have
350  * moved forward enough that it jumped to another file after
351  * we looked. If checkpointer signalled the process to
352  * termination, then it's definitely lost; but if a process is
353  * still alive, then "unreserved" seems more appropriate.
354  *
355  * If we do change it, save the state for safe_wal_size below.
356  */
357  if (!XLogRecPtrIsInvalid(slot_contents.data.restart_lsn))
358  {
359  int pid;
360 
361  SpinLockAcquire(&slot->mutex);
362  pid = slot->active_pid;
363  slot_contents.data.restart_lsn = slot->data.restart_lsn;
364  SpinLockRelease(&slot->mutex);
365  if (pid != 0)
366  {
367  values[i++] = CStringGetTextDatum("unreserved");
368  walstate = WALAVAIL_UNRESERVED;
369  break;
370  }
371  }
372  values[i++] = CStringGetTextDatum("lost");
373  break;
374  }
375 
376  /*
377  * safe_wal_size is only computed for slots that have not been lost,
378  * and only if there's a configured maximum size.
379  */
380  if (walstate == WALAVAIL_REMOVED || max_slot_wal_keep_size_mb < 0)
381  nulls[i++] = true;
382  else
383  {
384  XLogSegNo targetSeg;
385  uint64 slotKeepSegs;
386  uint64 keepSegs;
387  XLogSegNo failSeg;
388  XLogRecPtr failLSN;
389 
390  XLByteToSeg(slot_contents.data.restart_lsn, targetSeg, wal_segment_size);
391 
392  /* determine how many segments can be kept by slots */
394  /* ditto for wal_keep_size */
396 
397  /* if currpos reaches failLSN, we lose our segment */
398  failSeg = targetSeg + Max(slotKeepSegs, keepSegs) + 1;
399  XLogSegNoOffsetToRecPtr(failSeg, 0, wal_segment_size, failLSN);
400 
401  values[i++] = Int64GetDatum(failLSN - currlsn);
402  }
403 
404  values[i++] = BoolGetDatum(slot_contents.data.two_phase);
405 
407 
408  tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc,
409  values, nulls);
410  }
411 
412  LWLockRelease(ReplicationSlotControlLock);
413 
414  return (Datum) 0;
415 }
416 
417 /*
418  * Helper function for advancing our physical replication slot forward.
419  *
420  * The LSN position to move to is compared simply to the slot's restart_lsn,
421  * knowing that any position older than that would be removed by successive
422  * checkpoints.
423  */
424 static XLogRecPtr
426 {
428  XLogRecPtr retlsn = startlsn;
429 
430  Assert(moveto != InvalidXLogRecPtr);
431 
432  if (startlsn < moveto)
433  {
437  retlsn = moveto;
438 
439  /*
440  * Dirty the slot so as it is written out at the next checkpoint. Note
441  * that the LSN position advanced may still be lost in the event of a
442  * crash, but this makes the data consistent after a clean shutdown.
443  */
445  }
446 
447  return retlsn;
448 }
449 
450 /*
451  * Helper function for advancing our logical replication slot forward.
452  *
453  * The slot's restart_lsn is used as start point for reading records, while
454  * confirmed_flush is used as base point for the decoding context.
455  *
456  * We cannot just do LogicalConfirmReceivedLocation to update confirmed_flush,
457  * because we need to digest WAL to advance restart_lsn allowing to recycle
458  * WAL and removal of old catalog tuples. As decoding is done in fast_forward
459  * mode, no changes are generated anyway.
460  */
461 static XLogRecPtr
463 {
465  ResourceOwner old_resowner = CurrentResourceOwner;
466  XLogRecPtr retlsn;
467 
468  Assert(moveto != InvalidXLogRecPtr);
469 
470  PG_TRY();
471  {
472  /*
473  * Create our decoding context in fast_forward mode, passing start_lsn
474  * as InvalidXLogRecPtr, so that we start processing from my slot's
475  * confirmed_flush.
476  */
478  NIL,
479  true, /* fast_forward */
480  XL_ROUTINE(.page_read = read_local_xlog_page,
481  .segment_open = wal_segment_open,
482  .segment_close = wal_segment_close),
483  NULL, NULL, NULL);
484 
485  /*
486  * Start reading at the slot's restart_lsn, which we know to point to
487  * a valid record.
488  */
490 
491  /* invalidate non-timetravel entries */
493 
494  /* Decode at least one record, until we run out of records */
495  while (ctx->reader->EndRecPtr < moveto)
496  {
497  char *errm = NULL;
498  XLogRecord *record;
499 
500  /*
501  * Read records. No changes are generated in fast_forward mode,
502  * but snapbuilder/slot statuses are updated properly.
503  */
504  record = XLogReadRecord(ctx->reader, &errm);
505  if (errm)
506  elog(ERROR, "could not find record while advancing replication slot: %s",
507  errm);
508 
509  /*
510  * Process the record. Storage-level changes are ignored in
511  * fast_forward mode, but other modules (such as snapbuilder)
512  * might still have critical updates to do.
513  */
514  if (record)
516 
517  /* Stop once the requested target has been reached */
518  if (moveto <= ctx->reader->EndRecPtr)
519  break;
520 
522  }
523 
524  /*
525  * Logical decoding could have clobbered CurrentResourceOwner during
526  * transaction management, so restore the executor's value. (This is
527  * a kluge, but it's not worth cleaning up right now.)
528  */
529  CurrentResourceOwner = old_resowner;
530 
531  if (ctx->reader->EndRecPtr != InvalidXLogRecPtr)
532  {
534 
535  /*
536  * If only the confirmed_flush LSN has changed the slot won't get
537  * marked as dirty by the above. Callers on the walsender
538  * interface are expected to keep track of their own progress and
539  * don't need it written out. But SQL-interface users cannot
540  * specify their own start positions and it's harder for them to
541  * keep track of their progress, so we should make more of an
542  * effort to save it for them.
543  *
544  * Dirty the slot so it is written out at the next checkpoint. The
545  * LSN position advanced to may still be lost on a crash but this
546  * makes the data consistent after a clean shutdown.
547  */
549  }
550 
552 
553  /* free context, call shutdown callback */
554  FreeDecodingContext(ctx);
555 
557  }
558  PG_CATCH();
559  {
560  /* clear all timetravel entries */
562 
563  PG_RE_THROW();
564  }
565  PG_END_TRY();
566 
567  return retlsn;
568 }
569 
570 /*
571  * SQL function for moving the position in a replication slot.
572  */
573 Datum
575 {
576  Name slotname = PG_GETARG_NAME(0);
577  XLogRecPtr moveto = PG_GETARG_LSN(1);
578  XLogRecPtr endlsn;
579  XLogRecPtr minlsn;
580  TupleDesc tupdesc;
581  Datum values[2];
582  bool nulls[2];
583  HeapTuple tuple;
584  Datum result;
585 
587 
589 
590  if (XLogRecPtrIsInvalid(moveto))
591  ereport(ERROR,
592  (errmsg("invalid target WAL LSN")));
593 
594  /* Build a tuple descriptor for our result type */
595  if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
596  elog(ERROR, "return type must be a row type");
597 
598  /*
599  * We can't move slot past what's been flushed/replayed so clamp the
600  * target position accordingly.
601  */
602  if (!RecoveryInProgress())
603  moveto = Min(moveto, GetFlushRecPtr(NULL));
604  else
605  moveto = Min(moveto, GetXLogReplayRecPtr(NULL));
606 
607  /* Acquire the slot so we "own" it */
608  ReplicationSlotAcquire(NameStr(*slotname), true);
609 
610  /* A slot whose restart_lsn has never been reserved cannot be advanced */
612  ereport(ERROR,
613  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
614  errmsg("replication slot \"%s\" cannot be advanced",
615  NameStr(*slotname)),
616  errdetail("This slot has never previously reserved WAL, or it has been invalidated.")));
617 
618  /*
619  * Check if the slot is not moving backwards. Physical slots rely simply
620  * on restart_lsn as a minimum point, while logical slots have confirmed
621  * consumption up to confirmed_flush, meaning that in both cases data
622  * older than that is not available anymore.
623  */
626  else
628 
629  if (moveto < minlsn)
630  ereport(ERROR,
631  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
632  errmsg("cannot advance replication slot to %X/%X, minimum is %X/%X",
633  LSN_FORMAT_ARGS(moveto), LSN_FORMAT_ARGS(minlsn))));
634 
635  /* Do the actual slot update, depending on the slot type */
637  endlsn = pg_logical_replication_slot_advance(moveto);
638  else
639  endlsn = pg_physical_replication_slot_advance(moveto);
640 
642  nulls[0] = false;
643 
644  /*
645  * Recompute the minimum LSN and xmin across all slots to adjust with the
646  * advancing potentially done.
647  */
650 
652 
653  /* Return the reached position. */
654  values[1] = LSNGetDatum(endlsn);
655  nulls[1] = false;
656 
657  tuple = heap_form_tuple(tupdesc, values, nulls);
658  result = HeapTupleGetDatum(tuple);
659 
660  PG_RETURN_DATUM(result);
661 }
662 
663 /*
664  * Helper function of copying a replication slot.
665  */
666 static Datum
667 copy_replication_slot(FunctionCallInfo fcinfo, bool logical_slot)
668 {
669  Name src_name = PG_GETARG_NAME(0);
670  Name dst_name = PG_GETARG_NAME(1);
671  ReplicationSlot *src = NULL;
672  ReplicationSlot first_slot_contents;
673  ReplicationSlot second_slot_contents;
674  XLogRecPtr src_restart_lsn;
675  bool src_islogical;
676  bool temporary;
677  char *plugin;
678  Datum values[2];
679  bool nulls[2];
680  Datum result;
681  TupleDesc tupdesc;
682  HeapTuple tuple;
683 
684  if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
685  elog(ERROR, "return type must be a row type");
686 
688 
689  if (logical_slot)
691  else
693 
694  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
695 
696  /*
697  * We need to prevent the source slot's reserved WAL from being removed,
698  * but we don't want to lock that slot for very long, and it can advance
699  * in the meantime. So obtain the source slot's data, and create a new
700  * slot using its restart_lsn. Afterwards we lock the source slot again
701  * and verify that the data we copied (name, type) has not changed
702  * incompatibly. No inconvenient WAL removal can occur once the new slot
703  * is created -- but since WAL removal could have occurred before we
704  * managed to create the new slot, we advance the new slot's restart_lsn
705  * to the source slot's updated restart_lsn the second time we lock it.
706  */
707  for (int i = 0; i < max_replication_slots; i++)
708  {
710 
711  if (s->in_use && strcmp(NameStr(s->data.name), NameStr(*src_name)) == 0)
712  {
713  /* Copy the slot contents while holding spinlock */
714  SpinLockAcquire(&s->mutex);
715  first_slot_contents = *s;
716  SpinLockRelease(&s->mutex);
717  src = s;
718  break;
719  }
720  }
721 
722  LWLockRelease(ReplicationSlotControlLock);
723 
724  if (src == NULL)
725  ereport(ERROR,
726  (errcode(ERRCODE_UNDEFINED_OBJECT),
727  errmsg("replication slot \"%s\" does not exist", NameStr(*src_name))));
728 
729  src_islogical = SlotIsLogical(&first_slot_contents);
730  src_restart_lsn = first_slot_contents.data.restart_lsn;
731  temporary = (first_slot_contents.data.persistency == RS_TEMPORARY);
732  plugin = logical_slot ? NameStr(first_slot_contents.data.plugin) : NULL;
733 
734  /* Check type of replication slot */
735  if (src_islogical != logical_slot)
736  ereport(ERROR,
737  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
738  src_islogical ?
739  errmsg("cannot copy physical replication slot \"%s\" as a logical replication slot",
740  NameStr(*src_name)) :
741  errmsg("cannot copy logical replication slot \"%s\" as a physical replication slot",
742  NameStr(*src_name))));
743 
744  /* Copying non-reserved slot doesn't make sense */
745  if (XLogRecPtrIsInvalid(src_restart_lsn))
746  ereport(ERROR,
747  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
748  errmsg("cannot copy a replication slot that doesn't reserve WAL")));
749 
750  /* Overwrite params from optional arguments */
751  if (PG_NARGS() >= 3)
752  temporary = PG_GETARG_BOOL(2);
753  if (PG_NARGS() >= 4)
754  {
755  Assert(logical_slot);
756  plugin = NameStr(*(PG_GETARG_NAME(3)));
757  }
758 
759  /* Create new slot and acquire it */
760  if (logical_slot)
761  {
762  /*
763  * We must not try to read WAL, since we haven't reserved it yet --
764  * hence pass find_startpoint false. confirmed_flush will be set
765  * below, by copying from the source slot.
766  */
768  plugin,
769  temporary,
770  false,
771  src_restart_lsn,
772  false);
773  }
774  else
776  true,
777  temporary,
778  src_restart_lsn);
779 
780  /*
781  * Update the destination slot to current values of the source slot;
782  * recheck that the source slot is still the one we saw previously.
783  */
784  {
785  TransactionId copy_effective_xmin;
786  TransactionId copy_effective_catalog_xmin;
787  TransactionId copy_xmin;
788  TransactionId copy_catalog_xmin;
789  XLogRecPtr copy_restart_lsn;
790  XLogRecPtr copy_confirmed_flush;
791  bool copy_islogical;
792  char *copy_name;
793 
794  /* Copy data of source slot again */
795  SpinLockAcquire(&src->mutex);
796  second_slot_contents = *src;
797  SpinLockRelease(&src->mutex);
798 
799  copy_effective_xmin = second_slot_contents.effective_xmin;
800  copy_effective_catalog_xmin = second_slot_contents.effective_catalog_xmin;
801 
802  copy_xmin = second_slot_contents.data.xmin;
803  copy_catalog_xmin = second_slot_contents.data.catalog_xmin;
804  copy_restart_lsn = second_slot_contents.data.restart_lsn;
805  copy_confirmed_flush = second_slot_contents.data.confirmed_flush;
806 
807  /* for existence check */
808  copy_name = NameStr(second_slot_contents.data.name);
809  copy_islogical = SlotIsLogical(&second_slot_contents);
810 
811  /*
812  * Check if the source slot still exists and is valid. We regard it as
813  * invalid if the type of replication slot or name has been changed,
814  * or the restart_lsn either is invalid or has gone backward. (The
815  * restart_lsn could go backwards if the source slot is dropped and
816  * copied from an older slot during installation.)
817  *
818  * Since erroring out will release and drop the destination slot we
819  * don't need to release it here.
820  */
821  if (copy_restart_lsn < src_restart_lsn ||
822  src_islogical != copy_islogical ||
823  strcmp(copy_name, NameStr(*src_name)) != 0)
824  ereport(ERROR,
825  (errmsg("could not copy replication slot \"%s\"",
826  NameStr(*src_name)),
827  errdetail("The source replication slot was modified incompatibly during the copy operation.")));
828 
829  /* The source slot must have a consistent snapshot */
830  if (src_islogical && XLogRecPtrIsInvalid(copy_confirmed_flush))
831  ereport(ERROR,
832  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
833  errmsg("cannot copy unfinished logical replication slot \"%s\"",
834  NameStr(*src_name)),
835  errhint("Retry when the source replication slot's confirmed_flush_lsn is valid.")));
836 
837  /* Install copied values again */
839  MyReplicationSlot->effective_xmin = copy_effective_xmin;
840  MyReplicationSlot->effective_catalog_xmin = copy_effective_catalog_xmin;
841 
842  MyReplicationSlot->data.xmin = copy_xmin;
843  MyReplicationSlot->data.catalog_xmin = copy_catalog_xmin;
844  MyReplicationSlot->data.restart_lsn = copy_restart_lsn;
845  MyReplicationSlot->data.confirmed_flush = copy_confirmed_flush;
847 
852 
853 #ifdef USE_ASSERT_CHECKING
854  /* Check that the restart_lsn is available */
855  {
856  XLogSegNo segno;
857 
858  XLByteToSeg(copy_restart_lsn, segno, wal_segment_size);
859  Assert(XLogGetLastRemovedSegno() < segno);
860  }
861 #endif
862  }
863 
864  /* target slot fully created, mark as persistent if needed */
865  if (logical_slot && !temporary)
867 
868  /* All done. Set up the return values */
869  values[0] = NameGetDatum(dst_name);
870  nulls[0] = false;
872  {
874  nulls[1] = false;
875  }
876  else
877  nulls[1] = true;
878 
879  tuple = heap_form_tuple(tupdesc, values, nulls);
880  result = HeapTupleGetDatum(tuple);
881 
883 
884  PG_RETURN_DATUM(result);
885 }
886 
887 /* The wrappers below are all to appease opr_sanity */
888 Datum
890 {
891  return copy_replication_slot(fcinfo, true);
892 }
893 
894 Datum
896 {
897  return copy_replication_slot(fcinfo, true);
898 }
899 
900 Datum
902 {
903  return copy_replication_slot(fcinfo, true);
904 }
905 
906 Datum
908 {
909  return copy_replication_slot(fcinfo, false);
910 }
911 
912 Datum
914 {
915  return copy_replication_slot(fcinfo, false);
916 }
static Datum values[MAXATTR]
Definition: bootstrap.c:156
#define CStringGetTextDatum(s)
Definition: builtins.h:85
#define NameStr(name)
Definition: c.h:681
#define Min(x, y)
Definition: c.h:986
#define Max(x, y)
Definition: c.h:980
uint32 TransactionId
Definition: c.h:587
#define OidIsValid(objectId)
Definition: c.h:710
void LogicalDecodingProcessRecord(LogicalDecodingContext *ctx, XLogReaderState *record)
Definition: decode.c:91
int errdetail(const char *fmt,...)
Definition: elog.c:1037
int errhint(const char *fmt,...)
Definition: elog.c:1151
int errcode(int sqlerrcode)
Definition: elog.c:693
int errmsg(const char *fmt,...)
Definition: elog.c:904
#define PG_RE_THROW()
Definition: elog.h:340
#define PG_END_TRY()
Definition: elog.h:324
#define PG_TRY()
Definition: elog.h:299
#define ERROR
Definition: elog.h:33
#define elog(elevel,...)
Definition: elog.h:218
#define PG_CATCH()
Definition: elog.h:309
#define ereport(elevel,...)
Definition: elog.h:143
const char * name
Definition: encode.c:561
Datum Int64GetDatum(int64 X)
Definition: fmgr.c:1683
#define PG_RETURN_VOID()
Definition: fmgr.h:349
#define PG_NARGS()
Definition: fmgr.h:203
#define PG_GETARG_NAME(n)
Definition: fmgr.h:278
#define PG_GETARG_BOOL(n)
Definition: fmgr.h:274
#define PG_RETURN_DATUM(x)
Definition: fmgr.h:353
#define PG_FUNCTION_ARGS
Definition: fmgr.h:193
void SetSingleFuncCall(FunctionCallInfo fcinfo, bits32 flags)
Definition: funcapi.c:76
TypeFuncClass get_call_result_type(FunctionCallInfo fcinfo, Oid *resultTypeId, TupleDesc *resultTupleDesc)
Definition: funcapi.c:276
@ TYPEFUNC_COMPOSITE
Definition: funcapi.h:149
#define HeapTupleGetDatum(tuple)
Definition: funcapi.h:220
HeapTuple heap_form_tuple(TupleDesc tupleDescriptor, Datum *values, bool *isnull)
Definition: heaptuple.c:1020
void InvalidateSystemCaches(void)
Definition: inval.c:702
int i
Definition: isn.c:73
Assert(fmt[strlen(fmt) - 1] !='\n')
void LogicalConfirmReceivedLocation(XLogRecPtr lsn)
Definition: logical.c:1736
void FreeDecodingContext(LogicalDecodingContext *ctx)
Definition: logical.c:634
void DecodingContextFindStartpoint(LogicalDecodingContext *ctx)
Definition: logical.c:590
void CheckLogicalDecodingRequirements(void)
Definition: logical.c:103
LogicalDecodingContext * CreateInitDecodingContext(const char *plugin, List *output_plugin_options, bool need_full_snapshot, XLogRecPtr restart_lsn, XLogReaderRoutine *xl_routine, LogicalOutputPluginWriterPrepareWrite prepare_write, LogicalOutputPluginWriterWrite do_write, LogicalOutputPluginWriterUpdateProgress update_progress)
Definition: logical.c:319
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:479
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1196
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1800
@ LW_SHARED
Definition: lwlock.h:105
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:121
#define NIL
Definition: pg_list.h:65
#define PG_GETARG_LSN(n)
Definition: pg_lsn.h:24
#define LSNGetDatum(X)
Definition: pg_lsn.h:22
static bool two_phase
static const char * plugin
#define TransactionIdGetDatum(X)
Definition: postgres.h:565
uintptr_t Datum
Definition: postgres.h:411
#define ObjectIdGetDatum(X)
Definition: postgres.h:551
#define BoolGetDatum(X)
Definition: postgres.h:446
#define Int32GetDatum(X)
Definition: postgres.h:523
#define NameGetDatum(X)
Definition: postgres.h:639
#define InvalidOid
Definition: postgres_ext.h:36
ResourceOwner CurrentResourceOwner
Definition: resowner.c:146
void ReplicationSlotMarkDirty(void)
Definition: slot.c:798
void ReplicationSlotReserveWal(void)
Definition: slot.c:1151
void ReplicationSlotAcquire(const char *name, bool nowait)
Definition: slot.c:425
void ReplicationSlotsComputeRequiredXmin(bool already_locked)
Definition: slot.c:837
void ReplicationSlotPersist(void)
Definition: slot.c:815
ReplicationSlot * MyReplicationSlot
Definition: slot.c:97
void ReplicationSlotDrop(const char *name, bool nowait)
Definition: slot.c:625
void ReplicationSlotSave(void)
Definition: slot.c:780
void CheckSlotPermissions(void)
Definition: slot.c:1136
void ReplicationSlotRelease(void)
Definition: slot.c:522
int max_replication_slots
Definition: slot.c:100
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:94
void ReplicationSlotsComputeRequiredLSN(void)
Definition: slot.c:887
void ReplicationSlotCreate(const char *name, bool db_specific, ReplicationSlotPersistency persistency, bool two_phase)
Definition: slot.c:255
void CheckSlotRequirements(void)
Definition: slot.c:1114
@ RS_PERSISTENT
Definition: slot.h:35
@ RS_EPHEMERAL
Definition: slot.h:36
@ RS_TEMPORARY
Definition: slot.h:37
#define SlotIsLogical(slot)
Definition: slot.h:169
Datum pg_get_replication_slots(PG_FUNCTION_ARGS)
Definition: slotfuncs.c:232
Datum pg_copy_physical_replication_slot_a(PG_FUNCTION_ARGS)
Definition: slotfuncs.c:907
static void create_logical_replication_slot(char *name, char *plugin, bool temporary, bool two_phase, XLogRecPtr restart_lsn, bool find_startpoint)
Definition: slotfuncs.c:118
Datum pg_create_logical_replication_slot(PG_FUNCTION_ARGS)
Definition: slotfuncs.c:168
#define PG_GET_REPLICATION_SLOTS_COLS
static Datum copy_replication_slot(FunctionCallInfo fcinfo, bool logical_slot)
Definition: slotfuncs.c:667
Datum pg_copy_logical_replication_slot_c(PG_FUNCTION_ARGS)
Definition: slotfuncs.c:901
Datum pg_create_physical_replication_slot(PG_FUNCTION_ARGS)
Definition: slotfuncs.c:66
Datum pg_copy_logical_replication_slot_a(PG_FUNCTION_ARGS)
Definition: slotfuncs.c:889
Datum pg_copy_physical_replication_slot_b(PG_FUNCTION_ARGS)
Definition: slotfuncs.c:913
Datum pg_copy_logical_replication_slot_b(PG_FUNCTION_ARGS)
Definition: slotfuncs.c:895
static XLogRecPtr pg_logical_replication_slot_advance(XLogRecPtr moveto)
Definition: slotfuncs.c:462
static XLogRecPtr pg_physical_replication_slot_advance(XLogRecPtr moveto)
Definition: slotfuncs.c:425
Datum pg_replication_slot_advance(PG_FUNCTION_ARGS)
Definition: slotfuncs.c:574
static void create_physical_replication_slot(char *name, bool immediately_reserve, bool temporary, XLogRecPtr restart_lsn)
Definition: slotfuncs.c:38
Datum pg_drop_replication_slot(PG_FUNCTION_ARGS)
Definition: slotfuncs.c:215
#define SpinLockRelease(lock)
Definition: spin.h:64
#define SpinLockAcquire(lock)
Definition: spin.h:62
XLogReaderState * reader
Definition: logical.h:42
ReplicationSlot replication_slots[1]
Definition: slot.h:180
TransactionId xmin
Definition: slot.h:62
TransactionId catalog_xmin
Definition: slot.h:70
XLogRecPtr restart_lsn
Definition: slot.h:73
XLogRecPtr confirmed_flush
Definition: slot.h:84
XLogRecPtr invalidated_at
Definition: slot.h:76
ReplicationSlotPersistency persistency
Definition: slot.h:54
TransactionId effective_catalog_xmin
Definition: slot.h:144
slock_t mutex
Definition: slot.h:120
pid_t active_pid
Definition: slot.h:126
bool in_use
Definition: slot.h:123
TransactionId effective_xmin
Definition: slot.h:143
ReplicationSlotPersistentData data
Definition: slot.h:147
TupleDesc setDesc
Definition: execnodes.h:317
Tuplestorestate * setResult
Definition: execnodes.h:316
XLogRecPtr EndRecPtr
Definition: xlogreader.h:207
Definition: c.h:676
#define InvalidTransactionId
Definition: transam.h:31
void tuplestore_putvalues(Tuplestorestate *state, TupleDesc tdesc, Datum *values, bool *isnull)
Definition: tuplestore.c:750
bool RecoveryInProgress(void)
Definition: xlog.c:5753
XLogSegNo XLogGetLastRemovedSegno(void)
Definition: xlog.c:3485
int wal_keep_size_mb
Definition: xlog.c:117
int wal_segment_size
Definition: xlog.c:144
WALAvailability GetWALAvailability(XLogRecPtr targetLSN)
Definition: xlog.c:7180
int max_slot_wal_keep_size_mb
Definition: xlog.c:136
XLogRecPtr GetFlushRecPtr(TimeLineID *insertTLI)
Definition: xlog.c:5918
XLogRecPtr GetXLogWriteRecPtr(void)
Definition: xlog.c:8819
WALAvailability
Definition: xlog.h:182
@ WALAVAIL_REMOVED
Definition: xlog.h:188
@ WALAVAIL_RESERVED
Definition: xlog.h:184
@ WALAVAIL_UNRESERVED
Definition: xlog.h:187
@ WALAVAIL_EXTENDED
Definition: xlog.h:185
@ WALAVAIL_INVALID_LSN
Definition: xlog.h:183
#define XLogSegNoOffsetToRecPtr(segno, offset, wal_segsz_bytes, dest)
#define XLogMBVarToSegs(mbvar, wal_segsz_bytes)
#define XLByteToSeg(xlrp, logSegNo, wal_segsz_bytes)
#define LSN_FORMAT_ARGS(lsn)
Definition: xlogdefs.h:43
#define XLogRecPtrIsInvalid(r)
Definition: xlogdefs.h:29
uint64 XLogRecPtr
Definition: xlogdefs.h:21
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
uint64 XLogSegNo
Definition: xlogdefs.h:48
XLogRecord * XLogReadRecord(XLogReaderState *state, char **errormsg)
Definition: xlogreader.c:418
void XLogBeginRead(XLogReaderState *state, XLogRecPtr RecPtr)
Definition: xlogreader.c:264
#define XL_ROUTINE(...)
Definition: xlogreader.h:117
XLogRecPtr GetXLogReplayRecPtr(TimeLineID *replayTLI)
void wal_segment_close(XLogReaderState *state)
Definition: xlogutils.c:856
void wal_segment_open(XLogReaderState *state, XLogSegNo nextSegNo, TimeLineID *tli_p)
Definition: xlogutils.c:831
int read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen, XLogRecPtr targetRecPtr, char *cur_page)
Definition: xlogutils.c:875