/*-------------------------------------------------------------------------

 *

 * execReplication.c

 *    miscellaneous executor routines for logical replication

 *

 * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group

 * Portions Copyright (c) 1994, Regents of the University of California

 *

 * IDENTIFICATION

 *    src/backend/executor/execReplication.c

 *

 *-------------------------------------------------------------------------

 */

#include "postgres.h"

#include "access/amapi.h"

#include "access/commit_ts.h"

#include "access/genam.h"

#include "access/gist.h"

#include "access/relscan.h"

#include "access/tableam.h"

#include "access/transam.h"

#include "access/xact.h"

#include "access/heapam.h"

#include "catalog/pg_am_d.h"

#include "commands/trigger.h"

#include "executor/executor.h"

#include "executor/nodeModifyTable.h"

#include "replication/conflict.h"

#include "replication/logicalrelation.h"

#include "storage/lmgr.h"

#include "utils/builtins.h"

#include "utils/lsyscache.h"

#include "utils/rel.h"

#include "utils/snapmgr.h"

#include "utils/syscache.h"

#include "utils/typcache.h"

static bool tuples_equal(TupleTableSlot *slot1, TupleTableSlot *slot2,

             TypeCacheEntry **eq, Bitmapset *columns);

/*

 * Setup a ScanKey for a search in the relation 'rel' for a tuple 'key' that

 * is setup to match 'rel' (*NOT* idxrel!).

 *

 * Returns how many columns to use for the index scan.

 *

 * This is not a generic routine, idxrel must be PK, RI, or an index that can be

 * used for a REPLICA IDENTITY FULL table. See FindUsableIndexForReplicaIdentityFull()

 * for details.

 *

 * By definition, replication identity of a rel meets all limitations associated

 * with that. Note that any other index could also meet these limitations.

 */

static int

build_replindex_scan_key(ScanKey skey, Relation rel, Relation idxrel,

             TupleTableSlot *searchslot)

{

  int     index_attoff;

  int     skey_attoff = 0;

  Datum   indclassDatum;

  oidvector  *opclass;

  int2vector *indkey = &idxrel->rd_index->indkey;

  indclassDatum = SysCacheGetAttrNotNull(INDEXRELID, idxrel->rd_indextuple,

                       Anum_pg_index_indclass);

  opclass = (oidvector *) DatumGetPointer(indclassDatum);

  /* Build scankey for every non-expression attribute in the index. */

  for (index_attoff = 0; index_attoff < IndexRelationGetNumberOfKeyAttributes(idxrel);

     index_attoff++)

  {

    Oid     operator;

    Oid     optype;

    Oid     opfamily;

    RegProcedure regop;

    int     table_attno = indkey->values[index_attoff];

    StrategyNumber eq_strategy;

    if (!AttributeNumberIsValid(table_attno))

    {

      /*

       * XXX: Currently, we don't support expressions in the scan key,

       * see code below.

       */

      continue;

    }

    /*

     * Load the operator info.  We need this to get the equality operator

     * function for the scan key.

     */

    optype = get_opclass_input_type(opclass->values[index_attoff]);

    opfamily = get_opclass_family(opclass->values[index_attoff]);

    eq_strategy = IndexAmTranslateCompareType(COMPARE_EQ, idxrel->rd_rel->relam, opfamily, false);

    operator = get_opfamily_member(opfamily, optype,

                     optype,

                     eq_strategy);

    if (!OidIsValid(operator))

      elog(ERROR, "missing operator %d(%u,%u) in opfamily %u",

         eq_strategy, optype, optype, opfamily);

    regop = get_opcode(operator);

    /* Initialize the scankey. */

    ScanKeyInit(&skey[skey_attoff],

          index_attoff + 1,

          eq_strategy,

          regop,

          searchslot->tts_values[table_attno - 1]);

    skey[skey_attoff].sk_collation = idxrel->rd_indcollation[index_attoff];

    /* Check for null value. */

    if (searchslot->tts_isnull[table_attno - 1])

      skey[skey_attoff].sk_flags |= (SK_ISNULL | SK_SEARCHNULL);

    skey_attoff++;

  }

  /* There must always be at least one attribute for the index scan. */

  Assert(skey_attoff > 0);

  return skey_attoff;

}

/*

 * Helper function to check if it is necessary to re-fetch and lock the tuple

 * due to concurrent modifications. This function should be called after

 * invoking table_tuple_lock.

 */

static bool

should_refetch_tuple(TM_Result res, TM_FailureData *tmfd)

{

  bool    refetch = false;

  switch (res)

  {

    case TM_Ok:

      break;

    case TM_Updated:

      /* XXX: Improve handling here */

      if (ItemPointerIndicatesMovedPartitions(&tmfd->ctid))

        ereport(LOG,

            (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),

             errmsg("tuple to be locked was already moved to another partition due to concurrent update, retrying")));

      else

        ereport(LOG,

            (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),

             errmsg("concurrent update, retrying")));

      refetch = true;

      break;

    case TM_Deleted:

      /* XXX: Improve handling here */

      ereport(LOG,

          (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),

           errmsg("concurrent delete, retrying")));

      refetch = true;

      break;

    case TM_Invisible:

      elog(ERROR, "attempted to lock invisible tuple");

      break;

    default:

      elog(ERROR, "unexpected table_tuple_lock status: %u", res);

      break;

  }

  return refetch;

}

/*

 * Search the relation 'rel' for tuple using the index.

 *

 * If a matching tuple is found, lock it with lockmode, fill the slot with its

 * contents, and return true.  Return false otherwise.

 */

bool

RelationFindReplTupleByIndex(Relation rel, Oid idxoid,

               LockTupleMode lockmode,

               TupleTableSlot *searchslot,

               TupleTableSlot *outslot)

{

  ScanKeyData skey[INDEX_MAX_KEYS];

  int     skey_attoff;

  IndexScanDesc scan;

  SnapshotData snap;

  TransactionId xwait;

  Relation  idxrel;

  bool    found;

  TypeCacheEntry **eq = NULL;

  bool    isIdxSafeToSkipDuplicates;

  /* Open the index. */

  idxrel = index_open(idxoid, RowExclusiveLock);

  isIdxSafeToSkipDuplicates = (GetRelationIdentityOrPK(rel) == idxoid);

  InitDirtySnapshot(snap);

  /* Build scan key. */

  skey_attoff = build_replindex_scan_key(skey, rel, idxrel, searchslot);

  /* Start an index scan. */

  scan = index_beginscan(rel, idxrel,

               &snap, NULL, skey_attoff, 0, SO_NONE);

retry:

  found = false;

  index_rescan(scan, skey, skey_attoff, NULL, 0);

  /* Try to find the tuple */

  while (index_getnext_slot(scan, ForwardScanDirection, outslot))

  {

    /*

     * Avoid expensive equality check if the index is primary key or

     * replica identity index.

     */

    if (!isIdxSafeToSkipDuplicates)

    {

      if (eq == NULL)

        eq = palloc0_array(TypeCacheEntry *, outslot->tts_tupleDescriptor->natts);

      if (!tuples_equal(outslot, searchslot, eq, NULL))

        continue;

    }

    ExecMaterializeSlot(outslot);

    xwait = TransactionIdIsValid(snap.xmin) ?

      snap.xmin : snap.xmax;

    /*

     * If the tuple is locked, wait for locking transaction to finish and

     * retry.

     */

    if (TransactionIdIsValid(xwait))

    {

      XactLockTableWait(xwait, NULL, NULL, XLTW_None);

      goto retry;

    }

    /* Found our tuple and it's not locked */

    found = true;

    break;

  }

  /* Found tuple, try to lock it in the lockmode. */

  if (found)

  {

    TM_FailureData tmfd;

    TM_Result res;

    PushActiveSnapshot(GetLatestSnapshot());

    res = table_tuple_lock(rel, &(outslot->tts_tid), GetActiveSnapshot(),

                 outslot,

                 GetCurrentCommandId(false),

                 lockmode,

                 LockWaitBlock,

                 0 /* don't follow updates */ ,

                 &tmfd);

    PopActiveSnapshot();

    if (should_refetch_tuple(res, &tmfd))

      goto retry;

  }

  index_endscan(scan);

  /* Don't release lock until commit. */

  index_close(idxrel, NoLock);

  return found;

}

/*

 * Compare the tuples in the slots by checking if they have equal values.

 *

 * If 'columns' is not null, only the columns specified within it will be

 * considered for the equality check, ignoring all other columns.

 */

static bool

tuples_equal(TupleTableSlot *slot1, TupleTableSlot *slot2,

       TypeCacheEntry **eq, Bitmapset *columns)

{

  int     attrnum;

  Assert(slot1->tts_tupleDescriptor->natts ==

       slot2->tts_tupleDescriptor->natts);

  slot_getallattrs(slot1);

  slot_getallattrs(slot2);

  /* Check equality of the attributes. */

  for (attrnum = 0; attrnum < slot1->tts_tupleDescriptor->natts; attrnum++)

  {

    Form_pg_attribute att;

    TypeCacheEntry *typentry;

    att = TupleDescAttr(slot1->tts_tupleDescriptor, attrnum);

    /*

     * Ignore dropped and generated columns as the publisher doesn't send

     * those

     */

    if (att->attisdropped || att->attgenerated)

      continue;

    /*

     * Ignore columns that are not listed for checking.

     */

    if (columns &&

      !bms_is_member(att->attnum - FirstLowInvalidHeapAttributeNumber,

               columns))

      continue;

    /*

     * If one value is NULL and other is not, then they are certainly not

     * equal

     */

    if (slot1->tts_isnull[attrnum] != slot2->tts_isnull[attrnum])

      return false;

    /*

     * If both are NULL, they can be considered equal.

     */

    if (slot1->tts_isnull[attrnum] || slot2->tts_isnull[attrnum])

      continue;

    typentry = eq[attrnum];

    if (typentry == NULL)

    {

      typentry = lookup_type_cache(att->atttypid,

                     TYPECACHE_EQ_OPR_FINFO);

      if (!OidIsValid(typentry->eq_opr_finfo.fn_oid))

        ereport(ERROR,

            (errcode(ERRCODE_UNDEFINED_FUNCTION),

             errmsg("could not identify an equality operator for type %s",

                format_type_be(att->atttypid))));

      eq[attrnum] = typentry;

    }

    if (!DatumGetBool(FunctionCall2Coll(&typentry->eq_opr_finfo,

                      att->attcollation,

                      slot1->tts_values[attrnum],

                      slot2->tts_values[attrnum])))

      return false;

  }

  return true;

}

/*

 * Search the relation 'rel' for tuple using the sequential scan.

 *

 * If a matching tuple is found, lock it with lockmode, fill the slot with its

 * contents, and return true.  Return false otherwise.

 *

 * Note that this stops on the first matching tuple.

 *

 * This can obviously be quite slow on tables that have more than few rows.

 */

bool

RelationFindReplTupleSeq(Relation rel, LockTupleMode lockmode,

             TupleTableSlot *searchslot, TupleTableSlot *outslot)

{

  TupleTableSlot *scanslot;

  TableScanDesc scan;

  SnapshotData snap;

  TypeCacheEntry **eq;

  TransactionId xwait;

  bool    found;

  TupleDesc desc PG_USED_FOR_ASSERTS_ONLY = RelationGetDescr(rel);

  Assert(equalTupleDescs(desc, outslot->tts_tupleDescriptor));

  eq = palloc0_array(TypeCacheEntry *, outslot->tts_tupleDescriptor->natts);

  /* Start a heap scan. */

  InitDirtySnapshot(snap);

  scan = table_beginscan(rel, &snap, 0, NULL,

               SO_NONE);

  scanslot = table_slot_create(rel, NULL);

retry:

  found = false;

  table_rescan(scan, NULL);

  /* Try to find the tuple */

  while (table_scan_getnextslot(scan, ForwardScanDirection, scanslot))

  {

    if (!tuples_equal(scanslot, searchslot, eq, NULL))

      continue;

    found = true;

    ExecCopySlot(outslot, scanslot);

    xwait = TransactionIdIsValid(snap.xmin) ?

      snap.xmin : snap.xmax;

    /*

     * If the tuple is locked, wait for locking transaction to finish and

     * retry.

     */

    if (TransactionIdIsValid(xwait))

    {

      XactLockTableWait(xwait, NULL, NULL, XLTW_None);

      goto retry;

    }

    /* Found our tuple and it's not locked */

    break;

  }

  /* Found tuple, try to lock it in the lockmode. */

  if (found)

  {

    TM_FailureData tmfd;

    TM_Result res;

    PushActiveSnapshot(GetLatestSnapshot());

    res = table_tuple_lock(rel, &(outslot->tts_tid), GetActiveSnapshot(),

                 outslot,

                 GetCurrentCommandId(false),

                 lockmode,

                 LockWaitBlock,

                 0 /* don't follow updates */ ,

                 &tmfd);

    PopActiveSnapshot();

    if (should_refetch_tuple(res, &tmfd))

      goto retry;

  }

  table_endscan(scan);

  ExecDropSingleTupleTableSlot(scanslot);

  return found;

}

/*

 * Build additional index information necessary for conflict detection.

 */

static void

BuildConflictIndexInfo(ResultRelInfo *resultRelInfo, Oid conflictindex)

{

  for (int i = 0; i < resultRelInfo->ri_NumIndices; i++)

  {

    Relation  indexRelation = resultRelInfo->ri_IndexRelationDescs[i];

    IndexInfo  *indexRelationInfo = resultRelInfo->ri_IndexRelationInfo[i];

    if (conflictindex != RelationGetRelid(indexRelation))

      continue;

    /*

     * This Assert will fail if BuildSpeculativeIndexInfo() is called

     * twice for the given index.

     */

    Assert(indexRelationInfo->ii_UniqueOps == NULL);

    BuildSpeculativeIndexInfo(indexRelation, indexRelationInfo);

  }

}

/*

 * If the tuple is recently dead and was deleted by a transaction with a newer

 * commit timestamp than previously recorded, update the associated transaction

 * ID, commit time, and origin. This helps ensure that conflict detection uses

 * the most recent and relevant deletion metadata.

 */

static void

update_most_recent_deletion_info(TupleTableSlot *scanslot,

                 TransactionId oldestxmin,

                 TransactionId *delete_xid,

                 TimestampTz *delete_time,

                 ReplOriginId *delete_origin)

{

  BufferHeapTupleTableSlot *hslot;

  HeapTuple tuple;

  Buffer    buf;

  bool    recently_dead = false;

  TransactionId xmax;

  TimestampTz localts;

  ReplOriginId localorigin;

  hslot = (BufferHeapTupleTableSlot *) scanslot;

  tuple = ExecFetchSlotHeapTuple(scanslot, false, NULL);

  buf = hslot->buffer;

  LockBuffer(buf, BUFFER_LOCK_SHARE);

  /*

   * We do not consider HEAPTUPLE_DEAD status because it indicates either

   * tuples whose inserting transaction was aborted (meaning there is no

   * commit timestamp or origin), or tuples deleted by a transaction older

   * than oldestxmin, making it safe to ignore them during conflict

   * detection (See comments atop worker.c for details).

   */

  if (HeapTupleSatisfiesVacuum(tuple, oldestxmin, buf) == HEAPTUPLE_RECENTLY_DEAD)

    recently_dead = true;

  LockBuffer(buf, BUFFER_LOCK_UNLOCK);

  if (!recently_dead)

    return;

  xmax = HeapTupleHeaderGetUpdateXid(tuple->t_data);

  if (!TransactionIdIsValid(xmax))

    return;

  /* Select the dead tuple with the most recent commit timestamp */

  if (TransactionIdGetCommitTsData(xmax, &localts, &localorigin) &&

    TimestampDifferenceExceeds(*delete_time, localts, 0))

  {

    *delete_xid = xmax;

    *delete_time = localts;

    *delete_origin = localorigin;

  }

}

/*

 * Searches the relation 'rel' for the most recently deleted tuple that matches

 * the values in 'searchslot' and is not yet removable by VACUUM. The function

 * returns the transaction ID, origin, and commit timestamp of the transaction

 * that deleted this tuple.

 *

 * 'oldestxmin' acts as a cutoff transaction ID. Tuples deleted by transactions

 * with IDs >= 'oldestxmin' are considered recently dead and are eligible for

 * conflict detection.

 *

 * Instead of stopping at the first match, we scan all matching dead tuples to

 * identify most recent deletion. This is crucial because only the latest

 * deletion is relevant for resolving conflicts.

 *

 * For example, consider a scenario on the subscriber where a row is deleted,

 * re-inserted, and then deleted again only on the subscriber:

 *

 *   - (pk, 1) - deleted at 9:00,

 *   - (pk, 1) - deleted at 9:02,

 *

 * Now, a remote update arrives: (pk, 1) -> (pk, 2), timestamped at 9:01.

 *

 * If we mistakenly return the older deletion (9:00), the system may wrongly

 * apply the remote update using a last-update-wins strategy. Instead, we must

 * recognize the more recent deletion at 9:02 and skip the update. See

 * comments atop worker.c for details. Note, as of now, conflict resolution

 * is not implemented. Consequently, the system may incorrectly report the

 * older tuple as the conflicted one, leading to misleading results.

 *

 * The commit timestamp of the deleting transaction is used to determine which

 * tuple was deleted most recently.

 */

bool

RelationFindDeletedTupleInfoSeq(Relation rel, TupleTableSlot *searchslot,

                TransactionId oldestxmin,

                TransactionId *delete_xid,

                ReplOriginId *delete_origin,

                TimestampTz *delete_time)

{

  TupleTableSlot *scanslot;

  TableScanDesc scan;

  TypeCacheEntry **eq;

  Bitmapset  *indexbitmap;

  TupleDesc desc PG_USED_FOR_ASSERTS_ONLY = RelationGetDescr(rel);

  Assert(equalTupleDescs(desc, searchslot->tts_tupleDescriptor));

  *delete_xid = InvalidTransactionId;

  *delete_origin = InvalidReplOriginId;

  *delete_time = 0;

  /*

   * If the relation has a replica identity key or a primary key that is

   * unusable for locating deleted tuples (see

   * IsIndexUsableForFindingDeletedTuple), a full table scan becomes

   * necessary. In such cases, comparing the entire tuple is not required,

   * since the remote tuple might not include all column values. Instead,

   * the indexed columns alone are sufficient to identify the target tuple

   * (see logicalrep_rel_mark_updatable).

   */

  indexbitmap = RelationGetIndexAttrBitmap(rel,

                       INDEX_ATTR_BITMAP_IDENTITY_KEY);

  /* fallback to PK if no replica identity */

  if (!indexbitmap)

    indexbitmap = RelationGetIndexAttrBitmap(rel,

                         INDEX_ATTR_BITMAP_PRIMARY_KEY);

  eq = palloc0_array(TypeCacheEntry *, searchslot->tts_tupleDescriptor->natts);

  /*

   * Start a heap scan using SnapshotAny to identify dead tuples that are

   * not visible under a standard MVCC snapshot. Tuples from transactions

   * not yet committed or those just committed prior to the scan are

   * excluded in update_most_recent_deletion_info().

   */

  scan = table_beginscan(rel, SnapshotAny, 0, NULL,

               SO_NONE);

  scanslot = table_slot_create(rel, NULL);

  table_rescan(scan, NULL);

  /* Try to find the tuple */

  while (table_scan_getnextslot(scan, ForwardScanDirection, scanslot))

  {

    if (!tuples_equal(scanslot, searchslot, eq, indexbitmap))

      continue;

    update_most_recent_deletion_info(scanslot, oldestxmin, delete_xid,

                     delete_time, delete_origin);

  }

  table_endscan(scan);

  ExecDropSingleTupleTableSlot(scanslot);

  return *delete_time != 0;

}

/*

 * Similar to RelationFindDeletedTupleInfoSeq() but using index scan to locate

 * the deleted tuple.

 */

bool

RelationFindDeletedTupleInfoByIndex(Relation rel, Oid idxoid,

                  TupleTableSlot *searchslot,

                  TransactionId oldestxmin,

                  TransactionId *delete_xid,

                  ReplOriginId *delete_origin,

                  TimestampTz *delete_time)

{

  Relation  idxrel;

  ScanKeyData skey[INDEX_MAX_KEYS];

  int     skey_attoff;

  IndexScanDesc scan;

  TupleTableSlot *scanslot;

  TypeCacheEntry **eq = NULL;

  bool    isIdxSafeToSkipDuplicates;

  TupleDesc desc PG_USED_FOR_ASSERTS_ONLY = RelationGetDescr(rel);

  Assert(equalTupleDescs(desc, searchslot->tts_tupleDescriptor));

  Assert(OidIsValid(idxoid));

  *delete_xid = InvalidTransactionId;

  *delete_time = 0;

  *delete_origin = InvalidReplOriginId;

  isIdxSafeToSkipDuplicates = (GetRelationIdentityOrPK(rel) == idxoid);

  scanslot = table_slot_create(rel, NULL);

  idxrel = index_open(idxoid, RowExclusiveLock);

  /* Build scan key. */

  skey_attoff = build_replindex_scan_key(skey, rel, idxrel, searchslot);

  /*

   * Start an index scan using SnapshotAny to identify dead tuples that are

   * not visible under a standard MVCC snapshot. Tuples from transactions

   * not yet committed or those just committed prior to the scan are

   * excluded in update_most_recent_deletion_info().

   */

  scan = index_beginscan(rel, idxrel,

               SnapshotAny, NULL, skey_attoff, 0, SO_NONE);

  index_rescan(scan, skey, skey_attoff, NULL, 0);

  /* Try to find the tuple */

  while (index_getnext_slot(scan, ForwardScanDirection, scanslot))

  {

    /*

     * Avoid expensive equality check if the index is primary key or

     * replica identity index.

     */

    if (!isIdxSafeToSkipDuplicates)

    {

      if (eq == NULL)

        eq = palloc0_array(TypeCacheEntry *, scanslot->tts_tupleDescriptor->natts);

      if (!tuples_equal(scanslot, searchslot, eq, NULL))

        continue;

    }

    update_most_recent_deletion_info(scanslot, oldestxmin, delete_xid,

                     delete_time, delete_origin);

  }

  index_endscan(scan);

  index_close(idxrel, NoLock);

  ExecDropSingleTupleTableSlot(scanslot);

  return *delete_time != 0;

}

/*

 * Find the tuple that violates the passed unique index (conflictindex).

 *

 * If the conflicting tuple is found return true, otherwise false.

 *

 * We lock the tuple to avoid getting it deleted before the caller can fetch

 * the required information. Note that if the tuple is deleted before a lock

 * is acquired, we will retry to find the conflicting tuple again.

 */

static bool

FindConflictTuple(ResultRelInfo *resultRelInfo, EState *estate,

          Oid conflictindex, TupleTableSlot *slot,

          TupleTableSlot **conflictslot)

{

  Relation  rel = resultRelInfo->ri_RelationDesc;

  ItemPointerData conflictTid;

  TM_FailureData tmfd;

  TM_Result res;

  *conflictslot = NULL;

  /*

   * Build additional information required to check constraints violations.

   * See check_exclusion_or_unique_constraint().

   */

  BuildConflictIndexInfo(resultRelInfo, conflictindex);

retry:

  if (ExecCheckIndexConstraints(resultRelInfo, slot, estate,

                  &conflictTid, &slot->tts_tid,

                  list_make1_oid(conflictindex)))

  {

    if (*conflictslot)

      ExecDropSingleTupleTableSlot(*conflictslot);

    *conflictslot = NULL;

    return false;

  }

  *conflictslot = table_slot_create(rel, NULL);

  PushActiveSnapshot(GetLatestSnapshot());

  res = table_tuple_lock(rel, &conflictTid, GetActiveSnapshot(),

               *conflictslot,

               GetCurrentCommandId(false),

               LockTupleShare,

               LockWaitBlock,

               0 /* don't follow updates */ ,

               &tmfd);

  PopActiveSnapshot();

  if (should_refetch_tuple(res, &tmfd))

    goto retry;

  return true;

}

/*

 * Check all the unique indexes in 'recheckIndexes' for conflict with the

 * tuple in 'remoteslot' and report if found.

 */

static void

CheckAndReportConflict(ResultRelInfo *resultRelInfo, EState *estate,

             ConflictType type, List *recheckIndexes,

             TupleTableSlot *searchslot, TupleTableSlot *remoteslot)

{

  List     *conflicttuples = NIL;

  TupleTableSlot *conflictslot;

  /* Check all the unique indexes for conflicts */

  foreach_oid(uniqueidx, resultRelInfo->ri_onConflictArbiterIndexes)

  {

    if (list_member_oid(recheckIndexes, uniqueidx) &&

      FindConflictTuple(resultRelInfo, estate, uniqueidx, remoteslot,

                &conflictslot))

    {

      ConflictTupleInfo *conflicttuple = palloc0_object(ConflictTupleInfo);

      conflicttuple->slot = conflictslot;

      conflicttuple->indexoid = uniqueidx;

      GetTupleTransactionInfo(conflictslot, &conflicttuple->xmin,

                  &conflicttuple->origin, &conflicttuple->ts);

      conflicttuples = lappend(conflicttuples, conflicttuple);

    }

  }

  /* Report the conflict, if found */

  if (conflicttuples)

    ReportApplyConflict(estate, resultRelInfo, ERROR,

              list_length(conflicttuples) > 1 ? CT_MULTIPLE_UNIQUE_CONFLICTS : type,

              searchslot, remoteslot, conflicttuples);

}

/*

 * Insert tuple represented in the slot to the relation, update the indexes,

 * and execute any constraints and per-row triggers.

 *

 * Caller is responsible for opening the indexes.

 */

void

ExecSimpleRelationInsert(ResultRelInfo *resultRelInfo,

             EState *estate, TupleTableSlot *slot)

{

  bool    skip_tuple = false;

  Relation  rel = resultRelInfo->ri_RelationDesc;

  /* For now we support only tables. */

  Assert(rel->rd_rel->relkind == RELKIND_RELATION);

  CheckCmdReplicaIdentity(rel, CMD_INSERT);

  /* BEFORE ROW INSERT Triggers */

  if (resultRelInfo->ri_TrigDesc &&

    resultRelInfo->ri_TrigDesc->trig_insert_before_row)

  {

    if (!ExecBRInsertTriggers(estate, resultRelInfo, slot))

      skip_tuple = true; /* "do nothing" */

  }

  if (!skip_tuple)

  {

    List     *recheckIndexes = NIL;

    List     *conflictindexes;

    bool    conflict = false;

    /* Compute stored generated columns */

    if (rel->rd_att->constr &&

      rel->rd_att->constr->has_generated_stored)

      ExecComputeStoredGenerated(resultRelInfo, estate, slot,

                     CMD_INSERT);

    /* Check the constraints of the tuple */

    if (rel->rd_att->constr)

      ExecConstraints(resultRelInfo, slot, estate);

    if (rel->rd_rel->relispartition)

      ExecPartitionCheck(resultRelInfo, slot, estate, true);

    /* OK, store the tuple and create index entries for it */

    simple_table_tuple_insert(resultRelInfo->ri_RelationDesc, slot);

    conflictindexes = resultRelInfo->ri_onConflictArbiterIndexes;

    if (resultRelInfo->ri_NumIndices > 0)

    {

      uint32    flags;

      if (conflictindexes != NIL)

        flags = EIIT_NO_DUPE_ERROR;

      else

        flags = 0;

      recheckIndexes = ExecInsertIndexTuples(resultRelInfo,

                           estate, flags,

                           slot, conflictindexes,

                           &conflict);

    }

    /*

     * Checks the conflict indexes to fetch the conflicting local row and

     * reports the conflict. We perform this check here, instead of

     * performing an additional index scan before the actual insertion and

     * reporting the conflict if any conflicting rows are found. This is

     * to avoid the overhead of executing the extra scan for each INSERT

     * operation, even when no conflict arises, which could introduce

     * significant overhead to replication, particularly in cases where

     * conflicts are rare.

     *

     * XXX OTOH, this could lead to clean-up effort for dead tuples added

     * in heap and index in case of conflicts. But as conflicts shouldn't

     * be a frequent thing so we preferred to save the performance

     * overhead of extra scan before each insertion.

     */

    if (conflict)

      CheckAndReportConflict(resultRelInfo, estate, CT_INSERT_EXISTS,

                   recheckIndexes, NULL, slot);

    /* AFTER ROW INSERT Triggers */

    ExecARInsertTriggers(estate, resultRelInfo, slot,

               recheckIndexes, NULL);

    /*

     * XXX we should in theory pass a TransitionCaptureState object to the

     * above to capture transition tuples, but after statement triggers

     * don't actually get fired by replication yet anyway

     */

    list_free(recheckIndexes);

  }

}

/*

 * Find the searchslot tuple and update it with data in the slot,

 * update the indexes, and execute any constraints and per-row triggers.

 *

 * Caller is responsible for opening the indexes.

 */

void

ExecSimpleRelationUpdate(ResultRelInfo *resultRelInfo,

             EState *estate, EPQState *epqstate,

             TupleTableSlot *searchslot, TupleTableSlot *slot)

{

  bool    skip_tuple = false;

  Relation  rel = resultRelInfo->ri_RelationDesc;

  ItemPointer tid = &(searchslot->tts_tid);

  /*

   * We support only non-system tables, with

   * check_publication_add_relation() accountable.

   */

  Assert(rel->rd_rel->relkind == RELKIND_RELATION);

  Assert(!IsCatalogRelation(rel));

  CheckCmdReplicaIdentity(rel, CMD_UPDATE);

  /* BEFORE ROW UPDATE Triggers */

  if (resultRelInfo->ri_TrigDesc &&

    resultRelInfo->ri_TrigDesc->trig_update_before_row)

  {

    if (!ExecBRUpdateTriggers(estate, epqstate, resultRelInfo,

                  tid, NULL, slot, NULL, NULL, false))

      skip_tuple = true; /* "do nothing" */

  }

  if (!skip_tuple)

  {

    List     *recheckIndexes = NIL;

    TU_UpdateIndexes update_indexes;