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 skipped_xact)
 
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 792 of file logical.c.

793 {
794  LogicalDecodingContext *ctx = cache->private_data;
796  ErrorContextCallback errcallback;
797 
798  Assert(!ctx->fast_forward);
799 
800  /* Push callback + info on the error context stack */
801  state.ctx = ctx;
802  state.callback_name = "begin";
803  state.report_location = txn->first_lsn;
805  errcallback.arg = (void *) &state;
806  errcallback.previous = error_context_stack;
807  error_context_stack = &errcallback;
808 
809  /* set output state */
810  ctx->accept_writes = true;
811  ctx->write_xid = txn->xid;
812  ctx->write_location = txn->first_lsn;
813  ctx->end_xact = false;
814 
815  /* do the actual work: call callback */
816  ctx->callbacks.begin_cb(ctx, txn);
817 
818  /* Pop the error context stack */
819  error_context_stack = errcallback.previous;
820 }
ErrorContextCallback * error_context_stack
Definition: elog.c:94
Assert(fmt[strlen(fmt) - 1] !='\n')
static void output_plugin_error_callback(void *arg)
Definition: logical.c:712
struct ErrorContextCallback * previous
Definition: elog.h:234
void(* callback)(void *arg)
Definition: elog.h:235
XLogRecPtr write_location
Definition: logical.h:108
OutputPluginCallbacks callbacks
Definition: logical.h:53
TransactionId write_xid
Definition: logical.h:109
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, LogicalDecodingContext::end_xact, 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 862 of file logical.c.

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

References LogicalDecodingContext::accept_writes, ErrorContextCallback::arg, Assert(), OutputPluginCallbacks::begin_prepare_cb, ErrorContextCallback::callback, LogicalDecodingContext::callbacks, LogicalDecodingContext::end_xact, 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 1043 of file logical.c.

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

References LogicalDecodingContext::accept_writes, ErrorContextCallback::arg, Assert(), ErrorContextCallback::callback, LogicalDecodingContext::callbacks, OutputPluginCallbacks::change_cb, LogicalDecodingContext::end_xact, 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:89
#define InvalidOid
Definition: postgres_ext.h:36
void CheckSlotRequirements(void)
Definition: slot.c:1119
bool RecoveryInProgress(void)
Definition: xlog.c:5912
int wal_level
Definition: xlog.c:134
@ WAL_LEVEL_LOGICAL
Definition: xlog.h:71

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 823 of file logical.c.

825 {
826  LogicalDecodingContext *ctx = cache->private_data;
828  ErrorContextCallback errcallback;
829 
830  Assert(!ctx->fast_forward);
831 
832  /* Push callback + info on the error context stack */
833  state.ctx = ctx;
834  state.callback_name = "commit";
835  state.report_location = txn->final_lsn; /* beginning of commit record */
837  errcallback.arg = (void *) &state;
838  errcallback.previous = error_context_stack;
839  error_context_stack = &errcallback;
840 
841  /* set output state */
842  ctx->accept_writes = true;
843  ctx->write_xid = txn->xid;
844  ctx->write_location = txn->end_lsn; /* points to the end of the record */
845  ctx->end_xact = true;
846 
847  /* do the actual work: call callback */
848  ctx->callbacks.commit_cb(ctx, txn, commit_lsn);
849 
850  /* Pop the error context stack */
851  error_context_stack = errcallback.previous;
852 }
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, LogicalDecodingContext::end_xact, 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 951 of file logical.c.

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

References LogicalDecodingContext::accept_writes, ErrorContextCallback::arg, Assert(), ErrorContextCallback::callback, LogicalDecodingContext::callbacks, OutputPluginCallbacks::commit_prepared_cb, ReorderBufferTXN::end_lsn, LogicalDecodingContext::end_xact, 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:682
int errdetail(const char *fmt,...)
Definition: elog.c:1039
#define LOG
Definition: elog.h:27
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:731
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
Definition: palloc.h:135
const void * data
void ReplicationSlotMarkDirty(void)
Definition: slot.c:796
ReplicationSlot * MyReplicationSlot
Definition: slot.c:98
void ReplicationSlotSave(void)
Definition: slot.c:778
#define SlotIsPhysical(slot)
Definition: slot.h:168
void SnapBuildSetTwoPhaseAt(SnapBuild *builder, XLogRecPtr ptr)
Definition: snapbuild.c:420
OutputPluginOptions options
Definition: logical.h:54
MemoryContext context
Definition: logical.h:36
struct SnapBuild * snapshot_builder
Definition: logical.h:44
struct ReorderBuffer * reorder
Definition: logical.h:43
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:588
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1194
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1802
@ LW_EXCLUSIVE
Definition: lwlock.h:112
void namestrcpy(Name name, const char *str)
Definition: name.c:233
#define NIL
Definition: pg_list.h:66
static const char * plugin
TransactionId GetOldestSafeDecodingTransactionId(bool catalogOnly)
Definition: procarray.c:2969
void ReplicationSlotReserveWal(void)
Definition: slot.c:1156
void ReplicationSlotsComputeRequiredXmin(bool already_locked)
Definition: slot.c:835
#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:677
bool IsTransactionState(void)
Definition: xact.c:377
TransactionId GetTopTransactionIdIfAny(void)
Definition: xact.c:431
#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:26
bool DecodingContextReady(LogicalDecodingContext *ctx)
Definition: logical.c:581
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:121
XLogReaderState * reader
Definition: logical.h:42
ReplicationSlot * slot
Definition: logical.h:39
XLogRecPtr EndRecPtr
Definition: xlogreader.h:207
XLogRecord * XLogReadRecord(XLogReaderState *state, char **errormsg)
Definition: xlogreader.c:422
void XLogBeginRead(XLogReaderState *state, XLogRecPtr RecPtr)
Definition: xlogreader.c:264

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:402
@ 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 1156 of file logical.c.

1157 {
1159  ErrorContextCallback errcallback;
1160  bool ret;
1161 
1162  Assert(!ctx->fast_forward);
1163 
1164  /* Push callback + info on the error context stack */
1165  state.ctx = ctx;
1166  state.callback_name = "filter_by_origin";
1167  state.report_location = InvalidXLogRecPtr;
1168  errcallback.callback = output_plugin_error_callback;
1169  errcallback.arg = (void *) &state;
1170  errcallback.previous = error_context_stack;
1171  error_context_stack = &errcallback;
1172 
1173  /* set output state */
1174  ctx->accept_writes = false;
1175  ctx->end_xact = false;
1176 
1177  /* do the actual work: call callback */
1178  ret = ctx->callbacks.filter_by_origin_cb(ctx, origin_id);
1179 
1180  /* Pop the error context stack */
1181  error_context_stack = errcallback.previous;
1182 
1183  return ret;
1184 }
LogicalDecodeFilterByOriginCB filter_by_origin_cb

References LogicalDecodingContext::accept_writes, ErrorContextCallback::arg, Assert(), ErrorContextCallback::callback, LogicalDecodingContext::callbacks, LogicalDecodingContext::end_xact, 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 1124 of file logical.c.

1126 {
1128  ErrorContextCallback errcallback;
1129  bool ret;
1130 
1131  Assert(!ctx->fast_forward);
1132 
1133  /* Push callback + info on the error context stack */
1134  state.ctx = ctx;
1135  state.callback_name = "filter_prepare";
1136  state.report_location = InvalidXLogRecPtr;
1137  errcallback.callback = output_plugin_error_callback;
1138  errcallback.arg = (void *) &state;
1139  errcallback.previous = error_context_stack;
1140  error_context_stack = &errcallback;
1141 
1142  /* set output state */
1143  ctx->accept_writes = false;
1144  ctx->end_xact = false;
1145 
1146  /* do the actual work: call callback */
1147  ret = ctx->callbacks.filter_prepare_cb(ctx, xid, gid);
1148 
1149  /* Pop the error context stack */
1150  error_context_stack = errcallback.previous;
1151 
1152  return ret;
1153 }
LogicalDecodeFilterPrepareCB filter_prepare_cb

References LogicalDecodingContext::accept_writes, ErrorContextCallback::arg, Assert(), ErrorContextCallback::callback, LogicalDecodingContext::callbacks, LogicalDecodingContext::end_xact, 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:759
void MemoryContextDelete(MemoryContext context)
Definition: mcxt.c:376
void ReorderBufferFree(ReorderBuffer *rb)
void FreeSnapshotBuilder(SnapBuild *builder)
Definition: snapbuild.c:358
LogicalDecodeShutdownCB shutdown_cb
void XLogReaderFree(XLogReaderState *state)
Definition: xlogreader.c:170

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 690 of file logical.c.

691 {
692  LogicalOutputPluginInit plugin_init;
693 
694  plugin_init = (LogicalOutputPluginInit)
695  load_external_function(plugin, "_PG_output_plugin_init", false, NULL);
696 
697  if (plugin_init == NULL)
698  elog(ERROR, "output plugins have to declare the _PG_output_plugin_init symbol");
699 
700  /* ask the output plugin to fill the callback struct */
701  plugin_init(callbacks);
702 
703  if (callbacks->begin_cb == NULL)
704  elog(ERROR, "output plugins have to register a begin callback");
705  if (callbacks->change_cb == NULL)
706  elog(ERROR, "output plugins have to register a change callback");
707  if (callbacks->commit_cb == NULL)
708  elog(ERROR, "output plugins have to register a commit callback");
709 }
void * load_external_function(const char *filename, const char *funcname, bool signalNotFound, void **filehandle)
Definition: dfmgr.c:105
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 1736 of file logical.c.

1737 {
1738  Assert(lsn != InvalidXLogRecPtr);
1739 
1740  /* Do an unlocked check for candidate_lsn first. */
1743  {
1744  bool updated_xmin = false;
1745  bool updated_restart = false;
1746 
1748 
1750 
1751  /* if we're past the location required for bumping xmin, do so */
1754  {
1755  /*
1756  * We have to write the changed xmin to disk *before* we change
1757  * the in-memory value, otherwise after a crash we wouldn't know
1758  * that some catalog tuples might have been removed already.
1759  *
1760  * Ensure that by first writing to ->xmin and only update
1761  * ->effective_xmin once the new state is synced to disk. After a
1762  * crash ->effective_xmin is set to ->xmin.
1763  */
1766  {
1770  updated_xmin = true;
1771  }
1772  }
1773 
1776  {
1778 
1782  updated_restart = true;
1783  }
1784 
1786 
1787  /* first write new xmin to disk, so we know what's up after a crash */
1788  if (updated_xmin || updated_restart)
1789  {
1792  elog(DEBUG1, "updated xmin: %u restart: %u", updated_xmin, updated_restart);
1793  }
1794 
1795  /*
1796  * Now the new xmin is safely on disk, we can let the global value
1797  * advance. We do not take ProcArrayLock or similar since we only
1798  * advance xmin here and there's not much harm done by a concurrent
1799  * computation missing that.
1800  */
1801  if (updated_xmin)
1802  {
1806 
1809  }
1810  }
1811  else
1812  {
1816  }
1817 }
void ReplicationSlotsComputeRequiredLSN(void)
Definition: slot.c:892
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 1662 of file logical.c.

1663 {
1664  bool updated_lsn = false;
1665  ReplicationSlot *slot;
1666 
1667  slot = MyReplicationSlot;
1668 
1669  Assert(slot != NULL);
1670  Assert(restart_lsn != InvalidXLogRecPtr);
1671  Assert(current_lsn != InvalidXLogRecPtr);
1672 
1673  SpinLockAcquire(&slot->mutex);
1674 
1675  /* don't overwrite if have a newer restart lsn */
1676  if (restart_lsn <= slot->data.restart_lsn)
1677  {
1678  }
1679 
1680  /*
1681  * We might have already flushed far enough to directly accept this lsn,
1682  * in this case there is no need to check for existing candidate LSNs
1683  */
1684  else if (current_lsn <= slot->data.confirmed_flush)
1685  {
1686  slot->candidate_restart_valid = current_lsn;
1687  slot->candidate_restart_lsn = restart_lsn;
1688 
1689  /* our candidate can directly be used */
1690  updated_lsn = true;
1691  }
1692 
1693  /*
1694  * Only increase if the previous values have been applied, otherwise we
1695  * might never end up updating if the receiver acks too slowly. A missed
1696  * value here will just cause some extra effort after reconnecting.
1697  */
1699  {
1700  slot->candidate_restart_valid = current_lsn;
1701  slot->candidate_restart_lsn = restart_lsn;
1702  SpinLockRelease(&slot->mutex);
1703 
1704  elog(DEBUG1, "got new restart lsn %X/%X at %X/%X",
1705  LSN_FORMAT_ARGS(restart_lsn),
1706  LSN_FORMAT_ARGS(current_lsn));
1707  }
1708  else
1709  {
1710  XLogRecPtr candidate_restart_lsn;
1711  XLogRecPtr candidate_restart_valid;
1712  XLogRecPtr confirmed_flush;
1713 
1714  candidate_restart_lsn = slot->candidate_restart_lsn;
1715  candidate_restart_valid = slot->candidate_restart_valid;
1716  confirmed_flush = slot->data.confirmed_flush;
1717  SpinLockRelease(&slot->mutex);
1718 
1719  elog(DEBUG1, "failed to increase restart lsn: proposed %X/%X, after %X/%X, current candidate %X/%X, current after %X/%X, flushed up to %X/%X",
1720  LSN_FORMAT_ARGS(restart_lsn),
1721  LSN_FORMAT_ARGS(current_lsn),
1722  LSN_FORMAT_ARGS(candidate_restart_lsn),
1723  LSN_FORMAT_ARGS(candidate_restart_valid),
1724  LSN_FORMAT_ARGS(confirmed_flush));
1725  }
1726 
1727  /* candidates are already valid with the current flush position, apply */
1728  if (updated_lsn)
1730 }
void LogicalConfirmReceivedLocation(XLogRecPtr lsn)
Definition: logical.c:1736
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 1594 of file logical.c.

1595 {
1596  bool updated_xmin = false;
1597  ReplicationSlot *slot;
1598  bool got_new_xmin = false;
1599 
1600  slot = MyReplicationSlot;
1601 
1602  Assert(slot != NULL);
1603 
1604  SpinLockAcquire(&slot->mutex);
1605 
1606  /*
1607  * don't overwrite if we already have a newer xmin. This can happen if we
1608  * restart decoding in a slot.
1609  */
1611  {
1612  }
1613 
1614  /*
1615  * If the client has already confirmed up to this lsn, we directly can
1616  * mark this as accepted. This can happen if we restart decoding in a
1617  * slot.
1618  */
1619  else if (current_lsn <= slot->data.confirmed_flush)
1620  {
1621  slot->candidate_catalog_xmin = xmin;
1622  slot->candidate_xmin_lsn = current_lsn;
1623 
1624  /* our candidate can directly be used */
1625  updated_xmin = true;
1626  }
1627 
1628  /*
1629  * Only increase if the previous values have been applied, otherwise we
1630  * might never end up updating if the receiver acks too slowly.
1631  */
1632  else if (slot->candidate_xmin_lsn == InvalidXLogRecPtr)
1633  {
1634  slot->candidate_catalog_xmin = xmin;
1635  slot->candidate_xmin_lsn = current_lsn;
1636 
1637  /*
1638  * Log new xmin at an appropriate log level after releasing the
1639  * spinlock.
1640  */
1641  got_new_xmin = true;
1642  }
1643  SpinLockRelease(&slot->mutex);
1644 
1645  if (got_new_xmin)
1646  elog(DEBUG1, "got new catalog xmin %u at %X/%X", xmin,
1647  LSN_FORMAT_ARGS(current_lsn));
1648 
1649  /* candidate already valid with the current flush position, apply */
1650  if (updated_xmin)
1652 }
bool TransactionIdPrecedesOrEquals(TransactionId id1, TransactionId id2)
Definition: transam.c:292

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 1187 of file logical.c.

1190 {
1191  LogicalDecodingContext *ctx = cache->private_data;
1193  ErrorContextCallback errcallback;
1194 
1195  Assert(!ctx->fast_forward);
1196 
1197  if (ctx->callbacks.message_cb == NULL)
1198  return;
1199 
1200  /* Push callback + info on the error context stack */
1201  state.ctx = ctx;
1202  state.callback_name = "message";
1203  state.report_location = message_lsn;
1204  errcallback.callback = output_plugin_error_callback;
1205  errcallback.arg = (void *) &state;
1206  errcallback.previous = error_context_stack;
1207  error_context_stack = &errcallback;
1208 
1209  /* set output state */
1210  ctx->accept_writes = true;
1211  ctx->write_xid = txn != NULL ? txn->xid : InvalidTransactionId;
1212  ctx->write_location = message_lsn;
1213  ctx->end_xact = false;
1214 
1215  /* do the actual work: call callback */
1216  ctx->callbacks.message_cb(ctx, txn, message_lsn, transactional, prefix,
1217  message_size, message);
1218 
1219  /* Pop the error context stack */
1220  error_context_stack = errcallback.previous;
1221 }
LogicalDecodeMessageCB message_cb

References LogicalDecodingContext::accept_writes, ErrorContextCallback::arg, Assert(), ErrorContextCallback::callback, LogicalDecodingContext::callbacks, LogicalDecodingContext::end_xact, 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 712 of file logical.c.

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

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_change(), pgoutput_commit_prepared_txn(), pgoutput_commit_txn(), pgoutput_message(), pgoutput_prepare_txn(), pgoutput_rollback_prepared_txn(), pgoutput_send_begin(), 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,
bool  skipped_xact 
)

Definition at line 675 of file logical.c.

677 {
678  if (!ctx->update_progress)
679  return;
680 
681  ctx->update_progress(ctx, ctx->write_location, ctx->write_xid,
682  skipped_xact);
683 }
LogicalOutputPluginWriterUpdateProgress update_progress
Definition: logical.h:66

References LogicalDecodingContext::update_progress, LogicalDecodingContext::write_location, and LogicalDecodingContext::write_xid.

Referenced by update_replication_progress().

◆ OutputPluginWrite()

◆ prepare_cb_wrapper()

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

Definition at line 906 of file logical.c.

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

References LogicalDecodingContext::accept_writes, ErrorContextCallback::arg, Assert(), ErrorContextCallback::callback, LogicalDecodingContext::callbacks, ReorderBufferTXN::end_lsn, LogicalDecodingContext::end_xact, 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 1823 of file logical.c.

1824 {
1826  bsysscan = false;
1827 }
bool bsysscan
Definition: xact.c:99
TransactionId CheckXidAlive
Definition: xact.c:98

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 996 of file logical.c.

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

References LogicalDecodingContext::accept_writes, ErrorContextCallback::arg, Assert(), ErrorContextCallback::callback, LogicalDecodingContext::callbacks, ReorderBufferTXN::end_lsn, LogicalDecodingContext::end_xact, 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 759 of file logical.c.

760 {
762  ErrorContextCallback errcallback;
763 
764  Assert(!ctx->fast_forward);
765 
766  /* Push callback + info on the error context stack */
767  state.ctx = ctx;
768  state.callback_name = "shutdown";
769  state.report_location = InvalidXLogRecPtr;
771  errcallback.arg = (void *) &state;
772  errcallback.previous = error_context_stack;
773  error_context_stack = &errcallback;
774 
775  /* set output state */
776  ctx->accept_writes = false;
777  ctx->end_xact = false;
778 
779  /* do the actual work: call callback */
780  ctx->callbacks.shutdown_cb(ctx);
781 
782  /* Pop the error context stack */
783  error_context_stack = errcallback.previous;
784 }

References LogicalDecodingContext::accept_writes, ErrorContextCallback::arg, Assert(), ErrorContextCallback::callback, LogicalDecodingContext::callbacks, LogicalDecodingContext::end_xact, 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 731 of file logical.c.

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

References LogicalDecodingContext::accept_writes, ErrorContextCallback::arg, Assert(), ErrorContextCallback::callback, LogicalDecodingContext::callbacks, LogicalDecodingContext::end_xact, 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:1043
static void commit_prepared_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr commit_lsn)
Definition: logical.c:951
static void stream_start_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr first_lsn)
Definition: logical.c:1224
static void commit_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr commit_lsn)
Definition: logical.c:823
static void begin_prepare_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn)
Definition: logical.c:862
static void stream_prepare_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr prepare_lsn)
Definition: logical.c:1363
static void prepare_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr prepare_lsn)
Definition: logical.c:906
static void stream_truncate_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, int nrelations, Relation relations[], ReorderBufferChange *change)
Definition: logical.c:1539
static void truncate_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, int nrelations, Relation relations[], ReorderBufferChange *change)
Definition: logical.c:1082
static void begin_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn)
Definition: logical.c:792
static void rollback_prepared_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr prepare_end_lsn, TimestampTz prepare_time)
Definition: logical.c:996
static void stream_change_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, Relation relation, ReorderBufferChange *change)
Definition: logical.c:1449
static void stream_abort_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr abort_lsn)
Definition: logical.c:1322
static void stream_message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr message_lsn, bool transactional, const char *prefix, Size message_size, const char *message)
Definition: logical.c:1498
static void stream_commit_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr commit_lsn)
Definition: logical.c:1408
static void LoadOutputPlugin(OutputPluginCallbacks *callbacks, const char *plugin)
Definition: logical.c:690
static void stream_stop_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr last_lsn)
Definition: logical.c:1273
static void message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr message_lsn, bool transactional, const char *prefix, Size message_size, const char *message)
Definition: logical.c:1187
void * palloc0(Size size)
Definition: mcxt.c:1230
MemoryContext CurrentMemoryContext
Definition: mcxt.c:124
#define AllocSetContextCreate
Definition: memutils.h:129
#define ALLOCSET_DEFAULT_SIZES
Definition: memutils.h:153
#define PROC_IN_LOGICAL_DECODING
Definition: proc.h:60
ReorderBuffer * ReorderBufferAllocate(void)
SnapBuild * AllocateSnapshotBuilder(ReorderBuffer *reorder, TransactionId xmin_horizon, XLogRecPtr start_lsn, bool need_full_snapshot, XLogRecPtr two_phase_at)
Definition: snapbuild.c:312
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:71
List * output_plugin_options
Definition: logical.h:59
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:233
int pgxactoff
Definition: proc.h:188
uint8 * statusFlags
Definition: proc.h:377
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:4814
int wal_segment_size
Definition: xlog.c:146
XLogReaderState * XLogReaderAllocate(int wal_segment_size, const char *waldir, XLogReaderRoutine *routine, void *private_data)
Definition: xlogreader.c:108

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 1322 of file logical.c.

1324 {
1325  LogicalDecodingContext *ctx = cache->private_data;
1327  ErrorContextCallback errcallback;
1328 
1329  Assert(!ctx->fast_forward);
1330 
1331  /* We're only supposed to call this when streaming is supported. */
1332  Assert(ctx->streaming);
1333 
1334  /* Push callback + info on the error context stack */
1335  state.ctx = ctx;
1336  state.callback_name = "stream_abort";
1337  state.report_location = abort_lsn;
1338  errcallback.callback = output_plugin_error_callback;
1339  errcallback.arg = (void *) &state;
1340  errcallback.previous = error_context_stack;
1341  error_context_stack = &errcallback;
1342 
1343  /* set output state */
1344  ctx->accept_writes = true;
1345  ctx->write_xid = txn->xid;
1346  ctx->write_location = abort_lsn;
1347  ctx->end_xact = true;
1348 
1349  /* in streaming mode, stream_abort_cb is required */
1350  if (ctx->callbacks.stream_abort_cb == NULL)
1351  ereport(ERROR,
1352  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1353  errmsg("logical streaming requires a %s callback",
1354  "stream_abort_cb")));
1355 
1356  ctx->callbacks.stream_abort_cb(ctx, txn, abort_lsn);
1357 
1358  /* Pop the error context stack */
1359  error_context_stack = errcallback.previous;
1360 }

References LogicalDecodingContext::accept_writes, ErrorContextCallback::arg, Assert(), ErrorContextCallback::callback, LogicalDecodingContext::callbacks, LogicalDecodingContext::end_xact, 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 1449 of file logical.c.

1451 {
1452  LogicalDecodingContext *ctx = cache->private_data;
1454  ErrorContextCallback errcallback;
1455 
1456  Assert(!ctx->fast_forward);
1457 
1458  /* We're only supposed to call this when streaming is supported. */
1459  Assert(ctx->streaming);
1460 
1461  /* Push callback + info on the error context stack */
1462  state.ctx = ctx;
1463  state.callback_name = "stream_change";
1464  state.report_location = change->lsn;
1465  errcallback.callback = output_plugin_error_callback;
1466  errcallback.arg = (void *) &state;
1467  errcallback.previous = error_context_stack;
1468  error_context_stack = &errcallback;
1469 
1470  /* set output state */
1471  ctx->accept_writes = true;
1472  ctx->write_xid = txn->xid;
1473 
1474  /*
1475  * Report this change's lsn so replies from clients can give an up-to-date
1476  * answer. This won't ever be enough (and shouldn't be!) to confirm
1477  * receipt of this transaction, but it might allow another transaction's
1478  * commit to be confirmed with one message.
1479  */
1480  ctx->write_location = change->lsn;
1481 
1482  ctx->end_xact = false;
1483 
1484  /* in streaming mode, stream_change_cb is required */
1485  if (ctx->callbacks.stream_change_cb == NULL)
1486  ereport(ERROR,
1487  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1488  errmsg("logical streaming requires a %s callback",
1489  "stream_change_cb")));
1490 
1491  ctx->callbacks.stream_change_cb(ctx, txn, relation, change);
1492 
1493  /* Pop the error context stack */
1494  error_context_stack = errcallback.previous;
1495 }

References LogicalDecodingContext::accept_writes, ErrorContextCallback::arg, Assert(), ErrorContextCallback::callback, LogicalDecodingContext::callbacks, LogicalDecodingContext::end_xact, 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 1408 of file logical.c.

1410 {
1411  LogicalDecodingContext *ctx = cache->private_data;
1413  ErrorContextCallback errcallback;
1414 
1415  Assert(!ctx->fast_forward);
1416 
1417  /* We're only supposed to call this when streaming is supported. */
1418  Assert(ctx->streaming);
1419 
1420  /* Push callback + info on the error context stack */
1421  state.ctx = ctx;
1422  state.callback_name = "stream_commit";
1423  state.report_location = txn->final_lsn;
1424  errcallback.callback = output_plugin_error_callback;
1425  errcallback.arg = (void *) &state;
1426  errcallback.previous = error_context_stack;
1427  error_context_stack = &errcallback;
1428 
1429  /* set output state */
1430  ctx->accept_writes = true;
1431  ctx->write_xid = txn->xid;
1432  ctx->write_location = txn->end_lsn;
1433  ctx->end_xact = true;
1434 
1435  /* in streaming mode, stream_commit_cb is required */
1436  if (ctx->callbacks.stream_commit_cb == NULL)
1437  ereport(ERROR,
1438  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1439  errmsg("logical streaming requires a %s callback",
1440  "stream_commit_cb")));
1441 
1442  ctx->callbacks.stream_commit_cb(ctx, txn, commit_lsn);
1443 
1444  /* Pop the error context stack */
1445  error_context_stack = errcallback.previous;
1446 }

References LogicalDecodingContext::accept_writes, ErrorContextCallback::arg, Assert(), ErrorContextCallback::callback, LogicalDecodingContext::callbacks, ReorderBufferTXN::end_lsn, LogicalDecodingContext::end_xact, 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 1498 of file logical.c.

1501 {
1502  LogicalDecodingContext *ctx = cache->private_data;
1504  ErrorContextCallback errcallback;
1505 
1506  Assert(!ctx->fast_forward);
1507 
1508  /* We're only supposed to call this when streaming is supported. */
1509  Assert(ctx->streaming);
1510 
1511  /* this callback is optional */
1512  if (ctx->callbacks.stream_message_cb == NULL)
1513  return;
1514 
1515  /* Push callback + info on the error context stack */
1516  state.ctx = ctx;
1517  state.callback_name = "stream_message";
1518  state.report_location = message_lsn;
1519  errcallback.callback = output_plugin_error_callback;
1520  errcallback.arg = (void *) &state;
1521  errcallback.previous = error_context_stack;
1522  error_context_stack = &errcallback;
1523 
1524  /* set output state */
1525  ctx->accept_writes = true;
1526  ctx->write_xid = txn != NULL ? txn->xid : InvalidTransactionId;
1527  ctx->write_location = message_lsn;
1528  ctx->end_xact = false;
1529 
1530  /* do the actual work: call callback */
1531  ctx->callbacks.stream_message_cb(ctx, txn, message_lsn, transactional, prefix,
1532  message_size, message);
1533 
1534  /* Pop the error context stack */
1535  error_context_stack = errcallback.previous;
1536 }

References LogicalDecodingContext::accept_writes, ErrorContextCallback::arg, Assert(), ErrorContextCallback::callback, LogicalDecodingContext::callbacks, LogicalDecodingContext::end_xact, 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 1363 of file logical.c.

1365 {
1366  LogicalDecodingContext *ctx = cache->private_data;
1368  ErrorContextCallback errcallback;
1369 
1370  Assert(!ctx->fast_forward);
1371 
1372  /*
1373  * We're only supposed to call this when streaming and two-phase commits
1374  * are supported.
1375  */
1376  Assert(ctx->streaming);
1377  Assert(ctx->twophase);
1378 
1379  /* Push callback + info on the error context stack */
1380  state.ctx = ctx;
1381  state.callback_name = "stream_prepare";
1382  state.report_location = txn->final_lsn;
1383  errcallback.callback = output_plugin_error_callback;
1384  errcallback.arg = (void *) &state;
1385  errcallback.previous = error_context_stack;
1386  error_context_stack = &errcallback;
1387 
1388  /* set output state */
1389  ctx->accept_writes = true;
1390  ctx->write_xid = txn->xid;
1391  ctx->write_location = txn->end_lsn;
1392  ctx->end_xact = true;
1393 
1394  /* in streaming mode with two-phase commits, stream_prepare_cb is required */
1395  if (ctx->callbacks.stream_prepare_cb == NULL)
1396  ereport(ERROR,
1397  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1398  errmsg("logical streaming at prepare time requires a %s callback",
1399  "stream_prepare_cb")));
1400 
1401  ctx->callbacks.stream_prepare_cb(ctx, txn, prepare_lsn);
1402 
1403  /* Pop the error context stack */
1404  error_context_stack = errcallback.previous;
1405 }

References LogicalDecodingContext::accept_writes, ErrorContextCallback::arg, Assert(), ErrorContextCallback::callback, LogicalDecodingContext::callbacks, ReorderBufferTXN::end_lsn, LogicalDecodingContext::end_xact, 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 1224 of file logical.c.

1226 {
1227  LogicalDecodingContext *ctx = cache->private_data;
1229  ErrorContextCallback errcallback;
1230 
1231  Assert(!ctx->fast_forward);
1232 
1233  /* We're only supposed to call this when streaming is supported. */
1234  Assert(ctx->streaming);
1235 
1236  /* Push callback + info on the error context stack */
1237  state.ctx = ctx;
1238  state.callback_name = "stream_start";
1239  state.report_location = first_lsn;
1240  errcallback.callback = output_plugin_error_callback;
1241  errcallback.arg = (void *) &state;
1242  errcallback.previous = error_context_stack;
1243  error_context_stack = &errcallback;
1244 
1245  /* set output state */
1246  ctx->accept_writes = true;
1247  ctx->write_xid = txn->xid;
1248 
1249  /*
1250  * Report this message's lsn so replies from clients can give an
1251  * up-to-date answer. This won't ever be enough (and shouldn't be!) to
1252  * confirm receipt of this transaction, but it might allow another
1253  * transaction's commit to be confirmed with one message.
1254  */
1255  ctx->write_location = first_lsn;
1256 
1257  ctx->end_xact = false;
1258 
1259  /* in streaming mode, stream_start_cb is required */
1260  if (ctx->callbacks.stream_start_cb == NULL)
1261  ereport(ERROR,
1262  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1263  errmsg("logical streaming requires a %s callback",
1264  "stream_start_cb")));
1265 
1266  ctx->callbacks.stream_start_cb(ctx, txn);
1267 
1268  /* Pop the error context stack */
1269  error_context_stack = errcallback.previous;
1270 }

References LogicalDecodingContext::accept_writes, ErrorContextCallback::arg, Assert(), ErrorContextCallback::callback, LogicalDecodingContext::callbacks, LogicalDecodingContext::end_xact, 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 1273 of file logical.c.

1275 {
1276  LogicalDecodingContext *ctx = cache->private_data;
1278  ErrorContextCallback errcallback;
1279 
1280  Assert(!ctx->fast_forward);
1281 
1282  /* We're only supposed to call this when streaming is supported. */
1283  Assert(ctx->streaming);
1284 
1285  /* Push callback + info on the error context stack */
1286  state.ctx = ctx;
1287  state.callback_name = "stream_stop";
1288  state.report_location = last_lsn;
1289  errcallback.callback = output_plugin_error_callback;
1290  errcallback.arg = (void *) &state;
1291  errcallback.previous = error_context_stack;
1292  error_context_stack = &errcallback;
1293 
1294  /* set output state */
1295  ctx->accept_writes = true;
1296  ctx->write_xid = txn->xid;
1297 
1298  /*
1299  * Report this message's lsn so replies from clients can give an
1300  * up-to-date answer. This won't ever be enough (and shouldn't be!) to
1301  * confirm receipt of this transaction, but it might allow another
1302  * transaction's commit to be confirmed with one message.
1303  */
1304  ctx->write_location = last_lsn;
1305 
1306  ctx->end_xact = false;
1307 
1308  /* in streaming mode, stream_stop_cb is required */
1309  if (ctx->callbacks.stream_stop_cb == NULL)
1310  ereport(ERROR,
1311  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1312  errmsg("logical streaming requires a %s callback",
1313  "stream_stop_cb")));
1314 
1315  ctx->callbacks.stream_stop_cb(ctx, txn);
1316 
1317  /* Pop the error context stack */
1318  error_context_stack = errcallback.previous;
1319 }

References LogicalDecodingContext::accept_writes, ErrorContextCallback::arg, Assert(), ErrorContextCallback::callback, LogicalDecodingContext::callbacks, LogicalDecodingContext::end_xact, 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 1539 of file logical.c.

1542 {
1543  LogicalDecodingContext *ctx = cache->private_data;
1545  ErrorContextCallback errcallback;
1546 
1547  Assert(!ctx->fast_forward);
1548 
1549  /* We're only supposed to call this when streaming is supported. */
1550  Assert(ctx->streaming);
1551 
1552  /* this callback is optional */
1553  if (!ctx->callbacks.stream_truncate_cb)
1554  return;
1555 
1556  /* Push callback + info on the error context stack */
1557  state.ctx = ctx;
1558  state.callback_name = "stream_truncate";
1559  state.report_location = change->lsn;
1560  errcallback.callback = output_plugin_error_callback;
1561  errcallback.arg = (void *) &state;
1562  errcallback.previous = error_context_stack;
1563  error_context_stack = &errcallback;
1564 
1565  /* set output state */
1566  ctx->accept_writes = true;
1567  ctx->write_xid = txn->xid;
1568 
1569  /*
1570  * Report this change's lsn so replies from clients can give an up-to-date
1571  * answer. This won't ever be enough (and shouldn't be!) to confirm
1572  * receipt of this transaction, but it might allow another transaction's
1573  * commit to be confirmed with one message.
1574  */
1575  ctx->write_location = change->lsn;
1576 
1577  ctx->end_xact = false;
1578 
1579  ctx->callbacks.stream_truncate_cb(ctx, txn, nrelations, relations, change);
1580 
1581  /* Pop the error context stack */
1582  error_context_stack = errcallback.previous;
1583 }

References LogicalDecodingContext::accept_writes, ErrorContextCallback::arg, Assert(), ErrorContextCallback::callback, LogicalDecodingContext::callbacks, LogicalDecodingContext::end_xact, 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 1082 of file logical.c.

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

References LogicalDecodingContext::accept_writes, ErrorContextCallback::arg, Assert(), ErrorContextCallback::callback, LogicalDecodingContext::callbacks, LogicalDecodingContext::end_xact, 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 1833 of file logical.c.

1834 {
1835  ReorderBuffer *rb = ctx->reorder;
1836  PgStat_StatReplSlotEntry repSlotStat;
1837 
1838  /* Nothing to do if we don't have any replication stats to be sent. */
1839  if (rb->spillBytes <= 0 && rb->streamBytes <= 0 && rb->totalBytes <= 0)
1840  return;
1841 
1842  elog(DEBUG2, "UpdateDecodingStats: updating stats %p %lld %lld %lld %lld %lld %lld %lld %lld",
1843  rb,
1844  (long long) rb->spillTxns,
1845  (long long) rb->spillCount,
1846  (long long) rb->spillBytes,
1847  (long long) rb->streamTxns,
1848  (long long) rb->streamCount,
1849  (long long) rb->streamBytes,
1850  (long long) rb->totalTxns,
1851  (long long) rb->totalBytes);
1852 
1853  repSlotStat.spill_txns = rb->spillTxns;
1854  repSlotStat.spill_count = rb->spillCount;
1855  repSlotStat.spill_bytes = rb->spillBytes;
1856  repSlotStat.stream_txns = rb->streamTxns;
1857  repSlotStat.stream_count = rb->streamCount;
1858  repSlotStat.stream_bytes = rb->streamBytes;
1859  repSlotStat.total_txns = rb->totalTxns;
1860  repSlotStat.total_bytes = rb->totalBytes;
1861 
1862  pgstat_report_replslot(ctx->slot, &repSlotStat);
1863 
1864  rb->spillTxns = 0;
1865  rb->spillCount = 0;
1866  rb->spillBytes = 0;
1867  rb->streamTxns = 0;
1868  rb->streamCount = 0;
1869  rb->streamBytes = 0;
1870  rb->totalTxns = 0;
1871  rb->totalBytes = 0;
1872 }
#define DEBUG2
Definition: elog.h:25
void pgstat_report_replslot(ReplicationSlot *slot, const PgStat_StatReplSlotEntry *repSlotStat)
PgStat_Counter stream_count
Definition: pgstat.h:328
PgStat_Counter total_txns
Definition: pgstat.h:330
PgStat_Counter total_bytes
Definition: pgstat.h:331
PgStat_Counter spill_txns
Definition: pgstat.h:324
PgStat_Counter stream_txns
Definition: pgstat.h:327
PgStat_Counter spill_count
Definition: pgstat.h:325
PgStat_Counter stream_bytes
Definition: pgstat.h:329
PgStat_Counter spill_bytes
Definition: pgstat.h:326

References DEBUG2, elog(), pgstat_report_replslot(), LogicalDecodingContext::reorder, LogicalDecodingContext::slot, 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().