/* -------------------------------------------------------------------------
 *
 * decode.c
 *  This module decodes WAL records read using xlogreader.h's APIs for the
 *  purpose of logical decoding by passing information to the
 *  reorderbuffer module (containing the actual changes) and to the
 *  snapbuild module to build a fitting catalog snapshot (to be able to
 *  properly decode the changes in the reorderbuffer).
 *
 * NOTE:
 *  This basically tries to handle all low level xlog stuff for
 *  reorderbuffer.c and snapbuild.c. There's some minor leakage where a
 *  specific record's struct is used to pass data along, but those just
 *  happen to contain the right amount of data in a convenient
 *  format. There isn't and shouldn't be much intelligence about the
 *  contents of records in here except turning them into a more usable
 *  format.
 *
 * Portions Copyright (c) 1996-2025, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * IDENTIFICATION
 *   src/backend/replication/logical/decode.c
 *
 * -------------------------------------------------------------------------
 */
#include "postgres.h"

#include "access/heapam_xlog.h"
#include "access/transam.h"
#include "access/xact.h"
#include "access/xlog_internal.h"
#include "access/xlogreader.h"
#include "access/xlogrecord.h"
#include "catalog/pg_control.h"
#include "replication/decode.h"
#include "replication/logical.h"
#include "replication/message.h"
#include "replication/reorderbuffer.h"
#include "replication/snapbuild.h"
#include "storage/standbydefs.h"

/* individual record(group)'s handlers */
static void DecodeInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
static void DecodeUpdate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
static void DecodeDelete(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
static void DecodeTruncate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
static void DecodeMultiInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
static void DecodeSpecConfirm(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);

static void DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
       xl_xact_parsed_commit *parsed, TransactionId xid,
       bool two_phase);
static void DecodeAbort(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
      xl_xact_parsed_abort *parsed, TransactionId xid,
      bool two_phase);
static void DecodePrepare(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
        xl_xact_parsed_prepare *parsed);


/* common function to decode tuples */
static void DecodeXLogTuple(char *data, Size len, HeapTuple tuple);

/* helper functions for decoding transactions */
static inline bool FilterPrepare(LogicalDecodingContext *ctx,
         TransactionId xid, const char *gid);
static bool DecodeTXNNeedSkip(LogicalDecodingContext *ctx,
         XLogRecordBuffer *buf, Oid txn_dbid,
         RepOriginId origin_id);

/*
 * Take every XLogReadRecord()ed record and perform the actions required to
 * decode it using the output plugin already setup in the logical decoding
 * context.
 *
 * NB: Note that every record's xid needs to be processed by reorderbuffer
 * (xids contained in the content of records are not relevant for this rule).
 * That means that for records which'd otherwise not go through the
 * reorderbuffer ReorderBufferProcessXid() has to be called. We don't want to
 * call ReorderBufferProcessXid for each record type by default, because
 * e.g. empty xacts can be handled more efficiently if there's no previous
 * state for them.
 *
 * We also support the ability to fast forward thru records, skipping some
 * record types completely - see individual record types for details.
 */
void
LogicalDecodingProcessRecord(LogicalDecodingContext *ctx, XLogReaderState *record)
{
 XLogRecordBuffer buf;
 TransactionId txid;
 RmgrData rmgr;

 buf.origptr = ctx->reader->ReadRecPtr;
 buf.endptr = ctx->reader->EndRecPtr;
 buf.record = record;

 txid = XLogRecGetTopXid(record);

 /*
  * If the top-level xid is valid, we need to assign the subxact to the
  * top-level xact. We need to do this for all records, hence we do it
  * before the switch.
 */
 if (TransactionIdIsValid(txid))
 {
  ReorderBufferAssignChild(ctx->reorder,
         txid,
         XLogRecGetXid(record),
         buf.origptr);
 }

 rmgr = GetRmgr(XLogRecGetRmid(record));

 if (rmgr.rm_decode != NULL)
  rmgr.rm_decode(ctx, &buf);
 else
 {
  /* just deal with xid, and done */
  ReorderBufferProcessXid(ctx->reorder, XLogRecGetXid(record),
        buf.origptr);
 }
}

/*
 * Handle rmgr XLOG_ID records for LogicalDecodingProcessRecord().
 */
void
xlog_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
{
 SnapBuild  *builder = ctx->snapshot_builder;
 uint8  info = XLogRecGetInfo(buf->record) & ~XLR_INFO_MASK;

 ReorderBufferProcessXid(ctx->reorder, XLogRecGetXid(buf->record),
       buf->origptr);

 switch (info)
 {
   /* this is also used in END_OF_RECOVERY checkpoints */
  case XLOG_CHECKPOINT_SHUTDOWN:
  case XLOG_END_OF_RECOVERY:
   SnapBuildSerializationPoint(builder, buf->origptr);

   break;
  case XLOG_CHECKPOINT_ONLINE:

   /*
    * a RUNNING_XACTS record will have been logged near to this, we
    * can restart from there.
 */
   break;
  case XLOG_PARAMETER_CHANGE:
   {
    xl_parameter_change *xlrec =
     (xl_parameter_change *) XLogRecGetData(buf->record);

    /*
     * If wal_level on the primary is reduced to less than
     * logical, we want to prevent existing logical slots from
     * being used.  Existing logical slots on the standby get
     * invalidated when this WAL record is replayed; and further,
     * slot creation fails when wal_level is not sufficient; but
     * all these operations are not synchronized, so a logical
     * slot may creep in while the wal_level is being reduced.
     * Hence this extra check.
 */
    if (xlrec->wal_level < WAL_LEVEL_LOGICAL)
    {
     /*
      * This can occur only on a standby, as a primary would
      * not allow to restart after changing wal_level < logical
      * if there is pre-existing logical slot.
 */
     Assert(RecoveryInProgress());
     ereport(ERROR,
       (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
        errmsg("logical decoding on standby requires \"wal_level\" >= \"logical\" on the primary")));
    }
    break;
   }
  case XLOG_NOOP:
  case XLOG_NEXTOID:
  case XLOG_SWITCH:
  case XLOG_BACKUP_END:
  case XLOG_RESTORE_POINT:
  case XLOG_FPW_CHANGE:
  case XLOG_FPI_FOR_HINT:
  case XLOG_FPI:
  case XLOG_OVERWRITE_CONTRECORD:
  case XLOG_CHECKPOINT_REDO:
   break;
  default:
   elog(ERROR, "unexpected RM_XLOG_ID record type: %u", info);
 }
}

/*
 * Handle rmgr XACT_ID records for LogicalDecodingProcessRecord().
 */
void
xact_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
{
 SnapBuild  *builder = ctx->snapshot_builder;
 ReorderBuffer *reorder = ctx->reorder;
 XLogReaderState *r = buf->record;
 uint8  info = XLogRecGetInfo(r) & XLOG_XACT_OPMASK;

 /*
  * If the snapshot isn't yet fully built, we cannot decode anything, so
  * bail out.
 */
 if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT)
  return;

 switch (info)
 {
  case XLOG_XACT_COMMIT:
  case XLOG_XACT_COMMIT_PREPARED:
   {
    xl_xact_commit *xlrec;
    xl_xact_parsed_commit parsed;
    TransactionId xid;
    bool  two_phase = false;

    xlrec = (xl_xact_commit *) XLogRecGetData(r);
    ParseCommitRecord(XLogRecGetInfo(buf->record), xlrec, &parsed);

    if (!TransactionIdIsValid(parsed.twophase_xid))
     xid = XLogRecGetXid(r);
    else
     xid = parsed.twophase_xid;

    /*
     * We would like to process the transaction in a two-phase
     * manner iff output plugin supports two-phase commits and
     * doesn't filter the transaction at prepare time.
 */
    if (info == XLOG_XACT_COMMIT_PREPARED)
     two_phase = !(FilterPrepare(ctx, xid,
            parsed.twophase_gid));

    DecodeCommit(ctx, buf, &parsed, xid, two_phase);
    break;
   }
  case XLOG_XACT_ABORT:
  case XLOG_XACT_ABORT_PREPARED:
   {
    xl_xact_abort *xlrec;
    xl_xact_parsed_abort parsed;
    TransactionId xid;
    bool  two_phase = false;

    xlrec = (xl_xact_abort *) XLogRecGetData(r);
    ParseAbortRecord(XLogRecGetInfo(buf->record), xlrec, &parsed);

    if (!TransactionIdIsValid(parsed.twophase_xid))
     xid = XLogRecGetXid(r);
    else
     xid = parsed.twophase_xid;

    /*
     * We would like to process the transaction in a two-phase
     * manner iff output plugin supports two-phase commits and
     * doesn't filter the transaction at prepare time.
 */
    if (info == XLOG_XACT_ABORT_PREPARED)
     two_phase = !(FilterPrepare(ctx, xid,
            parsed.twophase_gid));

    DecodeAbort(ctx, buf, &parsed, xid, two_phase);
    break;
   }
  case XLOG_XACT_ASSIGNMENT:

   /*
    * We assign subxact to the toplevel xact while processing each
    * record if required.  So, we don't need to do anything here. See
    * LogicalDecodingProcessRecord.
 */
   break;
  case XLOG_XACT_INVALIDATIONS:
   {
    TransactionId xid;
    xl_xact_invals *invals;

    xid = XLogRecGetXid(r);
    invals = (xl_xact_invals *) XLogRecGetData(r);

    /*
     * Execute the invalidations for xid-less transactions,
     * otherwise, accumulate them so that they can be processed at
     * the commit time.
 */
    if (TransactionIdIsValid(xid))
    {
     if (!ctx->fast_forward)
      ReorderBufferAddInvalidations(reorder, xid,
               buf->origptr,
               invals->nmsgs,
               invals->msgs);
     ReorderBufferXidSetCatalogChanges(ctx->reorder, xid,
               buf->origptr);
    }
    else if (!ctx->fast_forward)
     ReorderBufferImmediateInvalidation(ctx->reorder,
                invals->nmsgs,
                invals->msgs);

    break;
   }
  case XLOG_XACT_PREPARE:
   {
    xl_xact_parsed_prepare parsed;
    xl_xact_prepare *xlrec;

    /* ok, parse it */
    xlrec = (xl_xact_prepare *) XLogRecGetData(r);
    ParsePrepareRecord(XLogRecGetInfo(buf->record),
           xlrec, &parsed);

    /*
     * We would like to process the transaction in a two-phase
     * manner iff output plugin supports two-phase commits and
     * doesn't filter the transaction at prepare time.
 */
    if (FilterPrepare(ctx, parsed.twophase_xid,
          parsed.twophase_gid))
    {
     ReorderBufferProcessXid(reorder, parsed.twophase_xid,
           buf->origptr);
     break;
    }

    /*
     * Note that if the prepared transaction has locked [user]
     * catalog tables exclusively then decoding prepare can block
     * till the main transaction is committed because it needs to
     * lock the catalog tables.
     *
     * XXX Now, this can even lead to a deadlock if the prepare
     * transaction is waiting to get it logically replicated for
     * distributed 2PC. This can be avoided by disallowing
     * preparing transactions that have locked [user] catalog
     * tables exclusively but as of now, we ask users not to do
     * such an operation.
 */
    DecodePrepare(ctx, buf, &parsed);
    break;
   }
  default:
   elog(ERROR, "unexpected RM_XACT_ID record type: %u", info);
 }
}

/*
 * Handle rmgr STANDBY_ID records for LogicalDecodingProcessRecord().
 */
void
standby_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
{
 SnapBuild  *builder = ctx->snapshot_builder;
 XLogReaderState *r = buf->record;
 uint8  info = XLogRecGetInfo(r) & ~XLR_INFO_MASK;

 ReorderBufferProcessXid(ctx->reorder, XLogRecGetXid(r), buf->origptr);

 switch (info)
 {
  case XLOG_RUNNING_XACTS:
   {
    xl_running_xacts *running = (xl_running_xacts *) XLogRecGetData(r);

    SnapBuildProcessRunningXacts(builder, buf->origptr, running);

    /*
     * Abort all transactions that we keep track of, that are
     * older than the record's oldestRunningXid. This is the most
     * convenient spot for doing so since, in contrast to shutdown
     * or end-of-recovery checkpoints, we have information about
     * all running transactions which includes prepared ones,
     * while shutdown checkpoints just know that no non-prepared
     * transactions are in progress.
 */
    ReorderBufferAbortOld(ctx->reorder, running->oldestRunningXid);
   }
   break;
  case XLOG_STANDBY_LOCK:
   break;
  case XLOG_INVALIDATIONS:

   /*
    * We are processing the invalidations at the command level via
    * XLOG_XACT_INVALIDATIONS.  So we don't need to do anything here.
 */
   break;
  default:
   elog(ERROR, "unexpected RM_STANDBY_ID record type: %u", info);
 }
}

/*
 * Handle rmgr HEAP2_ID records for LogicalDecodingProcessRecord().
 */
void
heap2_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
{
 uint8  info = XLogRecGetInfo(buf->record) & XLOG_HEAP_OPMASK;
 TransactionId xid = XLogRecGetXid(buf->record);
 SnapBuild  *builder = ctx->snapshot_builder;

 ReorderBufferProcessXid(ctx->reorder, xid, buf->origptr);

 /*
  * If we don't have snapshot or we are just fast-forwarding, there is no
  * point in decoding data changes. However, it's crucial to build the base
  * snapshot during fast-forward mode (as is done in
  * SnapBuildProcessChange()) because we require the snapshot's xmin when
  * determining the candidate catalog_xmin for the replication slot. See
  * SnapBuildProcessRunningXacts().
 */
 if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT)
  return;

 switch (info)
 {
  case XLOG_HEAP2_MULTI_INSERT:
   if (SnapBuildProcessChange(builder, xid, buf->origptr) &&
    !ctx->fast_forward)
    DecodeMultiInsert(ctx, buf);
   break;
  case XLOG_HEAP2_NEW_CID:
   if (!ctx->fast_forward)
   {
    xl_heap_new_cid *xlrec;

    xlrec = (xl_heap_new_cid *) XLogRecGetData(buf->record);
    SnapBuildProcessNewCid(builder, xid, buf->origptr, xlrec);

    break;
   }
  case XLOG_HEAP2_REWRITE:

   /*
    * Although these records only exist to serve the needs of logical
    * decoding, all the work happens as part of crash or archive
    * recovery, so we don't need to do anything here.
 */
   break;

   /*
    * Everything else here is just low level physical stuff we're not
    * interested in.
 */
  case XLOG_HEAP2_PRUNE_ON_ACCESS:
  case XLOG_HEAP2_PRUNE_VACUUM_SCAN:
  case XLOG_HEAP2_PRUNE_VACUUM_CLEANUP:
  case XLOG_HEAP2_VISIBLE:
  case XLOG_HEAP2_LOCK_UPDATED:
   break;
  default:
   elog(ERROR, "unexpected RM_HEAP2_ID record type: %u", info);
 }
}

/*
 * Handle rmgr HEAP_ID records for LogicalDecodingProcessRecord().
 */
void
heap_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
{
 uint8  info = XLogRecGetInfo(buf->record) & XLOG_HEAP_OPMASK;
 TransactionId xid = XLogRecGetXid(buf->record);
 SnapBuild  *builder = ctx->snapshot_builder;

 ReorderBufferProcessXid(ctx->reorder, xid, buf->origptr);

 /*
  * If we don't have snapshot or we are just fast-forwarding, there is no
  * point in decoding data changes. However, it's crucial to build the base
  * snapshot during fast-forward mode (as is done in
  * SnapBuildProcessChange()) because we require the snapshot's xmin when
  * determining the candidate catalog_xmin for the replication slot. See
  * SnapBuildProcessRunningXacts().
 */
 if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT)
  return;

 switch (info)
 {
  case XLOG_HEAP_INSERT:
   if (SnapBuildProcessChange(builder, xid, buf->origptr) &&
    !ctx->fast_forward)
    DecodeInsert(ctx, buf);
   break;

   /*
    * Treat HOT update as normal updates. There is no useful
    * information in the fact that we could make it a HOT update
    * locally and the WAL layout is compatible.
 */
  case XLOG_HEAP_HOT_UPDATE:
  case XLOG_HEAP_UPDATE:
   if (SnapBuildProcessChange(builder, xid, buf->origptr) &&
    !ctx->fast_forward)
    DecodeUpdate(ctx, buf);
   break;

  case XLOG_HEAP_DELETE:
   if (SnapBuildProcessChange(builder, xid, buf->origptr) &&
    !ctx->fast_forward)
    DecodeDelete(ctx, buf);
   break;

  case XLOG_HEAP_TRUNCATE:
   if (SnapBuildProcessChange(builder, xid, buf->origptr) &&
    !ctx->fast_forward)
    DecodeTruncate(ctx, buf);
   break;

  case XLOG_HEAP_INPLACE:

   /*
    * Inplace updates are only ever performed on catalog tuples and
    * can, per definition, not change tuple visibility.  Since we
    * also don't decode catalog tuples, we're not interested in the
    * record's contents.
 */
   break;

  case XLOG_HEAP_CONFIRM:
   if (SnapBuildProcessChange(builder, xid, buf->origptr) &&
    !ctx->fast_forward)
    DecodeSpecConfirm(ctx, buf);
   break;

  case XLOG_HEAP_LOCK:
   /* we don't care about row level locks for now */
   break;

  default:
   elog(ERROR, "unexpected RM_HEAP_ID record type: %u", info);
   break;
 }
}

/*
 * Ask output plugin whether we want to skip this PREPARE and send
 * this transaction as a regular commit later.
 */
static inline bool
FilterPrepare(LogicalDecodingContext *ctx, TransactionId xid,
     const char *gid)
{
 /*
  * Skip if decoding of two-phase transactions at PREPARE time is not
  * enabled. In that case, all two-phase transactions are considered
  * filtered out and will be applied as regular transactions at COMMIT
  * PREPARED.
 */
 if (!ctx->twophase)
  return true;

 /*
  * The filter_prepare callback is optional. When not supplied, all
  * prepared transactions should go through.
 */
 if (ctx->callbacks.filter_prepare_cb == NULL)
  return false;

 return filter_prepare_cb_wrapper(ctx, xid, gid);
}

static inline bool
FilterByOrigin(LogicalDecodingContext *ctx, RepOriginId origin_id)
{
 if (ctx->callbacks.filter_by_origin_cb == NULL)
  return false;

 return filter_by_origin_cb_wrapper(ctx, origin_id);
}

/*
 * Handle rmgr LOGICALMSG_ID records for LogicalDecodingProcessRecord().
 */
void
logicalmsg_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
{
 SnapBuild  *builder = ctx->snapshot_builder;
 XLogReaderState *r = buf->record;
 TransactionId xid = XLogRecGetXid(r);
 uint8  info = XLogRecGetInfo(r) & ~XLR_INFO_MASK;
 RepOriginId origin_id = XLogRecGetOrigin(r);
 Snapshot snapshot = NULL;
 xl_logical_message *message;

 if (info != XLOG_LOGICAL_MESSAGE)
  elog(ERROR, "unexpected RM_LOGICALMSG_ID record type: %u", info);

 ReorderBufferProcessXid(ctx->reorder, XLogRecGetXid(r), buf->origptr);

 /* If we don't have snapshot, there is no point in decoding messages */
 if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT)
  return;

 message = (xl_logical_message *) XLogRecGetData(r);

 if (message->dbId != ctx->slot->data.database ||
  FilterByOrigin(ctx, origin_id))
  return;

 if (message->transactional &&
  !SnapBuildProcessChange(builder, xid, buf->origptr))
  return;
 else if (!message->transactional &&
    (SnapBuildCurrentState(builder) != SNAPBUILD_CONSISTENT ||
     SnapBuildXactNeedsSkip(builder, buf->origptr)))
  return;

 /*
  * We also skip decoding in fast_forward mode. This check must be last
  * because we don't want to set the processing_required flag unless we
  * have a decodable message.
 */
 if (ctx->fast_forward)
 {
  /*
   * We need to set processing_required flag to notify the message's
   * existence to the caller. Usually, the flag is set when either the
   * COMMIT or ABORT records are decoded, but this must be turned on
   * here because the non-transactional logical message is decoded
   * without waiting for these records.
 */
  if (!message->transactional)
   ctx->processing_required = true;

  return;
 }

 /*
  * If this is a non-transactional change, get the snapshot we're expected
  * to use. We only get here when the snapshot is consistent, and the
  * change is not meant to be skipped.
  *
  * For transactional changes we don't need a snapshot, we'll use the
  * regular snapshot maintained by ReorderBuffer. We just leave it NULL.
 */
 if (!message->transactional)
  snapshot = SnapBuildGetOrBuildSnapshot(builder);

 ReorderBufferQueueMessage(ctx->reorder, xid, snapshot, buf->endptr,
         message->transactional,
         message->message, /* first part of message is
 * prefix */
         message->message_size,
         message->message + message->prefix_size);
}

/*
 * Consolidated commit record handling between the different form of commit
 * records.
 *
 * 'two_phase' indicates that caller wants to process the transaction in two
 * phases, first process prepare if not already done and then process
 * commit_prepared.
 */
static void
DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
    xl_xact_parsed_commit *parsed, TransactionId xid,
    bool two_phase)
{
 XLogRecPtr origin_lsn = InvalidXLogRecPtr;
 TimestampTz commit_time = parsed->xact_time;
 RepOriginId origin_id = XLogRecGetOrigin(buf->record);
 int   i;

 if (parsed->xinfo & XACT_XINFO_HAS_ORIGIN)
 {
  origin_lsn = parsed->origin_lsn;
  commit_time = parsed->origin_timestamp;
 }

 SnapBuildCommitTxn(ctx->snapshot_builder, buf->origptr, xid,
        parsed->nsubxacts, parsed->subxacts,
        parsed->xinfo);

 /* ----
  * Check whether we are interested in this specific transaction, and tell
  * the reorderbuffer to forget the content of the (sub-)transactions
  * if not.
  *
  * We can't just use ReorderBufferAbort() here, because we need to execute
  * the transaction's invalidations.  This currently won't be needed if
  * we're just skipping over the transaction because currently we only do
  * so during startup, to get to the first transaction the client needs. As
  * we have reset the catalog caches before starting to read WAL, and we
  * haven't yet touched any catalogs, there can't be anything to invalidate.
  * But if we're "forgetting" this commit because it happened in another
  * database, the invalidations might be important, because they could be
  * for shared catalogs and we might have loaded data into the relevant
  * syscaches.
  * ---
 */
 if (DecodeTXNNeedSkip(ctx, buf, parsed->dbId, origin_id))
 {
  for (i = 0; i < parsed->nsubxacts; i++)
  {
   ReorderBufferForget(ctx->reorder, parsed->subxacts[i], buf->origptr);
  }
  ReorderBufferForget(ctx->reorder, xid, buf->origptr);

  return;
 }

 /* tell the reorderbuffer about the surviving subtransactions */
 for (i = 0; i < parsed->nsubxacts; i++)
 {
  ReorderBufferCommitChild(ctx->reorder, xid, parsed->subxacts[i],
         buf->origptr, buf->endptr);
 }

 /*
  * Send the final commit record if the transaction data is already
  * decoded, otherwise, process the entire transaction.
 */
 if (two_phase)
 {
  ReorderBufferFinishPrepared(ctx->reorder, xid, buf->origptr, buf->endptr,
         SnapBuildGetTwoPhaseAt(ctx->snapshot_builder),
         commit_time, origin_id, origin_lsn,
         parsed->twophase_gid, true);
 }
 else
 {
  ReorderBufferCommit(ctx->reorder, xid, buf->origptr, buf->endptr,
       commit_time, origin_id, origin_lsn);
 }

 /*
  * Update the decoding stats at transaction prepare/commit/abort.
  * Additionally we send the stats when we spill or stream the changes to
  * avoid losing them in case the decoding is interrupted. It is not clear
  * that sending more or less frequently than this would be better.
 */
 UpdateDecodingStats(ctx);
}

/*
 * Decode PREPARE record. Similar logic as in DecodeCommit.
 *
 * Note that we don't skip prepare even if have detected concurrent abort
 * because it is quite possible that we had already sent some changes before we
 * detect abort in which case we need to abort those changes in the subscriber.
 * To abort such changes, we do send the prepare and then the rollback prepared
 * which is what happened on the publisher-side as well. Now, we can invent a
 * new abort API wherein in such cases we send abort and skip sending prepared
 * and rollback prepared but then it is not that straightforward because we
 * might have streamed this transaction by that time in which case it is
 * handled when the rollback is encountered. It is not impossible to optimize
 * the concurrent abort case but it can introduce design complexity w.r.t
 * handling different cases so leaving it for now as it doesn't seem worth it.
 */
static void
DecodePrepare(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
     xl_xact_parsed_prepare *parsed)
{
 SnapBuild  *builder = ctx->snapshot_builder;
 XLogRecPtr origin_lsn = parsed->origin_lsn;
 TimestampTz prepare_time = parsed->xact_time;
 RepOriginId origin_id = XLogRecGetOrigin(buf->record);
 int   i;
 TransactionId xid = parsed->twophase_xid;

 if (parsed->origin_timestamp != 0)
  prepare_time = parsed->origin_timestamp;

 /*
  * Remember the prepare info for a txn so that it can be used later in
  * commit prepared if required. See ReorderBufferFinishPrepared.
 */
 if (!ReorderBufferRememberPrepareInfo(ctx->reorder, xid, buf->origptr,
            buf->endptr, prepare_time, origin_id,
            origin_lsn))
  return;

 /* We can't start streaming unless a consistent state is reached. */
 if (SnapBuildCurrentState(builder) < SNAPBUILD_CONSISTENT)
 {
  ReorderBufferSkipPrepare(ctx->reorder, xid);
  return;
 }

 /*
  * Check whether we need to process this transaction. See
  * DecodeTXNNeedSkip for the reasons why we sometimes want to skip the
  * transaction.
  *
  * We can't call ReorderBufferForget as we did in DecodeCommit as the txn
  * hasn't yet been committed, removing this txn before a commit might
  * result in the computation of an incorrect restart_lsn. See
  * SnapBuildProcessRunningXacts. But we need to process cache
  * invalidations if there are any for the reasons mentioned in
  * DecodeCommit.
 */
 if (DecodeTXNNeedSkip(ctx, buf, parsed->dbId, origin_id))
 {
  ReorderBufferSkipPrepare(ctx->reorder, xid);
  ReorderBufferInvalidate(ctx->reorder, xid, buf->origptr);
  return;
 }

 /* Tell the reorderbuffer about the surviving subtransactions. */
 for (i = 0; i < parsed->nsubxacts; i++)
 {
  ReorderBufferCommitChild(ctx->reorder, xid, parsed->subxacts[i],
         buf->origptr, buf->endptr);
 }

 /* replay actions of all transaction + subtransactions in order */
 ReorderBufferPrepare(ctx->reorder, xid, parsed->twophase_gid);

 /*
  * Update the decoding stats at transaction prepare/commit/abort.
  * Additionally we send the stats when we spill or stream the changes to
  * avoid losing them in case the decoding is interrupted. It is not clear
  * that sending more or less frequently than this would be better.
 */
 UpdateDecodingStats(ctx);
}


/*
 * Get the data from the various forms of abort records and pass it on to
 * snapbuild.c and reorderbuffer.c.
 *
 * 'two_phase' indicates to finish prepared transaction.
 */
static void
DecodeAbort(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
   xl_xact_parsed_abort *parsed, TransactionId xid,
   bool two_phase)
{
 int   i;
 XLogRecPtr origin_lsn = InvalidXLogRecPtr;
 TimestampTz abort_time = parsed->xact_time;
 RepOriginId origin_id = XLogRecGetOrigin(buf->record);
 bool  skip_xact;

 if (parsed->xinfo & XACT_XINFO_HAS_ORIGIN)
 {
  origin_lsn = parsed->origin_lsn;
  abort_time = parsed->origin_timestamp;
 }

 /*
  * Check whether we need to process this transaction. See
  * DecodeTXNNeedSkip for the reasons why we sometimes want to skip the
  * transaction.
 */
 skip_xact = DecodeTXNNeedSkip(ctx, buf, parsed->dbId, origin_id);

 /*
  * Send the final rollback record for a prepared transaction unless we
  * need to skip it. For non-two-phase xacts, simply forget the xact.
 */
 if (two_phase && !skip_xact)
 {
  ReorderBufferFinishPrepared(ctx->reorder, xid, buf->origptr, buf->endptr,
         InvalidXLogRecPtr,
         abort_time, origin_id, origin_lsn,
         parsed->twophase_gid, false);
 }
 else
 {
  for (i = 0; i < parsed->nsubxacts; i++)
  {
   ReorderBufferAbort(ctx->reorder, parsed->subxacts[i],
          buf->record->EndRecPtr, abort_time);
  }

  ReorderBufferAbort(ctx->reorder, xid, buf->record->EndRecPtr,
         abort_time);
 }

 /* update the decoding stats */
 UpdateDecodingStats(ctx);
}

/*
 * Parse XLOG_HEAP_INSERT (not MULTI_INSERT!) records into tuplebufs.
 *
 * Inserts can contain the new tuple.
 */
static void
DecodeInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
{
 Size  datalen;
 char    *tupledata;
 Size  tuplelen;
 XLogReaderState *r = buf->record;
 xl_heap_insert *xlrec;
 ReorderBufferChange *change;
 RelFileLocator target_locator;

 xlrec = (xl_heap_insert *) XLogRecGetData(r);

 /*
  * Ignore insert records without new tuples (this does happen when
  * raw_heap_insert marks the TOAST record as HEAP_INSERT_NO_LOGICAL).
 */
 if (!(xlrec->flags & XLH_INSERT_CONTAINS_NEW_TUPLE))
  return;

 /* only interested in our database */
 XLogRecGetBlockTag(r, 0, &target_locator, NULL, NULL);
 if (target_locator.dbOid != ctx->slot->data.database)
  return;

 /* output plugin doesn't look for this origin, no need to queue */
 if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
  return;

 change = ReorderBufferAllocChange(ctx->reorder);
 if (!(xlrec->flags & XLH_INSERT_IS_SPECULATIVE))
  change->action = REORDER_BUFFER_CHANGE_INSERT;
 else
  change->action = REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT;
 change->origin_id = XLogRecGetOrigin(r);

 memcpy(&change->data.tp.rlocator, &target_locator, sizeof(RelFileLocator));

 tupledata = XLogRecGetBlockData(r, 0, &datalen);
 tuplelen = datalen - SizeOfHeapHeader;

 change->data.tp.newtuple =
  ReorderBufferAllocTupleBuf(ctx->reorder, tuplelen);

 DecodeXLogTuple(tupledata, datalen, change->data.tp.newtuple);

 change->data.tp.clear_toast_afterwards = true;

 ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r), buf->origptr,
        change,
        xlrec->flags & XLH_INSERT_ON_TOAST_RELATION);
}

/*
 * Parse XLOG_HEAP_UPDATE and XLOG_HEAP_HOT_UPDATE, which have the same layout
 * in the record, from wal into proper tuplebufs.
 *
 * Updates can possibly contain a new tuple and the old primary key.
 */
static void
DecodeUpdate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
{
 XLogReaderState *r = buf->record;
 xl_heap_update *xlrec;
 ReorderBufferChange *change;
 char    *data;
 RelFileLocator target_locator;

 xlrec = (xl_heap_update *) XLogRecGetData(r);

 /* only interested in our database */
 XLogRecGetBlockTag(r, 0, &target_locator, NULL, NULL);
 if (target_locator.dbOid != ctx->slot->data.database)
  return;

 /* output plugin doesn't look for this origin, no need to queue */
 if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
  return;

 change = ReorderBufferAllocChange(ctx->reorder);
 change->action = REORDER_BUFFER_CHANGE_UPDATE;
 change->origin_id = XLogRecGetOrigin(r);
 memcpy(&change->data.tp.rlocator, &target_locator, sizeof(RelFileLocator));

 if (xlrec->flags & XLH_UPDATE_CONTAINS_NEW_TUPLE)
 {
  Size  datalen;
  Size  tuplelen;

  data = XLogRecGetBlockData(r, 0, &datalen);

  tuplelen = datalen - SizeOfHeapHeader;

  change->data.tp.newtuple =
   ReorderBufferAllocTupleBuf(ctx->reorder, tuplelen);

  DecodeXLogTuple(data, datalen, change->data.tp.newtuple);
 }

 if (xlrec->flags & XLH_UPDATE_CONTAINS_OLD)
 {
  Size  datalen;
  Size  tuplelen;

  /* caution, remaining data in record is not aligned */
  data = XLogRecGetData(r) + SizeOfHeapUpdate;
  datalen = XLogRecGetDataLen(r) - SizeOfHeapUpdate;
  tuplelen = datalen - SizeOfHeapHeader;

  change->data.tp.oldtuple =
   ReorderBufferAllocTupleBuf(ctx->reorder, tuplelen);

  DecodeXLogTuple(data, datalen, change->data.tp.oldtuple);
 }

 change->data.tp.clear_toast_afterwards = true;

 ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r), buf->origptr,
        change, false);
}

/*
 * Parse XLOG_HEAP_DELETE from wal into proper tuplebufs.
 *
 * Deletes can possibly contain the old primary key.
 */
static void
DecodeDelete(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
{
 XLogReaderState *r = buf->record;
 xl_heap_delete *xlrec;
 ReorderBufferChange *change;
 RelFileLocator target_locator;

 xlrec = (xl_heap_delete *) XLogRecGetData(r);

 /* only interested in our database */
 XLogRecGetBlockTag(r, 0, &target_locator, NULL, NULL);
 if (target_locator.dbOid != ctx->slot->data.database)
  return;

 /* output plugin doesn't look for this origin, no need to queue */
 if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
  return;

 change = ReorderBufferAllocChange(ctx->reorder);

 if (xlrec->flags & XLH_DELETE_IS_SUPER)
  change->action = REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT;
 else
  change->action = REORDER_BUFFER_CHANGE_DELETE;

 change->origin_id = XLogRecGetOrigin(r);

 memcpy(&change->data.tp.rlocator, &target_locator, sizeof(RelFileLocator));

 /* old primary key stored */
 if (xlrec->flags & XLH_DELETE_CONTAINS_OLD)
 {
  Size  datalen = XLogRecGetDataLen(r) - SizeOfHeapDelete;
  Size  tuplelen = datalen - SizeOfHeapHeader;

  Assert(XLogRecGetDataLen(r) > (SizeOfHeapDelete + SizeOfHeapHeader));

  change->data.tp.oldtuple =
   ReorderBufferAllocTupleBuf(ctx->reorder, tuplelen);

  DecodeXLogTuple((char *) xlrec + SizeOfHeapDelete,
      datalen, change->data.tp.oldtuple);
 }

 change->data.tp.clear_toast_afterwards = true;

 ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r), buf->origptr,
        change, false);
}

/*
 * Parse XLOG_HEAP_TRUNCATE from wal
 */
static void
DecodeTruncate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
{
 XLogReaderState *r = buf->record;
 xl_heap_truncate *xlrec;
 ReorderBufferChange *change;

 xlrec = (xl_heap_truncate *) XLogRecGetData(r);

 /* only interested in our database */
 if (xlrec->dbId != ctx->slot->data.database)
  return;

 /* output plugin doesn't look for this origin, no need to queue */
 if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
  return;

 change = ReorderBufferAllocChange(ctx->reorder);
 change->action = REORDER_BUFFER_CHANGE_TRUNCATE;
 change->origin_id = XLogRecGetOrigin(r);
 if (xlrec->flags & XLH_TRUNCATE_CASCADE)
  change->data.truncate.cascade = true;
 if (xlrec->flags & XLH_TRUNCATE_RESTART_SEQS)
  change->data.truncate.restart_seqs = true;
 change->data.truncate.nrelids = xlrec->nrelids;
 change->data.truncate.relids = ReorderBufferAllocRelids(ctx->reorder,
               xlrec->nrelids);
 memcpy(change->data.truncate.relids, xlrec->relids,
     xlrec->nrelids * sizeof(Oid));
 ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r),
        buf->origptr, change, false);
}

/*
 * Decode XLOG_HEAP2_MULTI_INSERT record into multiple tuplebufs.
 *
 * Currently MULTI_INSERT will always contain the full tuples.
 */
static void
DecodeMultiInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
{
 XLogReaderState *r = buf->record;
 xl_heap_multi_insert *xlrec;
 int   i;
 char    *data;
 char    *tupledata;
 Size  tuplelen;
 RelFileLocator rlocator;

 xlrec = (xl_heap_multi_insert *) XLogRecGetData(r);

 /*
  * Ignore insert records without new tuples.  This happens when a
  * multi_insert is done on a catalog or on a non-persistent relation.
 */
 if (!(xlrec->flags & XLH_INSERT_CONTAINS_NEW_TUPLE))
  return;

 /* only interested in our database */
 XLogRecGetBlockTag(r, 0, &rlocator, NULL, NULL);
 if (rlocator.dbOid != ctx->slot->data.database)
  return;

 /* output plugin doesn't look for this origin, no need to queue */
 if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
  return;

 /*
  * We know that this multi_insert isn't for a catalog, so the block should
  * always have data even if a full-page write of it is taken.
 */
 tupledata = XLogRecGetBlockData(r, 0, &tuplelen);
 Assert(tupledata != NULL);

 data = tupledata;
 for (i = 0; i < xlrec->ntuples; i++)
 {
  ReorderBufferChange *change;
  xl_multi_insert_tuple *xlhdr;
  int   datalen;
  HeapTuple tuple;
  HeapTupleHeader header;

  change = ReorderBufferAllocChange(ctx->reorder);
  change->action = REORDER_BUFFER_CHANGE_INSERT;
  change->origin_id = XLogRecGetOrigin(r);

  memcpy(&change->data.tp.rlocator, &rlocator, sizeof(RelFileLocator));

  xlhdr = (xl_multi_insert_tuple *) SHORTALIGN(data);
  data = ((char *) xlhdr) + SizeOfMultiInsertTuple;
  datalen = xlhdr->datalen;

  change->data.tp.newtuple =
   ReorderBufferAllocTupleBuf(ctx->reorder, datalen);

  tuple = change->data.tp.newtuple;
  header = tuple->t_data;

  /* not a disk based tuple */
  ItemPointerSetInvalid(&tuple->t_self);

  /*
   * We can only figure this out after reassembling the transactions.
 */
  tuple->t_tableOid = InvalidOid;

  tuple->t_len = datalen + SizeofHeapTupleHeader;

  memset(header, 0, SizeofHeapTupleHeader);

  memcpy((char *) tuple->t_data + SizeofHeapTupleHeader, data, datalen);
  header->t_infomask = xlhdr->t_infomask;
  header->t_infomask2 = xlhdr->t_infomask2;
  header->t_hoff = xlhdr->t_hoff;

  /*
   * Reset toast reassembly state only after the last row in the last
   * xl_multi_insert_tuple record emitted by one heap_multi_insert()
   * call.
 */
  if (xlrec->flags & XLH_INSERT_LAST_IN_MULTI &&
   (i + 1) == xlrec->ntuples)
   change->data.tp.clear_toast_afterwards = true;
  else
   change->data.tp.clear_toast_afterwards = false;

  ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r),
         buf->origptr, change, false);

  /* move to the next xl_multi_insert_tuple entry */
  data += datalen;
 }
 Assert(data == tupledata + tuplelen);
}

/*
 * Parse XLOG_HEAP_CONFIRM from wal into a confirmation change.
 *
 * This is pretty trivial, all the state essentially already setup by the
 * speculative insertion.
 */
static void
DecodeSpecConfirm(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
{
 XLogReaderState *r = buf->record;
 ReorderBufferChange *change;
 RelFileLocator target_locator;

 /* only interested in our database */
 XLogRecGetBlockTag(r, 0, &target_locator, NULL, NULL);
 if (target_locator.dbOid != ctx->slot->data.database)
  return;

 /* output plugin doesn't look for this origin, no need to queue */
 if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
  return;

 change = ReorderBufferAllocChange(ctx->reorder);
 change->action = REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM;
 change->origin_id = XLogRecGetOrigin(r);

 memcpy(&change->data.tp.rlocator, &target_locator, sizeof(RelFileLocator));

 change->data.tp.clear_toast_afterwards = true;

 ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r), buf->origptr,
        change, false);
}


/*
 * Read a HeapTuple as WAL logged by heap_insert, heap_update and heap_delete
 * (but not by heap_multi_insert) into a tuplebuf.
 *
 * The size 'len' and the pointer 'data' in the record need to be
 * computed outside as they are record specific.
 */
static void
DecodeXLogTuple(char *data, Size len, HeapTuple tuple)
{
 xl_heap_header xlhdr;
 int   datalen = len - SizeOfHeapHeader;
 HeapTupleHeader header;

 Assert(datalen >= 0);

 tuple->t_len = datalen + SizeofHeapTupleHeader;
 header = tuple->t_data;

 /* not a disk based tuple */
 ItemPointerSetInvalid(&tuple->t_self);

 /* we can only figure this out after reassembling the transactions */
 tuple->t_tableOid = InvalidOid;

 /* data is not stored aligned, copy to aligned storage */
 memcpy(&xlhdr, data, SizeOfHeapHeader);

 memset(header, 0, SizeofHeapTupleHeader);

 memcpy(((char *) tuple->t_data) + SizeofHeapTupleHeader,
     data + SizeOfHeapHeader,
     datalen);

 header->t_infomask = xlhdr.t_infomask;
 header->t_infomask2 = xlhdr.t_infomask2;
 header->t_hoff = xlhdr.t_hoff;
}

/*
 * Check whether we are interested in this specific transaction.
 *
 * There can be several reasons we might not be interested in this
 * transaction:
 * 1) We might not be interested in decoding transactions up to this
 *   LSN. This can happen because we previously decoded it and now just
 *   are restarting or if we haven't assembled a consistent snapshot yet.
 * 2) The transaction happened in another database.
 * 3) The output plugin is not interested in the origin.
 * 4) We are doing fast-forwarding
 */
static bool
DecodeTXNNeedSkip(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
      Oid txn_dbid, RepOriginId origin_id)
{
 if (SnapBuildXactNeedsSkip(ctx->snapshot_builder, buf->origptr) ||
  (txn_dbid != InvalidOid && txn_dbid != ctx->slot->data.database) ||
  FilterByOrigin(ctx, origin_id))
  return true;

 /*
  * We also skip decoding in fast_forward mode. In passing set the
  * processing_required flag to indicate that if it were not for
  * fast_forward mode, processing would have been required.
 */
 if (ctx->fast_forward)
 {
  ctx->processing_required = true;
  return true;
 }

 return false;
}