PostgreSQL Source Code  git master
logical.c
Go to the documentation of this file.
1 /*-------------------------------------------------------------------------
2  * logical.c
3  * PostgreSQL logical decoding coordination
4  *
5  * Copyright (c) 2012-2020, PostgreSQL Global Development Group
6  *
7  * IDENTIFICATION
8  * src/backend/replication/logical/logical.c
9  *
10  * NOTES
11  * This file coordinates interaction between the various modules that
12  * together provide logical decoding, primarily by providing so
13  * called LogicalDecodingContexts. The goal is to encapsulate most of the
14  * internal complexity for consumers of logical decoding, so they can
15  * create and consume a changestream with a low amount of code. Builtin
16  * consumers are the walsender and SQL SRF interface, but it's possible to
17  * add further ones without changing core code, e.g. to consume changes in
18  * a bgworker.
19  *
20  * The idea is that a consumer provides three callbacks, one to read WAL,
21  * one to prepare a data write, and a final one for actually writing since
22  * their implementation depends on the type of consumer. Check
23  * logicalfuncs.c for an example implementation of a fairly simple consumer
24  * and an implementation of a WAL reading callback that's suitable for
25  * simple consumers.
26  *-------------------------------------------------------------------------
27  */
28 
29 #include "postgres.h"
30 
31 #include "access/xact.h"
32 #include "access/xlog_internal.h"
33 #include "fmgr.h"
34 #include "miscadmin.h"
35 #include "replication/decode.h"
36 #include "replication/logical.h"
37 #include "replication/origin.h"
39 #include "replication/snapbuild.h"
40 #include "storage/proc.h"
41 #include "storage/procarray.h"
42 #include "utils/builtins.h"
43 #include "utils/memutils.h"
44 
45 /* data for errcontext callback */
47 {
49  const char *callback_name;
52 
53 /* wrappers around output plugin callbacks */
54 static void output_plugin_error_callback(void *arg);
56  bool is_init);
58 static void begin_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn);
59 static void commit_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
60  XLogRecPtr commit_lsn);
61 static void change_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
62  Relation relation, ReorderBufferChange *change);
63 static void truncate_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
64  int nrelations, Relation relations[], ReorderBufferChange *change);
65 static void message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
66  XLogRecPtr message_lsn, bool transactional,
67  const char *prefix, Size message_size, const char *message);
68 
69 /* streaming callbacks */
71  XLogRecPtr first_lsn);
73  XLogRecPtr last_lsn);
75  XLogRecPtr abort_lsn);
77  XLogRecPtr commit_lsn);
79  Relation relation, ReorderBufferChange *change);
81  XLogRecPtr message_lsn, bool transactional,
82  const char *prefix, Size message_size, const char *message);
84  int nrelations, Relation relations[], ReorderBufferChange *change);
85 
86 static void LoadOutputPlugin(OutputPluginCallbacks *callbacks, const char *plugin);
87 
88 /*
89  * Make sure the current settings & environment are capable of doing logical
90  * decoding.
91  */
92 void
94 {
96 
97  /*
98  * NB: Adding a new requirement likely means that RestoreSlotFromDisk()
99  * needs the same check.
100  */
101 
103  ereport(ERROR,
104  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
105  errmsg("logical decoding requires wal_level >= logical")));
106 
107  if (MyDatabaseId == InvalidOid)
108  ereport(ERROR,
109  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
110  errmsg("logical decoding requires a database connection")));
111 
112  /* ----
113  * TODO: We got to change that someday soon...
114  *
115  * There's basically three things missing to allow this:
116  * 1) We need to be able to correctly and quickly identify the timeline a
117  * LSN belongs to
118  * 2) We need to force hot_standby_feedback to be enabled at all times so
119  * the primary cannot remove rows we need.
120  * 3) support dropping replication slots referring to a database, in
121  * dbase_redo. There can't be any active ones due to HS recovery
122  * conflicts, so that should be relatively easy.
123  * ----
124  */
125  if (RecoveryInProgress())
126  ereport(ERROR,
127  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
128  errmsg("logical decoding cannot be used while in recovery")));
129 }
130 
131 /*
132  * Helper function for CreateInitDecodingContext() and
133  * CreateDecodingContext() performing common tasks.
134  */
135 static LogicalDecodingContext *
136 StartupDecodingContext(List *output_plugin_options,
137  XLogRecPtr start_lsn,
138  TransactionId xmin_horizon,
139  bool need_full_snapshot,
140  bool fast_forward,
141  XLogReaderRoutine *xl_routine,
145 {
146  ReplicationSlot *slot;
147  MemoryContext context,
148  old_context;
150 
151  /* shorter lines... */
152  slot = MyReplicationSlot;
153 
155  "Logical decoding context",
157  old_context = MemoryContextSwitchTo(context);
158  ctx = palloc0(sizeof(LogicalDecodingContext));
159 
160  ctx->context = context;
161 
162  /*
163  * (re-)load output plugins, so we detect a bad (removed) output plugin
164  * now.
165  */
166  if (!fast_forward)
168 
169  /*
170  * Now that the slot's xmin has been set, we can announce ourselves as a
171  * logical decoding backend which doesn't need to be checked individually
172  * when computing the xmin horizon because the xmin is enforced via
173  * replication slots.
174  *
175  * We can only do so if we're outside of a transaction (i.e. the case when
176  * streaming changes via walsender), otherwise an already setup
177  * snapshot/xid would end up being ignored. That's not a particularly
178  * bothersome restriction since the SQL interface can't be used for
179  * streaming anyway.
180  */
182  {
183  LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
186  LWLockRelease(ProcArrayLock);
187  }
188 
189  ctx->slot = slot;
190 
191  ctx->reader = XLogReaderAllocate(wal_segment_size, NULL, xl_routine, ctx);
192  if (!ctx->reader)
193  ereport(ERROR,
194  (errcode(ERRCODE_OUT_OF_MEMORY),
195  errmsg("out of memory")));
196 
198  ctx->snapshot_builder =
199  AllocateSnapshotBuilder(ctx->reorder, xmin_horizon, start_lsn,
200  need_full_snapshot);
201 
202  ctx->reorder->private_data = ctx;
203 
204  /* wrap output plugin callbacks, so we can add error context information */
210 
211  /*
212  * To support streaming, we require start/stop/abort/commit/change
213  * callbacks. The message and truncate callbacks are optional, similar to
214  * regular output plugins. We however enable streaming when at least one
215  * of the methods is enabled so that we can easily identify missing
216  * methods.
217  *
218  * We decide it here, but only check it later in the wrappers.
219  */
220  ctx->streaming = (ctx->callbacks.stream_start_cb != NULL) ||
221  (ctx->callbacks.stream_stop_cb != NULL) ||
222  (ctx->callbacks.stream_abort_cb != NULL) ||
223  (ctx->callbacks.stream_commit_cb != NULL) ||
224  (ctx->callbacks.stream_change_cb != NULL) ||
225  (ctx->callbacks.stream_message_cb != NULL) ||
226  (ctx->callbacks.stream_truncate_cb != NULL);
227 
228  /*
229  * streaming callbacks
230  *
231  * stream_message and stream_truncate callbacks are optional, so we do not
232  * fail with ERROR when missing, but the wrappers simply do nothing. We
233  * must set the ReorderBuffer callbacks to something, otherwise the calls
234  * from there will crash (we don't want to move the checks there).
235  */
243 
244  ctx->out = makeStringInfo();
245  ctx->prepare_write = prepare_write;
246  ctx->write = do_write;
247  ctx->update_progress = update_progress;
248 
249  ctx->output_plugin_options = output_plugin_options;
250 
251  ctx->fast_forward = fast_forward;
252 
253  MemoryContextSwitchTo(old_context);
254 
255  return ctx;
256 }
257 
258 /*
259  * Create a new decoding context, for a new logical slot.
260  *
261  * plugin -- contains the name of the output plugin
262  * output_plugin_options -- contains options passed to the output plugin
263  * need_full_snapshot -- if true, must obtain a snapshot able to read all
264  * tables; if false, one that can read only catalogs is acceptable.
265  * restart_lsn -- if given as invalid, it's this routine's responsibility to
266  * mark WAL as reserved by setting a convenient restart_lsn for the slot.
267  * Otherwise, we set for decoding to start from the given LSN without
268  * marking WAL reserved beforehand. In that scenario, it's up to the
269  * caller to guarantee that WAL remains available.
270  * xl_routine -- XLogReaderRoutine for underlying XLogReader
271  * prepare_write, do_write, update_progress --
272  * callbacks that perform the use-case dependent, actual, work.
273  *
274  * Needs to be called while in a memory context that's at least as long lived
275  * as the decoding context because further memory contexts will be created
276  * inside it.
277  *
278  * Returns an initialized decoding context after calling the output plugin's
279  * startup function.
280  */
283  List *output_plugin_options,
284  bool need_full_snapshot,
285  XLogRecPtr restart_lsn,
286  XLogReaderRoutine *xl_routine,
290 {
291  TransactionId xmin_horizon = InvalidTransactionId;
292  ReplicationSlot *slot;
293  NameData plugin_name;
295  MemoryContext old_context;
296 
297  /* shorter lines... */
298  slot = MyReplicationSlot;
299 
300  /* first some sanity checks that are unlikely to be violated */
301  if (slot == NULL)
302  elog(ERROR, "cannot perform logical decoding without an acquired slot");
303 
304  if (plugin == NULL)
305  elog(ERROR, "cannot initialize logical decoding without a specified plugin");
306 
307  /* Make sure the passed slot is suitable. These are user facing errors. */
308  if (SlotIsPhysical(slot))
309  ereport(ERROR,
310  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
311  errmsg("cannot use physical replication slot for logical decoding")));
312 
313  if (slot->data.database != MyDatabaseId)
314  ereport(ERROR,
315  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
316  errmsg("replication slot \"%s\" was not created in this database",
317  NameStr(slot->data.name))));
318 
319  if (IsTransactionState() &&
321  ereport(ERROR,
322  (errcode(ERRCODE_ACTIVE_SQL_TRANSACTION),
323  errmsg("cannot create logical replication slot in transaction that has performed writes")));
324 
325  /*
326  * Register output plugin name with slot. We need the mutex to avoid
327  * concurrent reading of a partially copied string. But we don't want any
328  * complicated code while holding a spinlock, so do namestrcpy() outside.
329  */
330  namestrcpy(&plugin_name, plugin);
331  SpinLockAcquire(&slot->mutex);
332  slot->data.plugin = plugin_name;
333  SpinLockRelease(&slot->mutex);
334 
335  if (XLogRecPtrIsInvalid(restart_lsn))
337  else
338  {
339  SpinLockAcquire(&slot->mutex);
340  slot->data.restart_lsn = restart_lsn;
341  SpinLockRelease(&slot->mutex);
342  }
343 
344  /* ----
345  * This is a bit tricky: We need to determine a safe xmin horizon to start
346  * decoding from, to avoid starting from a running xacts record referring
347  * to xids whose rows have been vacuumed or pruned
348  * already. GetOldestSafeDecodingTransactionId() returns such a value, but
349  * without further interlock its return value might immediately be out of
350  * date.
351  *
352  * So we have to acquire the ProcArrayLock to prevent computation of new
353  * xmin horizons by other backends, get the safe decoding xid, and inform
354  * the slot machinery about the new limit. Once that's done the
355  * ProcArrayLock can be released as the slot machinery now is
356  * protecting against vacuum.
357  *
358  * Note that, temporarily, the data, not just the catalog, xmin has to be
359  * reserved if a data snapshot is to be exported. Otherwise the initial
360  * data snapshot created here is not guaranteed to be valid. After that
361  * the data xmin doesn't need to be managed anymore and the global xmin
362  * should be recomputed. As we are fine with losing the pegged data xmin
363  * after crash - no chance a snapshot would get exported anymore - we can
364  * get away with just setting the slot's
365  * effective_xmin. ReplicationSlotRelease will reset it again.
366  *
367  * ----
368  */
369  LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
370 
371  xmin_horizon = GetOldestSafeDecodingTransactionId(!need_full_snapshot);
372 
373  SpinLockAcquire(&slot->mutex);
374  slot->effective_catalog_xmin = xmin_horizon;
375  slot->data.catalog_xmin = xmin_horizon;
376  if (need_full_snapshot)
377  slot->effective_xmin = xmin_horizon;
378  SpinLockRelease(&slot->mutex);
379 
381 
382  LWLockRelease(ProcArrayLock);
383 
386 
387  ctx = StartupDecodingContext(NIL, restart_lsn, xmin_horizon,
388  need_full_snapshot, false,
389  xl_routine, prepare_write, do_write,
390  update_progress);
391 
392  /* call output plugin initialization callback */
393  old_context = MemoryContextSwitchTo(ctx->context);
394  if (ctx->callbacks.startup_cb != NULL)
395  startup_cb_wrapper(ctx, &ctx->options, true);
396  MemoryContextSwitchTo(old_context);
397 
399 
400  return ctx;
401 }
402 
403 /*
404  * Create a new decoding context, for a logical slot that has previously been
405  * used already.
406  *
407  * start_lsn
408  * The LSN at which to start decoding. If InvalidXLogRecPtr, restart
409  * from the slot's confirmed_flush; otherwise, start from the specified
410  * location (but move it forwards to confirmed_flush if it's older than
411  * that, see below).
412  *
413  * output_plugin_options
414  * options passed to the output plugin.
415  *
416  * fast_forward
417  * bypass the generation of logical changes.
418  *
419  * xl_routine
420  * XLogReaderRoutine used by underlying xlogreader
421  *
422  * prepare_write, do_write, update_progress
423  * callbacks that have to be filled to perform the use-case dependent,
424  * actual work.
425  *
426  * Needs to be called while in a memory context that's at least as long lived
427  * as the decoding context because further memory contexts will be created
428  * inside it.
429  *
430  * Returns an initialized decoding context after calling the output plugin's
431  * startup function.
432  */
435  List *output_plugin_options,
436  bool fast_forward,
437  XLogReaderRoutine *xl_routine,
441 {
443  ReplicationSlot *slot;
444  MemoryContext old_context;
445 
446  /* shorter lines... */
447  slot = MyReplicationSlot;
448 
449  /* first some sanity checks that are unlikely to be violated */
450  if (slot == NULL)
451  elog(ERROR, "cannot perform logical decoding without an acquired slot");
452 
453  /* make sure the passed slot is suitable, these are user facing errors */
454  if (SlotIsPhysical(slot))
455  ereport(ERROR,
456  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
457  errmsg("cannot use physical replication slot for logical decoding")));
458 
459  if (slot->data.database != MyDatabaseId)
460  ereport(ERROR,
461  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
462  errmsg("replication slot \"%s\" was not created in this database",
463  NameStr(slot->data.name))));
464 
465  if (start_lsn == InvalidXLogRecPtr)
466  {
467  /* continue from last position */
468  start_lsn = slot->data.confirmed_flush;
469  }
470  else if (start_lsn < slot->data.confirmed_flush)
471  {
472  /*
473  * It might seem like we should error out in this case, but it's
474  * pretty common for a client to acknowledge a LSN it doesn't have to
475  * do anything for, and thus didn't store persistently, because the
476  * xlog records didn't result in anything relevant for logical
477  * decoding. Clients have to be able to do that to support synchronous
478  * replication.
479  */
480  elog(DEBUG1, "cannot stream from %X/%X, minimum is %X/%X, forwarding",
481  (uint32) (start_lsn >> 32), (uint32) start_lsn,
482  (uint32) (slot->data.confirmed_flush >> 32),
483  (uint32) slot->data.confirmed_flush);
484 
485  start_lsn = slot->data.confirmed_flush;
486  }
487 
488  ctx = StartupDecodingContext(output_plugin_options,
489  start_lsn, InvalidTransactionId, false,
490  fast_forward, xl_routine, prepare_write,
491  do_write, update_progress);
492 
493  /* call output plugin initialization callback */
494  old_context = MemoryContextSwitchTo(ctx->context);
495  if (ctx->callbacks.startup_cb != NULL)
496  startup_cb_wrapper(ctx, &ctx->options, false);
497  MemoryContextSwitchTo(old_context);
498 
500 
501  ereport(LOG,
502  (errmsg("starting logical decoding for slot \"%s\"",
503  NameStr(slot->data.name)),
504  errdetail("Streaming transactions committing after %X/%X, reading WAL from %X/%X.",
505  (uint32) (slot->data.confirmed_flush >> 32),
506  (uint32) slot->data.confirmed_flush,
507  (uint32) (slot->data.restart_lsn >> 32),
508  (uint32) slot->data.restart_lsn)));
509 
510  return ctx;
511 }
512 
513 /*
514  * Returns true if a consistent initial decoding snapshot has been built.
515  */
516 bool
518 {
520 }
521 
522 /*
523  * Read from the decoding slot, until it is ready to start extracting changes.
524  */
525 void
527 {
528  ReplicationSlot *slot = ctx->slot;
529 
530  /* Initialize from where to start reading WAL. */
531  XLogBeginRead(ctx->reader, slot->data.restart_lsn);
532 
533  elog(DEBUG1, "searching for logical decoding starting point, starting at %X/%X",
534  (uint32) (slot->data.restart_lsn >> 32),
535  (uint32) slot->data.restart_lsn);
536 
537  /* Wait for a consistent starting point */
538  for (;;)
539  {
540  XLogRecord *record;
541  char *err = NULL;
542 
543  /* the read_page callback waits for new WAL */
544  record = XLogReadRecord(ctx->reader, &err);
545  if (err)
546  elog(ERROR, "%s", err);
547  if (!record)
548  elog(ERROR, "no record found"); /* shouldn't happen */
549 
551 
552  /* only continue till we found a consistent spot */
553  if (DecodingContextReady(ctx))
554  break;
555 
557  }
558 
559  SpinLockAcquire(&slot->mutex);
560  slot->data.confirmed_flush = ctx->reader->EndRecPtr;
561  SpinLockRelease(&slot->mutex);
562 }
563 
564 /*
565  * Free a previously allocated decoding context, invoking the shutdown
566  * callback if necessary.
567  */
568 void
570 {
571  if (ctx->callbacks.shutdown_cb != NULL)
572  shutdown_cb_wrapper(ctx);
573 
576  XLogReaderFree(ctx->reader);
578 }
579 
580 /*
581  * Prepare a write using the context's output routine.
582  */
583 void
585 {
586  if (!ctx->accept_writes)
587  elog(ERROR, "writes are only accepted in commit, begin and change callbacks");
588 
589  ctx->prepare_write(ctx, ctx->write_location, ctx->write_xid, last_write);
590  ctx->prepared_write = true;
591 }
592 
593 /*
594  * Perform a write using the context's output routine.
595  */
596 void
598 {
599  if (!ctx->prepared_write)
600  elog(ERROR, "OutputPluginPrepareWrite needs to be called before OutputPluginWrite");
601 
602  ctx->write(ctx, ctx->write_location, ctx->write_xid, last_write);
603  ctx->prepared_write = false;
604 }
605 
606 /*
607  * Update progress tracking (if supported).
608  */
609 void
611 {
612  if (!ctx->update_progress)
613  return;
614 
615  ctx->update_progress(ctx, ctx->write_location, ctx->write_xid);
616 }
617 
618 /*
619  * Load the output plugin, lookup its output plugin init function, and check
620  * that it provides the required callbacks.
621  */
622 static void
624 {
625  LogicalOutputPluginInit plugin_init;
626 
627  plugin_init = (LogicalOutputPluginInit)
628  load_external_function(plugin, "_PG_output_plugin_init", false, NULL);
629 
630  if (plugin_init == NULL)
631  elog(ERROR, "output plugins have to declare the _PG_output_plugin_init symbol");
632 
633  /* ask the output plugin to fill the callback struct */
634  plugin_init(callbacks);
635 
636  if (callbacks->begin_cb == NULL)
637  elog(ERROR, "output plugins have to register a begin callback");
638  if (callbacks->change_cb == NULL)
639  elog(ERROR, "output plugins have to register a change callback");
640  if (callbacks->commit_cb == NULL)
641  elog(ERROR, "output plugins have to register a commit callback");
642 }
643 
644 static void
646 {
648 
649  /* not all callbacks have an associated LSN */
650  if (state->report_location != InvalidXLogRecPtr)
651  errcontext("slot \"%s\", output plugin \"%s\", in the %s callback, associated LSN %X/%X",
652  NameStr(state->ctx->slot->data.name),
653  NameStr(state->ctx->slot->data.plugin),
654  state->callback_name,
655  (uint32) (state->report_location >> 32),
656  (uint32) state->report_location);
657  else
658  errcontext("slot \"%s\", output plugin \"%s\", in the %s callback",
659  NameStr(state->ctx->slot->data.name),
660  NameStr(state->ctx->slot->data.plugin),
661  state->callback_name);
662 }
663 
664 static void
666 {
668  ErrorContextCallback errcallback;
669 
670  Assert(!ctx->fast_forward);
671 
672  /* Push callback + info on the error context stack */
673  state.ctx = ctx;
674  state.callback_name = "startup";
677  errcallback.arg = (void *) &state;
678  errcallback.previous = error_context_stack;
679  error_context_stack = &errcallback;
680 
681  /* set output state */
682  ctx->accept_writes = false;
683 
684  /* do the actual work: call callback */
685  ctx->callbacks.startup_cb(ctx, opt, is_init);
686 
687  /* Pop the error context stack */
688  error_context_stack = errcallback.previous;
689 }
690 
691 static void
693 {
695  ErrorContextCallback errcallback;
696 
697  Assert(!ctx->fast_forward);
698 
699  /* Push callback + info on the error context stack */
700  state.ctx = ctx;
701  state.callback_name = "shutdown";
704  errcallback.arg = (void *) &state;
705  errcallback.previous = error_context_stack;
706  error_context_stack = &errcallback;
707 
708  /* set output state */
709  ctx->accept_writes = false;
710 
711  /* do the actual work: call callback */
712  ctx->callbacks.shutdown_cb(ctx);
713 
714  /* Pop the error context stack */
715  error_context_stack = errcallback.previous;
716 }
717 
718 
719 /*
720  * Callbacks for ReorderBuffer which add in some more information and then call
721  * output_plugin.h plugins.
722  */
723 static void
725 {
728  ErrorContextCallback errcallback;
729 
730  Assert(!ctx->fast_forward);
731 
732  /* Push callback + info on the error context stack */
733  state.ctx = ctx;
734  state.callback_name = "begin";
735  state.report_location = txn->first_lsn;
737  errcallback.arg = (void *) &state;
738  errcallback.previous = error_context_stack;
739  error_context_stack = &errcallback;
740 
741  /* set output state */
742  ctx->accept_writes = true;
743  ctx->write_xid = txn->xid;
744  ctx->write_location = txn->first_lsn;
745 
746  /* do the actual work: call callback */
747  ctx->callbacks.begin_cb(ctx, txn);
748 
749  /* Pop the error context stack */
750  error_context_stack = errcallback.previous;
751 }
752 
753 static void
755  XLogRecPtr commit_lsn)
756 {
759  ErrorContextCallback errcallback;
760 
761  Assert(!ctx->fast_forward);
762 
763  /* Push callback + info on the error context stack */
764  state.ctx = ctx;
765  state.callback_name = "commit";
766  state.report_location = txn->final_lsn; /* beginning of commit record */
768  errcallback.arg = (void *) &state;
769  errcallback.previous = error_context_stack;
770  error_context_stack = &errcallback;
771 
772  /* set output state */
773  ctx->accept_writes = true;
774  ctx->write_xid = txn->xid;
775  ctx->write_location = txn->end_lsn; /* points to the end of the record */
776 
777  /* do the actual work: call callback */
778  ctx->callbacks.commit_cb(ctx, txn, commit_lsn);
779 
780  /* Pop the error context stack */
781  error_context_stack = errcallback.previous;
782 }
783 
784 static void
786  Relation relation, ReorderBufferChange *change)
787 {
790  ErrorContextCallback errcallback;
791 
792  Assert(!ctx->fast_forward);
793 
794  /* Push callback + info on the error context stack */
795  state.ctx = ctx;
796  state.callback_name = "change";
797  state.report_location = change->lsn;
799  errcallback.arg = (void *) &state;
800  errcallback.previous = error_context_stack;
801  error_context_stack = &errcallback;
802 
803  /* set output state */
804  ctx->accept_writes = true;
805  ctx->write_xid = txn->xid;
806 
807  /*
808  * report this change's lsn so replies from clients can give an up2date
809  * answer. This won't ever be enough (and shouldn't be!) to confirm
810  * receipt of this transaction, but it might allow another transaction's
811  * commit to be confirmed with one message.
812  */
813  ctx->write_location = change->lsn;
814 
815  ctx->callbacks.change_cb(ctx, txn, relation, change);
816 
817  /* Pop the error context stack */
818  error_context_stack = errcallback.previous;
819 }
820 
821 static void
823  int nrelations, Relation relations[], ReorderBufferChange *change)
824 {
827  ErrorContextCallback errcallback;
828 
829  Assert(!ctx->fast_forward);
830 
831  if (!ctx->callbacks.truncate_cb)
832  return;
833 
834  /* Push callback + info on the error context stack */
835  state.ctx = ctx;
836  state.callback_name = "truncate";
837  state.report_location = change->lsn;
839  errcallback.arg = (void *) &state;
840  errcallback.previous = error_context_stack;
841  error_context_stack = &errcallback;
842 
843  /* set output state */
844  ctx->accept_writes = true;
845  ctx->write_xid = txn->xid;
846 
847  /*
848  * report this change's lsn so replies from clients can give an up2date
849  * answer. This won't ever be enough (and shouldn't be!) to confirm
850  * receipt of this transaction, but it might allow another transaction's
851  * commit to be confirmed with one message.
852  */
853  ctx->write_location = change->lsn;
854 
855  ctx->callbacks.truncate_cb(ctx, txn, nrelations, relations, change);
856 
857  /* Pop the error context stack */
858  error_context_stack = errcallback.previous;
859 }
860 
861 bool
863 {
865  ErrorContextCallback errcallback;
866  bool ret;
867 
868  Assert(!ctx->fast_forward);
869 
870  /* Push callback + info on the error context stack */
871  state.ctx = ctx;
872  state.callback_name = "filter_by_origin";
875  errcallback.arg = (void *) &state;
876  errcallback.previous = error_context_stack;
877  error_context_stack = &errcallback;
878 
879  /* set output state */
880  ctx->accept_writes = false;
881 
882  /* do the actual work: call callback */
883  ret = ctx->callbacks.filter_by_origin_cb(ctx, origin_id);
884 
885  /* Pop the error context stack */
886  error_context_stack = errcallback.previous;
887 
888  return ret;
889 }
890 
891 static void
893  XLogRecPtr message_lsn, bool transactional,
894  const char *prefix, Size message_size, const char *message)
895 {
898  ErrorContextCallback errcallback;
899 
900  Assert(!ctx->fast_forward);
901 
902  if (ctx->callbacks.message_cb == NULL)
903  return;
904 
905  /* Push callback + info on the error context stack */
906  state.ctx = ctx;
907  state.callback_name = "message";
908  state.report_location = message_lsn;
910  errcallback.arg = (void *) &state;
911  errcallback.previous = error_context_stack;
912  error_context_stack = &errcallback;
913 
914  /* set output state */
915  ctx->accept_writes = true;
916  ctx->write_xid = txn != NULL ? txn->xid : InvalidTransactionId;
917  ctx->write_location = message_lsn;
918 
919  /* do the actual work: call callback */
920  ctx->callbacks.message_cb(ctx, txn, message_lsn, transactional, prefix,
921  message_size, message);
922 
923  /* Pop the error context stack */
924  error_context_stack = errcallback.previous;
925 }
926 
927 static void
929  XLogRecPtr first_lsn)
930 {
933  ErrorContextCallback errcallback;
934 
935  Assert(!ctx->fast_forward);
936 
937  /* We're only supposed to call this when streaming is supported. */
938  Assert(ctx->streaming);
939 
940  /* Push callback + info on the error context stack */
941  state.ctx = ctx;
942  state.callback_name = "stream_start";
943  state.report_location = first_lsn;
945  errcallback.arg = (void *) &state;
946  errcallback.previous = error_context_stack;
947  error_context_stack = &errcallback;
948 
949  /* set output state */
950  ctx->accept_writes = true;
951  ctx->write_xid = txn->xid;
952 
953  /*
954  * report this message's lsn so replies from clients can give an up2date
955  * answer. This won't ever be enough (and shouldn't be!) to confirm
956  * receipt of this transaction, but it might allow another transaction's
957  * commit to be confirmed with one message.
958  */
959  ctx->write_location = first_lsn;
960 
961  /* in streaming mode, stream_start_cb is required */
962  if (ctx->callbacks.stream_start_cb == NULL)
963  ereport(ERROR,
964  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
965  errmsg("logical streaming requires a stream_start_cb callback")));
966 
967  ctx->callbacks.stream_start_cb(ctx, txn);
968 
969  /* Pop the error context stack */
970  error_context_stack = errcallback.previous;
971 }
972 
973 static void
975  XLogRecPtr last_lsn)
976 {
979  ErrorContextCallback errcallback;
980 
981  Assert(!ctx->fast_forward);
982 
983  /* We're only supposed to call this when streaming is supported. */
984  Assert(ctx->streaming);
985 
986  /* Push callback + info on the error context stack */
987  state.ctx = ctx;
988  state.callback_name = "stream_stop";
989  state.report_location = last_lsn;
991  errcallback.arg = (void *) &state;
992  errcallback.previous = error_context_stack;
993  error_context_stack = &errcallback;
994 
995  /* set output state */
996  ctx->accept_writes = true;
997  ctx->write_xid = txn->xid;
998 
999  /*
1000  * report this message's lsn so replies from clients can give an up2date
1001  * answer. This won't ever be enough (and shouldn't be!) to confirm
1002  * receipt of this transaction, but it might allow another transaction's
1003  * commit to be confirmed with one message.
1004  */
1005  ctx->write_location = last_lsn;
1006 
1007  /* in streaming mode, stream_stop_cb is required */
1008  if (ctx->callbacks.stream_stop_cb == NULL)
1009  ereport(ERROR,
1010  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1011  errmsg("logical streaming requires a stream_stop_cb callback")));
1012 
1013  ctx->callbacks.stream_stop_cb(ctx, txn);
1014 
1015  /* Pop the error context stack */
1016  error_context_stack = errcallback.previous;
1017 }
1018 
1019 static void
1021  XLogRecPtr abort_lsn)
1022 {
1025  ErrorContextCallback errcallback;
1026 
1027  Assert(!ctx->fast_forward);
1028 
1029  /* We're only supposed to call this when streaming is supported. */
1030  Assert(ctx->streaming);
1031 
1032  /* Push callback + info on the error context stack */
1033  state.ctx = ctx;
1034  state.callback_name = "stream_abort";
1035  state.report_location = abort_lsn;
1036  errcallback.callback = output_plugin_error_callback;
1037  errcallback.arg = (void *) &state;
1038  errcallback.previous = error_context_stack;
1039  error_context_stack = &errcallback;
1040 
1041  /* set output state */
1042  ctx->accept_writes = true;
1043  ctx->write_xid = txn->xid;
1044  ctx->write_location = abort_lsn;
1045 
1046  /* in streaming mode, stream_abort_cb is required */
1047  if (ctx->callbacks.stream_abort_cb == NULL)
1048  ereport(ERROR,
1049  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1050  errmsg("logical streaming requires a stream_abort_cb callback")));
1051 
1052  ctx->callbacks.stream_abort_cb(ctx, txn, abort_lsn);
1053 
1054  /* Pop the error context stack */
1055  error_context_stack = errcallback.previous;
1056 }
1057 
1058 static void
1060  XLogRecPtr commit_lsn)
1061 {
1064  ErrorContextCallback errcallback;
1065 
1066  Assert(!ctx->fast_forward);
1067 
1068  /* We're only supposed to call this when streaming is supported. */
1069  Assert(ctx->streaming);
1070 
1071  /* Push callback + info on the error context stack */
1072  state.ctx = ctx;
1073  state.callback_name = "stream_commit";
1074  state.report_location = txn->final_lsn;
1075  errcallback.callback = output_plugin_error_callback;
1076  errcallback.arg = (void *) &state;
1077  errcallback.previous = error_context_stack;
1078  error_context_stack = &errcallback;
1079 
1080  /* set output state */
1081  ctx->accept_writes = true;
1082  ctx->write_xid = txn->xid;
1083  ctx->write_location = txn->end_lsn;
1084 
1085  /* in streaming mode, stream_abort_cb is required */
1086  if (ctx->callbacks.stream_commit_cb == NULL)
1087  ereport(ERROR,
1088  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1089  errmsg("logical streaming requires a stream_commit_cb callback")));
1090 
1091  ctx->callbacks.stream_commit_cb(ctx, txn, commit_lsn);
1092 
1093  /* Pop the error context stack */
1094  error_context_stack = errcallback.previous;
1095 }
1096 
1097 static void
1099  Relation relation, ReorderBufferChange *change)
1100 {
1103  ErrorContextCallback errcallback;
1104 
1105  Assert(!ctx->fast_forward);
1106 
1107  /* We're only supposed to call this when streaming is supported. */
1108  Assert(ctx->streaming);
1109 
1110  /* Push callback + info on the error context stack */
1111  state.ctx = ctx;
1112  state.callback_name = "stream_change";
1113  state.report_location = change->lsn;
1114  errcallback.callback = output_plugin_error_callback;
1115  errcallback.arg = (void *) &state;
1116  errcallback.previous = error_context_stack;
1117  error_context_stack = &errcallback;
1118 
1119  /* set output state */
1120  ctx->accept_writes = true;
1121  ctx->write_xid = txn->xid;
1122 
1123  /*
1124  * report this change's lsn so replies from clients can give an up2date
1125  * answer. This won't ever be enough (and shouldn't be!) to confirm
1126  * receipt of this transaction, but it might allow another transaction's
1127  * commit to be confirmed with one message.
1128  */
1129  ctx->write_location = change->lsn;
1130 
1131  /* in streaming mode, stream_change_cb is required */
1132  if (ctx->callbacks.stream_change_cb == NULL)
1133  ereport(ERROR,
1134  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1135  errmsg("logical streaming requires a stream_change_cb callback")));
1136 
1137  ctx->callbacks.stream_change_cb(ctx, txn, relation, change);
1138 
1139  /* Pop the error context stack */
1140  error_context_stack = errcallback.previous;
1141 }
1142 
1143 static void
1145  XLogRecPtr message_lsn, bool transactional,
1146  const char *prefix, Size message_size, const char *message)
1147 {
1150  ErrorContextCallback errcallback;
1151 
1152  Assert(!ctx->fast_forward);
1153 
1154  /* We're only supposed to call this when streaming is supported. */
1155  Assert(ctx->streaming);
1156 
1157  /* this callback is optional */
1158  if (ctx->callbacks.stream_message_cb == NULL)
1159  return;
1160 
1161  /* Push callback + info on the error context stack */
1162  state.ctx = ctx;
1163  state.callback_name = "stream_message";
1164  state.report_location = message_lsn;
1165  errcallback.callback = output_plugin_error_callback;
1166  errcallback.arg = (void *) &state;
1167  errcallback.previous = error_context_stack;
1168  error_context_stack = &errcallback;
1169 
1170  /* set output state */
1171  ctx->accept_writes = true;
1172  ctx->write_xid = txn != NULL ? txn->xid : InvalidTransactionId;
1173  ctx->write_location = message_lsn;
1174 
1175  /* do the actual work: call callback */
1176  ctx->callbacks.stream_message_cb(ctx, txn, message_lsn, transactional, prefix,
1177  message_size, message);
1178 
1179  /* Pop the error context stack */
1180  error_context_stack = errcallback.previous;
1181 }
1182 
1183 static void
1185  int nrelations, Relation relations[],
1186  ReorderBufferChange *change)
1187 {
1190  ErrorContextCallback errcallback;
1191 
1192  Assert(!ctx->fast_forward);
1193 
1194  /* We're only supposed to call this when streaming is supported. */
1195  Assert(ctx->streaming);
1196 
1197  /* this callback is optional */
1198  if (!ctx->callbacks.stream_truncate_cb)
1199  return;
1200 
1201  /* Push callback + info on the error context stack */
1202  state.ctx = ctx;
1203  state.callback_name = "stream_truncate";
1204  state.report_location = change->lsn;
1205  errcallback.callback = output_plugin_error_callback;
1206  errcallback.arg = (void *) &state;
1207  errcallback.previous = error_context_stack;
1208  error_context_stack = &errcallback;
1209 
1210  /* set output state */
1211  ctx->accept_writes = true;
1212  ctx->write_xid = txn->xid;
1213 
1214  /*
1215  * report this change's lsn so replies from clients can give an up2date
1216  * answer. This won't ever be enough (and shouldn't be!) to confirm
1217  * receipt of this transaction, but it might allow another transaction's
1218  * commit to be confirmed with one message.
1219  */
1220  ctx->write_location = change->lsn;
1221 
1222  ctx->callbacks.stream_truncate_cb(ctx, txn, nrelations, relations, change);
1223 
1224  /* Pop the error context stack */
1225  error_context_stack = errcallback.previous;
1226 }
1227 
1228 /*
1229  * Set the required catalog xmin horizon for historic snapshots in the current
1230  * replication slot.
1231  *
1232  * Note that in the most cases, we won't be able to immediately use the xmin
1233  * to increase the xmin horizon: we need to wait till the client has confirmed
1234  * receiving current_lsn with LogicalConfirmReceivedLocation().
1235  */
1236 void
1238 {
1239  bool updated_xmin = false;
1240  ReplicationSlot *slot;
1241 
1242  slot = MyReplicationSlot;
1243 
1244  Assert(slot != NULL);
1245 
1246  SpinLockAcquire(&slot->mutex);
1247 
1248  /*
1249  * don't overwrite if we already have a newer xmin. This can happen if we
1250  * restart decoding in a slot.
1251  */
1253  {
1254  }
1255 
1256  /*
1257  * If the client has already confirmed up to this lsn, we directly can
1258  * mark this as accepted. This can happen if we restart decoding in a
1259  * slot.
1260  */
1261  else if (current_lsn <= slot->data.confirmed_flush)
1262  {
1263  slot->candidate_catalog_xmin = xmin;
1264  slot->candidate_xmin_lsn = current_lsn;
1265 
1266  /* our candidate can directly be used */
1267  updated_xmin = true;
1268  }
1269 
1270  /*
1271  * Only increase if the previous values have been applied, otherwise we
1272  * might never end up updating if the receiver acks too slowly.
1273  */
1274  else if (slot->candidate_xmin_lsn == InvalidXLogRecPtr)
1275  {
1276  slot->candidate_catalog_xmin = xmin;
1277  slot->candidate_xmin_lsn = current_lsn;
1278  }
1279  SpinLockRelease(&slot->mutex);
1280 
1281  /* candidate already valid with the current flush position, apply */
1282  if (updated_xmin)
1284 }
1285 
1286 /*
1287  * Mark the minimal LSN (restart_lsn) we need to read to replay all
1288  * transactions that have not yet committed at current_lsn.
1289  *
1290  * Just like LogicalIncreaseXminForSlot this only takes effect when the
1291  * client has confirmed to have received current_lsn.
1292  */
1293 void
1295 {
1296  bool updated_lsn = false;
1297  ReplicationSlot *slot;
1298 
1299  slot = MyReplicationSlot;
1300 
1301  Assert(slot != NULL);
1302  Assert(restart_lsn != InvalidXLogRecPtr);
1303  Assert(current_lsn != InvalidXLogRecPtr);
1304 
1305  SpinLockAcquire(&slot->mutex);
1306 
1307  /* don't overwrite if have a newer restart lsn */
1308  if (restart_lsn <= slot->data.restart_lsn)
1309  {
1310  }
1311 
1312  /*
1313  * We might have already flushed far enough to directly accept this lsn,
1314  * in this case there is no need to check for existing candidate LSNs
1315  */
1316  else if (current_lsn <= slot->data.confirmed_flush)
1317  {
1318  slot->candidate_restart_valid = current_lsn;
1319  slot->candidate_restart_lsn = restart_lsn;
1320 
1321  /* our candidate can directly be used */
1322  updated_lsn = true;
1323  }
1324 
1325  /*
1326  * Only increase if the previous values have been applied, otherwise we
1327  * might never end up updating if the receiver acks too slowly. A missed
1328  * value here will just cause some extra effort after reconnecting.
1329  */
1331  {
1332  slot->candidate_restart_valid = current_lsn;
1333  slot->candidate_restart_lsn = restart_lsn;
1334  SpinLockRelease(&slot->mutex);
1335 
1336  elog(DEBUG1, "got new restart lsn %X/%X at %X/%X",
1337  (uint32) (restart_lsn >> 32), (uint32) restart_lsn,
1338  (uint32) (current_lsn >> 32), (uint32) current_lsn);
1339  }
1340  else
1341  {
1342  XLogRecPtr candidate_restart_lsn;
1343  XLogRecPtr candidate_restart_valid;
1344  XLogRecPtr confirmed_flush;
1345 
1346  candidate_restart_lsn = slot->candidate_restart_lsn;
1347  candidate_restart_valid = slot->candidate_restart_valid;
1348  confirmed_flush = slot->data.confirmed_flush;
1349  SpinLockRelease(&slot->mutex);
1350 
1351  elog(DEBUG1, "failed to increase restart lsn: proposed %X/%X, after %X/%X, current candidate %X/%X, current after %X/%X, flushed up to %X/%X",
1352  (uint32) (restart_lsn >> 32), (uint32) restart_lsn,
1353  (uint32) (current_lsn >> 32), (uint32) current_lsn,
1354  (uint32) (candidate_restart_lsn >> 32),
1355  (uint32) candidate_restart_lsn,
1356  (uint32) (candidate_restart_valid >> 32),
1357  (uint32) candidate_restart_valid,
1358  (uint32) (confirmed_flush >> 32),
1359  (uint32) confirmed_flush);
1360  }
1361 
1362  /* candidates are already valid with the current flush position, apply */
1363  if (updated_lsn)
1365 }
1366 
1367 /*
1368  * Handle a consumer's confirmation having received all changes up to lsn.
1369  */
1370 void
1372 {
1373  Assert(lsn != InvalidXLogRecPtr);
1374 
1375  /* Do an unlocked check for candidate_lsn first. */
1378  {
1379  bool updated_xmin = false;
1380  bool updated_restart = false;
1381 
1383 
1385 
1386  /* if we're past the location required for bumping xmin, do so */
1389  {
1390  /*
1391  * We have to write the changed xmin to disk *before* we change
1392  * the in-memory value, otherwise after a crash we wouldn't know
1393  * that some catalog tuples might have been removed already.
1394  *
1395  * Ensure that by first writing to ->xmin and only update
1396  * ->effective_xmin once the new state is synced to disk. After a
1397  * crash ->effective_xmin is set to ->xmin.
1398  */
1401  {
1405  updated_xmin = true;
1406  }
1407  }
1408 
1411  {
1413 
1417  updated_restart = true;
1418  }
1419 
1421 
1422  /* first write new xmin to disk, so we know what's up after a crash */
1423  if (updated_xmin || updated_restart)
1424  {
1427  elog(DEBUG1, "updated xmin: %u restart: %u", updated_xmin, updated_restart);
1428  }
1429 
1430  /*
1431  * Now the new xmin is safely on disk, we can let the global value
1432  * advance. We do not take ProcArrayLock or similar since we only
1433  * advance xmin here and there's not much harm done by a concurrent
1434  * computation missing that.
1435  */
1436  if (updated_xmin)
1437  {
1441 
1444  }
1445  }
1446  else
1447  {
1451  }
1452 }
1453 
1454 /*
1455  * Clear logical streaming state during (sub)transaction abort.
1456  */
1457 void
1459 {
1461  bsysscan = false;
1462 }
XLogRecPtr first_lsn
LogicalDecodeTruncateCB truncate_cb
#define NIL
Definition: pg_list.h:65
TransactionId write_xid
Definition: logical.h:93
static void stream_stop_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr last_lsn)
Definition: logical.c:974
static const char * plugin
void CheckSlotRequirements(void)
Definition: slot.c:1032
TransactionId candidate_catalog_xmin
Definition: slot.h:158
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
static void stream_abort_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr abort_lsn)
Definition: logical.c:1020
TransactionId GetOldestSafeDecodingTransactionId(bool catalogOnly)
Definition: procarray.c:2794
ReorderBufferApplyChangeCB apply_change
void MemoryContextDelete(MemoryContext context)
Definition: mcxt.c:212
#define AllocSetContextCreate
Definition: memutils.h:170
void * private_data
#define DEBUG1
Definition: elog.h:25
static void change_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, Relation relation, ReorderBufferChange *change)
Definition: logical.c:785
struct ReorderBuffer * reorder
Definition: logical.h:42
#define PROC_IN_LOGICAL_DECODING
Definition: proc.h:57
int wal_segment_size
Definition: xlog.c:117
uint32 TransactionId
Definition: c.h:520
SnapBuildState SnapBuildCurrentState(SnapBuild *builder)
Definition: snapbuild.c:395
void namestrcpy(Name name, const char *str)
Definition: name.c:233
bool DecodingContextReady(LogicalDecodingContext *ctx)
Definition: logical.c:517
PGPROC * MyProc
Definition: proc.c:67
struct LogicalErrorCallbackState LogicalErrorCallbackState
static void commit_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr commit_lsn)
Definition: logical.c:754
OutputPluginOptions options
Definition: logical.h:53
void LogicalIncreaseRestartDecodingForSlot(XLogRecPtr current_lsn, XLogRecPtr restart_lsn)
Definition: logical.c:1294
ReorderBufferStreamAbortCB stream_abort
int wal_level
Definition: xlog.c:107
StringInfo makeStringInfo(void)
Definition: stringinfo.c:41
LogicalDecodeMessageCB message_cb
OutputPluginCallbacks callbacks
Definition: logical.h:52
LogicalDecodeStreamMessageCB stream_message_cb
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
Definition: palloc.h:109
void ReorderBufferFree(ReorderBuffer *rb)
uint16 RepOriginId
Definition: xlogdefs.h:58
int errcode(int sqlerrcode)
Definition: elog.c:610
bool IsTransactionOrTransactionBlock(void)
Definition: xact.c:4702
PROC_HDR * ProcGlobal
Definition: proc.c:79
MemoryContext context
Definition: logical.h:35
void ReplicationSlotSave(void)
Definition: slot.c:698
LogicalDecodeStreamAbortCB stream_abort_cb
List * output_plugin_options
Definition: logical.h:58
ReorderBufferCommitCB commit
ReplicationSlotPersistentData data
Definition: slot.h:143
#define LOG
Definition: elog.h:26
bool RecoveryInProgress(void)
Definition: xlog.c:8074
ReorderBufferStreamCommitCB stream_commit
void(* callback)(void *arg)
Definition: elog.h:229
struct ErrorContextCallback * previous
Definition: elog.h:228
#define SlotIsPhysical(slot)
Definition: slot.h:164
LogicalOutputPluginWriterWrite LogicalOutputPluginWriterPrepareWrite
Definition: logical.h:25
void(* LogicalOutputPluginWriterUpdateProgress)(struct LogicalDecodingContext *lr, XLogRecPtr Ptr, TransactionId xid)
Definition: logical.h:27
const char * callback_name
Definition: logical.c:49
XLogRecPtr confirmed_flush
Definition: slot.h:91
XLogRecPtr write_location
Definition: logical.h:92
XLogRecPtr EndRecPtr
Definition: xlogreader.h:176
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1812
ErrorContextCallback * error_context_stack
Definition: elog.c:92
#define SpinLockAcquire(lock)
Definition: spin.h:62
ReorderBufferStreamMessageCB stream_message
void ReplicationSlotReserveWal(void)
Definition: slot.c:1057
void DecodingContextFindStartpoint(LogicalDecodingContext *ctx)
Definition: logical.c:526
void ReplicationSlotsComputeRequiredLSN(void)
Definition: slot.c:805
XLogRecord * XLogReadRecord(XLogReaderState *state, char **errormsg)
Definition: xlogreader.c:268
static void stream_commit_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr commit_lsn)
Definition: logical.c:1059
bool TransactionIdPrecedesOrEquals(TransactionId id1, TransactionId id2)
Definition: transam.c:319
#define ERROR
Definition: elog.h:43
void LogicalDecodingProcessRecord(LogicalDecodingContext *ctx, XLogReaderState *record)
Definition: decode.c:94
static void startup_cb_wrapper(LogicalDecodingContext *ctx, OutputPluginOptions *opt, bool is_init)
Definition: logical.c:665
LogicalOutputPluginWriterPrepareWrite prepare_write
Definition: logical.h:63
bool bsysscan
Definition: xact.c:96
void ResetLogicalStreamingState(void)
Definition: logical.c:1458
LogicalDecodeCommitCB commit_cb
TransactionId effective_xmin
Definition: slot.h:139
void(* LogicalOutputPluginInit)(struct OutputPluginCallbacks *cb)
Definition: output_plugin.h:36
#define ALLOCSET_DEFAULT_SIZES
Definition: memutils.h:192
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:434
static void shutdown_cb_wrapper(LogicalDecodingContext *ctx)
Definition: logical.c:692
Definition: c.h:616
XLogRecPtr candidate_restart_valid
Definition: slot.h:160
int errdetail(const char *fmt,...)
Definition: elog.c:957
TransactionId catalog_xmin
Definition: slot.h:77
#define InvalidTransactionId
Definition: transam.h:31
void XLogBeginRead(XLogReaderState *state, XLogRecPtr RecPtr)
Definition: xlogreader.c:240
LogicalOutputPluginWriterUpdateProgress update_progress
Definition: logical.h:65
unsigned int uint32
Definition: c.h:374
XLogRecPtr final_lsn
MemoryContext CurrentMemoryContext
Definition: mcxt.c:38
bool filter_by_origin_cb_wrapper(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: logical.c:862
XLogReaderState * XLogReaderAllocate(int wal_segment_size, const char *waldir, XLogReaderRoutine *routine, void *private_data)
Definition: xlogreader.c:75
TransactionId GetTopTransactionIdIfAny(void)
Definition: xact.c:425
ReorderBufferMessageCB message
ReorderBufferStreamChangeCB stream_change
ReorderBufferStreamTruncateCB stream_truncate
void * load_external_function(const char *filename, const char *funcname, bool signalNotFound, void **filehandle)
Definition: dfmgr.c:107
#define XLogRecPtrIsInvalid(r)
Definition: xlogdefs.h:29
static void stream_message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr message_lsn, bool transactional, const char *prefix, Size message_size, const char *message)
Definition: logical.c:1144
TransactionId CheckXidAlive
Definition: xact.c:95
void XLogReaderFree(XLogReaderState *state)
Definition: xlogreader.c:139
#define SpinLockRelease(lock)
Definition: spin.h:64
void OutputPluginUpdateProgress(struct LogicalDecodingContext *ctx)
Definition: logical.c:610
void * palloc0(Size size)
Definition: mcxt.c:981
static void stream_start_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr first_lsn)
Definition: logical.c:928
LogicalDecodeChangeCB change_cb
TransactionId effective_catalog_xmin
Definition: slot.h:140
LogicalDecodeStreamTruncateCB stream_truncate_cb
TransactionId xid
Oid MyDatabaseId
Definition: globals.c:85
static void stream_truncate_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, int nrelations, Relation relations[], ReorderBufferChange *change)
Definition: logical.c:1184
#define InvalidOid
Definition: postgres_ext.h:36
ReorderBufferStreamStartCB stream_start
#define ereport(elevel,...)
Definition: elog.h:144
void FreeSnapshotBuilder(SnapBuild *builder)
Definition: snapbuild.c:351
static void begin_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn)
Definition: logical.c:724
void OutputPluginPrepareWrite(struct LogicalDecodingContext *ctx, bool last_write)
Definition: logical.c:584
struct SnapBuild * snapshot_builder
Definition: logical.h:43
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
uint64 XLogRecPtr
Definition: xlogdefs.h:21
#define Assert(condition)
Definition: c.h:745
Definition: regguts.h:298
void FreeDecodingContext(LogicalDecodingContext *ctx)
Definition: logical.c:569
XLogRecPtr restart_lsn
Definition: slot.h:80
void LogicalIncreaseXminForSlot(XLogRecPtr current_lsn, TransactionId xmin)
Definition: logical.c:1237
XLogRecPtr end_lsn
size_t Size
Definition: c.h:473
LogicalDecodeShutdownCB shutdown_cb
LogicalDecodeStreamCommitCB stream_commit_cb
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1208
bool IsTransactionState(void)
Definition: xact.c:371
ReorderBuffer * ReorderBufferAllocate(void)
void(* LogicalOutputPluginWriterWrite)(struct LogicalDecodingContext *lr, XLogRecPtr Ptr, TransactionId xid, bool last_write)
Definition: logical.h:19
void LogicalConfirmReceivedLocation(XLogRecPtr lsn)
Definition: logical.c:1371
LogicalDecodeStartupCB startup_cb
XLogRecPtr candidate_xmin_lsn
Definition: slot.h:159
XLogRecPtr report_location
Definition: logical.c:50
LogicalDecodeStreamStartCB stream_start_cb
static void truncate_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, int nrelations, Relation relations[], ReorderBufferChange *change)
Definition: logical.c:822
int errmsg(const char *fmt,...)
Definition: elog.c:824
XLogReaderState * reader
Definition: logical.h:41
ReplicationSlot * slot
Definition: logical.h:38
ReorderBufferApplyTruncateCB apply_truncate
void OutputPluginWrite(struct LogicalDecodingContext *ctx, bool last_write)
Definition: logical.c:597
LogicalOutputPluginWriterWrite write
Definition: logical.h:64
#define elog(elevel,...)
Definition: elog.h:214
StringInfo out
Definition: logical.h:70
int pgxactoff
Definition: proc.h:139
uint8 * vacuumFlags
Definition: proc.h:321
#define errcontext
Definition: elog.h:185
#define NameStr(name)
Definition: c.h:622
void * arg
uint8 vacuumFlags
Definition: proc.h:178
LogicalDecodeBeginCB begin_cb
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:99
static void message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr message_lsn, bool transactional, const char *prefix, Size message_size, const char *message)
Definition: logical.c:892
LogicalDecodeStreamStopCB stream_stop_cb
slock_t mutex
Definition: slot.h:116
#define TransactionIdIsValid(xid)
Definition: transam.h:41
LogicalDecodingContext * ctx
Definition: logical.c:48
ReorderBufferBeginCB begin
static void output_plugin_error_callback(void *arg)
Definition: logical.c:645
static void LoadOutputPlugin(OutputPluginCallbacks *callbacks, const char *plugin)
Definition: logical.c:623
LogicalDecodeFilterByOriginCB filter_by_origin_cb
Definition: pg_list.h:50
void CheckLogicalDecodingRequirements(void)
Definition: logical.c:93
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:282
LogicalDecodeStreamChangeCB stream_change_cb
SnapBuild * AllocateSnapshotBuilder(ReorderBuffer *reorder, TransactionId xmin_horizon, XLogRecPtr start_lsn, bool need_full_snapshot)
Definition: snapbuild.c:310
void ReplicationSlotsComputeRequiredXmin(bool already_locked)
Definition: slot.c:755
XLogRecPtr candidate_restart_lsn
Definition: slot.h:161
void ReplicationSlotMarkDirty(void)
Definition: slot.c:716
static void stream_change_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, Relation relation, ReorderBufferChange *change)
Definition: logical.c:1098
static LogicalDecodingContext * StartupDecodingContext(List *output_plugin_options, XLogRecPtr start_lsn, TransactionId xmin_horizon, bool need_full_snapshot, bool fast_forward, XLogReaderRoutine *xl_routine, LogicalOutputPluginWriterPrepareWrite prepare_write, LogicalOutputPluginWriterWrite do_write, LogicalOutputPluginWriterUpdateProgress update_progress)
Definition: logical.c:136
ReorderBufferStreamStopCB stream_stop