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-2022, 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 "pgstat.h"
36 #include "replication/decode.h"
37 #include "replication/logical.h"
38 #include "replication/origin.h"
40 #include "replication/snapbuild.h"
41 #include "storage/proc.h"
42 #include "storage/procarray.h"
43 #include "utils/builtins.h"
44 #include "utils/memutils.h"
45 
46 /* data for errcontext callback */
48 {
50  const char *callback_name;
53 
54 /* wrappers around output plugin callbacks */
55 static void output_plugin_error_callback(void *arg);
57  bool is_init);
59 static void begin_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn);
60 static void commit_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
61  XLogRecPtr commit_lsn);
63 static void prepare_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
64  XLogRecPtr prepare_lsn);
66  XLogRecPtr commit_lsn);
68  XLogRecPtr prepare_end_lsn, TimestampTz prepare_time);
69 static void change_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
70  Relation relation, ReorderBufferChange *change);
71 static void truncate_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
72  int nrelations, Relation relations[], ReorderBufferChange *change);
73 static void message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
74  XLogRecPtr message_lsn, bool transactional,
75  const char *prefix, Size message_size, const char *message);
76 
77 /* streaming callbacks */
79  XLogRecPtr first_lsn);
81  XLogRecPtr last_lsn);
83  XLogRecPtr abort_lsn);
85  XLogRecPtr prepare_lsn);
87  XLogRecPtr commit_lsn);
89  Relation relation, ReorderBufferChange *change);
91  XLogRecPtr message_lsn, bool transactional,
92  const char *prefix, Size message_size, const char *message);
94  int nrelations, Relation relations[], ReorderBufferChange *change);
95 
96 static void LoadOutputPlugin(OutputPluginCallbacks *callbacks, const char *plugin);
97 
98 /*
99  * Make sure the current settings & environment are capable of doing logical
100  * decoding.
101  */
102 void
104 {
106 
107  /*
108  * NB: Adding a new requirement likely means that RestoreSlotFromDisk()
109  * needs the same check.
110  */
111 
113  ereport(ERROR,
114  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
115  errmsg("logical decoding requires wal_level >= logical")));
116 
117  if (MyDatabaseId == InvalidOid)
118  ereport(ERROR,
119  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
120  errmsg("logical decoding requires a database connection")));
121 
122  /* ----
123  * TODO: We got to change that someday soon...
124  *
125  * There's basically three things missing to allow this:
126  * 1) We need to be able to correctly and quickly identify the timeline a
127  * LSN belongs to
128  * 2) We need to force hot_standby_feedback to be enabled at all times so
129  * the primary cannot remove rows we need.
130  * 3) support dropping replication slots referring to a database, in
131  * dbase_redo. There can't be any active ones due to HS recovery
132  * conflicts, so that should be relatively easy.
133  * ----
134  */
135  if (RecoveryInProgress())
136  ereport(ERROR,
137  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
138  errmsg("logical decoding cannot be used while in recovery")));
139 }
140 
141 /*
142  * Helper function for CreateInitDecodingContext() and
143  * CreateDecodingContext() performing common tasks.
144  */
145 static LogicalDecodingContext *
146 StartupDecodingContext(List *output_plugin_options,
147  XLogRecPtr start_lsn,
148  TransactionId xmin_horizon,
149  bool need_full_snapshot,
150  bool fast_forward,
151  XLogReaderRoutine *xl_routine,
155 {
156  ReplicationSlot *slot;
157  MemoryContext context,
158  old_context;
160 
161  /* shorter lines... */
162  slot = MyReplicationSlot;
163 
165  "Logical decoding context",
167  old_context = MemoryContextSwitchTo(context);
168  ctx = palloc0(sizeof(LogicalDecodingContext));
169 
170  ctx->context = context;
171 
172  /*
173  * (re-)load output plugins, so we detect a bad (removed) output plugin
174  * now.
175  */
176  if (!fast_forward)
178 
179  /*
180  * Now that the slot's xmin has been set, we can announce ourselves as a
181  * logical decoding backend which doesn't need to be checked individually
182  * when computing the xmin horizon because the xmin is enforced via
183  * replication slots.
184  *
185  * We can only do so if we're outside of a transaction (i.e. the case when
186  * streaming changes via walsender), otherwise an already setup
187  * snapshot/xid would end up being ignored. That's not a particularly
188  * bothersome restriction since the SQL interface can't be used for
189  * streaming anyway.
190  */
192  {
193  LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
196  LWLockRelease(ProcArrayLock);
197  }
198 
199  ctx->slot = slot;
200 
201  ctx->reader = XLogReaderAllocate(wal_segment_size, NULL, xl_routine, ctx);
202  if (!ctx->reader)
203  ereport(ERROR,
204  (errcode(ERRCODE_OUT_OF_MEMORY),
205  errmsg("out of memory"),
206  errdetail("Failed while allocating a WAL reading processor.")));
207 
209  ctx->snapshot_builder =
210  AllocateSnapshotBuilder(ctx->reorder, xmin_horizon, start_lsn,
211  need_full_snapshot, slot->data.two_phase_at);
212 
213  ctx->reorder->private_data = ctx;
214 
215  /* wrap output plugin callbacks, so we can add error context information */
221 
222  /*
223  * To support streaming, we require start/stop/abort/commit/change
224  * callbacks. The message and truncate callbacks are optional, similar to
225  * regular output plugins. We however enable streaming when at least one
226  * of the methods is enabled so that we can easily identify missing
227  * methods.
228  *
229  * We decide it here, but only check it later in the wrappers.
230  */
231  ctx->streaming = (ctx->callbacks.stream_start_cb != NULL) ||
232  (ctx->callbacks.stream_stop_cb != NULL) ||
233  (ctx->callbacks.stream_abort_cb != NULL) ||
234  (ctx->callbacks.stream_commit_cb != NULL) ||
235  (ctx->callbacks.stream_change_cb != NULL) ||
236  (ctx->callbacks.stream_message_cb != NULL) ||
237  (ctx->callbacks.stream_truncate_cb != NULL);
238 
239  /*
240  * streaming callbacks
241  *
242  * stream_message and stream_truncate callbacks are optional, so we do not
243  * fail with ERROR when missing, but the wrappers simply do nothing. We
244  * must set the ReorderBuffer callbacks to something, otherwise the calls
245  * from there will crash (we don't want to move the checks there).
246  */
255 
256 
257  /*
258  * To support two-phase logical decoding, we require
259  * begin_prepare/prepare/commit-prepare/abort-prepare callbacks. The
260  * filter_prepare callback is optional. We however enable two-phase
261  * logical decoding when at least one of the methods is enabled so that we
262  * can easily identify missing methods.
263  *
264  * We decide it here, but only check it later in the wrappers.
265  */
266  ctx->twophase = (ctx->callbacks.begin_prepare_cb != NULL) ||
267  (ctx->callbacks.prepare_cb != NULL) ||
268  (ctx->callbacks.commit_prepared_cb != NULL) ||
269  (ctx->callbacks.rollback_prepared_cb != NULL) ||
270  (ctx->callbacks.stream_prepare_cb != NULL) ||
271  (ctx->callbacks.filter_prepare_cb != NULL);
272 
273  /*
274  * Callback to support decoding at prepare time.
275  */
280 
281  ctx->out = makeStringInfo();
282  ctx->prepare_write = prepare_write;
283  ctx->write = do_write;
284  ctx->update_progress = update_progress;
285 
286  ctx->output_plugin_options = output_plugin_options;
287 
288  ctx->fast_forward = fast_forward;
289 
290  MemoryContextSwitchTo(old_context);
291 
292  return ctx;
293 }
294 
295 /*
296  * Create a new decoding context, for a new logical slot.
297  *
298  * plugin -- contains the name of the output plugin
299  * output_plugin_options -- contains options passed to the output plugin
300  * need_full_snapshot -- if true, must obtain a snapshot able to read all
301  * tables; if false, one that can read only catalogs is acceptable.
302  * restart_lsn -- if given as invalid, it's this routine's responsibility to
303  * mark WAL as reserved by setting a convenient restart_lsn for the slot.
304  * Otherwise, we set for decoding to start from the given LSN without
305  * marking WAL reserved beforehand. In that scenario, it's up to the
306  * caller to guarantee that WAL remains available.
307  * xl_routine -- XLogReaderRoutine for underlying XLogReader
308  * prepare_write, do_write, update_progress --
309  * callbacks that perform the use-case dependent, actual, work.
310  *
311  * Needs to be called while in a memory context that's at least as long lived
312  * as the decoding context because further memory contexts will be created
313  * inside it.
314  *
315  * Returns an initialized decoding context after calling the output plugin's
316  * startup function.
317  */
320  List *output_plugin_options,
321  bool need_full_snapshot,
322  XLogRecPtr restart_lsn,
323  XLogReaderRoutine *xl_routine,
327 {
328  TransactionId xmin_horizon = InvalidTransactionId;
329  ReplicationSlot *slot;
330  NameData plugin_name;
332  MemoryContext old_context;
333 
334  /* shorter lines... */
335  slot = MyReplicationSlot;
336 
337  /* first some sanity checks that are unlikely to be violated */
338  if (slot == NULL)
339  elog(ERROR, "cannot perform logical decoding without an acquired slot");
340 
341  if (plugin == NULL)
342  elog(ERROR, "cannot initialize logical decoding without a specified plugin");
343 
344  /* Make sure the passed slot is suitable. These are user facing errors. */
345  if (SlotIsPhysical(slot))
346  ereport(ERROR,
347  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
348  errmsg("cannot use physical replication slot for logical decoding")));
349 
350  if (slot->data.database != MyDatabaseId)
351  ereport(ERROR,
352  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
353  errmsg("replication slot \"%s\" was not created in this database",
354  NameStr(slot->data.name))));
355 
356  if (IsTransactionState() &&
358  ereport(ERROR,
359  (errcode(ERRCODE_ACTIVE_SQL_TRANSACTION),
360  errmsg("cannot create logical replication slot in transaction that has performed writes")));
361 
362  /*
363  * Register output plugin name with slot. We need the mutex to avoid
364  * concurrent reading of a partially copied string. But we don't want any
365  * complicated code while holding a spinlock, so do namestrcpy() outside.
366  */
367  namestrcpy(&plugin_name, plugin);
368  SpinLockAcquire(&slot->mutex);
369  slot->data.plugin = plugin_name;
370  SpinLockRelease(&slot->mutex);
371 
372  if (XLogRecPtrIsInvalid(restart_lsn))
374  else
375  {
376  SpinLockAcquire(&slot->mutex);
377  slot->data.restart_lsn = restart_lsn;
378  SpinLockRelease(&slot->mutex);
379  }
380 
381  /* ----
382  * This is a bit tricky: We need to determine a safe xmin horizon to start
383  * decoding from, to avoid starting from a running xacts record referring
384  * to xids whose rows have been vacuumed or pruned
385  * already. GetOldestSafeDecodingTransactionId() returns such a value, but
386  * without further interlock its return value might immediately be out of
387  * date.
388  *
389  * So we have to acquire the ProcArrayLock to prevent computation of new
390  * xmin horizons by other backends, get the safe decoding xid, and inform
391  * the slot machinery about the new limit. Once that's done the
392  * ProcArrayLock can be released as the slot machinery now is
393  * protecting against vacuum.
394  *
395  * Note that, temporarily, the data, not just the catalog, xmin has to be
396  * reserved if a data snapshot is to be exported. Otherwise the initial
397  * data snapshot created here is not guaranteed to be valid. After that
398  * the data xmin doesn't need to be managed anymore and the global xmin
399  * should be recomputed. As we are fine with losing the pegged data xmin
400  * after crash - no chance a snapshot would get exported anymore - we can
401  * get away with just setting the slot's
402  * effective_xmin. ReplicationSlotRelease will reset it again.
403  *
404  * ----
405  */
406  LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
407 
408  xmin_horizon = GetOldestSafeDecodingTransactionId(!need_full_snapshot);
409 
410  SpinLockAcquire(&slot->mutex);
411  slot->effective_catalog_xmin = xmin_horizon;
412  slot->data.catalog_xmin = xmin_horizon;
413  if (need_full_snapshot)
414  slot->effective_xmin = xmin_horizon;
415  SpinLockRelease(&slot->mutex);
416 
418 
419  LWLockRelease(ProcArrayLock);
420 
423 
424  ctx = StartupDecodingContext(NIL, restart_lsn, xmin_horizon,
425  need_full_snapshot, false,
426  xl_routine, prepare_write, do_write,
427  update_progress);
428 
429  /* call output plugin initialization callback */
430  old_context = MemoryContextSwitchTo(ctx->context);
431  if (ctx->callbacks.startup_cb != NULL)
432  startup_cb_wrapper(ctx, &ctx->options, true);
433  MemoryContextSwitchTo(old_context);
434 
435  /*
436  * We allow decoding of prepared transactions when the two_phase is
437  * enabled at the time of slot creation, or when the two_phase option is
438  * given at the streaming start, provided the plugin supports all the
439  * callbacks for two-phase.
440  */
441  ctx->twophase &= slot->data.two_phase;
442 
444 
445  return ctx;
446 }
447 
448 /*
449  * Create a new decoding context, for a logical slot that has previously been
450  * used already.
451  *
452  * start_lsn
453  * The LSN at which to start decoding. If InvalidXLogRecPtr, restart
454  * from the slot's confirmed_flush; otherwise, start from the specified
455  * location (but move it forwards to confirmed_flush if it's older than
456  * that, see below).
457  *
458  * output_plugin_options
459  * options passed to the output plugin.
460  *
461  * fast_forward
462  * bypass the generation of logical changes.
463  *
464  * xl_routine
465  * XLogReaderRoutine used by underlying xlogreader
466  *
467  * prepare_write, do_write, update_progress
468  * callbacks that have to be filled to perform the use-case dependent,
469  * actual work.
470  *
471  * Needs to be called while in a memory context that's at least as long lived
472  * as the decoding context because further memory contexts will be created
473  * inside it.
474  *
475  * Returns an initialized decoding context after calling the output plugin's
476  * startup function.
477  */
480  List *output_plugin_options,
481  bool fast_forward,
482  XLogReaderRoutine *xl_routine,
486 {
488  ReplicationSlot *slot;
489  MemoryContext old_context;
490 
491  /* shorter lines... */
492  slot = MyReplicationSlot;
493 
494  /* first some sanity checks that are unlikely to be violated */
495  if (slot == NULL)
496  elog(ERROR, "cannot perform logical decoding without an acquired slot");
497 
498  /* make sure the passed slot is suitable, these are user facing errors */
499  if (SlotIsPhysical(slot))
500  ereport(ERROR,
501  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
502  errmsg("cannot use physical replication slot for logical decoding")));
503 
504  if (slot->data.database != MyDatabaseId)
505  ereport(ERROR,
506  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
507  errmsg("replication slot \"%s\" was not created in this database",
508  NameStr(slot->data.name))));
509 
510  if (start_lsn == InvalidXLogRecPtr)
511  {
512  /* continue from last position */
513  start_lsn = slot->data.confirmed_flush;
514  }
515  else if (start_lsn < slot->data.confirmed_flush)
516  {
517  /*
518  * It might seem like we should error out in this case, but it's
519  * pretty common for a client to acknowledge a LSN it doesn't have to
520  * do anything for, and thus didn't store persistently, because the
521  * xlog records didn't result in anything relevant for logical
522  * decoding. Clients have to be able to do that to support synchronous
523  * replication.
524  *
525  * Starting at a different LSN than requested might not catch certain
526  * kinds of client errors; so the client may wish to check that
527  * confirmed_flush_lsn matches its expectations.
528  */
529  elog(LOG, "%X/%X has been already streamed, forwarding to %X/%X",
530  LSN_FORMAT_ARGS(start_lsn),
532 
533  start_lsn = slot->data.confirmed_flush;
534  }
535 
536  ctx = StartupDecodingContext(output_plugin_options,
537  start_lsn, InvalidTransactionId, false,
538  fast_forward, xl_routine, prepare_write,
539  do_write, update_progress);
540 
541  /* call output plugin initialization callback */
542  old_context = MemoryContextSwitchTo(ctx->context);
543  if (ctx->callbacks.startup_cb != NULL)
544  startup_cb_wrapper(ctx, &ctx->options, false);
545  MemoryContextSwitchTo(old_context);
546 
547  /*
548  * We allow decoding of prepared transactions when the two_phase is
549  * enabled at the time of slot creation, or when the two_phase option is
550  * given at the streaming start, provided the plugin supports all the
551  * callbacks for two-phase.
552  */
553  ctx->twophase &= (slot->data.two_phase || ctx->twophase_opt_given);
554 
555  /* Mark slot to allow two_phase decoding if not already marked */
556  if (ctx->twophase && !slot->data.two_phase)
557  {
558  slot->data.two_phase = true;
559  slot->data.two_phase_at = start_lsn;
562  SnapBuildSetTwoPhaseAt(ctx->snapshot_builder, start_lsn);
563  }
564 
566 
567  ereport(LOG,
568  (errmsg("starting logical decoding for slot \"%s\"",
569  NameStr(slot->data.name)),
570  errdetail("Streaming transactions committing after %X/%X, reading WAL from %X/%X.",
572  LSN_FORMAT_ARGS(slot->data.restart_lsn))));
573 
574  return ctx;
575 }
576 
577 /*
578  * Returns true if a consistent initial decoding snapshot has been built.
579  */
580 bool
582 {
584 }
585 
586 /*
587  * Read from the decoding slot, until it is ready to start extracting changes.
588  */
589 void
591 {
592  ReplicationSlot *slot = ctx->slot;
593 
594  /* Initialize from where to start reading WAL. */
595  XLogBeginRead(ctx->reader, slot->data.restart_lsn);
596 
597  elog(DEBUG1, "searching for logical decoding starting point, starting at %X/%X",
599 
600  /* Wait for a consistent starting point */
601  for (;;)
602  {
603  XLogRecord *record;
604  char *err = NULL;
605 
606  /* the read_page callback waits for new WAL */
607  record = XLogReadRecord(ctx->reader, &err);
608  if (err)
609  elog(ERROR, "could not find logical decoding starting point: %s", err);
610  if (!record)
611  elog(ERROR, "could not find logical decoding starting point");
612 
614 
615  /* only continue till we found a consistent spot */
616  if (DecodingContextReady(ctx))
617  break;
618 
620  }
621 
622  SpinLockAcquire(&slot->mutex);
623  slot->data.confirmed_flush = ctx->reader->EndRecPtr;
624  if (slot->data.two_phase)
625  slot->data.two_phase_at = ctx->reader->EndRecPtr;
626  SpinLockRelease(&slot->mutex);
627 }
628 
629 /*
630  * Free a previously allocated decoding context, invoking the shutdown
631  * callback if necessary.
632  */
633 void
635 {
636  if (ctx->callbacks.shutdown_cb != NULL)
637  shutdown_cb_wrapper(ctx);
638 
641  XLogReaderFree(ctx->reader);
643 }
644 
645 /*
646  * Prepare a write using the context's output routine.
647  */
648 void
650 {
651  if (!ctx->accept_writes)
652  elog(ERROR, "writes are only accepted in commit, begin and change callbacks");
653 
654  ctx->prepare_write(ctx, ctx->write_location, ctx->write_xid, last_write);
655  ctx->prepared_write = true;
656 }
657 
658 /*
659  * Perform a write using the context's output routine.
660  */
661 void
662 OutputPluginWrite(struct LogicalDecodingContext *ctx, bool last_write)
663 {
664  if (!ctx->prepared_write)
665  elog(ERROR, "OutputPluginPrepareWrite needs to be called before OutputPluginWrite");
666 
667  ctx->write(ctx, ctx->write_location, ctx->write_xid, last_write);
668  ctx->prepared_write = false;
669 }
670 
671 /*
672  * Update progress tracking (if supported).
673  */
674 void
676  bool skipped_xact)
677 {
678  if (!ctx->update_progress)
679  return;
680 
681  ctx->update_progress(ctx, ctx->write_location, ctx->write_xid,
682  skipped_xact);
683 }
684 
685 /*
686  * Load the output plugin, lookup its output plugin init function, and check
687  * that it provides the required callbacks.
688  */
689 static void
691 {
692  LogicalOutputPluginInit plugin_init;
693 
694  plugin_init = (LogicalOutputPluginInit)
695  load_external_function(plugin, "_PG_output_plugin_init", false, NULL);
696 
697  if (plugin_init == NULL)
698  elog(ERROR, "output plugins have to declare the _PG_output_plugin_init symbol");
699 
700  /* ask the output plugin to fill the callback struct */
701  plugin_init(callbacks);
702 
703  if (callbacks->begin_cb == NULL)
704  elog(ERROR, "output plugins have to register a begin callback");
705  if (callbacks->change_cb == NULL)
706  elog(ERROR, "output plugins have to register a change callback");
707  if (callbacks->commit_cb == NULL)
708  elog(ERROR, "output plugins have to register a commit callback");
709 }
710 
711 static void
713 {
715 
716  /* not all callbacks have an associated LSN */
717  if (state->report_location != InvalidXLogRecPtr)
718  errcontext("slot \"%s\", output plugin \"%s\", in the %s callback, associated LSN %X/%X",
719  NameStr(state->ctx->slot->data.name),
720  NameStr(state->ctx->slot->data.plugin),
721  state->callback_name,
722  LSN_FORMAT_ARGS(state->report_location));
723  else
724  errcontext("slot \"%s\", output plugin \"%s\", in the %s callback",
725  NameStr(state->ctx->slot->data.name),
726  NameStr(state->ctx->slot->data.plugin),
727  state->callback_name);
728 }
729 
730 static void
732 {
734  ErrorContextCallback errcallback;
735 
736  Assert(!ctx->fast_forward);
737 
738  /* Push callback + info on the error context stack */
739  state.ctx = ctx;
740  state.callback_name = "startup";
741  state.report_location = InvalidXLogRecPtr;
743  errcallback.arg = (void *) &state;
744  errcallback.previous = error_context_stack;
745  error_context_stack = &errcallback;
746 
747  /* set output state */
748  ctx->accept_writes = false;
749  ctx->end_xact = false;
750 
751  /* do the actual work: call callback */
752  ctx->callbacks.startup_cb(ctx, opt, is_init);
753 
754  /* Pop the error context stack */
755  error_context_stack = errcallback.previous;
756 }
757 
758 static void
760 {
762  ErrorContextCallback errcallback;
763 
764  Assert(!ctx->fast_forward);
765 
766  /* Push callback + info on the error context stack */
767  state.ctx = ctx;
768  state.callback_name = "shutdown";
769  state.report_location = InvalidXLogRecPtr;
771  errcallback.arg = (void *) &state;
772  errcallback.previous = error_context_stack;
773  error_context_stack = &errcallback;
774 
775  /* set output state */
776  ctx->accept_writes = false;
777  ctx->end_xact = false;
778 
779  /* do the actual work: call callback */
780  ctx->callbacks.shutdown_cb(ctx);
781 
782  /* Pop the error context stack */
783  error_context_stack = errcallback.previous;
784 }
785 
786 
787 /*
788  * Callbacks for ReorderBuffer which add in some more information and then call
789  * output_plugin.h plugins.
790  */
791 static void
793 {
794  LogicalDecodingContext *ctx = cache->private_data;
796  ErrorContextCallback errcallback;
797 
798  Assert(!ctx->fast_forward);
799 
800  /* Push callback + info on the error context stack */
801  state.ctx = ctx;
802  state.callback_name = "begin";
803  state.report_location = txn->first_lsn;
805  errcallback.arg = (void *) &state;
806  errcallback.previous = error_context_stack;
807  error_context_stack = &errcallback;
808 
809  /* set output state */
810  ctx->accept_writes = true;
811  ctx->write_xid = txn->xid;
812  ctx->write_location = txn->first_lsn;
813  ctx->end_xact = false;
814 
815  /* do the actual work: call callback */
816  ctx->callbacks.begin_cb(ctx, txn);
817 
818  /* Pop the error context stack */
819  error_context_stack = errcallback.previous;
820 }
821 
822 static void
824  XLogRecPtr commit_lsn)
825 {
826  LogicalDecodingContext *ctx = cache->private_data;
828  ErrorContextCallback errcallback;
829 
830  Assert(!ctx->fast_forward);
831 
832  /* Push callback + info on the error context stack */
833  state.ctx = ctx;
834  state.callback_name = "commit";
835  state.report_location = txn->final_lsn; /* beginning of commit record */
837  errcallback.arg = (void *) &state;
838  errcallback.previous = error_context_stack;
839  error_context_stack = &errcallback;
840 
841  /* set output state */
842  ctx->accept_writes = true;
843  ctx->write_xid = txn->xid;
844  ctx->write_location = txn->end_lsn; /* points to the end of the record */
845  ctx->end_xact = true;
846 
847  /* do the actual work: call callback */
848  ctx->callbacks.commit_cb(ctx, txn, commit_lsn);
849 
850  /* Pop the error context stack */
851  error_context_stack = errcallback.previous;
852 }
853 
854 /*
855  * The functionality of begin_prepare is quite similar to begin with the
856  * exception that this will have gid (global transaction id) information which
857  * can be used by plugin. Now, we thought about extending the existing begin
858  * but that would break the replication protocol and additionally this looks
859  * cleaner.
860  */
861 static void
863 {
864  LogicalDecodingContext *ctx = cache->private_data;
866  ErrorContextCallback errcallback;
867 
868  Assert(!ctx->fast_forward);
869 
870  /* We're only supposed to call this when two-phase commits are supported */
871  Assert(ctx->twophase);
872 
873  /* Push callback + info on the error context stack */
874  state.ctx = ctx;
875  state.callback_name = "begin_prepare";
876  state.report_location = txn->first_lsn;
878  errcallback.arg = (void *) &state;
879  errcallback.previous = error_context_stack;
880  error_context_stack = &errcallback;
881 
882  /* set output state */
883  ctx->accept_writes = true;
884  ctx->write_xid = txn->xid;
885  ctx->write_location = txn->first_lsn;
886  ctx->end_xact = false;
887 
888  /*
889  * If the plugin supports two-phase commits then begin prepare callback is
890  * mandatory
891  */
892  if (ctx->callbacks.begin_prepare_cb == NULL)
893  ereport(ERROR,
894  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
895  errmsg("logical replication at prepare time requires a %s callback",
896  "begin_prepare_cb")));
897 
898  /* do the actual work: call callback */
899  ctx->callbacks.begin_prepare_cb(ctx, txn);
900 
901  /* Pop the error context stack */
902  error_context_stack = errcallback.previous;
903 }
904 
905 static void
907  XLogRecPtr prepare_lsn)
908 {
909  LogicalDecodingContext *ctx = cache->private_data;
911  ErrorContextCallback errcallback;
912 
913  Assert(!ctx->fast_forward);
914 
915  /* We're only supposed to call this when two-phase commits are supported */
916  Assert(ctx->twophase);
917 
918  /* Push callback + info on the error context stack */
919  state.ctx = ctx;
920  state.callback_name = "prepare";
921  state.report_location = txn->final_lsn; /* beginning of prepare record */
923  errcallback.arg = (void *) &state;
924  errcallback.previous = error_context_stack;
925  error_context_stack = &errcallback;
926 
927  /* set output state */
928  ctx->accept_writes = true;
929  ctx->write_xid = txn->xid;
930  ctx->write_location = txn->end_lsn; /* points to the end of the record */
931  ctx->end_xact = true;
932 
933  /*
934  * If the plugin supports two-phase commits then prepare callback is
935  * mandatory
936  */
937  if (ctx->callbacks.prepare_cb == NULL)
938  ereport(ERROR,
939  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
940  errmsg("logical replication at prepare time requires a %s callback",
941  "prepare_cb")));
942 
943  /* do the actual work: call callback */
944  ctx->callbacks.prepare_cb(ctx, txn, prepare_lsn);
945 
946  /* Pop the error context stack */
947  error_context_stack = errcallback.previous;
948 }
949 
950 static void
952  XLogRecPtr commit_lsn)
953 {
954  LogicalDecodingContext *ctx = cache->private_data;
956  ErrorContextCallback errcallback;
957 
958  Assert(!ctx->fast_forward);
959 
960  /* We're only supposed to call this when two-phase commits are supported */
961  Assert(ctx->twophase);
962 
963  /* Push callback + info on the error context stack */
964  state.ctx = ctx;
965  state.callback_name = "commit_prepared";
966  state.report_location = txn->final_lsn; /* beginning of commit record */
968  errcallback.arg = (void *) &state;
969  errcallback.previous = error_context_stack;
970  error_context_stack = &errcallback;
971 
972  /* set output state */
973  ctx->accept_writes = true;
974  ctx->write_xid = txn->xid;
975  ctx->write_location = txn->end_lsn; /* points to the end of the record */
976  ctx->end_xact = true;
977 
978  /*
979  * If the plugin support two-phase commits then commit prepared callback
980  * is mandatory
981  */
982  if (ctx->callbacks.commit_prepared_cb == NULL)
983  ereport(ERROR,
984  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
985  errmsg("logical replication at prepare time requires a %s callback",
986  "commit_prepared_cb")));
987 
988  /* do the actual work: call callback */
989  ctx->callbacks.commit_prepared_cb(ctx, txn, commit_lsn);
990 
991  /* Pop the error context stack */
992  error_context_stack = errcallback.previous;
993 }
994 
995 static void
997  XLogRecPtr prepare_end_lsn,
998  TimestampTz prepare_time)
999 {
1000  LogicalDecodingContext *ctx = cache->private_data;
1002  ErrorContextCallback errcallback;
1003 
1004  Assert(!ctx->fast_forward);
1005 
1006  /* We're only supposed to call this when two-phase commits are supported */
1007  Assert(ctx->twophase);
1008 
1009  /* Push callback + info on the error context stack */
1010  state.ctx = ctx;
1011  state.callback_name = "rollback_prepared";
1012  state.report_location = txn->final_lsn; /* beginning of commit record */
1013  errcallback.callback = output_plugin_error_callback;
1014  errcallback.arg = (void *) &state;
1015  errcallback.previous = error_context_stack;
1016  error_context_stack = &errcallback;
1017 
1018  /* set output state */
1019  ctx->accept_writes = true;
1020  ctx->write_xid = txn->xid;
1021  ctx->write_location = txn->end_lsn; /* points to the end of the record */
1022  ctx->end_xact = true;
1023 
1024  /*
1025  * If the plugin support two-phase commits then rollback prepared callback
1026  * is mandatory
1027  */
1028  if (ctx->callbacks.rollback_prepared_cb == NULL)
1029  ereport(ERROR,
1030  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1031  errmsg("logical replication at prepare time requires a %s callback",
1032  "rollback_prepared_cb")));
1033 
1034  /* do the actual work: call callback */
1035  ctx->callbacks.rollback_prepared_cb(ctx, txn, prepare_end_lsn,
1036  prepare_time);
1037 
1038  /* Pop the error context stack */
1039  error_context_stack = errcallback.previous;
1040 }
1041 
1042 static void
1044  Relation relation, ReorderBufferChange *change)
1045 {
1046  LogicalDecodingContext *ctx = cache->private_data;
1048  ErrorContextCallback errcallback;
1049 
1050  Assert(!ctx->fast_forward);
1051 
1052  /* Push callback + info on the error context stack */
1053  state.ctx = ctx;
1054  state.callback_name = "change";
1055  state.report_location = change->lsn;
1056  errcallback.callback = output_plugin_error_callback;
1057  errcallback.arg = (void *) &state;
1058  errcallback.previous = error_context_stack;
1059  error_context_stack = &errcallback;
1060 
1061  /* set output state */
1062  ctx->accept_writes = true;
1063  ctx->write_xid = txn->xid;
1064 
1065  /*
1066  * Report this change's lsn so replies from clients can give an up-to-date
1067  * answer. This won't ever be enough (and shouldn't be!) to confirm
1068  * receipt of this transaction, but it might allow another transaction's
1069  * commit to be confirmed with one message.
1070  */
1071  ctx->write_location = change->lsn;
1072 
1073  ctx->end_xact = false;
1074 
1075  ctx->callbacks.change_cb(ctx, txn, relation, change);
1076 
1077  /* Pop the error context stack */
1078  error_context_stack = errcallback.previous;
1079 }
1080 
1081 static void
1083  int nrelations, Relation relations[], ReorderBufferChange *change)
1084 {
1085  LogicalDecodingContext *ctx = cache->private_data;
1087  ErrorContextCallback errcallback;
1088 
1089  Assert(!ctx->fast_forward);
1090 
1091  if (!ctx->callbacks.truncate_cb)
1092  return;
1093 
1094  /* Push callback + info on the error context stack */
1095  state.ctx = ctx;
1096  state.callback_name = "truncate";
1097  state.report_location = change->lsn;
1098  errcallback.callback = output_plugin_error_callback;
1099  errcallback.arg = (void *) &state;
1100  errcallback.previous = error_context_stack;
1101  error_context_stack = &errcallback;
1102 
1103  /* set output state */
1104  ctx->accept_writes = true;
1105  ctx->write_xid = txn->xid;
1106 
1107  /*
1108  * Report this change's lsn so replies from clients can give an up-to-date
1109  * answer. This won't ever be enough (and shouldn't be!) to confirm
1110  * receipt of this transaction, but it might allow another transaction's
1111  * commit to be confirmed with one message.
1112  */
1113  ctx->write_location = change->lsn;
1114 
1115  ctx->end_xact = false;
1116 
1117  ctx->callbacks.truncate_cb(ctx, txn, nrelations, relations, change);
1118 
1119  /* Pop the error context stack */
1120  error_context_stack = errcallback.previous;
1121 }
1122 
1123 bool
1125  const char *gid)
1126 {
1128  ErrorContextCallback errcallback;
1129  bool ret;
1130 
1131  Assert(!ctx->fast_forward);
1132 
1133  /* Push callback + info on the error context stack */
1134  state.ctx = ctx;
1135  state.callback_name = "filter_prepare";
1136  state.report_location = InvalidXLogRecPtr;
1137  errcallback.callback = output_plugin_error_callback;
1138  errcallback.arg = (void *) &state;
1139  errcallback.previous = error_context_stack;
1140  error_context_stack = &errcallback;
1141 
1142  /* set output state */
1143  ctx->accept_writes = false;
1144  ctx->end_xact = false;
1145 
1146  /* do the actual work: call callback */
1147  ret = ctx->callbacks.filter_prepare_cb(ctx, xid, gid);
1148 
1149  /* Pop the error context stack */
1150  error_context_stack = errcallback.previous;
1151 
1152  return ret;
1153 }
1154 
1155 bool
1157 {
1159  ErrorContextCallback errcallback;
1160  bool ret;
1161 
1162  Assert(!ctx->fast_forward);
1163 
1164  /* Push callback + info on the error context stack */
1165  state.ctx = ctx;
1166  state.callback_name = "filter_by_origin";
1167  state.report_location = InvalidXLogRecPtr;
1168  errcallback.callback = output_plugin_error_callback;
1169  errcallback.arg = (void *) &state;
1170  errcallback.previous = error_context_stack;
1171  error_context_stack = &errcallback;
1172 
1173  /* set output state */
1174  ctx->accept_writes = false;
1175  ctx->end_xact = false;
1176 
1177  /* do the actual work: call callback */
1178  ret = ctx->callbacks.filter_by_origin_cb(ctx, origin_id);
1179 
1180  /* Pop the error context stack */
1181  error_context_stack = errcallback.previous;
1182 
1183  return ret;
1184 }
1185 
1186 static void
1188  XLogRecPtr message_lsn, bool transactional,
1189  const char *prefix, Size message_size, const char *message)
1190 {
1191  LogicalDecodingContext *ctx = cache->private_data;
1193  ErrorContextCallback errcallback;
1194 
1195  Assert(!ctx->fast_forward);
1196 
1197  if (ctx->callbacks.message_cb == NULL)
1198  return;
1199 
1200  /* Push callback + info on the error context stack */
1201  state.ctx = ctx;
1202  state.callback_name = "message";
1203  state.report_location = message_lsn;
1204  errcallback.callback = output_plugin_error_callback;
1205  errcallback.arg = (void *) &state;
1206  errcallback.previous = error_context_stack;
1207  error_context_stack = &errcallback;
1208 
1209  /* set output state */
1210  ctx->accept_writes = true;
1211  ctx->write_xid = txn != NULL ? txn->xid : InvalidTransactionId;
1212  ctx->write_location = message_lsn;
1213  ctx->end_xact = false;
1214 
1215  /* do the actual work: call callback */
1216  ctx->callbacks.message_cb(ctx, txn, message_lsn, transactional, prefix,
1217  message_size, message);
1218 
1219  /* Pop the error context stack */
1220  error_context_stack = errcallback.previous;
1221 }
1222 
1223 static void
1225  XLogRecPtr first_lsn)
1226 {
1227  LogicalDecodingContext *ctx = cache->private_data;
1229  ErrorContextCallback errcallback;
1230 
1231  Assert(!ctx->fast_forward);
1232 
1233  /* We're only supposed to call this when streaming is supported. */
1234  Assert(ctx->streaming);
1235 
1236  /* Push callback + info on the error context stack */
1237  state.ctx = ctx;
1238  state.callback_name = "stream_start";
1239  state.report_location = first_lsn;
1240  errcallback.callback = output_plugin_error_callback;
1241  errcallback.arg = (void *) &state;
1242  errcallback.previous = error_context_stack;
1243  error_context_stack = &errcallback;
1244 
1245  /* set output state */
1246  ctx->accept_writes = true;
1247  ctx->write_xid = txn->xid;
1248 
1249  /*
1250  * Report this message's lsn so replies from clients can give an
1251  * up-to-date answer. This won't ever be enough (and shouldn't be!) to
1252  * confirm receipt of this transaction, but it might allow another
1253  * transaction's commit to be confirmed with one message.
1254  */
1255  ctx->write_location = first_lsn;
1256 
1257  ctx->end_xact = false;
1258 
1259  /* in streaming mode, stream_start_cb is required */
1260  if (ctx->callbacks.stream_start_cb == NULL)
1261  ereport(ERROR,
1262  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1263  errmsg("logical streaming requires a %s callback",
1264  "stream_start_cb")));
1265 
1266  ctx->callbacks.stream_start_cb(ctx, txn);
1267 
1268  /* Pop the error context stack */
1269  error_context_stack = errcallback.previous;
1270 }
1271 
1272 static void
1274  XLogRecPtr last_lsn)
1275 {
1276  LogicalDecodingContext *ctx = cache->private_data;
1278  ErrorContextCallback errcallback;
1279 
1280  Assert(!ctx->fast_forward);
1281 
1282  /* We're only supposed to call this when streaming is supported. */
1283  Assert(ctx->streaming);
1284 
1285  /* Push callback + info on the error context stack */
1286  state.ctx = ctx;
1287  state.callback_name = "stream_stop";
1288  state.report_location = last_lsn;
1289  errcallback.callback = output_plugin_error_callback;
1290  errcallback.arg = (void *) &state;
1291  errcallback.previous = error_context_stack;
1292  error_context_stack = &errcallback;
1293 
1294  /* set output state */
1295  ctx->accept_writes = true;
1296  ctx->write_xid = txn->xid;
1297 
1298  /*
1299  * Report this message's lsn so replies from clients can give an
1300  * up-to-date answer. This won't ever be enough (and shouldn't be!) to
1301  * confirm receipt of this transaction, but it might allow another
1302  * transaction's commit to be confirmed with one message.
1303  */
1304  ctx->write_location = last_lsn;
1305 
1306  ctx->end_xact = false;
1307 
1308  /* in streaming mode, stream_stop_cb is required */
1309  if (ctx->callbacks.stream_stop_cb == NULL)
1310  ereport(ERROR,
1311  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1312  errmsg("logical streaming requires a %s callback",
1313  "stream_stop_cb")));
1314 
1315  ctx->callbacks.stream_stop_cb(ctx, txn);
1316 
1317  /* Pop the error context stack */
1318  error_context_stack = errcallback.previous;
1319 }
1320 
1321 static void
1323  XLogRecPtr abort_lsn)
1324 {
1325  LogicalDecodingContext *ctx = cache->private_data;
1327  ErrorContextCallback errcallback;
1328 
1329  Assert(!ctx->fast_forward);
1330 
1331  /* We're only supposed to call this when streaming is supported. */
1332  Assert(ctx->streaming);
1333 
1334  /* Push callback + info on the error context stack */
1335  state.ctx = ctx;
1336  state.callback_name = "stream_abort";
1337  state.report_location = abort_lsn;
1338  errcallback.callback = output_plugin_error_callback;
1339  errcallback.arg = (void *) &state;
1340  errcallback.previous = error_context_stack;
1341  error_context_stack = &errcallback;
1342 
1343  /* set output state */
1344  ctx->accept_writes = true;
1345  ctx->write_xid = txn->xid;
1346  ctx->write_location = abort_lsn;
1347  ctx->end_xact = true;
1348 
1349  /* in streaming mode, stream_abort_cb is required */
1350  if (ctx->callbacks.stream_abort_cb == NULL)
1351  ereport(ERROR,
1352  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1353  errmsg("logical streaming requires a %s callback",
1354  "stream_abort_cb")));
1355 
1356  ctx->callbacks.stream_abort_cb(ctx, txn, abort_lsn);
1357 
1358  /* Pop the error context stack */
1359  error_context_stack = errcallback.previous;
1360 }
1361 
1362 static void
1364  XLogRecPtr prepare_lsn)
1365 {
1366  LogicalDecodingContext *ctx = cache->private_data;
1368  ErrorContextCallback errcallback;
1369 
1370  Assert(!ctx->fast_forward);
1371 
1372  /*
1373  * We're only supposed to call this when streaming and two-phase commits
1374  * are supported.
1375  */
1376  Assert(ctx->streaming);
1377  Assert(ctx->twophase);
1378 
1379  /* Push callback + info on the error context stack */
1380  state.ctx = ctx;
1381  state.callback_name = "stream_prepare";
1382  state.report_location = txn->final_lsn;
1383  errcallback.callback = output_plugin_error_callback;
1384  errcallback.arg = (void *) &state;
1385  errcallback.previous = error_context_stack;
1386  error_context_stack = &errcallback;
1387 
1388  /* set output state */
1389  ctx->accept_writes = true;
1390  ctx->write_xid = txn->xid;
1391  ctx->write_location = txn->end_lsn;
1392  ctx->end_xact = true;
1393 
1394  /* in streaming mode with two-phase commits, stream_prepare_cb is required */
1395  if (ctx->callbacks.stream_prepare_cb == NULL)
1396  ereport(ERROR,
1397  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1398  errmsg("logical streaming at prepare time requires a %s callback",
1399  "stream_prepare_cb")));
1400 
1401  ctx->callbacks.stream_prepare_cb(ctx, txn, prepare_lsn);
1402 
1403  /* Pop the error context stack */
1404  error_context_stack = errcallback.previous;
1405 }
1406 
1407 static void
1409  XLogRecPtr commit_lsn)
1410 {
1411  LogicalDecodingContext *ctx = cache->private_data;
1413  ErrorContextCallback errcallback;
1414 
1415  Assert(!ctx->fast_forward);
1416 
1417  /* We're only supposed to call this when streaming is supported. */
1418  Assert(ctx->streaming);
1419 
1420  /* Push callback + info on the error context stack */
1421  state.ctx = ctx;
1422  state.callback_name = "stream_commit";
1423  state.report_location = txn->final_lsn;
1424  errcallback.callback = output_plugin_error_callback;
1425  errcallback.arg = (void *) &state;
1426  errcallback.previous = error_context_stack;
1427  error_context_stack = &errcallback;
1428 
1429  /* set output state */
1430  ctx->accept_writes = true;
1431  ctx->write_xid = txn->xid;
1432  ctx->write_location = txn->end_lsn;
1433  ctx->end_xact = true;
1434 
1435  /* in streaming mode, stream_commit_cb is required */
1436  if (ctx->callbacks.stream_commit_cb == NULL)
1437  ereport(ERROR,
1438  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1439  errmsg("logical streaming requires a %s callback",
1440  "stream_commit_cb")));
1441 
1442  ctx->callbacks.stream_commit_cb(ctx, txn, commit_lsn);
1443 
1444  /* Pop the error context stack */
1445  error_context_stack = errcallback.previous;
1446 }
1447 
1448 static void
1450  Relation relation, ReorderBufferChange *change)
1451 {
1452  LogicalDecodingContext *ctx = cache->private_data;
1454  ErrorContextCallback errcallback;
1455 
1456  Assert(!ctx->fast_forward);
1457 
1458  /* We're only supposed to call this when streaming is supported. */
1459  Assert(ctx->streaming);
1460 
1461  /* Push callback + info on the error context stack */
1462  state.ctx = ctx;
1463  state.callback_name = "stream_change";
1464  state.report_location = change->lsn;
1465  errcallback.callback = output_plugin_error_callback;
1466  errcallback.arg = (void *) &state;
1467  errcallback.previous = error_context_stack;
1468  error_context_stack = &errcallback;
1469 
1470  /* set output state */
1471  ctx->accept_writes = true;
1472  ctx->write_xid = txn->xid;
1473 
1474  /*
1475  * Report this change's lsn so replies from clients can give an up-to-date
1476  * answer. This won't ever be enough (and shouldn't be!) to confirm
1477  * receipt of this transaction, but it might allow another transaction's
1478  * commit to be confirmed with one message.
1479  */
1480  ctx->write_location = change->lsn;
1481 
1482  ctx->end_xact = false;
1483 
1484  /* in streaming mode, stream_change_cb is required */
1485  if (ctx->callbacks.stream_change_cb == NULL)
1486  ereport(ERROR,
1487  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1488  errmsg("logical streaming requires a %s callback",
1489  "stream_change_cb")));
1490 
1491  ctx->callbacks.stream_change_cb(ctx, txn, relation, change);
1492 
1493  /* Pop the error context stack */
1494  error_context_stack = errcallback.previous;
1495 }
1496 
1497 static void
1499  XLogRecPtr message_lsn, bool transactional,
1500  const char *prefix, Size message_size, const char *message)
1501 {
1502  LogicalDecodingContext *ctx = cache->private_data;
1504  ErrorContextCallback errcallback;
1505 
1506  Assert(!ctx->fast_forward);
1507 
1508  /* We're only supposed to call this when streaming is supported. */
1509  Assert(ctx->streaming);
1510 
1511  /* this callback is optional */
1512  if (ctx->callbacks.stream_message_cb == NULL)
1513  return;
1514 
1515  /* Push callback + info on the error context stack */
1516  state.ctx = ctx;
1517  state.callback_name = "stream_message";
1518  state.report_location = message_lsn;
1519  errcallback.callback = output_plugin_error_callback;
1520  errcallback.arg = (void *) &state;
1521  errcallback.previous = error_context_stack;
1522  error_context_stack = &errcallback;
1523 
1524  /* set output state */
1525  ctx->accept_writes = true;
1526  ctx->write_xid = txn != NULL ? txn->xid : InvalidTransactionId;
1527  ctx->write_location = message_lsn;
1528  ctx->end_xact = false;
1529 
1530  /* do the actual work: call callback */
1531  ctx->callbacks.stream_message_cb(ctx, txn, message_lsn, transactional, prefix,
1532  message_size, message);
1533 
1534  /* Pop the error context stack */
1535  error_context_stack = errcallback.previous;
1536 }
1537 
1538 static void
1540  int nrelations, Relation relations[],
1541  ReorderBufferChange *change)
1542 {
1543  LogicalDecodingContext *ctx = cache->private_data;
1545  ErrorContextCallback errcallback;
1546 
1547  Assert(!ctx->fast_forward);
1548 
1549  /* We're only supposed to call this when streaming is supported. */
1550  Assert(ctx->streaming);
1551 
1552  /* this callback is optional */
1553  if (!ctx->callbacks.stream_truncate_cb)
1554  return;
1555 
1556  /* Push callback + info on the error context stack */
1557  state.ctx = ctx;
1558  state.callback_name = "stream_truncate";
1559  state.report_location = change->lsn;
1560  errcallback.callback = output_plugin_error_callback;
1561  errcallback.arg = (void *) &state;
1562  errcallback.previous = error_context_stack;
1563  error_context_stack = &errcallback;
1564 
1565  /* set output state */
1566  ctx->accept_writes = true;
1567  ctx->write_xid = txn->xid;
1568 
1569  /*
1570  * Report this change's lsn so replies from clients can give an up-to-date
1571  * answer. This won't ever be enough (and shouldn't be!) to confirm
1572  * receipt of this transaction, but it might allow another transaction's
1573  * commit to be confirmed with one message.
1574  */
1575  ctx->write_location = change->lsn;
1576 
1577  ctx->end_xact = false;
1578 
1579  ctx->callbacks.stream_truncate_cb(ctx, txn, nrelations, relations, change);
1580 
1581  /* Pop the error context stack */
1582  error_context_stack = errcallback.previous;
1583 }
1584 
1585 /*
1586  * Set the required catalog xmin horizon for historic snapshots in the current
1587  * replication slot.
1588  *
1589  * Note that in the most cases, we won't be able to immediately use the xmin
1590  * to increase the xmin horizon: we need to wait till the client has confirmed
1591  * receiving current_lsn with LogicalConfirmReceivedLocation().
1592  */
1593 void
1595 {
1596  bool updated_xmin = false;
1597  ReplicationSlot *slot;
1598  bool got_new_xmin = false;
1599 
1600  slot = MyReplicationSlot;
1601 
1602  Assert(slot != NULL);
1603 
1604  SpinLockAcquire(&slot->mutex);
1605 
1606  /*
1607  * don't overwrite if we already have a newer xmin. This can happen if we
1608  * restart decoding in a slot.
1609  */
1611  {
1612  }
1613 
1614  /*
1615  * If the client has already confirmed up to this lsn, we directly can
1616  * mark this as accepted. This can happen if we restart decoding in a
1617  * slot.
1618  */
1619  else if (current_lsn <= slot->data.confirmed_flush)
1620  {
1621  slot->candidate_catalog_xmin = xmin;
1622  slot->candidate_xmin_lsn = current_lsn;
1623 
1624  /* our candidate can directly be used */
1625  updated_xmin = true;
1626  }
1627 
1628  /*
1629  * Only increase if the previous values have been applied, otherwise we
1630  * might never end up updating if the receiver acks too slowly.
1631  */
1632  else if (slot->candidate_xmin_lsn == InvalidXLogRecPtr)
1633  {
1634  slot->candidate_catalog_xmin = xmin;
1635  slot->candidate_xmin_lsn = current_lsn;
1636 
1637  /*
1638  * Log new xmin at an appropriate log level after releasing the
1639  * spinlock.
1640  */
1641  got_new_xmin = true;
1642  }
1643  SpinLockRelease(&slot->mutex);
1644 
1645  if (got_new_xmin)
1646  elog(DEBUG1, "got new catalog xmin %u at %X/%X", xmin,
1647  LSN_FORMAT_ARGS(current_lsn));
1648 
1649  /* candidate already valid with the current flush position, apply */
1650  if (updated_xmin)
1652 }
1653 
1654 /*
1655  * Mark the minimal LSN (restart_lsn) we need to read to replay all
1656  * transactions that have not yet committed at current_lsn.
1657  *
1658  * Just like LogicalIncreaseXminForSlot this only takes effect when the
1659  * client has confirmed to have received current_lsn.
1660  */
1661 void
1663 {
1664  bool updated_lsn = false;
1665  ReplicationSlot *slot;
1666 
1667  slot = MyReplicationSlot;
1668 
1669  Assert(slot != NULL);
1670  Assert(restart_lsn != InvalidXLogRecPtr);
1671  Assert(current_lsn != InvalidXLogRecPtr);
1672 
1673  SpinLockAcquire(&slot->mutex);
1674 
1675  /* don't overwrite if have a newer restart lsn */
1676  if (restart_lsn <= slot->data.restart_lsn)
1677  {
1678  }
1679 
1680  /*
1681  * We might have already flushed far enough to directly accept this lsn,
1682  * in this case there is no need to check for existing candidate LSNs
1683  */
1684  else if (current_lsn <= slot->data.confirmed_flush)
1685  {
1686  slot->candidate_restart_valid = current_lsn;
1687  slot->candidate_restart_lsn = restart_lsn;
1688 
1689  /* our candidate can directly be used */
1690  updated_lsn = true;
1691  }
1692 
1693  /*
1694  * Only increase if the previous values have been applied, otherwise we
1695  * might never end up updating if the receiver acks too slowly. A missed
1696  * value here will just cause some extra effort after reconnecting.
1697  */
1699  {
1700  slot->candidate_restart_valid = current_lsn;
1701  slot->candidate_restart_lsn = restart_lsn;
1702  SpinLockRelease(&slot->mutex);
1703 
1704  elog(DEBUG1, "got new restart lsn %X/%X at %X/%X",
1705  LSN_FORMAT_ARGS(restart_lsn),
1706  LSN_FORMAT_ARGS(current_lsn));
1707  }
1708  else
1709  {
1710  XLogRecPtr candidate_restart_lsn;
1711  XLogRecPtr candidate_restart_valid;
1712  XLogRecPtr confirmed_flush;
1713 
1714  candidate_restart_lsn = slot->candidate_restart_lsn;
1715  candidate_restart_valid = slot->candidate_restart_valid;
1716  confirmed_flush = slot->data.confirmed_flush;
1717  SpinLockRelease(&slot->mutex);
1718 
1719  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",
1720  LSN_FORMAT_ARGS(restart_lsn),
1721  LSN_FORMAT_ARGS(current_lsn),
1722  LSN_FORMAT_ARGS(candidate_restart_lsn),
1723  LSN_FORMAT_ARGS(candidate_restart_valid),
1724  LSN_FORMAT_ARGS(confirmed_flush));
1725  }
1726 
1727  /* candidates are already valid with the current flush position, apply */
1728  if (updated_lsn)
1730 }
1731 
1732 /*
1733  * Handle a consumer's confirmation having received all changes up to lsn.
1734  */
1735 void
1737 {
1738  Assert(lsn != InvalidXLogRecPtr);
1739 
1740  /* Do an unlocked check for candidate_lsn first. */
1743  {
1744  bool updated_xmin = false;
1745  bool updated_restart = false;
1746 
1748 
1750 
1751  /* if we're past the location required for bumping xmin, do so */
1754  {
1755  /*
1756  * We have to write the changed xmin to disk *before* we change
1757  * the in-memory value, otherwise after a crash we wouldn't know
1758  * that some catalog tuples might have been removed already.
1759  *
1760  * Ensure that by first writing to ->xmin and only update
1761  * ->effective_xmin once the new state is synced to disk. After a
1762  * crash ->effective_xmin is set to ->xmin.
1763  */
1766  {
1770  updated_xmin = true;
1771  }
1772  }
1773 
1776  {
1778 
1782  updated_restart = true;
1783  }
1784 
1786 
1787  /* first write new xmin to disk, so we know what's up after a crash */
1788  if (updated_xmin || updated_restart)
1789  {
1792  elog(DEBUG1, "updated xmin: %u restart: %u", updated_xmin, updated_restart);
1793  }
1794 
1795  /*
1796  * Now the new xmin is safely on disk, we can let the global value
1797  * advance. We do not take ProcArrayLock or similar since we only
1798  * advance xmin here and there's not much harm done by a concurrent
1799  * computation missing that.
1800  */
1801  if (updated_xmin)
1802  {
1806 
1809  }
1810  }
1811  else
1812  {
1816  }
1817 }
1818 
1819 /*
1820  * Clear logical streaming state during (sub)transaction abort.
1821  */
1822 void
1824 {
1826  bsysscan = false;
1827 }
1828 
1829 /*
1830  * Report stats for a slot.
1831  */
1832 void
1834 {
1835  ReorderBuffer *rb = ctx->reorder;
1836  PgStat_StatReplSlotEntry repSlotStat;
1837 
1838  /* Nothing to do if we don't have any replication stats to be sent. */
1839  if (rb->spillBytes <= 0 && rb->streamBytes <= 0 && rb->totalBytes <= 0)
1840  return;
1841 
1842  elog(DEBUG2, "UpdateDecodingStats: updating stats %p %lld %lld %lld %lld %lld %lld %lld %lld",
1843  rb,
1844  (long long) rb->spillTxns,
1845  (long long) rb->spillCount,
1846  (long long) rb->spillBytes,
1847  (long long) rb->streamTxns,
1848  (long long) rb->streamCount,
1849  (long long) rb->streamBytes,
1850  (long long) rb->totalTxns,
1851  (long long) rb->totalBytes);
1852 
1853  repSlotStat.spill_txns = rb->spillTxns;
1854  repSlotStat.spill_count = rb->spillCount;
1855  repSlotStat.spill_bytes = rb->spillBytes;
1856  repSlotStat.stream_txns = rb->streamTxns;
1857  repSlotStat.stream_count = rb->streamCount;
1858  repSlotStat.stream_bytes = rb->streamBytes;
1859  repSlotStat.total_txns = rb->totalTxns;
1860  repSlotStat.total_bytes = rb->totalBytes;
1861 
1862  pgstat_report_replslot(ctx->slot, &repSlotStat);
1863 
1864  rb->spillTxns = 0;
1865  rb->spillCount = 0;
1866  rb->spillBytes = 0;
1867  rb->streamTxns = 0;
1868  rb->streamCount = 0;
1869  rb->streamBytes = 0;
1870  rb->totalTxns = 0;
1871  rb->totalBytes = 0;
1872 }
#define NameStr(name)
Definition: c.h:681
uint32 TransactionId
Definition: c.h:587
size_t Size
Definition: c.h:540
int64 TimestampTz
Definition: timestamp.h:39
void LogicalDecodingProcessRecord(LogicalDecodingContext *ctx, XLogReaderState *record)
Definition: decode.c:91
void * load_external_function(const char *filename, const char *funcname, bool signalNotFound, void **filehandle)
Definition: dfmgr.c:105
int errdetail(const char *fmt,...)
Definition: elog.c:1037
ErrorContextCallback * error_context_stack
Definition: elog.c:93
int errcode(int sqlerrcode)
Definition: elog.c:693
int errmsg(const char *fmt,...)
Definition: elog.c:904
#define LOG
Definition: elog.h:25
#define errcontext
Definition: elog.h:190
#define DEBUG2
Definition: elog.h:23
#define DEBUG1
Definition: elog.h:24
#define ERROR
Definition: elog.h:33
#define elog(elevel,...)
Definition: elog.h:218
#define ereport(elevel,...)
Definition: elog.h:143
Oid MyDatabaseId
Definition: globals.c:89
Assert(fmt[strlen(fmt) - 1] !='\n')
static void change_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, Relation relation, ReorderBufferChange *change)
Definition: logical.c:1043
static void commit_prepared_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr commit_lsn)
Definition: logical.c:951
void LogicalConfirmReceivedLocation(XLogRecPtr lsn)
Definition: logical.c:1736
void FreeDecodingContext(LogicalDecodingContext *ctx)
Definition: logical.c:634
static void stream_start_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr first_lsn)
Definition: logical.c:1224
static void commit_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr commit_lsn)
Definition: logical.c:823
static void output_plugin_error_callback(void *arg)
Definition: logical.c:712
static void begin_prepare_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn)
Definition: logical.c:862
static void stream_prepare_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr prepare_lsn)
Definition: logical.c:1363
static void prepare_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr prepare_lsn)
Definition: logical.c:906
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:146
void OutputPluginWrite(struct LogicalDecodingContext *ctx, bool last_write)
Definition: logical.c:662
static void stream_truncate_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, int nrelations, Relation relations[], ReorderBufferChange *change)
Definition: logical.c:1539
static void truncate_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, int nrelations, Relation relations[], ReorderBufferChange *change)
Definition: logical.c:1082
void DecodingContextFindStartpoint(LogicalDecodingContext *ctx)
Definition: logical.c:590
static void begin_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn)
Definition: logical.c:792
static void rollback_prepared_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr prepare_end_lsn, TimestampTz prepare_time)
Definition: logical.c:996
bool DecodingContextReady(LogicalDecodingContext *ctx)
Definition: logical.c:581
void OutputPluginUpdateProgress(struct LogicalDecodingContext *ctx, bool skipped_xact)
Definition: logical.c:675
static void startup_cb_wrapper(LogicalDecodingContext *ctx, OutputPluginOptions *opt, bool is_init)
Definition: logical.c:731
void UpdateDecodingStats(LogicalDecodingContext *ctx)
Definition: logical.c:1833
void LogicalIncreaseRestartDecodingForSlot(XLogRecPtr current_lsn, XLogRecPtr restart_lsn)
Definition: logical.c:1662
static void stream_change_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, Relation relation, ReorderBufferChange *change)
Definition: logical.c:1449
static void stream_abort_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr abort_lsn)
Definition: logical.c:1322
void ResetLogicalStreamingState(void)
Definition: logical.c:1823
void LogicalIncreaseXminForSlot(XLogRecPtr current_lsn, TransactionId xmin)
Definition: logical.c:1594
struct LogicalErrorCallbackState LogicalErrorCallbackState
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:1498
static void stream_commit_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr commit_lsn)
Definition: logical.c:1408
bool filter_prepare_cb_wrapper(LogicalDecodingContext *ctx, TransactionId xid, const char *gid)
Definition: logical.c:1124
static void shutdown_cb_wrapper(LogicalDecodingContext *ctx)
Definition: logical.c:759
void OutputPluginPrepareWrite(struct LogicalDecodingContext *ctx, bool last_write)
Definition: logical.c:649
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
bool filter_by_origin_cb_wrapper(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: logical.c:1156
static void LoadOutputPlugin(OutputPluginCallbacks *callbacks, const char *plugin)
Definition: logical.c:690
static void stream_stop_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr last_lsn)
Definition: logical.c:1273
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:1187
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
void(* LogicalOutputPluginWriterUpdateProgress)(struct LogicalDecodingContext *lr, XLogRecPtr Ptr, TransactionId xid, bool skipped_xact)
Definition: logical.h:27
void(* LogicalOutputPluginWriterWrite)(struct LogicalDecodingContext *lr, XLogRecPtr Ptr, TransactionId xid, bool last_write)
Definition: logical.h:19
LogicalOutputPluginWriterWrite LogicalOutputPluginWriterPrepareWrite
Definition: logical.h:25
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1196
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1800
@ LW_EXCLUSIVE
Definition: lwlock.h:104
void * palloc0(Size size)
Definition: mcxt.c:1099
MemoryContext CurrentMemoryContext
Definition: mcxt.c:42
void MemoryContextDelete(MemoryContext context)
Definition: mcxt.c:218
#define AllocSetContextCreate
Definition: memutils.h:173
#define ALLOCSET_DEFAULT_SIZES
Definition: memutils.h:197
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:121
void namestrcpy(Name name, const char *str)
Definition: name.c:233
void(* LogicalOutputPluginInit)(struct OutputPluginCallbacks *cb)
Definition: output_plugin.h:36
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
Definition: palloc.h:109
void * arg
const void * data
#define NIL
Definition: pg_list.h:65
static const char * plugin
void pgstat_report_replslot(ReplicationSlot *slot, const PgStat_StatReplSlotEntry *repSlotStat)
#define InvalidOid
Definition: postgres_ext.h:36
#define PROC_IN_LOGICAL_DECODING
Definition: proc.h:58
TransactionId GetOldestSafeDecodingTransactionId(bool catalogOnly)
Definition: procarray.c:2977
ReorderBuffer * ReorderBufferAllocate(void)
void ReorderBufferFree(ReorderBuffer *rb)
void ReplicationSlotMarkDirty(void)
Definition: slot.c:798
void ReplicationSlotReserveWal(void)
Definition: slot.c:1151
void ReplicationSlotsComputeRequiredXmin(bool already_locked)
Definition: slot.c:837
ReplicationSlot * MyReplicationSlot
Definition: slot.c:97
void ReplicationSlotSave(void)
Definition: slot.c:780
void ReplicationSlotsComputeRequiredLSN(void)
Definition: slot.c:887
void CheckSlotRequirements(void)
Definition: slot.c:1114
#define SlotIsPhysical(slot)
Definition: slot.h:168
void SnapBuildSetTwoPhaseAt(SnapBuild *builder, XLogRecPtr ptr)
Definition: snapbuild.c:385
SnapBuildState SnapBuildCurrentState(SnapBuild *builder)
Definition: snapbuild.c:367
void FreeSnapshotBuilder(SnapBuild *builder)
Definition: snapbuild.c:323
SnapBuild * AllocateSnapshotBuilder(ReorderBuffer *reorder, TransactionId xmin_horizon, XLogRecPtr start_lsn, bool need_full_snapshot, XLogRecPtr two_phase_at)
Definition: snapbuild.c:280
@ SNAPBUILD_CONSISTENT
Definition: snapbuild.h:46
#define SpinLockRelease(lock)
Definition: spin.h:64
#define SpinLockAcquire(lock)
Definition: spin.h:62
PGPROC * MyProc
Definition: proc.c:68
PROC_HDR * ProcGlobal
Definition: proc.c:80
StringInfo makeStringInfo(void)
Definition: stringinfo.c:41
struct ErrorContextCallback * previous
Definition: elog.h:232
void(* callback)(void *arg)
Definition: elog.h:233
Definition: pg_list.h:51
OutputPluginOptions options
Definition: logical.h:54
XLogReaderState * reader
Definition: logical.h:42
MemoryContext context
Definition: logical.h:36
struct SnapBuild * snapshot_builder
Definition: logical.h:44
StringInfo out
Definition: logical.h:71
XLogRecPtr write_location
Definition: logical.h:108
LogicalOutputPluginWriterPrepareWrite prepare_write
Definition: logical.h:64
OutputPluginCallbacks callbacks
Definition: logical.h:53
TransactionId write_xid
Definition: logical.h:109
List * output_plugin_options
Definition: logical.h:59
ReplicationSlot * slot
Definition: logical.h:39
LogicalOutputPluginWriterWrite write
Definition: logical.h:65
struct ReorderBuffer * reorder
Definition: logical.h:43
LogicalOutputPluginWriterUpdateProgress update_progress
Definition: logical.h:66
XLogRecPtr report_location
Definition: logical.c:51
LogicalDecodingContext * ctx
Definition: logical.c:49
const char * callback_name
Definition: logical.c:50
LogicalDecodeStreamChangeCB stream_change_cb
LogicalDecodeMessageCB message_cb
LogicalDecodeStreamTruncateCB stream_truncate_cb
LogicalDecodeStreamMessageCB stream_message_cb
LogicalDecodeFilterPrepareCB filter_prepare_cb
LogicalDecodeFilterByOriginCB filter_by_origin_cb
LogicalDecodeTruncateCB truncate_cb
LogicalDecodeStreamStopCB stream_stop_cb
LogicalDecodeStreamCommitCB stream_commit_cb
LogicalDecodeRollbackPreparedCB rollback_prepared_cb
LogicalDecodeStreamPrepareCB stream_prepare_cb
LogicalDecodeCommitPreparedCB commit_prepared_cb
LogicalDecodeStreamStartCB stream_start_cb
LogicalDecodePrepareCB prepare_cb
LogicalDecodeStartupCB startup_cb
LogicalDecodeCommitCB commit_cb
LogicalDecodeBeginCB begin_cb
LogicalDecodeStreamAbortCB stream_abort_cb
LogicalDecodeBeginPrepareCB begin_prepare_cb
LogicalDecodeChangeCB change_cb
LogicalDecodeShutdownCB shutdown_cb
uint8 statusFlags
Definition: proc.h:227
int pgxactoff
Definition: proc.h:186
uint8 * statusFlags
Definition: proc.h:371
PgStat_Counter stream_count
Definition: pgstat.h:329
PgStat_Counter total_txns
Definition: pgstat.h:331
PgStat_Counter total_bytes
Definition: pgstat.h:332
PgStat_Counter spill_txns
Definition: pgstat.h:325
PgStat_Counter stream_txns
Definition: pgstat.h:328
PgStat_Counter spill_count
Definition: pgstat.h:326
PgStat_Counter stream_bytes
Definition: pgstat.h:330
PgStat_Counter spill_bytes
Definition: pgstat.h:327
XLogRecPtr first_lsn
XLogRecPtr final_lsn
XLogRecPtr end_lsn
TransactionId xid
ReorderBufferStreamMessageCB stream_message
ReorderBufferStreamChangeCB stream_change
ReorderBufferBeginCB begin_prepare
ReorderBufferStreamTruncateCB stream_truncate
ReorderBufferCommitPreparedCB commit_prepared
ReorderBufferMessageCB message
ReorderBufferRollbackPreparedCB rollback_prepared
ReorderBufferPrepareCB prepare
ReorderBufferStreamStopCB stream_stop
ReorderBufferApplyChangeCB apply_change
ReorderBufferStreamPrepareCB stream_prepare
ReorderBufferStreamAbortCB stream_abort
ReorderBufferCommitCB commit
ReorderBufferStreamStartCB stream_start
ReorderBufferStreamCommitCB stream_commit
ReorderBufferApplyTruncateCB apply_truncate
ReorderBufferBeginCB begin
void * private_data
XLogRecPtr two_phase_at
Definition: slot.h:90
TransactionId catalog_xmin
Definition: slot.h:70
XLogRecPtr restart_lsn
Definition: slot.h:73
XLogRecPtr confirmed_flush
Definition: slot.h:84
XLogRecPtr candidate_xmin_lsn
Definition: slot.h:163
TransactionId effective_catalog_xmin
Definition: slot.h:144
slock_t mutex
Definition: slot.h:120
XLogRecPtr candidate_restart_valid
Definition: slot.h:164
TransactionId effective_xmin
Definition: slot.h:143
XLogRecPtr candidate_restart_lsn
Definition: slot.h:165
TransactionId candidate_catalog_xmin
Definition: slot.h:162
ReplicationSlotPersistentData data
Definition: slot.h:147
XLogRecPtr EndRecPtr
Definition: xlogreader.h:207
Definition: c.h:676
Definition: regguts.h:318
bool TransactionIdPrecedesOrEquals(TransactionId id1, TransactionId id2)
Definition: transam.c:319
#define InvalidTransactionId
Definition: transam.h:31
#define TransactionIdIsValid(xid)
Definition: transam.h:41
bool IsTransactionOrTransactionBlock(void)
Definition: xact.c:4784
bool bsysscan
Definition: xact.c:99
TransactionId CheckXidAlive
Definition: xact.c:98
bool IsTransactionState(void)
Definition: xact.c:374
TransactionId GetTopTransactionIdIfAny(void)
Definition: xact.c:428
bool RecoveryInProgress(void)
Definition: xlog.c:5753
int wal_level
Definition: xlog.c:132
int wal_segment_size
Definition: xlog.c:144
@ WAL_LEVEL_LOGICAL
Definition: xlog.h:71
#define LSN_FORMAT_ARGS(lsn)
Definition: xlogdefs.h:43
#define XLogRecPtrIsInvalid(r)
Definition: xlogdefs.h:29
uint16 RepOriginId
Definition: xlogdefs.h:65
uint64 XLogRecPtr
Definition: xlogdefs.h:21
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
XLogRecord * XLogReadRecord(XLogReaderState *state, char **errormsg)
Definition: xlogreader.c:418
void XLogReaderFree(XLogReaderState *state)
Definition: xlogreader.c:170
XLogReaderState * XLogReaderAllocate(int wal_segment_size, const char *waldir, XLogReaderRoutine *routine, void *private_data)
Definition: xlogreader.c:108
void XLogBeginRead(XLogReaderState *state, XLogRecPtr RecPtr)
Definition: xlogreader.c:264