PostgreSQL Source Code  git master
logical.c File Reference
#include "postgres.h"
#include "access/xact.h"
#include "access/xlog_internal.h"
#include "fmgr.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "replication/decode.h"
#include "replication/logical.h"
#include "replication/origin.h"
#include "replication/reorderbuffer.h"
#include "replication/snapbuild.h"
#include "storage/proc.h"
#include "storage/procarray.h"
#include "utils/builtins.h"
#include "utils/memutils.h"
Include dependency graph for logical.c:

Go to the source code of this file.

Data Structures

struct  LogicalErrorCallbackState
 

Typedefs

typedef struct LogicalErrorCallbackState LogicalErrorCallbackState
 

Functions

static void output_plugin_error_callback (void *arg)
 
static void startup_cb_wrapper (LogicalDecodingContext *ctx, OutputPluginOptions *opt, bool is_init)
 
static void shutdown_cb_wrapper (LogicalDecodingContext *ctx)
 
static void begin_cb_wrapper (ReorderBuffer *cache, ReorderBufferTXN *txn)
 
static void commit_cb_wrapper (ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr commit_lsn)
 
static void begin_prepare_cb_wrapper (ReorderBuffer *cache, ReorderBufferTXN *txn)
 
static void prepare_cb_wrapper (ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr prepare_lsn)
 
static void commit_prepared_cb_wrapper (ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr commit_lsn)
 
static void rollback_prepared_cb_wrapper (ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr prepare_end_lsn, TimestampTz prepare_time)
 
static void change_cb_wrapper (ReorderBuffer *cache, ReorderBufferTXN *txn, Relation relation, ReorderBufferChange *change)
 
static void truncate_cb_wrapper (ReorderBuffer *cache, ReorderBufferTXN *txn, int nrelations, Relation relations[], ReorderBufferChange *change)
 
static void message_cb_wrapper (ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr message_lsn, bool transactional, const char *prefix, Size message_size, const char *message)
 
static void stream_start_cb_wrapper (ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr first_lsn)
 
static void stream_stop_cb_wrapper (ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr last_lsn)
 
static void stream_abort_cb_wrapper (ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr abort_lsn)
 
static void stream_prepare_cb_wrapper (ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr prepare_lsn)
 
static void stream_commit_cb_wrapper (ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr commit_lsn)
 
static void stream_change_cb_wrapper (ReorderBuffer *cache, ReorderBufferTXN *txn, Relation relation, ReorderBufferChange *change)
 
static void stream_message_cb_wrapper (ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr message_lsn, bool transactional, const char *prefix, Size message_size, const char *message)
 
static void stream_truncate_cb_wrapper (ReorderBuffer *cache, ReorderBufferTXN *txn, int nrelations, Relation relations[], ReorderBufferChange *change)
 
static void LoadOutputPlugin (OutputPluginCallbacks *callbacks, const char *plugin)
 
void CheckLogicalDecodingRequirements (void)
 
static LogicalDecodingContextStartupDecodingContext (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)
 
LogicalDecodingContextCreateInitDecodingContext (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)
 
LogicalDecodingContextCreateDecodingContext (XLogRecPtr start_lsn, List *output_plugin_options, bool fast_forward, XLogReaderRoutine *xl_routine, LogicalOutputPluginWriterPrepareWrite prepare_write, LogicalOutputPluginWriterWrite do_write, LogicalOutputPluginWriterUpdateProgress update_progress)
 
bool DecodingContextReady (LogicalDecodingContext *ctx)
 
void DecodingContextFindStartpoint (LogicalDecodingContext *ctx)
 
void FreeDecodingContext (LogicalDecodingContext *ctx)
 
void OutputPluginPrepareWrite (struct LogicalDecodingContext *ctx, bool last_write)
 
void OutputPluginWrite (struct LogicalDecodingContext *ctx, bool last_write)
 
void OutputPluginUpdateProgress (struct LogicalDecodingContext *ctx)
 
bool filter_prepare_cb_wrapper (LogicalDecodingContext *ctx, TransactionId xid, const char *gid)
 
bool filter_by_origin_cb_wrapper (LogicalDecodingContext *ctx, RepOriginId origin_id)
 
void LogicalIncreaseXminForSlot (XLogRecPtr current_lsn, TransactionId xmin)
 
void LogicalIncreaseRestartDecodingForSlot (XLogRecPtr current_lsn, XLogRecPtr restart_lsn)
 
void LogicalConfirmReceivedLocation (XLogRecPtr lsn)
 
void ResetLogicalStreamingState (void)
 
void UpdateDecodingStats (LogicalDecodingContext *ctx)
 

Typedef Documentation

◆ LogicalErrorCallbackState

Function Documentation

◆ begin_cb_wrapper()

static void begin_cb_wrapper ( ReorderBuffer cache,
ReorderBufferTXN txn 
)
static

Definition at line 788 of file logical.c.

789 {
790  LogicalDecodingContext *ctx = cache->private_data;
792  ErrorContextCallback errcallback;
793 
794  Assert(!ctx->fast_forward);
795 
796  /* Push callback + info on the error context stack */
797  state.ctx = ctx;
798  state.callback_name = "begin";
799  state.report_location = txn->first_lsn;
801  errcallback.arg = (void *) &state;
802  errcallback.previous = error_context_stack;
803  error_context_stack = &errcallback;
804 
805  /* set output state */
806  ctx->accept_writes = true;
807  ctx->write_xid = txn->xid;
808  ctx->write_location = txn->first_lsn;
809 
810  /* do the actual work: call callback */
811  ctx->callbacks.begin_cb(ctx, txn);
812 
813  /* Pop the error context stack */
814  error_context_stack = errcallback.previous;
815 }
ErrorContextCallback * error_context_stack
Definition: elog.c:93
Assert(fmt[strlen(fmt) - 1] !='\n')
static void output_plugin_error_callback(void *arg)
Definition: logical.c:710
struct ErrorContextCallback * previous
Definition: elog.h:232
void(* callback)(void *arg)
Definition: elog.h:233
XLogRecPtr write_location
Definition: logical.h:107
OutputPluginCallbacks callbacks
Definition: logical.h:52
TransactionId write_xid
Definition: logical.h:108
LogicalDecodeBeginCB begin_cb
XLogRecPtr first_lsn
TransactionId xid
void * private_data
Definition: regguts.h:318

References LogicalDecodingContext::accept_writes, ErrorContextCallback::arg, Assert(), OutputPluginCallbacks::begin_cb, ErrorContextCallback::callback, LogicalDecodingContext::callbacks, error_context_stack, LogicalDecodingContext::fast_forward, ReorderBufferTXN::first_lsn, output_plugin_error_callback(), ErrorContextCallback::previous, ReorderBuffer::private_data, LogicalDecodingContext::write_location, LogicalDecodingContext::write_xid, and ReorderBufferTXN::xid.

Referenced by StartupDecodingContext().

◆ begin_prepare_cb_wrapper()

static void begin_prepare_cb_wrapper ( ReorderBuffer cache,
ReorderBufferTXN txn 
)
static

Definition at line 856 of file logical.c.

857 {
858  LogicalDecodingContext *ctx = cache->private_data;
860  ErrorContextCallback errcallback;
861 
862  Assert(!ctx->fast_forward);
863 
864  /* We're only supposed to call this when two-phase commits are supported */
865  Assert(ctx->twophase);
866 
867  /* Push callback + info on the error context stack */
868  state.ctx = ctx;
869  state.callback_name = "begin_prepare";
870  state.report_location = txn->first_lsn;
872  errcallback.arg = (void *) &state;
873  errcallback.previous = error_context_stack;
874  error_context_stack = &errcallback;
875 
876  /* set output state */
877  ctx->accept_writes = true;
878  ctx->write_xid = txn->xid;
879  ctx->write_location = txn->first_lsn;
880 
881  /*
882  * If the plugin supports two-phase commits then begin prepare callback is
883  * mandatory
884  */
885  if (ctx->callbacks.begin_prepare_cb == NULL)
886  ereport(ERROR,
887  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
888  errmsg("logical replication at prepare time requires a %s callback",
889  "begin_prepare_cb")));
890 
891  /* do the actual work: call callback */
892  ctx->callbacks.begin_prepare_cb(ctx, txn);
893 
894  /* Pop the error context stack */
895  error_context_stack = errcallback.previous;
896 }
int errcode(int sqlerrcode)
Definition: elog.c:693
int errmsg(const char *fmt,...)
Definition: elog.c:904
#define ERROR
Definition: elog.h:33
#define ereport(elevel,...)
Definition: elog.h:143
LogicalDecodeBeginPrepareCB begin_prepare_cb

References LogicalDecodingContext::accept_writes, ErrorContextCallback::arg, Assert(), OutputPluginCallbacks::begin_prepare_cb, ErrorContextCallback::callback, LogicalDecodingContext::callbacks, ereport, errcode(), errmsg(), ERROR, error_context_stack, LogicalDecodingContext::fast_forward, ReorderBufferTXN::first_lsn, output_plugin_error_callback(), ErrorContextCallback::previous, ReorderBuffer::private_data, LogicalDecodingContext::twophase, LogicalDecodingContext::write_location, LogicalDecodingContext::write_xid, and ReorderBufferTXN::xid.

Referenced by StartupDecodingContext().

◆ change_cb_wrapper()

static void change_cb_wrapper ( ReorderBuffer cache,
ReorderBufferTXN txn,
Relation  relation,
ReorderBufferChange change 
)
static

Definition at line 1033 of file logical.c.

1035 {
1036  LogicalDecodingContext *ctx = cache->private_data;
1038  ErrorContextCallback errcallback;
1039 
1040  Assert(!ctx->fast_forward);
1041 
1042  /* Push callback + info on the error context stack */
1043  state.ctx = ctx;
1044  state.callback_name = "change";
1045  state.report_location = change->lsn;
1046  errcallback.callback = output_plugin_error_callback;
1047  errcallback.arg = (void *) &state;
1048  errcallback.previous = error_context_stack;
1049  error_context_stack = &errcallback;
1050 
1051  /* set output state */
1052  ctx->accept_writes = true;
1053  ctx->write_xid = txn->xid;
1054 
1055  /*
1056  * Report this change's lsn so replies from clients can give an up-to-date
1057  * answer. This won't ever be enough (and shouldn't be!) to confirm
1058  * receipt of this transaction, but it might allow another transaction's
1059  * commit to be confirmed with one message.
1060  */
1061  ctx->write_location = change->lsn;
1062 
1063  ctx->callbacks.change_cb(ctx, txn, relation, change);
1064 
1065  /* Pop the error context stack */
1066  error_context_stack = errcallback.previous;
1067 }
LogicalDecodeChangeCB change_cb

References LogicalDecodingContext::accept_writes, ErrorContextCallback::arg, Assert(), ErrorContextCallback::callback, LogicalDecodingContext::callbacks, OutputPluginCallbacks::change_cb, error_context_stack, LogicalDecodingContext::fast_forward, ReorderBufferChange::lsn, output_plugin_error_callback(), ErrorContextCallback::previous, ReorderBuffer::private_data, LogicalDecodingContext::write_location, LogicalDecodingContext::write_xid, and ReorderBufferTXN::xid.

Referenced by StartupDecodingContext().

◆ CheckLogicalDecodingRequirements()

void CheckLogicalDecodingRequirements ( void  )

Definition at line 103 of file logical.c.

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 }
Oid MyDatabaseId
Definition: globals.c:88
#define InvalidOid
Definition: postgres_ext.h:36
void CheckSlotRequirements(void)
Definition: slot.c:1044
bool RecoveryInProgress(void)
Definition: xlog.c:8404
int wal_level
Definition: xlog.c:112
@ WAL_LEVEL_LOGICAL
Definition: xlog.h:123

References CheckSlotRequirements(), ereport, errcode(), errmsg(), ERROR, InvalidOid, MyDatabaseId, RecoveryInProgress(), wal_level, and WAL_LEVEL_LOGICAL.

Referenced by copy_replication_slot(), CreateReplicationSlot(), pg_create_logical_replication_slot(), pg_logical_slot_get_changes_guts(), and StartLogicalReplication().

◆ commit_cb_wrapper()

static void commit_cb_wrapper ( ReorderBuffer cache,
ReorderBufferTXN txn,
XLogRecPtr  commit_lsn 
)
static

Definition at line 818 of file logical.c.

820 {
821  LogicalDecodingContext *ctx = cache->private_data;
823  ErrorContextCallback errcallback;
824 
825  Assert(!ctx->fast_forward);
826 
827  /* Push callback + info on the error context stack */
828  state.ctx = ctx;
829  state.callback_name = "commit";
830  state.report_location = txn->final_lsn; /* beginning of commit record */
832  errcallback.arg = (void *) &state;
833  errcallback.previous = error_context_stack;
834  error_context_stack = &errcallback;
835 
836  /* set output state */
837  ctx->accept_writes = true;
838  ctx->write_xid = txn->xid;
839  ctx->write_location = txn->end_lsn; /* points to the end of the record */
840 
841  /* do the actual work: call callback */
842  ctx->callbacks.commit_cb(ctx, txn, commit_lsn);
843 
844  /* Pop the error context stack */
845  error_context_stack = errcallback.previous;
846 }
LogicalDecodeCommitCB commit_cb
XLogRecPtr final_lsn
XLogRecPtr end_lsn

References LogicalDecodingContext::accept_writes, ErrorContextCallback::arg, Assert(), ErrorContextCallback::callback, LogicalDecodingContext::callbacks, OutputPluginCallbacks::commit_cb, ReorderBufferTXN::end_lsn, error_context_stack, LogicalDecodingContext::fast_forward, ReorderBufferTXN::final_lsn, output_plugin_error_callback(), ErrorContextCallback::previous, ReorderBuffer::private_data, LogicalDecodingContext::write_location, LogicalDecodingContext::write_xid, and ReorderBufferTXN::xid.

Referenced by StartupDecodingContext().

◆ commit_prepared_cb_wrapper()

static void commit_prepared_cb_wrapper ( ReorderBuffer cache,
ReorderBufferTXN txn,
XLogRecPtr  commit_lsn 
)
static

Definition at line 943 of file logical.c.

945 {
946  LogicalDecodingContext *ctx = cache->private_data;
948  ErrorContextCallback errcallback;
949 
950  Assert(!ctx->fast_forward);
951 
952  /* We're only supposed to call this when two-phase commits are supported */
953  Assert(ctx->twophase);
954 
955  /* Push callback + info on the error context stack */
956  state.ctx = ctx;
957  state.callback_name = "commit_prepared";
958  state.report_location = txn->final_lsn; /* beginning of commit record */
960  errcallback.arg = (void *) &state;
961  errcallback.previous = error_context_stack;
962  error_context_stack = &errcallback;
963 
964  /* set output state */
965  ctx->accept_writes = true;
966  ctx->write_xid = txn->xid;
967  ctx->write_location = txn->end_lsn; /* points to the end of the record */
968 
969  /*
970  * If the plugin support two-phase commits then commit prepared callback
971  * is mandatory
972  */
973  if (ctx->callbacks.commit_prepared_cb == NULL)
974  ereport(ERROR,
975  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
976  errmsg("logical replication at prepare time requires a %s callback",
977  "commit_prepared_cb")));
978 
979  /* do the actual work: call callback */
980  ctx->callbacks.commit_prepared_cb(ctx, txn, commit_lsn);
981 
982  /* Pop the error context stack */
983  error_context_stack = errcallback.previous;
984 }
LogicalDecodeCommitPreparedCB commit_prepared_cb

References LogicalDecodingContext::accept_writes, ErrorContextCallback::arg, Assert(), ErrorContextCallback::callback, LogicalDecodingContext::callbacks, OutputPluginCallbacks::commit_prepared_cb, ReorderBufferTXN::end_lsn, ereport, errcode(), errmsg(), ERROR, error_context_stack, LogicalDecodingContext::fast_forward, ReorderBufferTXN::final_lsn, output_plugin_error_callback(), ErrorContextCallback::previous, ReorderBuffer::private_data, LogicalDecodingContext::twophase, LogicalDecodingContext::write_location, LogicalDecodingContext::write_xid, and ReorderBufferTXN::xid.

Referenced by StartupDecodingContext().

◆ CreateDecodingContext()

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 at line 479 of file logical.c.

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 }
#define NameStr(name)
Definition: c.h:681
int errdetail(const char *fmt,...)
Definition: elog.c:1037
#define LOG
Definition: elog.h:25
#define elog(elevel,...)
Definition: elog.h:218
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
static void startup_cb_wrapper(LogicalDecodingContext *ctx, OutputPluginOptions *opt, bool is_init)
Definition: logical.c:729
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
Definition: palloc.h:109
const void * data
void ReplicationSlotMarkDirty(void)
Definition: slot.c:728
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
void ReplicationSlotSave(void)
Definition: slot.c:710
#define SlotIsPhysical(slot)
Definition: slot.h:168
void SnapBuildSetTwoPhaseAt(SnapBuild *builder, XLogRecPtr ptr)
Definition: snapbuild.c:385
OutputPluginOptions options
Definition: logical.h:53
MemoryContext context
Definition: logical.h:35
struct SnapBuild * snapshot_builder
Definition: logical.h:43
struct ReorderBuffer * reorder
Definition: logical.h:42
LogicalDecodeStartupCB startup_cb
XLogRecPtr two_phase_at
Definition: slot.h:90
XLogRecPtr restart_lsn
Definition: slot.h:73
XLogRecPtr confirmed_flush
Definition: slot.h:84
ReplicationSlotPersistentData data
Definition: slot.h:147
#define InvalidTransactionId
Definition: transam.h:31
#define LSN_FORMAT_ARGS(lsn)
Definition: xlogdefs.h:43
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28

References LogicalDecodingContext::callbacks, ReplicationSlotPersistentData::confirmed_flush, LogicalDecodingContext::context, ReplicationSlot::data, data, ReplicationSlotPersistentData::database, elog, ereport, errcode(), errdetail(), errmsg(), ERROR, InvalidTransactionId, InvalidXLogRecPtr, LOG, LSN_FORMAT_ARGS, MemoryContextSwitchTo(), MyDatabaseId, MyReplicationSlot, ReplicationSlotPersistentData::name, NameStr, LogicalDecodingContext::options, ReorderBuffer::output_rewrites, OutputPluginOptions::receive_rewrites, LogicalDecodingContext::reorder, ReplicationSlotMarkDirty(), ReplicationSlotSave(), ReplicationSlotPersistentData::restart_lsn, SlotIsPhysical, SnapBuildSetTwoPhaseAt(), LogicalDecodingContext::snapshot_builder, OutputPluginCallbacks::startup_cb, startup_cb_wrapper(), StartupDecodingContext(), ReplicationSlotPersistentData::two_phase, ReplicationSlotPersistentData::two_phase_at, LogicalDecodingContext::twophase, and LogicalDecodingContext::twophase_opt_given.

Referenced by pg_logical_replication_slot_advance(), pg_logical_slot_get_changes_guts(), and StartLogicalReplication().

◆ CreateInitDecodingContext()

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 at line 319 of file logical.c.

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 }
uint32 TransactionId
Definition: c.h:587
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1199
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1803
@ LW_EXCLUSIVE
Definition: lwlock.h:104
void namestrcpy(Name name, const char *str)
Definition: name.c:233
#define NIL
Definition: pg_list.h:65
static const char * plugin
TransactionId GetOldestSafeDecodingTransactionId(bool catalogOnly)
Definition: procarray.c:2960
void ReplicationSlotReserveWal(void)
Definition: slot.c:1081
void ReplicationSlotsComputeRequiredXmin(bool already_locked)
Definition: slot.c:767
#define SpinLockRelease(lock)
Definition: spin.h:64
#define SpinLockAcquire(lock)
Definition: spin.h:62
TransactionId catalog_xmin
Definition: slot.h:70
TransactionId effective_catalog_xmin
Definition: slot.h:144
slock_t mutex
Definition: slot.h:120
TransactionId effective_xmin
Definition: slot.h:143
Definition: c.h:676
bool IsTransactionState(void)
Definition: xact.c:373
TransactionId GetTopTransactionIdIfAny(void)
Definition: xact.c:427
#define XLogRecPtrIsInvalid(r)
Definition: xlogdefs.h:29

References LogicalDecodingContext::callbacks, ReplicationSlotPersistentData::catalog_xmin, LogicalDecodingContext::context, ReplicationSlot::data, ReplicationSlotPersistentData::database, ReplicationSlot::effective_catalog_xmin, ReplicationSlot::effective_xmin, elog, ereport, errcode(), errmsg(), ERROR, GetOldestSafeDecodingTransactionId(), GetTopTransactionIdIfAny(), InvalidTransactionId, IsTransactionState(), LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MemoryContextSwitchTo(), ReplicationSlot::mutex, MyDatabaseId, MyReplicationSlot, ReplicationSlotPersistentData::name, NameStr, namestrcpy(), NIL, LogicalDecodingContext::options, ReorderBuffer::output_rewrites, plugin, ReplicationSlotPersistentData::plugin, OutputPluginOptions::receive_rewrites, LogicalDecodingContext::reorder, ReplicationSlotMarkDirty(), ReplicationSlotReserveWal(), ReplicationSlotSave(), ReplicationSlotsComputeRequiredXmin(), ReplicationSlotPersistentData::restart_lsn, SlotIsPhysical, SpinLockAcquire, SpinLockRelease, OutputPluginCallbacks::startup_cb, startup_cb_wrapper(), StartupDecodingContext(), ReplicationSlotPersistentData::two_phase, LogicalDecodingContext::twophase, and XLogRecPtrIsInvalid.

Referenced by create_logical_replication_slot(), and CreateReplicationSlot().

◆ DecodingContextFindStartpoint()

void DecodingContextFindStartpoint ( LogicalDecodingContext ctx)

Definition at line 590 of file logical.c.

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 }
void LogicalDecodingProcessRecord(LogicalDecodingContext *ctx, XLogReaderState *record)
Definition: decode.c:91
#define DEBUG1
Definition: elog.h:24
bool DecodingContextReady(LogicalDecodingContext *ctx)
Definition: logical.c:581
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:120
XLogReaderState * reader
Definition: logical.h:41
ReplicationSlot * slot
Definition: logical.h:38
XLogRecPtr EndRecPtr
Definition: xlogreader.h:176
XLogRecord * XLogReadRecord(XLogReaderState *state, char **errormsg)
Definition: xlogreader.c:271
void XLogBeginRead(XLogReaderState *state, XLogRecPtr RecPtr)
Definition: xlogreader.c:243

References CHECK_FOR_INTERRUPTS, ReplicationSlotPersistentData::confirmed_flush, ReplicationSlot::data, DEBUG1, DecodingContextReady(), elog, XLogReaderState::EndRecPtr, ERROR, LogicalDecodingProcessRecord(), LSN_FORMAT_ARGS, ReplicationSlot::mutex, LogicalDecodingContext::reader, ReplicationSlotPersistentData::restart_lsn, LogicalDecodingContext::slot, SpinLockAcquire, SpinLockRelease, ReplicationSlotPersistentData::two_phase, ReplicationSlotPersistentData::two_phase_at, XLogBeginRead(), and XLogReadRecord().

Referenced by create_logical_replication_slot(), and CreateReplicationSlot().

◆ DecodingContextReady()

bool DecodingContextReady ( LogicalDecodingContext ctx)

Definition at line 581 of file logical.c.

582 {
584 }
SnapBuildState SnapBuildCurrentState(SnapBuild *builder)
Definition: snapbuild.c:367
@ SNAPBUILD_CONSISTENT
Definition: snapbuild.h:46

References SNAPBUILD_CONSISTENT, SnapBuildCurrentState(), and LogicalDecodingContext::snapshot_builder.

Referenced by DecodingContextFindStartpoint().

◆ filter_by_origin_cb_wrapper()

bool filter_by_origin_cb_wrapper ( LogicalDecodingContext ctx,
RepOriginId  origin_id 
)

Definition at line 1141 of file logical.c.

1142 {
1144  ErrorContextCallback errcallback;
1145  bool ret;
1146 
1147  Assert(!ctx->fast_forward);
1148 
1149  /* Push callback + info on the error context stack */
1150  state.ctx = ctx;
1151  state.callback_name = "filter_by_origin";
1152  state.report_location = InvalidXLogRecPtr;
1153  errcallback.callback = output_plugin_error_callback;
1154  errcallback.arg = (void *) &state;
1155  errcallback.previous = error_context_stack;
1156  error_context_stack = &errcallback;
1157 
1158  /* set output state */
1159  ctx->accept_writes = false;
1160 
1161  /* do the actual work: call callback */
1162  ret = ctx->callbacks.filter_by_origin_cb(ctx, origin_id);
1163 
1164  /* Pop the error context stack */
1165  error_context_stack = errcallback.previous;
1166 
1167  return ret;
1168 }
LogicalDecodeFilterByOriginCB filter_by_origin_cb

References LogicalDecodingContext::accept_writes, ErrorContextCallback::arg, Assert(), ErrorContextCallback::callback, LogicalDecodingContext::callbacks, error_context_stack, LogicalDecodingContext::fast_forward, OutputPluginCallbacks::filter_by_origin_cb, InvalidXLogRecPtr, output_plugin_error_callback(), and ErrorContextCallback::previous.

Referenced by FilterByOrigin().

◆ filter_prepare_cb_wrapper()

bool filter_prepare_cb_wrapper ( LogicalDecodingContext ctx,
TransactionId  xid,
const char *  gid 
)

Definition at line 1110 of file logical.c.

1112 {
1114  ErrorContextCallback errcallback;
1115  bool ret;
1116 
1117  Assert(!ctx->fast_forward);
1118 
1119  /* Push callback + info on the error context stack */
1120  state.ctx = ctx;
1121  state.callback_name = "filter_prepare";
1122  state.report_location = InvalidXLogRecPtr;
1123  errcallback.callback = output_plugin_error_callback;
1124  errcallback.arg = (void *) &state;
1125  errcallback.previous = error_context_stack;
1126  error_context_stack = &errcallback;
1127 
1128  /* set output state */
1129  ctx->accept_writes = false;
1130 
1131  /* do the actual work: call callback */
1132  ret = ctx->callbacks.filter_prepare_cb(ctx, xid, gid);
1133 
1134  /* Pop the error context stack */
1135  error_context_stack = errcallback.previous;
1136 
1137  return ret;
1138 }
LogicalDecodeFilterPrepareCB filter_prepare_cb

References LogicalDecodingContext::accept_writes, ErrorContextCallback::arg, Assert(), ErrorContextCallback::callback, LogicalDecodingContext::callbacks, error_context_stack, LogicalDecodingContext::fast_forward, OutputPluginCallbacks::filter_prepare_cb, InvalidXLogRecPtr, output_plugin_error_callback(), and ErrorContextCallback::previous.

Referenced by FilterPrepare().

◆ FreeDecodingContext()

void FreeDecodingContext ( LogicalDecodingContext ctx)

Definition at line 634 of file logical.c.

635 {
636  if (ctx->callbacks.shutdown_cb != NULL)
637  shutdown_cb_wrapper(ctx);
638 
641  XLogReaderFree(ctx->reader);
643 }
static void shutdown_cb_wrapper(LogicalDecodingContext *ctx)
Definition: logical.c:756
void MemoryContextDelete(MemoryContext context)
Definition: mcxt.c:218
void ReorderBufferFree(ReorderBuffer *rb)
void FreeSnapshotBuilder(SnapBuild *builder)
Definition: snapbuild.c:323
LogicalDecodeShutdownCB shutdown_cb
void XLogReaderFree(XLogReaderState *state)
Definition: xlogreader.c:142

References LogicalDecodingContext::callbacks, LogicalDecodingContext::context, FreeSnapshotBuilder(), MemoryContextDelete(), LogicalDecodingContext::reader, LogicalDecodingContext::reorder, ReorderBufferFree(), OutputPluginCallbacks::shutdown_cb, shutdown_cb_wrapper(), LogicalDecodingContext::snapshot_builder, and XLogReaderFree().

Referenced by create_logical_replication_slot(), CreateReplicationSlot(), pg_logical_replication_slot_advance(), pg_logical_slot_get_changes_guts(), and StartLogicalReplication().

◆ LoadOutputPlugin()

static void LoadOutputPlugin ( OutputPluginCallbacks callbacks,
const char *  plugin 
)
static

Definition at line 688 of file logical.c.

689 {
690  LogicalOutputPluginInit plugin_init;
691 
692  plugin_init = (LogicalOutputPluginInit)
693  load_external_function(plugin, "_PG_output_plugin_init", false, NULL);
694 
695  if (plugin_init == NULL)
696  elog(ERROR, "output plugins have to declare the _PG_output_plugin_init symbol");
697 
698  /* ask the output plugin to fill the callback struct */
699  plugin_init(callbacks);
700 
701  if (callbacks->begin_cb == NULL)
702  elog(ERROR, "output plugins have to register a begin callback");
703  if (callbacks->change_cb == NULL)
704  elog(ERROR, "output plugins have to register a change callback");
705  if (callbacks->commit_cb == NULL)
706  elog(ERROR, "output plugins have to register a commit callback");
707 }
void * load_external_function(const char *filename, const char *funcname, bool signalNotFound, void **filehandle)
Definition: dfmgr.c:107
void(* LogicalOutputPluginInit)(struct OutputPluginCallbacks *cb)
Definition: output_plugin.h:36

References OutputPluginCallbacks::begin_cb, OutputPluginCallbacks::change_cb, OutputPluginCallbacks::commit_cb, elog, ERROR, load_external_function(), and plugin.

Referenced by StartupDecodingContext().

◆ LogicalConfirmReceivedLocation()

void LogicalConfirmReceivedLocation ( XLogRecPtr  lsn)

Definition at line 1707 of file logical.c.

1708 {
1709  Assert(lsn != InvalidXLogRecPtr);
1710 
1711  /* Do an unlocked check for candidate_lsn first. */
1714  {
1715  bool updated_xmin = false;
1716  bool updated_restart = false;
1717 
1719 
1721 
1722  /* if we're past the location required for bumping xmin, do so */
1725  {
1726  /*
1727  * We have to write the changed xmin to disk *before* we change
1728  * the in-memory value, otherwise after a crash we wouldn't know
1729  * that some catalog tuples might have been removed already.
1730  *
1731  * Ensure that by first writing to ->xmin and only update
1732  * ->effective_xmin once the new state is synced to disk. After a
1733  * crash ->effective_xmin is set to ->xmin.
1734  */
1737  {
1741  updated_xmin = true;
1742  }
1743  }
1744 
1747  {
1749 
1753  updated_restart = true;
1754  }
1755 
1757 
1758  /* first write new xmin to disk, so we know what's up after a crash */
1759  if (updated_xmin || updated_restart)
1760  {
1763  elog(DEBUG1, "updated xmin: %u restart: %u", updated_xmin, updated_restart);
1764  }
1765 
1766  /*
1767  * Now the new xmin is safely on disk, we can let the global value
1768  * advance. We do not take ProcArrayLock or similar since we only
1769  * advance xmin here and there's not much harm done by a concurrent
1770  * computation missing that.
1771  */
1772  if (updated_xmin)
1773  {
1777 
1780  }
1781  }
1782  else
1783  {
1787  }
1788 }
void ReplicationSlotsComputeRequiredLSN(void)
Definition: slot.c:817
XLogRecPtr candidate_xmin_lsn
Definition: slot.h:163
XLogRecPtr candidate_restart_valid
Definition: slot.h:164
XLogRecPtr candidate_restart_lsn
Definition: slot.h:165
TransactionId candidate_catalog_xmin
Definition: slot.h:162
#define TransactionIdIsValid(xid)
Definition: transam.h:41

References Assert(), ReplicationSlot::candidate_catalog_xmin, ReplicationSlot::candidate_restart_lsn, ReplicationSlot::candidate_restart_valid, ReplicationSlot::candidate_xmin_lsn, ReplicationSlotPersistentData::catalog_xmin, ReplicationSlotPersistentData::confirmed_flush, ReplicationSlot::data, DEBUG1, ReplicationSlot::effective_catalog_xmin, elog, InvalidTransactionId, InvalidXLogRecPtr, ReplicationSlot::mutex, MyReplicationSlot, ReplicationSlotMarkDirty(), ReplicationSlotSave(), ReplicationSlotsComputeRequiredLSN(), ReplicationSlotsComputeRequiredXmin(), ReplicationSlotPersistentData::restart_lsn, SpinLockAcquire, SpinLockRelease, and TransactionIdIsValid.

Referenced by LogicalIncreaseRestartDecodingForSlot(), LogicalIncreaseXminForSlot(), pg_logical_replication_slot_advance(), pg_logical_slot_get_changes_guts(), and ProcessStandbyReplyMessage().

◆ LogicalIncreaseRestartDecodingForSlot()

void LogicalIncreaseRestartDecodingForSlot ( XLogRecPtr  current_lsn,
XLogRecPtr  restart_lsn 
)

Definition at line 1633 of file logical.c.

1634 {
1635  bool updated_lsn = false;
1636  ReplicationSlot *slot;
1637 
1638  slot = MyReplicationSlot;
1639 
1640  Assert(slot != NULL);
1641  Assert(restart_lsn != InvalidXLogRecPtr);
1642  Assert(current_lsn != InvalidXLogRecPtr);
1643 
1644  SpinLockAcquire(&slot->mutex);
1645 
1646  /* don't overwrite if have a newer restart lsn */
1647  if (restart_lsn <= slot->data.restart_lsn)
1648  {
1649  }
1650 
1651  /*
1652  * We might have already flushed far enough to directly accept this lsn,
1653  * in this case there is no need to check for existing candidate LSNs
1654  */
1655  else if (current_lsn <= slot->data.confirmed_flush)
1656  {
1657  slot->candidate_restart_valid = current_lsn;
1658  slot->candidate_restart_lsn = restart_lsn;
1659 
1660  /* our candidate can directly be used */
1661  updated_lsn = true;
1662  }
1663 
1664  /*
1665  * Only increase if the previous values have been applied, otherwise we
1666  * might never end up updating if the receiver acks too slowly. A missed
1667  * value here will just cause some extra effort after reconnecting.
1668  */
1670  {
1671  slot->candidate_restart_valid = current_lsn;
1672  slot->candidate_restart_lsn = restart_lsn;
1673  SpinLockRelease(&slot->mutex);
1674 
1675  elog(DEBUG1, "got new restart lsn %X/%X at %X/%X",
1676  LSN_FORMAT_ARGS(restart_lsn),
1677  LSN_FORMAT_ARGS(current_lsn));
1678  }
1679  else
1680  {
1681  XLogRecPtr candidate_restart_lsn;
1682  XLogRecPtr candidate_restart_valid;
1683  XLogRecPtr confirmed_flush;
1684 
1685  candidate_restart_lsn = slot->candidate_restart_lsn;
1686  candidate_restart_valid = slot->candidate_restart_valid;
1687  confirmed_flush = slot->data.confirmed_flush;
1688  SpinLockRelease(&slot->mutex);
1689 
1690  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",
1691  LSN_FORMAT_ARGS(restart_lsn),
1692  LSN_FORMAT_ARGS(current_lsn),
1693  LSN_FORMAT_ARGS(candidate_restart_lsn),
1694  LSN_FORMAT_ARGS(candidate_restart_valid),
1695  LSN_FORMAT_ARGS(confirmed_flush));
1696  }
1697 
1698  /* candidates are already valid with the current flush position, apply */
1699  if (updated_lsn)
1701 }
void LogicalConfirmReceivedLocation(XLogRecPtr lsn)
Definition: logical.c:1707
uint64 XLogRecPtr
Definition: xlogdefs.h:21

References Assert(), ReplicationSlot::candidate_restart_lsn, ReplicationSlot::candidate_restart_valid, ReplicationSlotPersistentData::confirmed_flush, ReplicationSlot::data, data, DEBUG1, elog, InvalidXLogRecPtr, LogicalConfirmReceivedLocation(), LSN_FORMAT_ARGS, ReplicationSlot::mutex, MyReplicationSlot, SpinLockAcquire, and SpinLockRelease.

Referenced by SnapBuildProcessRunningXacts().

◆ LogicalIncreaseXminForSlot()

void LogicalIncreaseXminForSlot ( XLogRecPtr  current_lsn,
TransactionId  xmin 
)

Definition at line 1565 of file logical.c.

1566 {
1567  bool updated_xmin = false;
1568  ReplicationSlot *slot;
1569  bool got_new_xmin = false;
1570 
1571  slot = MyReplicationSlot;
1572 
1573  Assert(slot != NULL);
1574 
1575  SpinLockAcquire(&slot->mutex);
1576 
1577  /*
1578  * don't overwrite if we already have a newer xmin. This can happen if we
1579  * restart decoding in a slot.
1580  */
1582  {
1583  }
1584 
1585  /*
1586  * If the client has already confirmed up to this lsn, we directly can
1587  * mark this as accepted. This can happen if we restart decoding in a
1588  * slot.
1589  */
1590  else if (current_lsn <= slot->data.confirmed_flush)
1591  {
1592  slot->candidate_catalog_xmin = xmin;
1593  slot->candidate_xmin_lsn = current_lsn;
1594 
1595  /* our candidate can directly be used */
1596  updated_xmin = true;
1597  }
1598 
1599  /*
1600  * Only increase if the previous values have been applied, otherwise we
1601  * might never end up updating if the receiver acks too slowly.
1602  */
1603  else if (slot->candidate_xmin_lsn == InvalidXLogRecPtr)
1604  {
1605  slot->candidate_catalog_xmin = xmin;
1606  slot->candidate_xmin_lsn = current_lsn;
1607 
1608  /*
1609  * Log new xmin at an appropriate log level after releasing the
1610  * spinlock.
1611  */
1612  got_new_xmin = true;
1613  }
1614  SpinLockRelease(&slot->mutex);
1615 
1616  if (got_new_xmin)
1617  elog(DEBUG1, "got new catalog xmin %u at %X/%X", xmin,
1618  LSN_FORMAT_ARGS(current_lsn));
1619 
1620  /* candidate already valid with the current flush position, apply */
1621  if (updated_xmin)
1623 }
bool TransactionIdPrecedesOrEquals(TransactionId id1, TransactionId id2)
Definition: transam.c:319

References Assert(), ReplicationSlot::candidate_catalog_xmin, ReplicationSlot::candidate_xmin_lsn, ReplicationSlotPersistentData::catalog_xmin, ReplicationSlotPersistentData::confirmed_flush, ReplicationSlot::data, data, DEBUG1, elog, InvalidXLogRecPtr, LogicalConfirmReceivedLocation(), LSN_FORMAT_ARGS, ReplicationSlot::mutex, MyReplicationSlot, SpinLockAcquire, SpinLockRelease, and TransactionIdPrecedesOrEquals().

Referenced by SnapBuildProcessRunningXacts().

◆ message_cb_wrapper()

static void message_cb_wrapper ( ReorderBuffer cache,
ReorderBufferTXN txn,
XLogRecPtr  message_lsn,
bool  transactional,
const char *  prefix,
Size  message_size,
const char *  message 
)
static

Definition at line 1171 of file logical.c.

1174 {
1175  LogicalDecodingContext *ctx = cache->private_data;
1177  ErrorContextCallback errcallback;
1178 
1179  Assert(!ctx->fast_forward);
1180 
1181  if (ctx->callbacks.message_cb == NULL)
1182  return;
1183 
1184  /* Push callback + info on the error context stack */
1185  state.ctx = ctx;
1186  state.callback_name = "message";
1187  state.report_location = message_lsn;
1188  errcallback.callback = output_plugin_error_callback;
1189  errcallback.arg = (void *) &state;
1190  errcallback.previous = error_context_stack;
1191  error_context_stack = &errcallback;
1192 
1193  /* set output state */
1194  ctx->accept_writes = true;
1195  ctx->write_xid = txn != NULL ? txn->xid : InvalidTransactionId;
1196  ctx->write_location = message_lsn;
1197 
1198  /* do the actual work: call callback */
1199  ctx->callbacks.message_cb(ctx, txn, message_lsn, transactional, prefix,
1200  message_size, message);
1201 
1202  /* Pop the error context stack */
1203  error_context_stack = errcallback.previous;
1204 }
LogicalDecodeMessageCB message_cb

References LogicalDecodingContext::accept_writes, ErrorContextCallback::arg, Assert(), ErrorContextCallback::callback, LogicalDecodingContext::callbacks, error_context_stack, LogicalDecodingContext::fast_forward, InvalidTransactionId, OutputPluginCallbacks::message_cb, output_plugin_error_callback(), ErrorContextCallback::previous, ReorderBuffer::private_data, LogicalDecodingContext::write_location, LogicalDecodingContext::write_xid, and ReorderBufferTXN::xid.

Referenced by StartupDecodingContext().

◆ output_plugin_error_callback()

static void output_plugin_error_callback ( void *  arg)
static

Definition at line 710 of file logical.c.

711 {
713 
714  /* not all callbacks have an associated LSN */
715  if (state->report_location != InvalidXLogRecPtr)
716  errcontext("slot \"%s\", output plugin \"%s\", in the %s callback, associated LSN %X/%X",
717  NameStr(state->ctx->slot->data.name),
718  NameStr(state->ctx->slot->data.plugin),
719  state->callback_name,
720  LSN_FORMAT_ARGS(state->report_location));
721  else
722  errcontext("slot \"%s\", output plugin \"%s\", in the %s callback",
723  NameStr(state->ctx->slot->data.name),
724  NameStr(state->ctx->slot->data.plugin),
725  state->callback_name);
726 }
#define errcontext
Definition: elog.h:190
void * arg

References arg, errcontext, InvalidXLogRecPtr, LSN_FORMAT_ARGS, and NameStr.

Referenced by begin_cb_wrapper(), begin_prepare_cb_wrapper(), change_cb_wrapper(), commit_cb_wrapper(), commit_prepared_cb_wrapper(), filter_by_origin_cb_wrapper(), filter_prepare_cb_wrapper(), message_cb_wrapper(), prepare_cb_wrapper(), rollback_prepared_cb_wrapper(), shutdown_cb_wrapper(), startup_cb_wrapper(), stream_abort_cb_wrapper(), stream_change_cb_wrapper(), stream_commit_cb_wrapper(), stream_message_cb_wrapper(), stream_prepare_cb_wrapper(), stream_start_cb_wrapper(), stream_stop_cb_wrapper(), stream_truncate_cb_wrapper(), and truncate_cb_wrapper().

◆ OutputPluginPrepareWrite()

void OutputPluginPrepareWrite ( struct LogicalDecodingContext ctx,
bool  last_write 
)

Definition at line 649 of file logical.c.

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 }
LogicalOutputPluginWriterPrepareWrite prepare_write
Definition: logical.h:63

References LogicalDecodingContext::accept_writes, elog, ERROR, LogicalDecodingContext::prepare_write, LogicalDecodingContext::prepared_write, LogicalDecodingContext::write_location, and LogicalDecodingContext::write_xid.

Referenced by pg_decode_change(), pg_decode_commit_prepared_txn(), pg_decode_commit_txn(), pg_decode_message(), pg_decode_prepare_txn(), pg_decode_rollback_prepared_txn(), pg_decode_stream_abort(), pg_decode_stream_change(), pg_decode_stream_commit(), pg_decode_stream_message(), pg_decode_stream_prepare(), pg_decode_stream_stop(), pg_decode_stream_truncate(), pg_decode_truncate(), pg_output_begin(), pg_output_stream_start(), pgoutput_begin_prepare_txn(), pgoutput_begin_txn(), pgoutput_change(), pgoutput_commit_prepared_txn(), pgoutput_commit_txn(), pgoutput_message(), pgoutput_prepare_txn(), pgoutput_rollback_prepared_txn(), pgoutput_stream_abort(), pgoutput_stream_commit(), pgoutput_stream_prepare_txn(), pgoutput_stream_start(), pgoutput_stream_stop(), pgoutput_truncate(), send_relation_and_attrs(), and send_repl_origin().

◆ OutputPluginUpdateProgress()

void OutputPluginUpdateProgress ( struct LogicalDecodingContext ctx)

◆ OutputPluginWrite()

◆ prepare_cb_wrapper()

static void prepare_cb_wrapper ( ReorderBuffer cache,
ReorderBufferTXN txn,
XLogRecPtr  prepare_lsn 
)
static

Definition at line 899 of file logical.c.

901 {
902  LogicalDecodingContext *ctx = cache->private_data;
904  ErrorContextCallback errcallback;
905 
906  Assert(!ctx->fast_forward);
907 
908  /* We're only supposed to call this when two-phase commits are supported */
909  Assert(ctx->twophase);
910 
911  /* Push callback + info on the error context stack */
912  state.ctx = ctx;
913  state.callback_name = "prepare";
914  state.report_location = txn->final_lsn; /* beginning of prepare record */
916  errcallback.arg = (void *) &state;
917  errcallback.previous = error_context_stack;
918  error_context_stack = &errcallback;
919 
920  /* set output state */
921  ctx->accept_writes = true;
922  ctx->write_xid = txn->xid;
923  ctx->write_location = txn->end_lsn; /* points to the end of the record */
924 
925  /*
926  * If the plugin supports two-phase commits then prepare callback is
927  * mandatory
928  */
929  if (ctx->callbacks.prepare_cb == NULL)
930  ereport(ERROR,
931  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
932  errmsg("logical replication at prepare time requires a %s callback",
933  "prepare_cb")));
934 
935  /* do the actual work: call callback */
936  ctx->callbacks.prepare_cb(ctx, txn, prepare_lsn);
937 
938  /* Pop the error context stack */
939  error_context_stack = errcallback.previous;
940 }
LogicalDecodePrepareCB prepare_cb

References LogicalDecodingContext::accept_writes, ErrorContextCallback::arg, Assert(), ErrorContextCallback::callback, LogicalDecodingContext::callbacks, ReorderBufferTXN::end_lsn, ereport, errcode(), errmsg(), ERROR, error_context_stack, LogicalDecodingContext::fast_forward, ReorderBufferTXN::final_lsn, output_plugin_error_callback(), OutputPluginCallbacks::prepare_cb, ErrorContextCallback::previous, ReorderBuffer::private_data, LogicalDecodingContext::twophase, LogicalDecodingContext::write_location, LogicalDecodingContext::write_xid, and ReorderBufferTXN::xid.

Referenced by StartupDecodingContext().

◆ ResetLogicalStreamingState()

void ResetLogicalStreamingState ( void  )

Definition at line 1794 of file logical.c.

1795 {
1797  bsysscan = false;
1798 }
bool bsysscan
Definition: xact.c:98
TransactionId CheckXidAlive
Definition: xact.c:97

References bsysscan, CheckXidAlive, and InvalidTransactionId.

Referenced by AbortSubTransaction(), and AbortTransaction().

◆ rollback_prepared_cb_wrapper()

static void rollback_prepared_cb_wrapper ( ReorderBuffer cache,
ReorderBufferTXN txn,
XLogRecPtr  prepare_end_lsn,
TimestampTz  prepare_time 
)
static

Definition at line 987 of file logical.c.

990 {
991  LogicalDecodingContext *ctx = cache->private_data;
993  ErrorContextCallback errcallback;
994 
995  Assert(!ctx->fast_forward);
996 
997  /* We're only supposed to call this when two-phase commits are supported */
998  Assert(ctx->twophase);
999 
1000  /* Push callback + info on the error context stack */
1001  state.ctx = ctx;
1002  state.callback_name = "rollback_prepared";
1003  state.report_location = txn->final_lsn; /* beginning of commit record */
1004  errcallback.callback = output_plugin_error_callback;
1005  errcallback.arg = (void *) &state;
1006  errcallback.previous = error_context_stack;
1007  error_context_stack = &errcallback;
1008 
1009  /* set output state */
1010  ctx->accept_writes = true;
1011  ctx->write_xid = txn->xid;
1012  ctx->write_location = txn->end_lsn; /* points to the end of the record */
1013 
1014  /*
1015  * If the plugin support two-phase commits then rollback prepared callback
1016  * is mandatory
1017  */
1018  if (ctx->callbacks.rollback_prepared_cb == NULL)
1019  ereport(ERROR,
1020  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1021  errmsg("logical replication at prepare time requires a %s callback",
1022  "rollback_prepared_cb")));
1023 
1024  /* do the actual work: call callback */
1025  ctx->callbacks.rollback_prepared_cb(ctx, txn, prepare_end_lsn,
1026  prepare_time);
1027 
1028  /* Pop the error context stack */
1029  error_context_stack = errcallback.previous;
1030 }
LogicalDecodeRollbackPreparedCB rollback_prepared_cb

References LogicalDecodingContext::accept_writes, ErrorContextCallback::arg, Assert(), ErrorContextCallback::callback, LogicalDecodingContext::callbacks, ReorderBufferTXN::end_lsn, ereport, errcode(), errmsg(), ERROR, error_context_stack, LogicalDecodingContext::fast_forward, ReorderBufferTXN::final_lsn, output_plugin_error_callback(), ErrorContextCallback::previous, ReorderBuffer::private_data, OutputPluginCallbacks::rollback_prepared_cb, LogicalDecodingContext::twophase, LogicalDecodingContext::write_location, LogicalDecodingContext::write_xid, and ReorderBufferTXN::xid.

Referenced by StartupDecodingContext().

◆ shutdown_cb_wrapper()

static void shutdown_cb_wrapper ( LogicalDecodingContext ctx)
static

Definition at line 756 of file logical.c.

757 {
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 = "shutdown";
766  state.report_location = InvalidXLogRecPtr;
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 = false;
774 
775  /* do the actual work: call callback */
776  ctx->callbacks.shutdown_cb(ctx);
777 
778  /* Pop the error context stack */
779  error_context_stack = errcallback.previous;
780 }

References LogicalDecodingContext::accept_writes, ErrorContextCallback::arg, Assert(), ErrorContextCallback::callback, LogicalDecodingContext::callbacks, error_context_stack, LogicalDecodingContext::fast_forward, InvalidXLogRecPtr, output_plugin_error_callback(), ErrorContextCallback::previous, and OutputPluginCallbacks::shutdown_cb.

Referenced by FreeDecodingContext().

◆ startup_cb_wrapper()

static void startup_cb_wrapper ( LogicalDecodingContext ctx,
OutputPluginOptions opt,
bool  is_init 
)
static

Definition at line 729 of file logical.c.

730 {
732  ErrorContextCallback errcallback;
733 
734  Assert(!ctx->fast_forward);
735 
736  /* Push callback + info on the error context stack */
737  state.ctx = ctx;
738  state.callback_name = "startup";
739  state.report_location = InvalidXLogRecPtr;
741  errcallback.arg = (void *) &state;
742  errcallback.previous = error_context_stack;
743  error_context_stack = &errcallback;
744 
745  /* set output state */
746  ctx->accept_writes = false;
747 
748  /* do the actual work: call callback */
749  ctx->callbacks.startup_cb(ctx, opt, is_init);
750 
751  /* Pop the error context stack */
752  error_context_stack = errcallback.previous;
753 }

References LogicalDecodingContext::accept_writes, ErrorContextCallback::arg, Assert(), ErrorContextCallback::callback, LogicalDecodingContext::callbacks, error_context_stack, LogicalDecodingContext::fast_forward, InvalidXLogRecPtr, output_plugin_error_callback(), ErrorContextCallback::previous, and OutputPluginCallbacks::startup_cb.

Referenced by CreateDecodingContext(), and CreateInitDecodingContext().

◆ StartupDecodingContext()

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 
)
static

Definition at line 146 of file logical.c.

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 }
static void change_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, Relation relation, ReorderBufferChange *change)
Definition: logical.c:1033
static void commit_prepared_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr commit_lsn)
Definition: logical.c:943
static void stream_start_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr first_lsn)
Definition: logical.c:1207
static void commit_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr commit_lsn)
Definition: logical.c:818
static void begin_prepare_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn)
Definition: logical.c:856
static void stream_prepare_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr prepare_lsn)
Definition: logical.c:1341
static void prepare_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr prepare_lsn)
Definition: logical.c:899
static void stream_truncate_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, int nrelations, Relation relations[], ReorderBufferChange *change)
Definition: logical.c:1512
static void truncate_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, int nrelations, Relation relations[], ReorderBufferChange *change)
Definition: logical.c:1070
static void begin_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn)
Definition: logical.c:788
static void rollback_prepared_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr prepare_end_lsn, TimestampTz prepare_time)
Definition: logical.c:987
static void stream_change_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, Relation relation, ReorderBufferChange *change)
Definition: logical.c:1425
static void stream_abort_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr abort_lsn)
Definition: logical.c:1301
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:1472
static void stream_commit_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr commit_lsn)
Definition: logical.c:1385
static void LoadOutputPlugin(OutputPluginCallbacks *callbacks, const char *plugin)
Definition: logical.c:688
static void stream_stop_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr last_lsn)
Definition: logical.c:1254
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:1171
void * palloc0(Size size)
Definition: mcxt.c:1093
MemoryContext CurrentMemoryContext
Definition: mcxt.c:42
#define AllocSetContextCreate
Definition: memutils.h:173
#define ALLOCSET_DEFAULT_SIZES
Definition: memutils.h:195
#define PROC_IN_LOGICAL_DECODING
Definition: proc.h:58
ReorderBuffer * ReorderBufferAllocate(void)
SnapBuild * AllocateSnapshotBuilder(ReorderBuffer *reorder, TransactionId xmin_horizon, XLogRecPtr start_lsn, bool need_full_snapshot, XLogRecPtr two_phase_at)
Definition: snapbuild.c:280
PGPROC * MyProc
Definition: proc.c:68
PROC_HDR * ProcGlobal
Definition: proc.c:80
StringInfo makeStringInfo(void)
Definition: stringinfo.c:41
StringInfo out
Definition: logical.h:70
List * output_plugin_options
Definition: logical.h:58
LogicalDecodeStreamChangeCB stream_change_cb
LogicalDecodeStreamTruncateCB stream_truncate_cb
LogicalDecodeStreamMessageCB stream_message_cb
LogicalDecodeStreamStopCB stream_stop_cb
LogicalDecodeStreamCommitCB stream_commit_cb
LogicalDecodeStreamPrepareCB stream_prepare_cb
LogicalDecodeStreamStartCB stream_start_cb
LogicalDecodeStreamAbortCB stream_abort_cb
uint8 statusFlags
Definition: proc.h:192
int pgxactoff
Definition: proc.h:151
uint8 * statusFlags
Definition: proc.h:336
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
bool IsTransactionOrTransactionBlock(void)
Definition: xact.c:4772
int wal_segment_size
Definition: xlog.c:123
XLogReaderState * XLogReaderAllocate(int wal_segment_size, const char *waldir, XLogReaderRoutine *routine, void *private_data)
Definition: xlogreader.c:78

References AllocateSnapshotBuilder(), ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate, ReorderBuffer::apply_change, ReorderBuffer::apply_truncate, ReorderBuffer::begin, begin_cb_wrapper(), ReorderBuffer::begin_prepare, OutputPluginCallbacks::begin_prepare_cb, begin_prepare_cb_wrapper(), LogicalDecodingContext::callbacks, change_cb_wrapper(), ReorderBuffer::commit, commit_cb_wrapper(), ReorderBuffer::commit_prepared, OutputPluginCallbacks::commit_prepared_cb, commit_prepared_cb_wrapper(), LogicalDecodingContext::context, CurrentMemoryContext, ReplicationSlot::data, ereport, errcode(), errdetail(), errmsg(), ERROR, LogicalDecodingContext::fast_forward, OutputPluginCallbacks::filter_prepare_cb, IsTransactionOrTransactionBlock(), LoadOutputPlugin(), LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), makeStringInfo(), MemoryContextSwitchTo(), ReorderBuffer::message, message_cb_wrapper(), MyProc, MyReplicationSlot, NameStr, LogicalDecodingContext::out, LogicalDecodingContext::output_plugin_options, palloc0(), PGPROC::pgxactoff, ReplicationSlotPersistentData::plugin, ReorderBuffer::prepare, OutputPluginCallbacks::prepare_cb, prepare_cb_wrapper(), LogicalDecodingContext::prepare_write, ReorderBuffer::private_data, PROC_IN_LOGICAL_DECODING, ProcGlobal, LogicalDecodingContext::reader, LogicalDecodingContext::reorder, ReorderBufferAllocate(), ReorderBuffer::rollback_prepared, OutputPluginCallbacks::rollback_prepared_cb, rollback_prepared_cb_wrapper(), LogicalDecodingContext::slot, LogicalDecodingContext::snapshot_builder, PGPROC::statusFlags, PROC_HDR::statusFlags, ReorderBuffer::stream_abort, OutputPluginCallbacks::stream_abort_cb, stream_abort_cb_wrapper(), ReorderBuffer::stream_change, OutputPluginCallbacks::stream_change_cb, stream_change_cb_wrapper(), ReorderBuffer::stream_commit, OutputPluginCallbacks::stream_commit_cb, stream_commit_cb_wrapper(), ReorderBuffer::stream_message, OutputPluginCallbacks::stream_message_cb, stream_message_cb_wrapper(), ReorderBuffer::stream_prepare, OutputPluginCallbacks::stream_prepare_cb, stream_prepare_cb_wrapper(), ReorderBuffer::stream_start, OutputPluginCallbacks::stream_start_cb, stream_start_cb_wrapper(), ReorderBuffer::stream_stop, OutputPluginCallbacks::stream_stop_cb, stream_stop_cb_wrapper(), ReorderBuffer::stream_truncate, OutputPluginCallbacks::stream_truncate_cb, stream_truncate_cb_wrapper(), LogicalDecodingContext::streaming, truncate_cb_wrapper(), ReplicationSlotPersistentData::two_phase_at, LogicalDecodingContext::twophase, LogicalDecodingContext::update_progress, wal_segment_size, LogicalDecodingContext::write, and XLogReaderAllocate().

Referenced by CreateDecodingContext(), and CreateInitDecodingContext().

◆ stream_abort_cb_wrapper()

static void stream_abort_cb_wrapper ( ReorderBuffer cache,
ReorderBufferTXN txn,
XLogRecPtr  abort_lsn 
)
static

Definition at line 1301 of file logical.c.

1303 {
1304  LogicalDecodingContext *ctx = cache->private_data;
1306  ErrorContextCallback errcallback;
1307 
1308  Assert(!ctx->fast_forward);
1309 
1310  /* We're only supposed to call this when streaming is supported. */
1311  Assert(ctx->streaming);
1312 
1313  /* Push callback + info on the error context stack */
1314  state.ctx = ctx;
1315  state.callback_name = "stream_abort";
1316  state.report_location = abort_lsn;
1317  errcallback.callback = output_plugin_error_callback;
1318  errcallback.arg = (void *) &state;
1319  errcallback.previous = error_context_stack;
1320  error_context_stack = &errcallback;
1321 
1322  /* set output state */
1323  ctx->accept_writes = true;
1324  ctx->write_xid = txn->xid;
1325  ctx->write_location = abort_lsn;
1326 
1327  /* in streaming mode, stream_abort_cb is required */
1328  if (ctx->callbacks.stream_abort_cb == NULL)
1329  ereport(ERROR,
1330  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1331  errmsg("logical streaming requires a %s callback",
1332  "stream_abort_cb")));
1333 
1334  ctx->callbacks.stream_abort_cb(ctx, txn, abort_lsn);
1335 
1336  /* Pop the error context stack */
1337  error_context_stack = errcallback.previous;
1338 }

References LogicalDecodingContext::accept_writes, ErrorContextCallback::arg, Assert(), ErrorContextCallback::callback, LogicalDecodingContext::callbacks, ereport, errcode(), errmsg(), ERROR, error_context_stack, LogicalDecodingContext::fast_forward, output_plugin_error_callback(), ErrorContextCallback::previous, ReorderBuffer::private_data, OutputPluginCallbacks::stream_abort_cb, LogicalDecodingContext::streaming, LogicalDecodingContext::write_location, LogicalDecodingContext::write_xid, and ReorderBufferTXN::xid.

Referenced by StartupDecodingContext().

◆ stream_change_cb_wrapper()

static void stream_change_cb_wrapper ( ReorderBuffer cache,
ReorderBufferTXN txn,
Relation  relation,
ReorderBufferChange change 
)
static

Definition at line 1425 of file logical.c.

1427 {
1428  LogicalDecodingContext *ctx = cache->private_data;
1430  ErrorContextCallback errcallback;
1431 
1432  Assert(!ctx->fast_forward);
1433 
1434  /* We're only supposed to call this when streaming is supported. */
1435  Assert(ctx->streaming);
1436 
1437  /* Push callback + info on the error context stack */
1438  state.ctx = ctx;
1439  state.callback_name = "stream_change";
1440  state.report_location = change->lsn;
1441  errcallback.callback = output_plugin_error_callback;
1442  errcallback.arg = (void *) &state;
1443  errcallback.previous = error_context_stack;
1444  error_context_stack = &errcallback;
1445 
1446  /* set output state */
1447  ctx->accept_writes = true;
1448  ctx->write_xid = txn->xid;
1449 
1450  /*
1451  * Report this change's lsn so replies from clients can give an up-to-date
1452  * answer. This won't ever be enough (and shouldn't be!) to confirm
1453  * receipt of this transaction, but it might allow another transaction's
1454  * commit to be confirmed with one message.
1455  */
1456  ctx->write_location = change->lsn;
1457 
1458  /* in streaming mode, stream_change_cb is required */
1459  if (ctx->callbacks.stream_change_cb == NULL)
1460  ereport(ERROR,
1461  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1462  errmsg("logical streaming requires a %s callback",
1463  "stream_change_cb")));
1464 
1465  ctx->callbacks.stream_change_cb(ctx, txn, relation, change);
1466 
1467  /* Pop the error context stack */
1468  error_context_stack = errcallback.previous;
1469 }

References LogicalDecodingContext::accept_writes, ErrorContextCallback::arg, Assert(), ErrorContextCallback::callback, LogicalDecodingContext::callbacks, ereport, errcode(), errmsg(), ERROR, error_context_stack, LogicalDecodingContext::fast_forward, ReorderBufferChange::lsn, output_plugin_error_callback(), ErrorContextCallback::previous, ReorderBuffer::private_data, OutputPluginCallbacks::stream_change_cb, LogicalDecodingContext::streaming, LogicalDecodingContext::write_location, LogicalDecodingContext::write_xid, and ReorderBufferTXN::xid.

Referenced by StartupDecodingContext().

◆ stream_commit_cb_wrapper()

static void stream_commit_cb_wrapper ( ReorderBuffer cache,
ReorderBufferTXN txn,
XLogRecPtr  commit_lsn 
)
static

Definition at line 1385 of file logical.c.

1387 {
1388  LogicalDecodingContext *ctx = cache->private_data;
1390  ErrorContextCallback errcallback;
1391 
1392  Assert(!ctx->fast_forward);
1393 
1394  /* We're only supposed to call this when streaming is supported. */
1395  Assert(ctx->streaming);
1396 
1397  /* Push callback + info on the error context stack */
1398  state.ctx = ctx;
1399  state.callback_name = "stream_commit";
1400  state.report_location = txn->final_lsn;
1401  errcallback.callback = output_plugin_error_callback;
1402  errcallback.arg = (void *) &state;
1403  errcallback.previous = error_context_stack;
1404  error_context_stack = &errcallback;
1405 
1406  /* set output state */
1407  ctx->accept_writes = true;
1408  ctx->write_xid = txn->xid;
1409  ctx->write_location = txn->end_lsn;
1410 
1411  /* in streaming mode, stream_commit_cb is required */
1412  if (ctx->callbacks.stream_commit_cb == NULL)
1413  ereport(ERROR,
1414  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1415  errmsg("logical streaming requires a %s callback",
1416  "stream_commit_cb")));
1417 
1418  ctx->callbacks.stream_commit_cb(ctx, txn, commit_lsn);
1419 
1420  /* Pop the error context stack */
1421  error_context_stack = errcallback.previous;
1422 }

References LogicalDecodingContext::accept_writes, ErrorContextCallback::arg, Assert(), ErrorContextCallback::callback, LogicalDecodingContext::callbacks, ReorderBufferTXN::end_lsn, ereport, errcode(), errmsg(), ERROR, error_context_stack, LogicalDecodingContext::fast_forward, ReorderBufferTXN::final_lsn, output_plugin_error_callback(), ErrorContextCallback::previous, ReorderBuffer::private_data, OutputPluginCallbacks::stream_commit_cb, LogicalDecodingContext::streaming, LogicalDecodingContext::write_location, LogicalDecodingContext::write_xid, and ReorderBufferTXN::xid.

Referenced by StartupDecodingContext().

◆ stream_message_cb_wrapper()

static void stream_message_cb_wrapper ( ReorderBuffer cache,
ReorderBufferTXN txn,
XLogRecPtr  message_lsn,
bool  transactional,
const char *  prefix,
Size  message_size,
const char *  message 
)
static

Definition at line 1472 of file logical.c.

1475 {
1476  LogicalDecodingContext *ctx = cache->private_data;
1478  ErrorContextCallback errcallback;
1479 
1480  Assert(!ctx->fast_forward);
1481 
1482  /* We're only supposed to call this when streaming is supported. */
1483  Assert(ctx->streaming);
1484 
1485  /* this callback is optional */
1486  if (ctx->callbacks.stream_message_cb == NULL)
1487  return;
1488 
1489  /* Push callback + info on the error context stack */
1490  state.ctx = ctx;
1491  state.callback_name = "stream_message";
1492  state.report_location = message_lsn;
1493  errcallback.callback = output_plugin_error_callback;
1494  errcallback.arg = (void *) &state;
1495  errcallback.previous = error_context_stack;
1496  error_context_stack = &errcallback;
1497 
1498  /* set output state */
1499  ctx->accept_writes = true;
1500  ctx->write_xid = txn != NULL ? txn->xid : InvalidTransactionId;
1501  ctx->write_location = message_lsn;
1502 
1503  /* do the actual work: call callback */
1504  ctx->callbacks.stream_message_cb(ctx, txn, message_lsn, transactional, prefix,
1505  message_size, message);
1506 
1507  /* Pop the error context stack */
1508  error_context_stack = errcallback.previous;
1509 }

References LogicalDecodingContext::accept_writes, ErrorContextCallback::arg, Assert(), ErrorContextCallback::callback, LogicalDecodingContext::callbacks, error_context_stack, LogicalDecodingContext::fast_forward, InvalidTransactionId, output_plugin_error_callback(), ErrorContextCallback::previous, ReorderBuffer::private_data, OutputPluginCallbacks::stream_message_cb, LogicalDecodingContext::streaming, LogicalDecodingContext::write_location, LogicalDecodingContext::write_xid, and ReorderBufferTXN::xid.

Referenced by StartupDecodingContext().

◆ stream_prepare_cb_wrapper()

static void stream_prepare_cb_wrapper ( ReorderBuffer cache,
ReorderBufferTXN txn,
XLogRecPtr  prepare_lsn 
)
static

Definition at line 1341 of file logical.c.

1343 {
1344  LogicalDecodingContext *ctx = cache->private_data;
1346  ErrorContextCallback errcallback;
1347 
1348  Assert(!ctx->fast_forward);
1349 
1350  /*
1351  * We're only supposed to call this when streaming and two-phase commits
1352  * are supported.
1353  */
1354  Assert(ctx->streaming);
1355  Assert(ctx->twophase);
1356 
1357  /* Push callback + info on the error context stack */
1358  state.ctx = ctx;
1359  state.callback_name = "stream_prepare";
1360  state.report_location = txn->final_lsn;
1361  errcallback.callback = output_plugin_error_callback;
1362  errcallback.arg = (void *) &state;
1363  errcallback.previous = error_context_stack;
1364  error_context_stack = &errcallback;
1365 
1366  /* set output state */
1367  ctx->accept_writes = true;
1368  ctx->write_xid = txn->xid;
1369  ctx->write_location = txn->end_lsn;
1370 
1371  /* in streaming mode with two-phase commits, stream_prepare_cb is required */
1372  if (ctx->callbacks.stream_prepare_cb == NULL)
1373  ereport(ERROR,
1374  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1375  errmsg("logical streaming at prepare time requires a %s callback",
1376  "stream_prepare_cb")));
1377 
1378  ctx->callbacks.stream_prepare_cb(ctx, txn, prepare_lsn);
1379 
1380  /* Pop the error context stack */
1381  error_context_stack = errcallback.previous;
1382 }

References LogicalDecodingContext::accept_writes, ErrorContextCallback::arg, Assert(), ErrorContextCallback::callback, LogicalDecodingContext::callbacks, ReorderBufferTXN::end_lsn, ereport, errcode(), errmsg(), ERROR, error_context_stack, LogicalDecodingContext::fast_forward, ReorderBufferTXN::final_lsn, output_plugin_error_callback(), ErrorContextCallback::previous, ReorderBuffer::private_data, OutputPluginCallbacks::stream_prepare_cb, LogicalDecodingContext::streaming, LogicalDecodingContext::twophase, LogicalDecodingContext::write_location, LogicalDecodingContext::write_xid, and ReorderBufferTXN::xid.

Referenced by StartupDecodingContext().

◆ stream_start_cb_wrapper()

static void stream_start_cb_wrapper ( ReorderBuffer cache,
ReorderBufferTXN txn,
XLogRecPtr  first_lsn 
)
static

Definition at line 1207 of file logical.c.

1209 {
1210  LogicalDecodingContext *ctx = cache->private_data;
1212  ErrorContextCallback errcallback;
1213 
1214  Assert(!ctx->fast_forward);
1215 
1216  /* We're only supposed to call this when streaming is supported. */
1217  Assert(ctx->streaming);
1218 
1219  /* Push callback + info on the error context stack */
1220  state.ctx = ctx;
1221  state.callback_name = "stream_start";
1222  state.report_location = first_lsn;
1223  errcallback.callback = output_plugin_error_callback;
1224  errcallback.arg = (void *) &state;
1225  errcallback.previous = error_context_stack;
1226  error_context_stack = &errcallback;
1227 
1228  /* set output state */
1229  ctx->accept_writes = true;
1230  ctx->write_xid = txn->xid;
1231 
1232  /*
1233  * Report this message's lsn so replies from clients can give an
1234  * up-to-date answer. This won't ever be enough (and shouldn't be!) to
1235  * confirm receipt of this transaction, but it might allow another
1236  * transaction's commit to be confirmed with one message.
1237  */
1238  ctx->write_location = first_lsn;
1239 
1240  /* in streaming mode, stream_start_cb is required */
1241  if (ctx->callbacks.stream_start_cb == NULL)
1242  ereport(ERROR,
1243  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1244  errmsg("logical streaming requires a %s callback",
1245  "stream_start_cb")));
1246 
1247  ctx->callbacks.stream_start_cb(ctx, txn);
1248 
1249  /* Pop the error context stack */
1250  error_context_stack = errcallback.previous;
1251 }

References LogicalDecodingContext::accept_writes, ErrorContextCallback::arg, Assert(), ErrorContextCallback::callback, LogicalDecodingContext::callbacks, ereport, errcode(), errmsg(), ERROR, error_context_stack, LogicalDecodingContext::fast_forward, output_plugin_error_callback(), ErrorContextCallback::previous, ReorderBuffer::private_data, OutputPluginCallbacks::stream_start_cb, LogicalDecodingContext::streaming, LogicalDecodingContext::write_location, LogicalDecodingContext::write_xid, and ReorderBufferTXN::xid.

Referenced by StartupDecodingContext().

◆ stream_stop_cb_wrapper()

static void stream_stop_cb_wrapper ( ReorderBuffer cache,
ReorderBufferTXN txn,
XLogRecPtr  last_lsn 
)
static

Definition at line 1254 of file logical.c.

1256 {
1257  LogicalDecodingContext *ctx = cache->private_data;
1259  ErrorContextCallback errcallback;
1260 
1261  Assert(!ctx->fast_forward);
1262 
1263  /* We're only supposed to call this when streaming is supported. */
1264  Assert(ctx->streaming);
1265 
1266  /* Push callback + info on the error context stack */
1267  state.ctx = ctx;
1268  state.callback_name = "stream_stop";
1269  state.report_location = last_lsn;
1270  errcallback.callback = output_plugin_error_callback;
1271  errcallback.arg = (void *) &state;
1272  errcallback.previous = error_context_stack;
1273  error_context_stack = &errcallback;
1274 
1275  /* set output state */
1276  ctx->accept_writes = true;
1277  ctx->write_xid = txn->xid;
1278 
1279  /*
1280  * Report this message's lsn so replies from clients can give an
1281  * up-to-date answer. This won't ever be enough (and shouldn't be!) to
1282  * confirm receipt of this transaction, but it might allow another
1283  * transaction's commit to be confirmed with one message.
1284  */
1285  ctx->write_location = last_lsn;
1286 
1287  /* in streaming mode, stream_stop_cb is required */
1288  if (ctx->callbacks.stream_stop_cb == NULL)
1289  ereport(ERROR,
1290  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1291  errmsg("logical streaming requires a %s callback",
1292  "stream_stop_cb")));
1293 
1294  ctx->callbacks.stream_stop_cb(ctx, txn);
1295 
1296  /* Pop the error context stack */
1297  error_context_stack = errcallback.previous;
1298 }

References LogicalDecodingContext::accept_writes, ErrorContextCallback::arg, Assert(), ErrorContextCallback::callback, LogicalDecodingContext::callbacks, ereport, errcode(), errmsg(), ERROR, error_context_stack, LogicalDecodingContext::fast_forward, output_plugin_error_callback(), ErrorContextCallback::previous, ReorderBuffer::private_data, OutputPluginCallbacks::stream_stop_cb, LogicalDecodingContext::streaming, LogicalDecodingContext::write_location, LogicalDecodingContext::write_xid, and ReorderBufferTXN::xid.

Referenced by StartupDecodingContext().

◆ stream_truncate_cb_wrapper()

static void stream_truncate_cb_wrapper ( ReorderBuffer cache,
ReorderBufferTXN txn,
int  nrelations,
Relation  relations[],
ReorderBufferChange change 
)
static

Definition at line 1512 of file logical.c.

1515 {
1516  LogicalDecodingContext *ctx = cache->private_data;
1518  ErrorContextCallback errcallback;
1519 
1520  Assert(!ctx->fast_forward);
1521 
1522  /* We're only supposed to call this when streaming is supported. */
1523  Assert(ctx->streaming);
1524 
1525  /* this callback is optional */
1526  if (!ctx->callbacks.stream_truncate_cb)
1527  return;
1528 
1529  /* Push callback + info on the error context stack */
1530  state.ctx = ctx;
1531  state.callback_name = "stream_truncate";
1532  state.report_location = change->lsn;
1533  errcallback.callback = output_plugin_error_callback;
1534  errcallback.arg = (void *) &state;
1535  errcallback.previous = error_context_stack;
1536  error_context_stack = &errcallback;
1537 
1538  /* set output state */
1539  ctx->accept_writes = true;
1540  ctx->write_xid = txn->xid;
1541 
1542  /*
1543  * Report this change's lsn so replies from clients can give an up-to-date
1544  * answer. This won't ever be enough (and shouldn't be!) to confirm
1545  * receipt of this transaction, but it might allow another transaction's
1546  * commit to be confirmed with one message.
1547  */
1548  ctx->write_location = change->lsn;
1549 
1550  ctx->callbacks.stream_truncate_cb(ctx, txn, nrelations, relations, change);
1551 
1552  /* Pop the error context stack */
1553  error_context_stack = errcallback.previous;
1554 }

References LogicalDecodingContext::accept_writes, ErrorContextCallback::arg, Assert(), ErrorContextCallback::callback, LogicalDecodingContext::callbacks, error_context_stack, LogicalDecodingContext::fast_forward, ReorderBufferChange::lsn, output_plugin_error_callback(), ErrorContextCallback::previous, ReorderBuffer::private_data, OutputPluginCallbacks::stream_truncate_cb, LogicalDecodingContext::streaming, LogicalDecodingContext::write_location, LogicalDecodingContext::write_xid, and ReorderBufferTXN::xid.

Referenced by StartupDecodingContext().

◆ truncate_cb_wrapper()

static void truncate_cb_wrapper ( ReorderBuffer cache,
ReorderBufferTXN txn,
int  nrelations,
Relation  relations[],
ReorderBufferChange change 
)
static

Definition at line 1070 of file logical.c.

1072 {
1073  LogicalDecodingContext *ctx = cache->private_data;
1075  ErrorContextCallback errcallback;
1076 
1077  Assert(!ctx->fast_forward);
1078 
1079  if (!ctx->callbacks.truncate_cb)
1080  return;
1081 
1082  /* Push callback + info on the error context stack */
1083  state.ctx = ctx;
1084  state.callback_name = "truncate";
1085  state.report_location = change->lsn;
1086  errcallback.callback = output_plugin_error_callback;
1087  errcallback.arg = (void *) &state;
1088  errcallback.previous = error_context_stack;
1089  error_context_stack = &errcallback;
1090 
1091  /* set output state */
1092  ctx->accept_writes = true;
1093  ctx->write_xid = txn->xid;
1094 
1095  /*
1096  * Report this change's lsn so replies from clients can give an up-to-date
1097  * answer. This won't ever be enough (and shouldn't be!) to confirm
1098  * receipt of this transaction, but it might allow another transaction's
1099  * commit to be confirmed with one message.
1100  */
1101  ctx->write_location = change->lsn;
1102 
1103  ctx->callbacks.truncate_cb(ctx, txn, nrelations, relations, change);
1104 
1105  /* Pop the error context stack */
1106  error_context_stack = errcallback.previous;
1107 }
LogicalDecodeTruncateCB truncate_cb

References LogicalDecodingContext::accept_writes, ErrorContextCallback::arg, Assert(), ErrorContextCallback::callback, LogicalDecodingContext::callbacks, error_context_stack, LogicalDecodingContext::fast_forward, ReorderBufferChange::lsn, output_plugin_error_callback(), ErrorContextCallback::previous, ReorderBuffer::private_data, OutputPluginCallbacks::truncate_cb, LogicalDecodingContext::write_location, LogicalDecodingContext::write_xid, and ReorderBufferTXN::xid.

Referenced by StartupDecodingContext().

◆ UpdateDecodingStats()

void UpdateDecodingStats ( LogicalDecodingContext ctx)

Definition at line 1804 of file logical.c.

1805 {
1806  ReorderBuffer *rb = ctx->reorder;
1807  PgStat_StatReplSlotEntry repSlotStat;
1808 
1809  /* Nothing to do if we don't have any replication stats to be sent. */
1810  if (rb->spillBytes <= 0 && rb->streamBytes <= 0 && rb->totalBytes <= 0)
1811  return;
1812 
1813  elog(DEBUG2, "UpdateDecodingStats: updating stats %p %lld %lld %lld %lld %lld %lld %lld %lld",
1814  rb,
1815  (long long) rb->spillTxns,
1816  (long long) rb->spillCount,
1817  (long long) rb->spillBytes,
1818  (long long) rb->streamTxns,
1819  (long long) rb->streamCount,
1820  (long long) rb->streamBytes,
1821  (long long) rb->totalTxns,
1822  (long long) rb->totalBytes);
1823 
1824  namestrcpy(&repSlotStat.slotname, NameStr(ctx->slot->data.name));
1825  repSlotStat.spill_txns = rb->spillTxns;
1826  repSlotStat.spill_count = rb->spillCount;
1827  repSlotStat.spill_bytes = rb->spillBytes;
1828  repSlotStat.stream_txns = rb->streamTxns;
1829  repSlotStat.stream_count = rb->streamCount;
1830  repSlotStat.stream_bytes = rb->streamBytes;
1831  repSlotStat.total_txns = rb->totalTxns;
1832  repSlotStat.total_bytes = rb->totalBytes;
1833 
1834  pgstat_report_replslot(&repSlotStat);
1835 
1836  rb->spillTxns = 0;
1837  rb->spillCount = 0;
1838  rb->spillBytes = 0;
1839  rb->streamTxns = 0;
1840  rb->streamCount = 0;
1841  rb->streamBytes = 0;
1842  rb->totalTxns = 0;
1843  rb->totalBytes = 0;
1844 }
#define DEBUG2
Definition: elog.h:23
void pgstat_report_replslot(const PgStat_StatReplSlotEntry *repSlotStat)
Definition: pgstat.c:1887
PgStat_Counter stream_count
Definition: pgstat.h:985
PgStat_Counter total_txns
Definition: pgstat.h:987
PgStat_Counter total_bytes
Definition: pgstat.h:988
PgStat_Counter spill_txns
Definition: pgstat.h:981
PgStat_Counter stream_txns
Definition: pgstat.h:984
PgStat_Counter spill_count
Definition: pgstat.h:982
PgStat_Counter stream_bytes
Definition: pgstat.h:986
PgStat_Counter spill_bytes
Definition: pgstat.h:983

References ReplicationSlot::data, DEBUG2, elog, ReplicationSlotPersistentData::name, NameStr, namestrcpy(), pgstat_report_replslot(), LogicalDecodingContext::reorder, LogicalDecodingContext::slot, PgStat_StatReplSlotEntry::slotname, PgStat_StatReplSlotEntry::spill_bytes, PgStat_StatReplSlotEntry::spill_count, PgStat_StatReplSlotEntry::spill_txns, ReorderBuffer::spillBytes, ReorderBuffer::spillCount, ReorderBuffer::spillTxns, PgStat_StatReplSlotEntry::stream_bytes, PgStat_StatReplSlotEntry::stream_count, PgStat_StatReplSlotEntry::stream_txns, ReorderBuffer::streamBytes, ReorderBuffer::streamCount, ReorderBuffer::streamTxns, PgStat_StatReplSlotEntry::total_bytes, PgStat_StatReplSlotEntry::total_txns, ReorderBuffer::totalBytes, and ReorderBuffer::totalTxns.

Referenced by DecodeAbort(), DecodeCommit(), DecodePrepare(), ReorderBufferSerializeTXN(), and ReorderBufferStreamTXN().