execReplication.c File Reference

#include "postgres.h"
#include "access/genam.h"
#include "access/relscan.h"
#include "access/tableam.h"
#include "access/transam.h"
#include "access/xact.h"
#include "commands/trigger.h"
#include "executor/executor.h"
#include "executor/nodeModifyTable.h"
#include "nodes/nodeFuncs.h"
#include "parser/parse_relation.h"
#include "parser/parsetree.h"
#include "storage/bufmgr.h"
#include "storage/lmgr.h"
#include "utils/builtins.h"
#include "utils/datum.h"
#include "utils/lsyscache.h"
#include "utils/memutils.h"
#include "utils/rel.h"
#include "utils/snapmgr.h"
#include "utils/syscache.h"
#include "utils/typcache.h"

Include dependency graph for execReplication.c:

Go to the source code of this file.

Functions
static bool	build_replindex_scan_key (ScanKey skey, Relation rel, Relation idxrel, TupleTableSlot *searchslot)
bool	RelationFindReplTupleByIndex (Relation rel, Oid idxoid, LockTupleMode lockmode, TupleTableSlot *searchslot, TupleTableSlot *outslot)
static bool	tuples_equal (TupleTableSlot *slot1, TupleTableSlot *slot2, TypeCacheEntry **eq)
bool	RelationFindReplTupleSeq (Relation rel, LockTupleMode lockmode, TupleTableSlot *searchslot, TupleTableSlot *outslot)
void	ExecSimpleRelationInsert (ResultRelInfo *resultRelInfo, EState *estate, TupleTableSlot *slot)
void	ExecSimpleRelationUpdate (ResultRelInfo *resultRelInfo, EState *estate, EPQState *epqstate, TupleTableSlot *searchslot, TupleTableSlot *slot)
void	ExecSimpleRelationDelete (ResultRelInfo *resultRelInfo, EState *estate, EPQState *epqstate, TupleTableSlot *searchslot)
void	CheckCmdReplicaIdentity (Relation rel, CmdType cmd)
void	CheckSubscriptionRelkind (char relkind, const char *nspname, const char *relname)

Function Documentation

build_replindex_scan_key()

static bool build_replindex_scan_key ( ScanKey  skey, Relation  rel, Relation  idxrel, TupleTableSlot *  searchslot  )

static

Definition at line 50 of file execReplication.c.

References Assert, BTEqualStrategyNumber, DatumGetPointer, elog, ERROR, get_opclass_family(), get_opclass_input_type(), get_opcode(), get_opfamily_member(), IndexRelationGetNumberOfKeyAttributes, INDEXRELID, OidIsValid, RelationData::rd_indcollation, RelationData::rd_index, RelationData::rd_indextuple, RelationGetPrimaryKeyIndex(), RelationGetRelid, RelationGetReplicaIndex(), ScanKeyInit(), ScanKeyData::sk_collation, ScanKeyData::sk_flags, SK_ISNULL, SysCacheGetAttr(), TupleTableSlot::tts_isnull, TupleTableSlot::tts_values, int2vector::values, and oidvector::values.

Referenced by RelationFindReplTupleByIndex().

{
    int         attoff;
    bool        isnull;
    Datum       indclassDatum;
    oidvector  *opclass;
    int2vector *indkey = &idxrel->rd_index->indkey;
    bool        hasnulls = false;

    Assert(RelationGetReplicaIndex(rel) == RelationGetRelid(idxrel) ||
           RelationGetPrimaryKeyIndex(rel) == RelationGetRelid(idxrel));

    indclassDatum = SysCacheGetAttr(INDEXRELID, idxrel->rd_indextuple,
                                    Anum_pg_index_indclass, &isnull);
    Assert(!isnull);
    opclass = (oidvector *) DatumGetPointer(indclassDatum);

    /* Build scankey for every attribute in the index. */
    for (attoff = 0; attoff < IndexRelationGetNumberOfKeyAttributes(idxrel); attoff++)
    {
        Oid         operator;
        Oid         opfamily;
        RegProcedure regop;
        int         pkattno = attoff + 1;
        int         mainattno = indkey->values[attoff];
        Oid         optype = get_opclass_input_type(opclass->values[attoff]);

        /*
         * Load the operator info.  We need this to get the equality operator
         * function for the scan key.
         */
        opfamily = get_opclass_family(opclass->values[attoff]);

        operator = get_opfamily_member(opfamily, optype,
                                       optype,
                                       BTEqualStrategyNumber);
        if (!OidIsValid(operator))
            elog(ERROR, "missing operator %d(%u,%u) in opfamily %u",
                 BTEqualStrategyNumber, optype, optype, opfamily);

        regop = get_opcode(operator);

        /* Initialize the scankey. */
        ScanKeyInit(&skey[attoff],
                    pkattno,
                    BTEqualStrategyNumber,
                    regop,
                    searchslot->tts_values[mainattno - 1]);

        skey[attoff].sk_collation = idxrel->rd_indcollation[attoff];

        /* Check for null value. */
        if (searchslot->tts_isnull[mainattno - 1])
        {
            hasnulls = true;
            skey[attoff].sk_flags |= SK_ISNULL;
        }
    }

    return hasnulls;
}

CheckCmdReplicaIdentity()

void CheckCmdReplicaIdentity	(	Relation	rel,
		CmdType	cmd
	)

Definition at line 568 of file execReplication.c.

References CMD_DELETE, CMD_UPDATE, ereport, errcode(), errhint(), errmsg(), ERROR, GetRelationPublicationActions(), OidIsValid, PublicationActions::pubdelete, PublicationActions::pubupdate, RelationData::rd_rel, RelationGetRelationName, and RelationGetReplicaIndex().

Referenced by CheckValidResultRel(), exec_rt_fetch(), ExecSimpleRelationDelete(), ExecSimpleRelationInsert(), and ExecSimpleRelationUpdate().

{
    PublicationActions *pubactions;

    /* We only need to do checks for UPDATE and DELETE. */
    if (cmd != CMD_UPDATE && cmd != CMD_DELETE)
        return;

    /* If relation has replica identity we are always good. */
    if (rel->rd_rel->relreplident == REPLICA_IDENTITY_FULL ||
        OidIsValid(RelationGetReplicaIndex(rel)))
        return;

    /*
     * This is either UPDATE OR DELETE and there is no replica identity.
     *
     * Check if the table publishes UPDATES or DELETES.
     */
    pubactions = GetRelationPublicationActions(rel);
    if (cmd == CMD_UPDATE && pubactions->pubupdate)
        ereport(ERROR,
                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                 errmsg("cannot update table \"%s\" because it does not have a replica identity and publishes updates",
                        RelationGetRelationName(rel)),
                 errhint("To enable updating the table, set REPLICA IDENTITY using ALTER TABLE.")));
    else if (cmd == CMD_DELETE && pubactions->pubdelete)
        ereport(ERROR,
                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                 errmsg("cannot delete from table \"%s\" because it does not have a replica identity and publishes deletes",
                        RelationGetRelationName(rel)),
                 errhint("To enable deleting from the table, set REPLICA IDENTITY using ALTER TABLE.")));
}

CheckSubscriptionRelkind()

void CheckSubscriptionRelkind	(	char	relkind,
		const char *	nspname,
		const char *	relname
	)

Definition at line 608 of file execReplication.c.

References ereport, errcode(), errdetail(), errmsg(), and ERROR.

Referenced by AlterSubscription_refresh(), CreateSubscription(), exec_rt_fetch(), and logicalrep_rel_open().

{
    /*
     * Give a more specific error for foreign tables.
     */
    if (relkind == RELKIND_FOREIGN_TABLE)
        ereport(ERROR,
                (errcode(ERRCODE_WRONG_OBJECT_TYPE),
                 errmsg("cannot use relation \"%s.%s\" as logical replication target",
                        nspname, relname),
                 errdetail("\"%s.%s\" is a foreign table.",
                           nspname, relname)));

    if (relkind != RELKIND_RELATION && relkind != RELKIND_PARTITIONED_TABLE)
        ereport(ERROR,
                (errcode(ERRCODE_WRONG_OBJECT_TYPE),
                 errmsg("cannot use relation \"%s.%s\" as logical replication target",
                        nspname, relname),
                 errdetail("\"%s.%s\" is not a table.",
                           nspname, relname)));
}

ExecSimpleRelationDelete()

void ExecSimpleRelationDelete	(	ResultRelInfo *	resultRelInfo,
		EState *	estate,
		EPQState *	epqstate,
		TupleTableSlot *	searchslot
	)

Definition at line 534 of file execReplication.c.

References CheckCmdReplicaIdentity(), CMD_DELETE, EState::es_snapshot, ExecARDeleteTriggers(), ExecBRDeleteTriggers(), ResultRelInfo::ri_RelationDesc, ResultRelInfo::ri_TrigDesc, simple_table_tuple_delete(), TriggerDesc::trig_delete_before_row, and TupleTableSlot::tts_tid.

Referenced by apply_handle_delete_internal(), and exec_rt_fetch().

{
    bool        skip_tuple = false;
    Relation    rel = resultRelInfo->ri_RelationDesc;
    ItemPointer tid = &searchslot->tts_tid;

    CheckCmdReplicaIdentity(rel, CMD_DELETE);

    /* BEFORE ROW DELETE Triggers */
    if (resultRelInfo->ri_TrigDesc &&
        resultRelInfo->ri_TrigDesc->trig_delete_before_row)
    {
        skip_tuple = !ExecBRDeleteTriggers(estate, epqstate, resultRelInfo,
                                           tid, NULL, NULL);

    }

    if (!skip_tuple)
    {
        /* OK, delete the tuple */
        simple_table_tuple_delete(rel, tid, estate->es_snapshot);

        /* AFTER ROW DELETE Triggers */
        ExecARDeleteTriggers(estate, resultRelInfo,
                             tid, NULL, NULL);
    }
}

ExecSimpleRelationInsert()

void ExecSimpleRelationInsert	(	ResultRelInfo *	resultRelInfo,
		EState *	estate,
		TupleTableSlot *	slot
	)

Definition at line 407 of file execReplication.c.

References Assert, CheckCmdReplicaIdentity(), CMD_INSERT, TupleDescData::constr, ExecARInsertTriggers(), ExecBRInsertTriggers(), ExecComputeStoredGenerated(), ExecConstraints(), ExecInsertIndexTuples(), ExecPartitionCheck(), TupleConstr::has_generated_stored, list_free(), NIL, RelationData::rd_att, RelationData::rd_rel, ResultRelInfo::ri_NumIndices, ResultRelInfo::ri_RelationDesc, ResultRelInfo::ri_TrigDesc, simple_table_tuple_insert(), and TriggerDesc::trig_insert_before_row.

Referenced by apply_handle_insert_internal(), and exec_rt_fetch().

{
    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;

        /* 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);

        if (resultRelInfo->ri_NumIndices > 0)
            recheckIndexes = ExecInsertIndexTuples(resultRelInfo,
                                                   slot, estate, false, false,
                                                   NULL, NIL);

        /* 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);
    }
}

ExecSimpleRelationUpdate()

void ExecSimpleRelationUpdate	(	ResultRelInfo *	resultRelInfo,
		EState *	estate,
		EPQState *	epqstate,
		TupleTableSlot *	searchslot,
		TupleTableSlot *	slot
	)

Definition at line 471 of file execReplication.c.

References Assert, CheckCmdReplicaIdentity(), CMD_UPDATE, TupleDescData::constr, EState::es_snapshot, ExecARUpdateTriggers(), ExecBRUpdateTriggers(), ExecComputeStoredGenerated(), ExecConstraints(), ExecInsertIndexTuples(), ExecPartitionCheck(), TupleConstr::has_generated_stored, list_free(), NIL, RelationData::rd_att, RelationData::rd_rel, ResultRelInfo::ri_NumIndices, ResultRelInfo::ri_RelationDesc, ResultRelInfo::ri_TrigDesc, simple_table_tuple_update(), TriggerDesc::trig_update_before_row, and TupleTableSlot::tts_tid.

Referenced by apply_handle_tuple_routing(), apply_handle_update_internal(), and exec_rt_fetch().

{
    bool        skip_tuple = false;
    Relation    rel = resultRelInfo->ri_RelationDesc;
    ItemPointer tid = &(searchslot->tts_tid);

    /* For now we support only tables. */
    Assert(rel->rd_rel->relkind == RELKIND_RELATION);

    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))
            skip_tuple = true;  /* "do nothing" */
    }

    if (!skip_tuple)
    {
        List       *recheckIndexes = NIL;
        bool        update_indexes;

        /* Compute stored generated columns */
        if (rel->rd_att->constr &&
            rel->rd_att->constr->has_generated_stored)
            ExecComputeStoredGenerated(resultRelInfo, estate, slot,
                                       CMD_UPDATE);

        /* 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);

        simple_table_tuple_update(rel, tid, slot, estate->es_snapshot,
                                  &update_indexes);

        if (resultRelInfo->ri_NumIndices > 0 && update_indexes)
            recheckIndexes = ExecInsertIndexTuples(resultRelInfo,
                                                   slot, estate, true, false,
                                                   NULL, NIL);

        /* AFTER ROW UPDATE Triggers */
        ExecARUpdateTriggers(estate, resultRelInfo,
                             tid, NULL, slot,
                             recheckIndexes, NULL);

        list_free(recheckIndexes);
    }
}

RelationFindReplTupleByIndex()

bool RelationFindReplTupleByIndex	(	Relation	rel,
		Oid	idxoid,
		LockTupleMode	lockmode,
		TupleTableSlot *	searchslot,
		TupleTableSlot *	outslot
	)

Definition at line 120 of file execReplication.c.

References build_replindex_scan_key(), TM_FailureData::ctid, elog, ereport, errcode(), errmsg(), ERROR, ExecMaterializeSlot(), ForwardScanDirection, GetCurrentCommandId(), GetLatestSnapshot(), index_beginscan(), index_close(), index_endscan(), index_getnext_slot(), INDEX_MAX_KEYS, index_open(), index_rescan(), IndexRelationGetNumberOfKeyAttributes, InitDirtySnapshot, ItemPointerIndicatesMovedPartitions, LockWaitBlock, LOG, NoLock, PopActiveSnapshot(), PushActiveSnapshot(), RowExclusiveLock, table_tuple_lock(), TM_Deleted, TM_Invisible, TM_Ok, TM_Updated, TransactionIdIsValid, TupleTableSlot::tts_tid, XactLockTableWait(), XLTW_None, SnapshotData::xmax, and SnapshotData::xmin.

Referenced by exec_rt_fetch(), and FindReplTupleInLocalRel().

{
    ScanKeyData skey[INDEX_MAX_KEYS];
    IndexScanDesc scan;
    SnapshotData snap;
    TransactionId xwait;
    Relation    idxrel;
    bool        found;

    /* Open the index. */
    idxrel = index_open(idxoid, RowExclusiveLock);

    /* Start an index scan. */
    InitDirtySnapshot(snap);
    scan = index_beginscan(rel, idxrel, &snap,
                           IndexRelationGetNumberOfKeyAttributes(idxrel),
                           0);

    /* Build scan key. */
    build_replindex_scan_key(skey, rel, idxrel, searchslot);

retry:
    found = false;

    index_rescan(scan, skey, IndexRelationGetNumberOfKeyAttributes(idxrel), NULL, 0);

    /* Try to find the tuple */
    if (index_getnext_slot(scan, ForwardScanDirection, outslot))
    {
        found = true;
        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 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), GetLatestSnapshot(),
                               outslot,
                               GetCurrentCommandId(false),
                               lockmode,
                               LockWaitBlock,
                               0 /* don't follow updates */ ,
                               &tmfd);

        PopActiveSnapshot();

        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")));
                goto retry;
            case TM_Deleted:
                /* XXX: Improve handling here */
                ereport(LOG,
                        (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
                         errmsg("concurrent delete, retrying")));
                goto retry;
            case TM_Invisible:
                elog(ERROR, "attempted to lock invisible tuple");
                break;
            default:
                elog(ERROR, "unexpected table_tuple_lock status: %u", res);
                break;
        }
    }

    index_endscan(scan);

    /* Don't release lock until commit. */
    index_close(idxrel, NoLock);

    return found;
}

RelationFindReplTupleSeq()

bool RelationFindReplTupleSeq	(	Relation	rel,
		LockTupleMode	lockmode,
		TupleTableSlot *	searchslot,
		TupleTableSlot *	outslot
	)

Definition at line 295 of file execReplication.c.

References Assert, TM_FailureData::ctid, elog, equalTupleDescs(), ereport, errcode(), errmsg(), ERROR, ExecCopySlot(), ExecDropSingleTupleTableSlot(), ForwardScanDirection, GetCurrentCommandId(), GetLatestSnapshot(), InitDirtySnapshot, ItemPointerIndicatesMovedPartitions, LockWaitBlock, LOG, TupleDescData::natts, palloc0(), PG_USED_FOR_ASSERTS_ONLY, PopActiveSnapshot(), PushActiveSnapshot(), RelationGetDescr, table_beginscan(), table_endscan(), table_rescan(), table_scan_getnextslot(), table_slot_create(), table_tuple_lock(), TM_Deleted, TM_Invisible, TM_Ok, TM_Updated, TransactionIdIsValid, TupleTableSlot::tts_tid, TupleTableSlot::tts_tupleDescriptor, tuples_equal(), XactLockTableWait(), XLTW_None, SnapshotData::xmax, and SnapshotData::xmin.

Referenced by exec_rt_fetch(), and FindReplTupleInLocalRel().

{
    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(sizeof(*eq) * outslot->tts_tupleDescriptor->natts);

    /* Start a heap scan. */
    InitDirtySnapshot(snap);
    scan = table_beginscan(rel, &snap, 0, NULL);
    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))
            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), GetLatestSnapshot(),
                               outslot,
                               GetCurrentCommandId(false),
                               lockmode,
                               LockWaitBlock,
                               0 /* don't follow updates */ ,
                               &tmfd);

        PopActiveSnapshot();

        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")));
                goto retry;
            case TM_Deleted:
                /* XXX: Improve handling here */
                ereport(LOG,
                        (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
                         errmsg("concurrent delete, retrying")));
                goto retry;
            case TM_Invisible:
                elog(ERROR, "attempted to lock invisible tuple");
                break;
            default:
                elog(ERROR, "unexpected table_tuple_lock status: %u", res);
                break;
        }
    }

    table_endscan(scan);
    ExecDropSingleTupleTableSlot(scanslot);

    return found;
}

tuples_equal()

static bool tuples_equal ( TupleTableSlot *  slot1, TupleTableSlot *  slot2, TypeCacheEntry **  eq  )

static

Definition at line 229 of file execReplication.c.

References Assert, DatumGetBool, TypeCacheEntry::eq_opr_finfo, ereport, errcode(), errmsg(), ERROR, FmgrInfo::fn_oid, format_type_be(), FunctionCall2Coll(), lookup_type_cache(), TupleDescData::natts, OidIsValid, slot_getallattrs(), TupleTableSlot::tts_isnull, TupleTableSlot::tts_tupleDescriptor, TupleTableSlot::tts_values, TupleDescAttr, and TYPECACHE_EQ_OPR_FINFO.

Referenced by RelationFindReplTupleSeq().

{
    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;

        /*
         * 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;

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

        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;
}