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