mirror of
https://github.com/postgres/postgres.git
synced 2026-02-27 20:00:50 -05:00
2f3304ce13
When creating and initializing a logical slot, the restart_lsn is set to the latest WAL insertion point (or the latest replay point on standbys). Subsequently, WAL records are decoded from that point to find the start point for extracting changes in the DecodingContextFindStartpoint() function. Since the initial restart_lsn could be in the middle of a transaction, the start point must be a consistent point where we won't see the data for partial transactions. Previously, when not building a full snapshot, serialized snapshots were restored, and the SnapBuild jumps to the consistent state even while finding the start point. Consequently, the slot's restart_lsn and confirmed_flush could be set to the middle of a transaction. This could lead to various unexpected consequences. Specifically, there were reports of logical decoding decoding partial transactions, and assertion failures occurred because only subtransactions were decoded without decoding their top-level transaction until decoding the commit record. To resolve this issue, the changes prevent restoring the serialized snapshot and jumping to the consistent state while finding the start point. On v17 and HEAD, a flag indicating whether snapshot restores should be skipped has been added to the SnapBuild struct, and SNAPBUILD_VERSION has been bumpded. On backbranches, the flag is stored in the LogicalDecodingContext instead, preserving on-disk compatibility. Backpatch to all supported versions. Reported-by: Drew Callahan Reviewed-by: Amit Kapila, Hayato Kuroda Discussion: https://postgr.es/m/2444AA15-D21B-4CCE-8052-52C7C2DAFE5C%40amazon.com Backpatch-through: 12
154 lines
4.5 KiB
C
154 lines
4.5 KiB
C
/*-------------------------------------------------------------------------
|
|
* logical.h
|
|
* PostgreSQL logical decoding coordination
|
|
*
|
|
* Copyright (c) 2012-2023, PostgreSQL Global Development Group
|
|
*
|
|
*-------------------------------------------------------------------------
|
|
*/
|
|
#ifndef LOGICAL_H
|
|
#define LOGICAL_H
|
|
|
|
#include "access/xlog.h"
|
|
#include "access/xlogreader.h"
|
|
#include "replication/output_plugin.h"
|
|
#include "replication/slot.h"
|
|
|
|
struct LogicalDecodingContext;
|
|
|
|
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
|
|
{
|
|
/* memory context this is all allocated in */
|
|
MemoryContext context;
|
|
|
|
/* The associated replication slot */
|
|
ReplicationSlot *slot;
|
|
|
|
/* infrastructure pieces for decoding */
|
|
XLogReaderState *reader;
|
|
struct ReorderBuffer *reorder;
|
|
struct SnapBuild *snapshot_builder;
|
|
|
|
/*
|
|
* Marks the logical decoding context as fast forward decoding one. Such a
|
|
* context does not have plugin loaded so most of the following properties
|
|
* are unused.
|
|
*/
|
|
bool fast_forward;
|
|
|
|
OutputPluginCallbacks callbacks;
|
|
OutputPluginOptions options;
|
|
|
|
/*
|
|
* User specified options
|
|
*/
|
|
List *output_plugin_options;
|
|
|
|
/*
|
|
* User-Provided callback for writing/streaming out data.
|
|
*/
|
|
LogicalOutputPluginWriterPrepareWrite prepare_write;
|
|
LogicalOutputPluginWriterWrite write;
|
|
LogicalOutputPluginWriterUpdateProgress update_progress;
|
|
|
|
/*
|
|
* Output buffer.
|
|
*/
|
|
StringInfo out;
|
|
|
|
/*
|
|
* Private data pointer of the output plugin.
|
|
*/
|
|
void *output_plugin_private;
|
|
|
|
/*
|
|
* Private data pointer for the data writer.
|
|
*/
|
|
void *output_writer_private;
|
|
|
|
/*
|
|
* Does the output plugin support streaming, and is it enabled?
|
|
*/
|
|
bool streaming;
|
|
|
|
/*
|
|
* Does the output plugin support two-phase decoding, and is it enabled?
|
|
*/
|
|
bool twophase;
|
|
|
|
/*
|
|
* Is two-phase option given by output plugin?
|
|
*
|
|
* This flag indicates that the plugin passed in the two-phase option as
|
|
* part of the START_STREAMING command. We can't rely solely on the
|
|
* twophase flag which only tells whether the plugin provided all the
|
|
* necessary two-phase callbacks.
|
|
*/
|
|
bool twophase_opt_given;
|
|
|
|
/*
|
|
* State for writing output.
|
|
*/
|
|
bool accept_writes;
|
|
bool prepared_write;
|
|
XLogRecPtr write_location;
|
|
TransactionId write_xid;
|
|
/* Are we processing the end LSN of a transaction? */
|
|
bool end_xact;
|
|
|
|
/*
|
|
* True if the logical decoding context being used for the creation
|
|
* of a logical replication slot.
|
|
*/
|
|
bool in_create;
|
|
} LogicalDecodingContext;
|
|
|
|
|
|
extern void CheckLogicalDecodingRequirements(void);
|
|
|
|
extern 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);
|
|
extern LogicalDecodingContext *CreateDecodingContext(XLogRecPtr start_lsn,
|
|
List *output_plugin_options,
|
|
bool fast_forward,
|
|
XLogReaderRoutine *xl_routine,
|
|
LogicalOutputPluginWriterPrepareWrite prepare_write,
|
|
LogicalOutputPluginWriterWrite do_write,
|
|
LogicalOutputPluginWriterUpdateProgress update_progress);
|
|
extern void DecodingContextFindStartpoint(LogicalDecodingContext *ctx);
|
|
extern bool DecodingContextReady(LogicalDecodingContext *ctx);
|
|
extern void FreeDecodingContext(LogicalDecodingContext *ctx);
|
|
|
|
extern void LogicalIncreaseXminForSlot(XLogRecPtr current_lsn,
|
|
TransactionId xmin);
|
|
extern void LogicalIncreaseRestartDecodingForSlot(XLogRecPtr current_lsn,
|
|
XLogRecPtr restart_lsn);
|
|
extern void LogicalConfirmReceivedLocation(XLogRecPtr lsn);
|
|
|
|
extern bool filter_prepare_cb_wrapper(LogicalDecodingContext *ctx,
|
|
TransactionId xid, const char *gid);
|
|
extern bool filter_by_origin_cb_wrapper(LogicalDecodingContext *ctx, RepOriginId origin_id);
|
|
extern void ResetLogicalStreamingState(void);
|
|
extern void UpdateDecodingStats(LogicalDecodingContext *ctx);
|
|
|
|
#endif
|