PostgreSQL Source Code git master
Data Structures | Typedefs | Functions
logical.h File Reference
#include "access/xlog.h"
#include "access/xlogreader.h"
#include "replication/output_plugin.h"
#include "replication/slot.h"
Include dependency graph for logical.h:
This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Data Structures

struct   LogicalDecodingContext
 

Typedefs

typedef void(*  LogicalOutputPluginWriterWrite) (struct LogicalDecodingContext *lr, XLogRecPtr Ptr, TransactionId xid, bool last_write)
 
typedef LogicalOutputPluginWriterWrite  LogicalOutputPluginWriterPrepareWrite
 
typedef void(*  LogicalOutputPluginWriterUpdateProgress) (struct LogicalDecodingContext *lr, XLogRecPtr Ptr, TransactionId xid, bool skipped_xact)
 
typedef struct LogicalDecodingContext  LogicalDecodingContext
 

Functions

void  CheckLogicalDecodingRequirements (void)
 
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)
 
LogicalDecodingContext *  CreateDecodingContext (XLogRecPtr start_lsn, List *output_plugin_options, bool fast_forward, XLogReaderRoutine *xl_routine, LogicalOutputPluginWriterPrepareWrite prepare_write, LogicalOutputPluginWriterWrite do_write, LogicalOutputPluginWriterUpdateProgress update_progress)
 
void  DecodingContextFindStartpoint (LogicalDecodingContext *ctx)
 
bool  DecodingContextReady (LogicalDecodingContext *ctx)
 
void  FreeDecodingContext (LogicalDecodingContext *ctx)
 
void  LogicalIncreaseXminForSlot (XLogRecPtr current_lsn, TransactionId xmin)
 
void  LogicalIncreaseRestartDecodingForSlot (XLogRecPtr current_lsn, XLogRecPtr restart_lsn)
 
void  LogicalConfirmReceivedLocation (XLogRecPtr lsn)
 
bool  filter_prepare_cb_wrapper (LogicalDecodingContext *ctx, TransactionId xid, const char *gid)
 
bool  filter_by_origin_cb_wrapper (LogicalDecodingContext *ctx, RepOriginId origin_id)
 
void  ResetLogicalStreamingState (void)
 
void  UpdateDecodingStats (LogicalDecodingContext *ctx)
 
bool  LogicalReplicationSlotHasPendingWal (XLogRecPtr end_of_wal)
 
XLogRecPtr  LogicalSlotAdvanceAndCheckSnapState (XLogRecPtr moveto, bool *found_consistent_snapshot)
 

Typedef Documentation

◆ LogicalDecodingContext

typedef struct LogicalDecodingContext LogicalDecodingContext

◆ LogicalOutputPluginWriterPrepareWrite

typedef LogicalOutputPluginWriterWrite LogicalOutputPluginWriterPrepareWrite

Definition at line 25 of file logical.h.

◆ LogicalOutputPluginWriterUpdateProgress

typedef void(* LogicalOutputPluginWriterUpdateProgress) (struct LogicalDecodingContext *lr, XLogRecPtr Ptr, TransactionId xid, bool skipped_xact)

Definition at line 27 of file logical.h.

◆ LogicalOutputPluginWriterWrite

typedef void(* LogicalOutputPluginWriterWrite) (struct LogicalDecodingContext *lr, XLogRecPtr Ptr, TransactionId xid, bool last_write)

Definition at line 19 of file logical.h.

Function Documentation

◆ CheckLogicalDecodingRequirements()

void CheckLogicalDecodingRequirements ( void  )

Definition at line 111 of file logical.c.

112{
114
115 /*
116 * NB: Adding a new requirement likely means that RestoreSlotFromDisk()
117 * needs the same check.
118 */
119
122 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
123 errmsg("logical decoding requires \"wal_level\" >= \"logical\"")));
124
127 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
128 errmsg("logical decoding requires a database connection")));
129
130 if (RecoveryInProgress())
131 {
132 /*
133 * This check may have race conditions, but whenever
134 * XLOG_PARAMETER_CHANGE indicates that wal_level has changed, we
135 * verify that there are no existing logical replication slots. And to
136 * avoid races around creating a new slot,
137 * CheckLogicalDecodingRequirements() is called once before creating
138 * the slot, and once when logical decoding is initially starting up.
139 */
142 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
143 errmsg("logical decoding on standby requires \"wal_level\" >= \"logical\" on the primary")));
144 }
145}
int errcode(int sqlerrcode)
Definition: elog.c:854
int errmsg(const char *fmt,...)
Definition: elog.c:1071
#define ERROR
Definition: elog.h:39
#define ereport(elevel,...)
Definition: elog.h:150
Oid MyDatabaseId
Definition: globals.c:94
#define InvalidOid
Definition: postgres_ext.h:37
void CheckSlotRequirements(void)
Definition: slot.c:1500
bool RecoveryInProgress(void)
Definition: xlog.c:6386
int wal_level
Definition: xlog.c:132
WalLevel GetActiveWalLevelOnStandby(void)
Definition: xlog.c:4901
@ WAL_LEVEL_LOGICAL
Definition: xlog.h:76

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

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

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

505{
507 ReplicationSlot *slot;
508 MemoryContext old_context;
509
510 /* shorter lines... */
511 slot = MyReplicationSlot;
512
513 /* first some sanity checks that are unlikely to be violated */
514 if (slot == NULL)
515 elog(ERROR, "cannot perform logical decoding without an acquired slot");
516
517 /* make sure the passed slot is suitable, these are user facing errors */
518 if (SlotIsPhysical(slot))
520 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
521 errmsg("cannot use physical replication slot for logical decoding")));
522
523 /*
524 * We need to access the system tables during decoding to build the
525 * logical changes unless we are in fast_forward mode where no changes are
526 * generated.
527 */
528 if (slot->data.database != MyDatabaseId && !fast_forward)
530 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
531 errmsg("replication slot \"%s\" was not created in this database",
532 NameStr(slot->data.name))));
533
534 /*
535 * The slots being synced from the primary can't be used for decoding as
536 * they are used after failover. However, we do allow advancing the LSNs
537 * during the synchronization of slots. See update_local_synced_slot.
538 */
541 errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
542 errmsg("cannot use replication slot \"%s\" for logical decoding",
543 NameStr(slot->data.name)),
544 errdetail("This replication slot is being synchronized from the primary server."),
545 errhint("Specify another replication slot."));
546
547 /* slot must be valid to allow decoding */
550
551 if (start_lsn == InvalidXLogRecPtr)
552 {
553 /* continue from last position */
554 start_lsn = slot->data.confirmed_flush;
555 }
556 else if (start_lsn < slot->data.confirmed_flush)
557 {
558 /*
559 * It might seem like we should error out in this case, but it's
560 * pretty common for a client to acknowledge a LSN it doesn't have to
561 * do anything for, and thus didn't store persistently, because the
562 * xlog records didn't result in anything relevant for logical
563 * decoding. Clients have to be able to do that to support synchronous
564 * replication.
565 *
566 * Starting at a different LSN than requested might not catch certain
567 * kinds of client errors; so the client may wish to check that
568 * confirmed_flush_lsn matches its expectations.
569 */
570 elog(LOG, "%X/%08X has been already streamed, forwarding to %X/%08X",
571 LSN_FORMAT_ARGS(start_lsn),
573
574 start_lsn = slot->data.confirmed_flush;
575 }
576
577 ctx = StartupDecodingContext(output_plugin_options,
578 start_lsn, InvalidTransactionId, false,
579 fast_forward, false, xl_routine, prepare_write,
580 do_write, update_progress);
581
582 /* call output plugin initialization callback */
583 old_context = MemoryContextSwitchTo(ctx->context);
584 if (ctx->callbacks.startup_cb != NULL)
585 startup_cb_wrapper(ctx, &ctx->options, false);
586 MemoryContextSwitchTo(old_context);
587
588 /*
589 * We allow decoding of prepared transactions when the two_phase is
590 * enabled at the time of slot creation, or when the two_phase option is
591 * given at the streaming start, provided the plugin supports all the
592 * callbacks for two-phase.
593 */
594 ctx->twophase &= (slot->data.two_phase || ctx->twophase_opt_given);
595
596 /* Mark slot to allow two_phase decoding if not already marked */
597 if (ctx->twophase && !slot->data.two_phase)
598 {
599 SpinLockAcquire(&slot->mutex);
600 slot->data.two_phase = true;
601 slot->data.two_phase_at = start_lsn;
602 SpinLockRelease(&slot->mutex);
606 }
607
609
610 ereport(LOG,
611 (errmsg("starting logical decoding for slot \"%s\"",
612 NameStr(slot->data.name)),
613 errdetail("Streaming transactions committing after %X/%08X, reading WAL from %X/%08X.",
616
617 return ctx;
618}
#define NameStr(name)
Definition: c.h:751
int errdetail(const char *fmt,...)
Definition: elog.c:1207
int errhint(const char *fmt,...)
Definition: elog.c:1321
#define LOG
Definition: elog.h:31
#define elog(elevel,...)
Definition: elog.h:226
Assert(PointerIsAligned(start, uint64))
static void startup_cb_wrapper(LogicalDecodingContext *ctx, OutputPluginOptions *opt, bool is_init)
Definition: logical.c:774
static LogicalDecodingContext * StartupDecodingContext(List *output_plugin_options, XLogRecPtr start_lsn, TransactionId xmin_horizon, bool need_full_snapshot, bool fast_forward, bool in_create, XLogReaderRoutine *xl_routine, LogicalOutputPluginWriterPrepareWrite prepare_write, LogicalOutputPluginWriterWrite do_write, LogicalOutputPluginWriterUpdateProgress update_progress)
Definition: logical.c:152
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
Definition: palloc.h:124
const void * data
void ReplicationSlotMarkDirty(void)
Definition: slot.c:1106
ReplicationSlot * MyReplicationSlot
Definition: slot.c:148
void ReplicationSlotSave(void)
Definition: slot.c:1088
#define SlotIsPhysical(slot)
Definition: slot.h:254
@ RS_INVAL_NONE
Definition: slot.h:60
bool IsSyncingReplicationSlots(void)
Definition: slotsync.c:1668
void SnapBuildSetTwoPhaseAt(SnapBuild *builder, XLogRecPtr ptr)
Definition: snapbuild.c:295
#define SpinLockRelease(lock)
Definition: spin.h:61
#define SpinLockAcquire(lock)
Definition: spin.h:59
OutputPluginOptions options
Definition: logical.h:54
MemoryContext context
Definition: logical.h:36
struct SnapBuild * snapshot_builder
Definition: logical.h:44
OutputPluginCallbacks callbacks
Definition: logical.h:53
struct ReorderBuffer * reorder
Definition: logical.h:43
bool twophase_opt_given
Definition: logical.h:101
LogicalDecodeStartupCB startup_cb
Definition: output_plugin.h:218
bool output_rewrites
Definition: reorderbuffer.h:651
XLogRecPtr two_phase_at
Definition: slot.h:124
XLogRecPtr restart_lsn
Definition: slot.h:107
XLogRecPtr confirmed_flush
Definition: slot.h:118
ReplicationSlotInvalidationCause invalidated
Definition: slot.h:110
slock_t mutex
Definition: slot.h:165
ReplicationSlotPersistentData data
Definition: slot.h:192
#define InvalidTransactionId
Definition: transam.h:31
#define LSN_FORMAT_ARGS(lsn)
Definition: xlogdefs.h:46
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28

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

Referenced by LogicalReplicationSlotHasPendingWal(), LogicalSlotAdvanceAndCheckSnapState(), 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 332 of file logical.c.

340{
341 TransactionId xmin_horizon = InvalidTransactionId;
342 ReplicationSlot *slot;
343 NameData plugin_name;
345 MemoryContext old_context;
346
347 /*
348 * On a standby, this check is also required while creating the slot.
349 * Check the comments in the function.
350 */
352
353 /* shorter lines... */
354 slot = MyReplicationSlot;
355
356 /* first some sanity checks that are unlikely to be violated */
357 if (slot == NULL)
358 elog(ERROR, "cannot perform logical decoding without an acquired slot");
359
360 if (plugin == NULL)
361 elog(ERROR, "cannot initialize logical decoding without a specified plugin");
362
363 /* Make sure the passed slot is suitable. These are user facing errors. */
364 if (SlotIsPhysical(slot))
366 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
367 errmsg("cannot use physical replication slot for logical decoding")));
368
369 if (slot->data.database != MyDatabaseId)
371 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
372 errmsg("replication slot \"%s\" was not created in this database",
373 NameStr(slot->data.name))));
374
375 if (IsTransactionState() &&
378 (errcode(ERRCODE_ACTIVE_SQL_TRANSACTION),
379 errmsg("cannot create logical replication slot in transaction that has performed writes")));
380
381 /*
382 * Register output plugin name with slot. We need the mutex to avoid
383 * concurrent reading of a partially copied string. But we don't want any
384 * complicated code while holding a spinlock, so do namestrcpy() outside.
385 */
386 namestrcpy(&plugin_name, plugin);
387 SpinLockAcquire(&slot->mutex);
388 slot->data.plugin = plugin_name;
389 SpinLockRelease(&slot->mutex);
390
391 if (XLogRecPtrIsInvalid(restart_lsn))
393 else
394 {
395 SpinLockAcquire(&slot->mutex);
396 slot->data.restart_lsn = restart_lsn;
397 SpinLockRelease(&slot->mutex);
398 }
399
400 /* ----
401 * This is a bit tricky: We need to determine a safe xmin horizon to start
402 * decoding from, to avoid starting from a running xacts record referring
403 * to xids whose rows have been vacuumed or pruned
404 * already. GetOldestSafeDecodingTransactionId() returns such a value, but
405 * without further interlock its return value might immediately be out of
406 * date.
407 *
408 * So we have to acquire the ProcArrayLock to prevent computation of new
409 * xmin horizons by other backends, get the safe decoding xid, and inform
410 * the slot machinery about the new limit. Once that's done the
411 * ProcArrayLock can be released as the slot machinery now is
412 * protecting against vacuum.
413 *
414 * Note that, temporarily, the data, not just the catalog, xmin has to be
415 * reserved if a data snapshot is to be exported. Otherwise the initial
416 * data snapshot created here is not guaranteed to be valid. After that
417 * the data xmin doesn't need to be managed anymore and the global xmin
418 * should be recomputed. As we are fine with losing the pegged data xmin
419 * after crash - no chance a snapshot would get exported anymore - we can
420 * get away with just setting the slot's
421 * effective_xmin. ReplicationSlotRelease will reset it again.
422 *
423 * ----
424 */
425 LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
426
427 xmin_horizon = GetOldestSafeDecodingTransactionId(!need_full_snapshot);
428
429 SpinLockAcquire(&slot->mutex);
430 slot->effective_catalog_xmin = xmin_horizon;
431 slot->data.catalog_xmin = xmin_horizon;
432 if (need_full_snapshot)
433 slot->effective_xmin = xmin_horizon;
434 SpinLockRelease(&slot->mutex);
435
437
438 LWLockRelease(ProcArrayLock);
439
442
443 ctx = StartupDecodingContext(NIL, restart_lsn, xmin_horizon,
444 need_full_snapshot, false, true,
445 xl_routine, prepare_write, do_write,
446 update_progress);
447
448 /* call output plugin initialization callback */
449 old_context = MemoryContextSwitchTo(ctx->context);
450 if (ctx->callbacks.startup_cb != NULL)
451 startup_cb_wrapper(ctx, &ctx->options, true);
452 MemoryContextSwitchTo(old_context);
453
454 /*
455 * We allow decoding of prepared transactions when the two_phase is
456 * enabled at the time of slot creation, or when the two_phase option is
457 * given at the streaming start, provided the plugin supports all the
458 * callbacks for two-phase.
459 */
460 ctx->twophase &= slot->data.two_phase;
461
463
464 return ctx;
465}
uint32 TransactionId
Definition: c.h:657
void CheckLogicalDecodingRequirements(void)
Definition: logical.c:111
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1174
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1894
@ LW_EXCLUSIVE
Definition: lwlock.h:112
void namestrcpy(Name name, const char *str)
Definition: name.c:233
#define NIL
Definition: pg_list.h:68
static const char * plugin
Definition: pg_recvlogical.c:61
TransactionId GetOldestSafeDecodingTransactionId(bool catalogOnly)
Definition: procarray.c:2907
void ReplicationSlotReserveWal(void)
Definition: slot.c:1539
void ReplicationSlotsComputeRequiredXmin(bool already_locked)
Definition: slot.c:1145
TransactionId catalog_xmin
Definition: slot.h:104
TransactionId effective_catalog_xmin
Definition: slot.h:189
TransactionId effective_xmin
Definition: slot.h:188
Definition: c.h:746
bool IsTransactionState(void)
Definition: xact.c:387
TransactionId GetTopTransactionIdIfAny(void)
Definition: xact.c:441
#define XLogRecPtrIsInvalid(r)
Definition: xlogdefs.h:29

References LogicalDecodingContext::callbacks, ReplicationSlotPersistentData::catalog_xmin, CheckLogicalDecodingRequirements(), 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 633 of file logical.c.

634{
635 ReplicationSlot *slot = ctx->slot;
636
637 /* Initialize from where to start reading WAL. */
639
640 elog(DEBUG1, "searching for logical decoding starting point, starting at %X/%08X",
642
643 /* Wait for a consistent starting point */
644 for (;;)
645 {
646 XLogRecord *record;
647 char *err = NULL;
648
649 /* the read_page callback waits for new WAL */
650 record = XLogReadRecord(ctx->reader, &err);
651 if (err)
652 elog(ERROR, "could not find logical decoding starting point: %s", err);
653 if (!record)
654 elog(ERROR, "could not find logical decoding starting point");
655
657
658 /* only continue till we found a consistent spot */
659 if (DecodingContextReady(ctx))
660 break;
661
663 }
664
665 SpinLockAcquire(&slot->mutex);
666 slot->data.confirmed_flush = ctx->reader->EndRecPtr;
667 if (slot->data.two_phase)
668 slot->data.two_phase_at = ctx->reader->EndRecPtr;
669 SpinLockRelease(&slot->mutex);
670}
void LogicalDecodingProcessRecord(LogicalDecodingContext *ctx, XLogReaderState *record)
Definition: decode.c:88
#define DEBUG1
Definition: elog.h:30
void err(int eval, const char *fmt,...)
Definition: err.c:43
bool DecodingContextReady(LogicalDecodingContext *ctx)
Definition: logical.c:624
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:122
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:390
void XLogBeginRead(XLogReaderState *state, XLogRecPtr RecPtr)
Definition: xlogreader.c:232

References CHECK_FOR_INTERRUPTS, ReplicationSlotPersistentData::confirmed_flush, ReplicationSlot::data, DEBUG1, DecodingContextReady(), elog, XLogReaderState::EndRecPtr, err(), 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 624 of file logical.c.

625{
627}
SnapBuildState SnapBuildCurrentState(SnapBuild *builder)
Definition: snapbuild.c:277
@ SNAPBUILD_CONSISTENT
Definition: snapbuild.h:50

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

Referenced by DecodingContextFindStartpoint(), and LogicalSlotAdvanceAndCheckSnapState().

◆ filter_by_origin_cb_wrapper()

bool filter_by_origin_cb_wrapper ( LogicalDecodingContext *  ctx,
RepOriginId  origin_id 
)

Definition at line 1199 of file logical.c.

1200{
1202 ErrorContextCallback errcallback;
1203 bool ret;
1204
1205 Assert(!ctx->fast_forward);
1206
1207 /* Push callback + info on the error context stack */
1208 state.ctx = ctx;
1209 state.callback_name = "filter_by_origin";
1210 state.report_location = InvalidXLogRecPtr;
1212 errcallback.arg = &state;
1213 errcallback.previous = error_context_stack;
1214 error_context_stack = &errcallback;
1215
1216 /* set output state */
1217 ctx->accept_writes = false;
1218 ctx->end_xact = false;
1219
1220 /* do the actual work: call callback */
1221 ret = ctx->callbacks.filter_by_origin_cb(ctx, origin_id);
1222
1223 /* Pop the error context stack */
1224 error_context_stack = errcallback.previous;
1225
1226 return ret;
1227}
ErrorContextCallback * error_context_stack
Definition: elog.c:95
static void output_plugin_error_callback(void *arg)
Definition: logical.c:755
struct ErrorContextCallback * previous
Definition: elog.h:297
void * arg
Definition: elog.h:299
void(* callback)(void *arg)
Definition: elog.h:298
LogicalDecodeFilterByOriginCB filter_by_origin_cb
Definition: output_plugin.h:224
Definition: regguts.h:323

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

1169{
1171 ErrorContextCallback errcallback;
1172 bool ret;
1173
1174 Assert(!ctx->fast_forward);
1175
1176 /* Push callback + info on the error context stack */
1177 state.ctx = ctx;
1178 state.callback_name = "filter_prepare";
1179 state.report_location = InvalidXLogRecPtr;
1181 errcallback.arg = &state;
1182 errcallback.previous = error_context_stack;
1183 error_context_stack = &errcallback;
1184
1185 /* set output state */
1186 ctx->accept_writes = false;
1187 ctx->end_xact = false;
1188
1189 /* do the actual work: call callback */
1190 ret = ctx->callbacks.filter_prepare_cb(ctx, xid, gid);
1191
1192 /* Pop the error context stack */
1193 error_context_stack = errcallback.previous;
1194
1195 return ret;
1196}
LogicalDecodeFilterPrepareCB filter_prepare_cb
Definition: output_plugin.h:228

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

678{
679 if (ctx->callbacks.shutdown_cb != NULL)
681
686}
static void shutdown_cb_wrapper(LogicalDecodingContext *ctx)
Definition: logical.c:802
void MemoryContextDelete(MemoryContext context)
Definition: mcxt.c:469
void ReorderBufferFree(ReorderBuffer *rb)
Definition: reorderbuffer.c:416
void FreeSnapshotBuilder(SnapBuild *builder)
Definition: snapbuild.c:233
LogicalDecodeShutdownCB shutdown_cb
Definition: output_plugin.h:225
void XLogReaderFree(XLogReaderState *state)
Definition: xlogreader.c:162

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(), LogicalReplicationSlotHasPendingWal(), LogicalSlotAdvanceAndCheckSnapState(), pg_logical_slot_get_changes_guts(), and StartLogicalReplication().

◆ LogicalConfirmReceivedLocation()

void LogicalConfirmReceivedLocation ( XLogRecPtr  lsn )

Definition at line 1820 of file logical.c.

1821{
1822 Assert(lsn != InvalidXLogRecPtr);
1823
1824 /* Do an unlocked check for candidate_lsn first. */
1827 {
1828 bool updated_xmin = false;
1829 bool updated_restart = false;
1830 XLogRecPtr restart_lsn pg_attribute_unused();
1831
1833
1834 /* remember the old restart lsn */
1835 restart_lsn = MyReplicationSlot->data.restart_lsn;
1836
1837 /*
1838 * Prevent moving the confirmed_flush backwards, as this could lead to
1839 * data duplication issues caused by replicating already replicated
1840 * changes.
1841 *
1842 * This can happen when a client acknowledges an LSN it doesn't have
1843 * to do anything for, and thus didn't store persistently. After a
1844 * restart, the client can send the prior LSN that it stored
1845 * persistently as an acknowledgement, but we need to ignore such an
1846 * LSN. See similar case handling in CreateDecodingContext.
1847 */
1850
1851 /* if we're past the location required for bumping xmin, do so */
1854 {
1855 /*
1856 * We have to write the changed xmin to disk *before* we change
1857 * the in-memory value, otherwise after a crash we wouldn't know
1858 * that some catalog tuples might have been removed already.
1859 *
1860 * Ensure that by first writing to ->xmin and only update
1861 * ->effective_xmin once the new state is synced to disk. After a
1862 * crash ->effective_xmin is set to ->xmin.
1863 */
1866 {
1870 updated_xmin = true;
1871 }
1872 }
1873
1876 {
1878
1882 updated_restart = true;
1883 }
1884
1886
1887 /* first write new xmin to disk, so we know what's up after a crash */
1888 if (updated_xmin || updated_restart)
1889 {
1890#ifdef USE_INJECTION_POINTS
1891 XLogSegNo seg1,
1892 seg2;
1893
1894 XLByteToSeg(restart_lsn, seg1, wal_segment_size);
1896
1897 /* trigger injection point, but only if segment changes */
1898 if (seg1 != seg2)
1899 INJECTION_POINT("logical-replication-slot-advance-segment", NULL);
1900#endif
1901
1904 elog(DEBUG1, "updated xmin: %u restart: %u", updated_xmin, updated_restart);
1905 }
1906
1907 /*
1908 * Now the new xmin is safely on disk, we can let the global value
1909 * advance. We do not take ProcArrayLock or similar since we only
1910 * advance xmin here and there's not much harm done by a concurrent
1911 * computation missing that.
1912 */
1913 if (updated_xmin)
1914 {
1918
1921 }
1922 }
1923 else
1924 {
1926
1927 /*
1928 * Prevent moving the confirmed_flush backwards. See comments above
1929 * for the details.
1930 */
1933
1935 }
1936}
#define pg_attribute_unused()
Definition: c.h:132
#define INJECTION_POINT(name, arg)
void ReplicationSlotsComputeRequiredLSN(void)
Definition: slot.c:1201
XLogRecPtr candidate_xmin_lsn
Definition: slot.h:208
XLogRecPtr candidate_restart_valid
Definition: slot.h:209
XLogRecPtr candidate_restart_lsn
Definition: slot.h:210
TransactionId candidate_catalog_xmin
Definition: slot.h:207
#define TransactionIdIsValid(xid)
Definition: transam.h:41
int wal_segment_size
Definition: xlog.c:144
#define XLByteToSeg(xlrp, logSegNo, wal_segsz_bytes)
Definition: xlog_internal.h:117
uint64 XLogRecPtr
Definition: xlogdefs.h:21
uint64 XLogSegNo
Definition: xlogdefs.h:51

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, INJECTION_POINT, InvalidTransactionId, InvalidXLogRecPtr, ReplicationSlot::mutex, MyReplicationSlot, pg_attribute_unused, ReplicationSlotMarkDirty(), ReplicationSlotSave(), ReplicationSlotsComputeRequiredLSN(), ReplicationSlotsComputeRequiredXmin(), ReplicationSlotPersistentData::restart_lsn, SpinLockAcquire, SpinLockRelease, TransactionIdIsValid, wal_segment_size, and XLByteToSeg.

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

◆ LogicalIncreaseRestartDecodingForSlot()

void LogicalIncreaseRestartDecodingForSlot ( XLogRecPtr  current_lsn,
XLogRecPtr  restart_lsn 
)

Definition at line 1744 of file logical.c.

1745{
1746 bool updated_lsn = false;
1747 ReplicationSlot *slot;
1748
1749 slot = MyReplicationSlot;
1750
1751 Assert(slot != NULL);
1752 Assert(restart_lsn != InvalidXLogRecPtr);
1753 Assert(current_lsn != InvalidXLogRecPtr);
1754
1755 SpinLockAcquire(&slot->mutex);
1756
1757 /* don't overwrite if have a newer restart lsn */
1758 if (restart_lsn <= slot->data.restart_lsn)
1759 {
1760 SpinLockRelease(&slot->mutex);
1761 }
1762
1763 /*
1764 * We might have already flushed far enough to directly accept this lsn,
1765 * in this case there is no need to check for existing candidate LSNs
1766 */
1767 else if (current_lsn <= slot->data.confirmed_flush)
1768 {
1769 slot->candidate_restart_valid = current_lsn;
1770 slot->candidate_restart_lsn = restart_lsn;
1771 SpinLockRelease(&slot->mutex);
1772
1773 /* our candidate can directly be used */
1774 updated_lsn = true;
1775 }
1776
1777 /*
1778 * Only increase if the previous values have been applied, otherwise we
1779 * might never end up updating if the receiver acks too slowly. A missed
1780 * value here will just cause some extra effort after reconnecting.
1781 */
1783 {
1784 slot->candidate_restart_valid = current_lsn;
1785 slot->candidate_restart_lsn = restart_lsn;
1786 SpinLockRelease(&slot->mutex);
1787
1788 elog(DEBUG1, "got new restart lsn %X/%08X at %X/%08X",
1789 LSN_FORMAT_ARGS(restart_lsn),
1790 LSN_FORMAT_ARGS(current_lsn));
1791 }
1792 else
1793 {
1794 XLogRecPtr candidate_restart_lsn;
1795 XLogRecPtr candidate_restart_valid;
1796 XLogRecPtr confirmed_flush;
1797
1798 candidate_restart_lsn = slot->candidate_restart_lsn;
1799 candidate_restart_valid = slot->candidate_restart_valid;
1800 confirmed_flush = slot->data.confirmed_flush;
1801 SpinLockRelease(&slot->mutex);
1802
1803 elog(DEBUG1, "failed to increase restart lsn: proposed %X/%08X, after %X/%08X, current candidate %X/%08X, current after %X/%08X, flushed up to %X/%08X",
1804 LSN_FORMAT_ARGS(restart_lsn),
1805 LSN_FORMAT_ARGS(current_lsn),
1806 LSN_FORMAT_ARGS(candidate_restart_lsn),
1807 LSN_FORMAT_ARGS(candidate_restart_valid),
1808 LSN_FORMAT_ARGS(confirmed_flush));
1809 }
1810
1811 /* candidates are already valid with the current flush position, apply */
1812 if (updated_lsn)
1814}
void LogicalConfirmReceivedLocation(XLogRecPtr lsn)
Definition: logical.c:1820

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

1677{
1678 bool updated_xmin = false;
1679 ReplicationSlot *slot;
1680 bool got_new_xmin = false;
1681
1682 slot = MyReplicationSlot;
1683
1684 Assert(slot != NULL);
1685
1686 SpinLockAcquire(&slot->mutex);
1687
1688 /*
1689 * don't overwrite if we already have a newer xmin. This can happen if we
1690 * restart decoding in a slot.
1691 */
1693 {
1694 }
1695
1696 /*
1697 * If the client has already confirmed up to this lsn, we directly can
1698 * mark this as accepted. This can happen if we restart decoding in a
1699 * slot.
1700 */
1701 else if (current_lsn <= slot->data.confirmed_flush)
1702 {
1703 slot->candidate_catalog_xmin = xmin;
1704 slot->candidate_xmin_lsn = current_lsn;
1705
1706 /* our candidate can directly be used */
1707 updated_xmin = true;
1708 }
1709
1710 /*
1711 * Only increase if the previous values have been applied, otherwise we
1712 * might never end up updating if the receiver acks too slowly.
1713 */
1714 else if (slot->candidate_xmin_lsn == InvalidXLogRecPtr)
1715 {
1716 slot->candidate_catalog_xmin = xmin;
1717 slot->candidate_xmin_lsn = current_lsn;
1718
1719 /*
1720 * Log new xmin at an appropriate log level after releasing the
1721 * spinlock.
1722 */
1723 got_new_xmin = true;
1724 }
1725 SpinLockRelease(&slot->mutex);
1726
1727 if (got_new_xmin)
1728 elog(DEBUG1, "got new catalog xmin %u at %X/%08X", xmin,
1729 LSN_FORMAT_ARGS(current_lsn));
1730
1731 /* candidate already valid with the current flush position, apply */
1732 if (updated_xmin)
1734}
bool TransactionIdPrecedesOrEquals(TransactionId id1, TransactionId id2)
Definition: transam.c:299

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().

◆ LogicalReplicationSlotHasPendingWal()

bool LogicalReplicationSlotHasPendingWal ( XLogRecPtr  end_of_wal )

Definition at line 1999 of file logical.c.

2000{
2001 bool has_pending_wal = false;
2002
2004
2005 PG_TRY();
2006 {
2008
2009 /*
2010 * Create our decoding context in fast_forward mode, passing start_lsn
2011 * as InvalidXLogRecPtr, so that we start processing from the slot's
2012 * confirmed_flush.
2013 */
2015 NIL,
2016 true, /* fast_forward */
2017 XL_ROUTINE(.page_read = read_local_xlog_page,
2018 .segment_open = wal_segment_open,
2019 .segment_close = wal_segment_close),
2020 NULL, NULL, NULL);
2021
2022 /*
2023 * Start reading at the slot's restart_lsn, which we know points to a
2024 * valid record.
2025 */
2027
2028 /* Invalidate non-timetravel entries */
2030
2031 /* Loop until the end of WAL or some changes are processed */
2032 while (!has_pending_wal && ctx->reader->EndRecPtr < end_of_wal)
2033 {
2034 XLogRecord *record;
2035 char *errm = NULL;
2036
2037 record = XLogReadRecord(ctx->reader, &errm);
2038
2039 if (errm)
2040 elog(ERROR, "could not find record for logical decoding: %s", errm);
2041
2042 if (record != NULL)
2044
2045 has_pending_wal = ctx->processing_required;
2046
2048 }
2049
2050 /* Clean up */
2053 }
2054 PG_CATCH();
2055 {
2056 /* clear all timetravel entries */
2058
2059 PG_RE_THROW();
2060 }
2061 PG_END_TRY();
2062
2063 return has_pending_wal;
2064}
#define PG_RE_THROW()
Definition: elog.h:405
#define PG_TRY(...)
Definition: elog.h:372
#define PG_END_TRY(...)
Definition: elog.h:397
#define PG_CATCH(...)
Definition: elog.h:382
void InvalidateSystemCaches(void)
Definition: inval.c:916
void FreeDecodingContext(LogicalDecodingContext *ctx)
Definition: logical.c:677
LogicalDecodingContext * CreateDecodingContext(XLogRecPtr start_lsn, List *output_plugin_options, bool fast_forward, XLogReaderRoutine *xl_routine, LogicalOutputPluginWriterPrepareWrite prepare_write, LogicalOutputPluginWriterWrite do_write, LogicalOutputPluginWriterUpdateProgress update_progress)
Definition: logical.c:498
bool processing_required
Definition: logical.h:114
#define XL_ROUTINE(...)
Definition: xlogreader.h:117
void wal_segment_close(XLogReaderState *state)
Definition: xlogutils.c:831
void wal_segment_open(XLogReaderState *state, XLogSegNo nextSegNo, TimeLineID *tli_p)
Definition: xlogutils.c:806
int read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen, XLogRecPtr targetRecPtr, char *cur_page)
Definition: xlogutils.c:845

References Assert(), CHECK_FOR_INTERRUPTS, CreateDecodingContext(), ReplicationSlot::data, elog, XLogReaderState::EndRecPtr, ERROR, FreeDecodingContext(), InvalidateSystemCaches(), InvalidXLogRecPtr, LogicalDecodingProcessRecord(), MyReplicationSlot, NIL, PG_CATCH, PG_END_TRY, PG_RE_THROW, PG_TRY, LogicalDecodingContext::processing_required, read_local_xlog_page(), LogicalDecodingContext::reader, ReplicationSlotPersistentData::restart_lsn, wal_segment_close(), wal_segment_open(), XL_ROUTINE, XLogBeginRead(), and XLogReadRecord().

Referenced by binary_upgrade_logical_slot_has_caught_up().

◆ LogicalSlotAdvanceAndCheckSnapState()

XLogRecPtr LogicalSlotAdvanceAndCheckSnapState ( XLogRecPtr  moveto,
bool *  found_consistent_snapshot 
)

Definition at line 2081 of file logical.c.

2083{
2086 XLogRecPtr retlsn;
2087
2088 Assert(moveto != InvalidXLogRecPtr);
2089
2090 if (found_consistent_snapshot)
2091 *found_consistent_snapshot = false;
2092
2093 PG_TRY();
2094 {
2095 /*
2096 * Create our decoding context in fast_forward mode, passing start_lsn
2097 * as InvalidXLogRecPtr, so that we start processing from my slot's
2098 * confirmed_flush.
2099 */
2101 NIL,
2102 true, /* fast_forward */
2103 XL_ROUTINE(.page_read = read_local_xlog_page,
2104 .segment_open = wal_segment_open,
2105 .segment_close = wal_segment_close),
2106 NULL, NULL, NULL);
2107
2108 /*
2109 * Wait for specified streaming replication standby servers (if any)
2110 * to confirm receipt of WAL up to moveto lsn.
2111 */
2113
2114 /*
2115 * Start reading at the slot's restart_lsn, which we know to point to
2116 * a valid record.
2117 */
2119
2120 /* invalidate non-timetravel entries */
2122
2123 /* Decode records until we reach the requested target */
2124 while (ctx->reader->EndRecPtr < moveto)
2125 {
2126 char *errm = NULL;
2127 XLogRecord *record;
2128
2129 /*
2130 * Read records. No changes are generated in fast_forward mode,
2131 * but snapbuilder/slot statuses are updated properly.
2132 */
2133 record = XLogReadRecord(ctx->reader, &errm);
2134 if (errm)
2135 elog(ERROR, "could not find record while advancing replication slot: %s",
2136 errm);
2137
2138 /*
2139 * Process the record. Storage-level changes are ignored in
2140 * fast_forward mode, but other modules (such as snapbuilder)
2141 * might still have critical updates to do.
2142 */
2143 if (record)
2144 {
2146
2147 /*
2148 * We used to have bugs where logical decoding would fail to
2149 * preserve the resource owner. That's important here, so
2150 * verify that that doesn't happen anymore. XXX this could be
2151 * removed once it's been battle-tested.
2152 */
2153 Assert(CurrentResourceOwner == old_resowner);
2154 }
2155
2157 }
2158
2159 if (found_consistent_snapshot && DecodingContextReady(ctx))
2160 *found_consistent_snapshot = true;
2161
2162 if (ctx->reader->EndRecPtr != InvalidXLogRecPtr)
2163 {
2165
2166 /*
2167 * If only the confirmed_flush LSN has changed the slot won't get
2168 * marked as dirty by the above. Callers on the walsender
2169 * interface are expected to keep track of their own progress and
2170 * don't need it written out. But SQL-interface users cannot
2171 * specify their own start positions and it's harder for them to
2172 * keep track of their progress, so we should make more of an
2173 * effort to save it for them.
2174 *
2175 * Dirty the slot so it is written out at the next checkpoint. The
2176 * LSN position advanced to may still be lost on a crash but this
2177 * makes the data consistent after a clean shutdown.
2178 */
2180 }
2181
2183
2184 /* free context, call shutdown callback */
2186
2188 }
2189 PG_CATCH();
2190 {
2191 /* clear all timetravel entries */
2193
2194 PG_RE_THROW();
2195 }
2196 PG_END_TRY();
2197
2198 return retlsn;
2199}
#define PG_USED_FOR_ASSERTS_ONLY
Definition: c.h:223
ResourceOwner CurrentResourceOwner
Definition: resowner.c:173
void WaitForStandbyConfirmation(XLogRecPtr wait_for_lsn)
Definition: slot.c:3059

References Assert(), CHECK_FOR_INTERRUPTS, ReplicationSlotPersistentData::confirmed_flush, CreateDecodingContext(), CurrentResourceOwner, ReplicationSlot::data, DecodingContextReady(), elog, XLogReaderState::EndRecPtr, ERROR, FreeDecodingContext(), InvalidateSystemCaches(), InvalidXLogRecPtr, LogicalConfirmReceivedLocation(), LogicalDecodingProcessRecord(), MyReplicationSlot, NIL, PG_CATCH, PG_END_TRY, PG_RE_THROW, PG_TRY, PG_USED_FOR_ASSERTS_ONLY, read_local_xlog_page(), LogicalDecodingContext::reader, ReplicationSlotMarkDirty(), ReplicationSlotPersistentData::restart_lsn, WaitForStandbyConfirmation(), wal_segment_close(), wal_segment_open(), XL_ROUTINE, XLogBeginRead(), and XLogReadRecord().

Referenced by pg_logical_replication_slot_advance(), and update_local_synced_slot().

◆ ResetLogicalStreamingState()

void ResetLogicalStreamingState ( void  )

Definition at line 1942 of file logical.c.

1943{
1945 bsysscan = false;
1946}
bool bsysscan
Definition: xact.c:100
TransactionId CheckXidAlive
Definition: xact.c:99

References bsysscan, CheckXidAlive, and InvalidTransactionId.

Referenced by AbortSubTransaction(), and AbortTransaction().

◆ UpdateDecodingStats()

void UpdateDecodingStats ( LogicalDecodingContext *  ctx )

Definition at line 1952 of file logical.c.

1953{
1954 ReorderBuffer *rb = ctx->reorder;
1955 PgStat_StatReplSlotEntry repSlotStat;
1956
1957 /* Nothing to do if we don't have any replication stats to be sent. */
1958 if (rb->spillBytes <= 0 && rb->streamBytes <= 0 && rb->totalBytes <= 0)
1959 return;
1960
1961 elog(DEBUG2, "UpdateDecodingStats: updating stats %p %" PRId64 " %" PRId64 " %" PRId64 " %" PRId64 " %" PRId64 " %" PRId64 " %" PRId64 " %" PRId64,
1962 rb,
1963 rb->spillTxns,
1964 rb->spillCount,
1965 rb->spillBytes,
1966 rb->streamTxns,
1967 rb->streamCount,
1968 rb->streamBytes,
1969 rb->totalTxns,
1970 rb->totalBytes);
1971
1972 repSlotStat.spill_txns = rb->spillTxns;
1973 repSlotStat.spill_count = rb->spillCount;
1974 repSlotStat.spill_bytes = rb->spillBytes;
1975 repSlotStat.stream_txns = rb->streamTxns;
1976 repSlotStat.stream_count = rb->streamCount;
1977 repSlotStat.stream_bytes = rb->streamBytes;
1978 repSlotStat.total_txns = rb->totalTxns;
1979 repSlotStat.total_bytes = rb->totalBytes;
1980
1981 pgstat_report_replslot(ctx->slot, &repSlotStat);
1982
1983 rb->spillTxns = 0;
1984 rb->spillCount = 0;
1985 rb->spillBytes = 0;
1986 rb->streamTxns = 0;
1987 rb->streamCount = 0;
1988 rb->streamBytes = 0;
1989 rb->totalTxns = 0;
1990 rb->totalBytes = 0;
1991}
#define DEBUG2
Definition: elog.h:29
void pgstat_report_replslot(ReplicationSlot *slot, const PgStat_StatReplSlotEntry *repSlotStat)
Definition: pgstat.h:390
PgStat_Counter stream_count
Definition: pgstat.h:395
PgStat_Counter total_txns
Definition: pgstat.h:397
PgStat_Counter total_bytes
Definition: pgstat.h:398
PgStat_Counter spill_txns
Definition: pgstat.h:391
PgStat_Counter stream_txns
Definition: pgstat.h:394
PgStat_Counter spill_count
Definition: pgstat.h:392
PgStat_Counter stream_bytes
Definition: pgstat.h:396
PgStat_Counter spill_bytes
Definition: pgstat.h:393
int64 streamBytes
Definition: reorderbuffer.h:691
int64 streamCount
Definition: reorderbuffer.h:690
int64 totalBytes
Definition: reorderbuffer.h:698
int64 streamTxns
Definition: reorderbuffer.h:689
int64 spillCount
Definition: reorderbuffer.h:685
int64 spillBytes
Definition: reorderbuffer.h:686
int64 totalTxns
Definition: reorderbuffer.h:697
int64 spillTxns
Definition: reorderbuffer.h:684

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().

AltStyle によって変換されたページ (->オリジナル) /