From 78b9a8ef04b647e95c8a595a245c96038e53da70 Mon Sep 17 00:00:00 2001 From: amitlan Date: Thu, 10 Dec 2020 20:21:29 +0900 Subject: [PATCH v6 2/2] Avoid using SPI for some RI checks This modifies the subroutines called by RI trigger functions that want to check if a given referenced value exists in the referenced relation to simply scan the foreign key constraint's unique index. That replaces the current way of issuing a `SELECT 1 FROM referenced_relation WHERE ref_key = $1` query through SPI to do the same. This saves a lot of work, especially when inserting into or updating a referencing relation. --- src/backend/utils/adt/ri_triggers.c | 596 ++++++++++++++++++---------- 1 file changed, 380 insertions(+), 216 deletions(-) diff --git a/src/backend/utils/adt/ri_triggers.c b/src/backend/utils/adt/ri_triggers.c index 6e3a41062f..581e283fda 100644 --- a/src/backend/utils/adt/ri_triggers.c +++ b/src/backend/utils/adt/ri_triggers.c @@ -23,22 +23,27 @@ #include "postgres.h" +#include "access/genam.h" #include "access/htup_details.h" +#include "access/skey.h" #include "access/sysattr.h" #include "access/table.h" #include "access/tableam.h" #include "access/xact.h" +#include "catalog/partition.h" #include "catalog/pg_collation.h" #include "catalog/pg_constraint.h" #include "catalog/pg_operator.h" #include "catalog/pg_type.h" #include "commands/trigger.h" +#include "executor/execPartition.h" #include "executor/executor.h" #include "executor/spi.h" #include "lib/ilist.h" #include "miscadmin.h" #include "parser/parse_coerce.h" #include "parser/parse_relation.h" +#include "partitioning/partdesc.h" #include "storage/bufmgr.h" #include "utils/acl.h" #include "utils/builtins.h" @@ -48,6 +53,7 @@ #include "utils/inval.h" #include "utils/lsyscache.h" #include "utils/memutils.h" +#include "utils/partcache.h" #include "utils/rel.h" #include "utils/rls.h" #include "utils/ruleutils.h" @@ -68,16 +74,11 @@ #define RI_KEYS_NONE_NULL 2 /* RI query type codes */ -/* these queries are executed against the PK (referenced) table: */ -#define RI_PLAN_CHECK_LOOKUPPK 1 -#define RI_PLAN_CHECK_LOOKUPPK_FROM_PK 2 -#define RI_PLAN_LAST_ON_PK RI_PLAN_CHECK_LOOKUPPK_FROM_PK -/* these queries are executed against the FK (referencing) table: */ -#define RI_PLAN_CASCADE_DEL_DODELETE 3 -#define RI_PLAN_CASCADE_UPD_DOUPDATE 4 -#define RI_PLAN_RESTRICT_CHECKREF 5 -#define RI_PLAN_SETNULL_DOUPDATE 6 -#define RI_PLAN_SETDEFAULT_DOUPDATE 7 +#define RI_PLAN_CASCADE_DEL_DODELETE 1 +#define RI_PLAN_CASCADE_UPD_DOUPDATE 2 +#define RI_PLAN_RESTRICT_CHECKREF 3 +#define RI_PLAN_SETNULL_DOUPDATE 4 +#define RI_PLAN_SETDEFAULT_DOUPDATE 5 #define MAX_QUOTED_NAME_LEN (NAMEDATALEN*2+3) #define MAX_QUOTED_REL_NAME_LEN (MAX_QUOTED_NAME_LEN*2) @@ -220,8 +221,334 @@ static void ri_ExtractValues(Relation rel, TupleTableSlot *slot, static void ri_ReportViolation(const RI_ConstraintInfo *riinfo, Relation pk_rel, Relation fk_rel, TupleTableSlot *violatorslot, TupleDesc tupdesc, - int queryno, bool partgone) pg_attribute_noreturn(); + bool on_fk, bool partgone) pg_attribute_noreturn(); +static Relation find_leaf_pk_rel(Relation root_pk_rel, const RI_ConstraintInfo *riinfo, + Datum *pk_vals, char *pk_nulls, + Oid root_idxoid, Oid *leaf_idxoid); +/* + * Checks whether a tuple containing the same unique key as extracted from the + * tuple provided in 'slot' exists in 'pk_rel'. The key is extracted using the + * constraint's index given in 'riinfo', which is also scanned to check the + * existence of the key. + * + * If 'pk_rel' is a partitioned table, the check is performed on its leaf + * partition that would contain the key. + * + * The provided tuple is either the one being inserted into the referencing + * relation ('fk_rel' is non-NULL), or the one being deleted from the + * referenced relation, that is, 'pk_rel' ('fk_rel' is NULL). + */ +static bool +ri_ReferencedKeyExists(Relation pk_rel, Relation fk_rel, + TupleTableSlot *slot, + const RI_ConstraintInfo *riinfo) +{ + Oid constr_id = riinfo->constraint_id; + Oid idxoid; + Relation idxrel; + Relation leaf_pk_rel = NULL; + int num_pk; + int i; + bool found; + const Oid *eq_oprs; + Datum pk_vals[INDEX_MAX_KEYS]; + char pk_nulls[INDEX_MAX_KEYS]; + ScanKeyData skey[INDEX_MAX_KEYS]; + IndexScanDesc scan; + TupleTableSlot *outslot; + + /* + * Extract the unique key from the provided slot and choose the equality + * operators to use when scanning the index below. + */ + if (fk_rel) + { + ri_ExtractValues(fk_rel, slot, riinfo, false, pk_vals, pk_nulls); + /* Use PK = FK equality operator. */ + eq_oprs = riinfo->pf_eq_oprs; + + /* + * May neeed to cast each of the individual values of the foreign key + * to the corresponding PK column's type if the equality operator + * demands it. + */ + for (i = 0; i < riinfo->nkeys; i++) + { + if (pk_nulls[i] != 'n') + { + Oid eq_opr = eq_oprs[i]; + Oid typeid = RIAttType(fk_rel, riinfo->fk_attnums[i]); + RI_CompareHashEntry *entry = ri_HashCompareOp(eq_opr, typeid); + + if (OidIsValid(entry->cast_func_finfo.fn_oid)) + pk_vals[i] = FunctionCall3(&entry->cast_func_finfo, + pk_vals[i], + Int32GetDatum(-1), /* typmod */ + BoolGetDatum(false)); /* implicit coercion */ + } + } + } + else + { + ri_ExtractValues(pk_rel, slot, riinfo, true, pk_vals, pk_nulls); + /* Use PK = PK equality operator. */ + eq_oprs = riinfo->pp_eq_oprs; + } + + /* Open the constraint index to be scanned. */ + idxoid = get_constraint_index(constr_id); + + /* Find the leaf partition if needed. */ + if (pk_rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE) + { + Oid leaf_idxoid; + + leaf_pk_rel = find_leaf_pk_rel(pk_rel, riinfo, + pk_vals, pk_nulls, + idxoid, &leaf_idxoid); + + /* + * If no suitable leaf partition exists, neither can the key we're + * looking for. + */ + if (leaf_pk_rel == NULL) + return false; + + pk_rel = leaf_pk_rel; + idxoid = leaf_idxoid; + } + + idxrel = index_open(idxoid, RowShareLock); + num_pk = IndexRelationGetNumberOfKeyAttributes(idxrel); + + /* Set up ScanKeys for the index scan. */ + for (i = 0; i < num_pk; i++) + { + int pkattno = i + 1; + Oid operator = eq_oprs[i]; + RegProcedure regop = get_opcode(operator); + + /* Initialize the scankey. */ + ScanKeyInit(&skey[i], + pkattno, + BTEqualStrategyNumber, + regop, + pk_vals[i]); + + skey[i].sk_collation = idxrel->rd_indcollation[i]; + + /* + * Check for null value. Should not occur, because callers currently + * take care of the cases in which they do occur. + */ + if (pk_nulls[i] == 'n') + skey[i].sk_flags |= SK_ISNULL; + } + + /* + * Start the scan. To make the changes of the current command visible to + * the scan and for subsequent locking of the tuple (if any) found, + * increment the command counter. + */ + CommandCounterIncrement(); + PushActiveSnapshot(GetTransactionSnapshot()); + scan = index_beginscan(pk_rel, idxrel, GetActiveSnapshot(), num_pk, 0); + outslot = table_slot_create(pk_rel, NULL); + + found = false; + index_rescan(scan, skey, num_pk, NULL, 0); + + /* Try to find the tuple */ + if (index_getnext_slot(scan, ForwardScanDirection, outslot)) + found = true; + + /* Found tuple, try to lock it in the lockmode. */ + if (found) + { + TM_FailureData tmfd; + TM_Result res; + int lockflags; + + lockflags = TUPLE_LOCK_FLAG_LOCK_UPDATE_IN_PROGRESS; + if (!IsolationUsesXactSnapshot()) + lockflags |= TUPLE_LOCK_FLAG_FIND_LAST_VERSION; + res = table_tuple_lock(pk_rel, &(outslot->tts_tid), GetActiveSnapshot(), + outslot, + GetCurrentCommandId(false), + LockTupleKeyShare, + LockWaitBlock, + lockflags, + &tmfd); + + switch (res) + { + case TM_Ok: + break; + + case TM_Updated: + case TM_Deleted: + if (IsolationUsesXactSnapshot()) + ereport(ERROR, + (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE), + errmsg("could not serialize access due to concurrent update"))); + found = false; + break; + + case TM_Invisible: + elog(ERROR, "attempted to lock invisible tuple"); + break; + + case TM_SelfModified: + case TM_BeingModified: + case TM_WouldBlock: + elog(ERROR, "unexpected table_tuple_lock status: %u", res); + break; + + default: + elog(ERROR, "unrecognized table_tuple_lock status: %u", res); + } + } + + PopActiveSnapshot(); + index_endscan(scan); + ExecDropSingleTupleTableSlot(outslot); + + /* Don't release lock until commit. */ + index_close(idxrel, NoLock); + + /* Close leaf partition relation if any. */ + if (leaf_pk_rel) + table_close(leaf_pk_rel, NoLock); + + return found; +} + +/* + * Finds the leaf partition of the partitioned relation 'root_pk_rel' that + * might contain the specified unique key. + * + * Returns NULL if no such leaf partition is found. + * + * This works because the unique key defined on the root relation always + * contains the partition key columns of all ancestors leading up to a + * given leaf partition. + */ +static Relation +find_leaf_pk_rel(Relation root_pk_rel, const RI_ConstraintInfo *riinfo, + Datum *pk_vals, char *pk_nulls, + Oid root_idxoid, Oid *leaf_idxoid) +{ + Relation pk_rel = root_pk_rel; + const AttrNumber *pk_attnums = riinfo->pk_attnums; + Oid constr_idxoid = root_idxoid; + + *leaf_idxoid = InvalidOid; + + /* + * Descend through partitioned parents to find the leaf partition that + * would accept a row with the provided key values. + */ + while (true) + { + PartitionKey partkey = RelationGetPartitionKey(pk_rel); + PartitionDesc partdesc = RelationGetPartitionDesc(pk_rel); + Datum partkey_vals[PARTITION_MAX_KEYS]; + bool partkey_isnull[PARTITION_MAX_KEYS]; + AttrNumber *root_partattrs = partkey->partattrs; + int i, + j; + int partidx; + Oid partoid; + + /* + * Collect partition key values from the unique key. + * + * Because we only have the root table's copy of pk_attnums, must map + * any non-root table's partition key attribute numbers to the root + * table's. + */ + if (pk_rel != root_pk_rel) + { + /* + * map->attnums will contain root table attribute numbers for each + * attribute of the current partitioned relation. + */ + AttrMap *map = build_attrmap_by_name_if_req(RelationGetDescr(root_pk_rel), + RelationGetDescr(pk_rel)); + + if (map) + { + root_partattrs = palloc(partkey->partnatts * + sizeof(AttrNumber)); + for (i = 0; i < partkey->partnatts; i++) + { + AttrNumber partattno = partkey->partattrs[i]; + + root_partattrs[i] = map->attnums[partattno - 1]; + } + + free_attrmap(map); + } + } + + /* + * Referenced key specification does not allow expressions, so there + * would not be expressions in the partition keys either. + */ + Assert(partkey->partexprs == NIL); + for (i = 0, j = 0; i < partkey->partnatts; i++) + { + int k; + + for (k = 0; k < riinfo->nkeys; k++) + { + if (root_partattrs[i] == pk_attnums[k]) + { + partkey_vals[j] = pk_vals[k]; + partkey_isnull[j] = (pk_nulls[k] == 'n' ? true : false); + j++; + break; + } + } + } + /* Had better have found values for all of the partition keys. */ + Assert(j == partkey->partnatts); + + if (root_partattrs != partkey->partattrs) + pfree(root_partattrs); + + partidx = get_partition_for_tuple(partkey, partdesc, + partkey_vals, partkey_isnull); + + /* close any intermediate parents we opened */ + if (pk_rel != root_pk_rel) + table_close(pk_rel, NoLock); + + /* No partition found. */ + if (partidx < 0) + return NULL; + + Assert(partidx < partdesc->nparts); + partoid = partdesc->oids[partidx]; + + pk_rel = table_open(partoid, RowShareLock); + constr_idxoid = index_get_partition(pk_rel, constr_idxoid); + + /* + * Return if the partition is a leaf, else find its partition in the + * next iteration. + */ + if (partdesc->is_leaf[partidx]) + { + *leaf_idxoid = constr_idxoid; + return pk_rel; + } + } + + Assert(false); + return NULL; +} /* * RI_FKey_check - @@ -235,8 +562,6 @@ RI_FKey_check(TriggerData *trigdata) Relation fk_rel; Relation pk_rel; TupleTableSlot *newslot; - RI_QueryKey qkey; - SPIPlanPtr qplan; riinfo = ri_FetchConstraintInfo(trigdata->tg_trigger, trigdata->tg_relation, false); @@ -316,9 +641,9 @@ RI_FKey_check(TriggerData *trigdata) /* * MATCH PARTIAL - all non-null columns must match. (not - * implemented, can be done by modifying the query below - * to only include non-null columns, or by writing a - * special version here) + * implemented, can be done by modifying + * ri_ReferencedKeyExists() to only include non-null + * columns. */ break; #endif @@ -333,70 +658,12 @@ RI_FKey_check(TriggerData *trigdata) break; } - if (SPI_connect() != SPI_OK_CONNECT) - elog(ERROR, "SPI_connect failed"); - - /* Fetch or prepare a saved plan for the real check */ - ri_BuildQueryKey(&qkey, riinfo, RI_PLAN_CHECK_LOOKUPPK); - - if ((qplan = ri_FetchPreparedPlan(&qkey)) == NULL) - { - StringInfoData querybuf; - char pkrelname[MAX_QUOTED_REL_NAME_LEN]; - char attname[MAX_QUOTED_NAME_LEN]; - char paramname[16]; - const char *querysep; - Oid queryoids[RI_MAX_NUMKEYS]; - const char *pk_only; - - /* ---------- - * The query string built is - * SELECT 1 FROM [ONLY] x WHERE pkatt1 = $1 [AND ...] - * FOR KEY SHARE OF x - * The type id's for the $ parameters are those of the - * corresponding FK attributes. - * ---------- - */ - initStringInfo(&querybuf); - pk_only = pk_rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE ? - "" : "ONLY "; - quoteRelationName(pkrelname, pk_rel); - appendStringInfo(&querybuf, "SELECT 1 FROM %s%s x", - pk_only, pkrelname); - querysep = "WHERE"; - for (int i = 0; i < riinfo->nkeys; i++) - { - Oid pk_type = RIAttType(pk_rel, riinfo->pk_attnums[i]); - Oid fk_type = RIAttType(fk_rel, riinfo->fk_attnums[i]); - - quoteOneName(attname, - RIAttName(pk_rel, riinfo->pk_attnums[i])); - sprintf(paramname, "$%d", i + 1); - ri_GenerateQual(&querybuf, querysep, - attname, pk_type, - riinfo->pf_eq_oprs[i], - paramname, fk_type); - querysep = "AND"; - queryoids[i] = fk_type; - } - appendStringInfoString(&querybuf, " FOR KEY SHARE OF x"); - - /* Prepare and save the plan */ - qplan = ri_PlanCheck(querybuf.data, riinfo->nkeys, queryoids, - &qkey, fk_rel, pk_rel); - } - - /* - * Now check that foreign key exists in PK table - */ - ri_PerformCheck(riinfo, &qkey, qplan, - fk_rel, pk_rel, - NULL, newslot, - false, - SPI_OK_SELECT); - - if (SPI_finish() != SPI_OK_FINISH) - elog(ERROR, "SPI_finish failed"); + if (!ri_ReferencedKeyExists(pk_rel, fk_rel, newslot, riinfo)) + ri_ReportViolation(riinfo, + pk_rel, fk_rel, + newslot, + NULL, + true, false); table_close(pk_rel, RowShareLock); @@ -451,81 +718,10 @@ ri_Check_Pk_Match(Relation pk_rel, Relation fk_rel, TupleTableSlot *oldslot, const RI_ConstraintInfo *riinfo) { - SPIPlanPtr qplan; - RI_QueryKey qkey; - bool result; - /* Only called for non-null rows */ Assert(ri_NullCheck(RelationGetDescr(pk_rel), oldslot, riinfo, true) == RI_KEYS_NONE_NULL); - if (SPI_connect() != SPI_OK_CONNECT) - elog(ERROR, "SPI_connect failed"); - - /* - * Fetch or prepare a saved plan for checking PK table with values coming - * from a PK row - */ - ri_BuildQueryKey(&qkey, riinfo, RI_PLAN_CHECK_LOOKUPPK_FROM_PK); - - if ((qplan = ri_FetchPreparedPlan(&qkey)) == NULL) - { - StringInfoData querybuf; - char pkrelname[MAX_QUOTED_REL_NAME_LEN]; - char attname[MAX_QUOTED_NAME_LEN]; - char paramname[16]; - const char *querysep; - const char *pk_only; - Oid queryoids[RI_MAX_NUMKEYS]; - - /* ---------- - * The query string built is - * SELECT 1 FROM [ONLY] x WHERE pkatt1 = $1 [AND ...] - * FOR KEY SHARE OF x - * The type id's for the $ parameters are those of the - * PK attributes themselves. - * ---------- - */ - initStringInfo(&querybuf); - pk_only = pk_rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE ? - "" : "ONLY "; - quoteRelationName(pkrelname, pk_rel); - appendStringInfo(&querybuf, "SELECT 1 FROM %s%s x", - pk_only, pkrelname); - querysep = "WHERE"; - for (int i = 0; i < riinfo->nkeys; i++) - { - Oid pk_type = RIAttType(pk_rel, riinfo->pk_attnums[i]); - - quoteOneName(attname, - RIAttName(pk_rel, riinfo->pk_attnums[i])); - sprintf(paramname, "$%d", i + 1); - ri_GenerateQual(&querybuf, querysep, - attname, pk_type, - riinfo->pp_eq_oprs[i], - paramname, pk_type); - querysep = "AND"; - queryoids[i] = pk_type; - } - appendStringInfoString(&querybuf, " FOR KEY SHARE OF x"); - - /* Prepare and save the plan */ - qplan = ri_PlanCheck(querybuf.data, riinfo->nkeys, queryoids, - &qkey, fk_rel, pk_rel); - } - - /* - * We have a plan now. Run it. - */ - result = ri_PerformCheck(riinfo, &qkey, qplan, - fk_rel, pk_rel, - oldslot, NULL, - true, /* treat like update */ - SPI_OK_SELECT); - - if (SPI_finish() != SPI_OK_FINISH) - elog(ERROR, "SPI_finish failed"); - - return result; + return ri_ReferencedKeyExists(pk_rel, NULL, oldslot, riinfo); } @@ -1541,15 +1737,10 @@ RI_Initial_Check(Trigger *trigger, Relation fk_rel, Relation pk_rel) errtableconstraint(fk_rel, NameStr(fake_riinfo.conname)))); - /* - * We tell ri_ReportViolation we were doing the RI_PLAN_CHECK_LOOKUPPK - * query, which isn't true, but will cause it to use - * fake_riinfo.fk_attnums as we need. - */ ri_ReportViolation(&fake_riinfo, pk_rel, fk_rel, slot, tupdesc, - RI_PLAN_CHECK_LOOKUPPK, false); + true, false); ExecDropSingleTupleTableSlot(slot); } @@ -1766,7 +1957,7 @@ RI_PartitionRemove_Check(Trigger *trigger, Relation fk_rel, Relation pk_rel) fake_riinfo.pk_attnums[i] = i + 1; ri_ReportViolation(&fake_riinfo, pk_rel, fk_rel, - slot, tupdesc, 0, true); + slot, tupdesc, true, true); } if (SPI_finish() != SPI_OK_FINISH) @@ -2136,19 +2327,11 @@ ri_PlanCheck(const char *querystr, int nargs, Oid *argtypes, RI_QueryKey *qkey, Relation fk_rel, Relation pk_rel) { SPIPlanPtr qplan; - Relation query_rel; + /* There are currently no queries that run on PK table. */ + Relation query_rel = fk_rel; Oid save_userid; int save_sec_context; - /* - * Use the query type code to determine whether the query is run against - * the PK or FK table; we'll do the check as that table's owner - */ - if (qkey->constr_queryno <= RI_PLAN_LAST_ON_PK) - query_rel = pk_rel; - else - query_rel = fk_rel; - /* Switch to proper UID to perform check as */ GetUserIdAndSecContext(&save_userid, &save_sec_context); SetUserIdAndSecContext(RelationGetForm(query_rel)->relowner, @@ -2181,9 +2364,10 @@ ri_PerformCheck(const RI_ConstraintInfo *riinfo, TupleTableSlot *oldslot, TupleTableSlot *newslot, bool detectNewRows, int expect_OK) { - Relation query_rel, - source_rel; - bool source_is_pk; + /* There are currently no queries that run on PK table. */ + Relation query_rel = fk_rel, + source_rel = pk_rel; + bool source_is_pk = true; Snapshot test_snapshot; Snapshot crosscheck_snapshot; int limit; @@ -2193,33 +2377,6 @@ ri_PerformCheck(const RI_ConstraintInfo *riinfo, Datum vals[RI_MAX_NUMKEYS * 2]; char nulls[RI_MAX_NUMKEYS * 2]; - /* - * Use the query type code to determine whether the query is run against - * the PK or FK table; we'll do the check as that table's owner - */ - if (qkey->constr_queryno <= RI_PLAN_LAST_ON_PK) - query_rel = pk_rel; - else - query_rel = fk_rel; - - /* - * The values for the query are taken from the table on which the trigger - * is called - it is normally the other one with respect to query_rel. An - * exception is ri_Check_Pk_Match(), which uses the PK table for both (and - * sets queryno to RI_PLAN_CHECK_LOOKUPPK_FROM_PK). We might eventually - * need some less klugy way to determine this. - */ - if (qkey->constr_queryno == RI_PLAN_CHECK_LOOKUPPK) - { - source_rel = fk_rel; - source_is_pk = false; - } - else - { - source_rel = pk_rel; - source_is_pk = true; - } - /* Extract the parameters to be passed into the query */ if (newslot) { @@ -2296,14 +2453,12 @@ ri_PerformCheck(const RI_ConstraintInfo *riinfo, errhint("This is most likely due to a rule having rewritten the query."))); /* XXX wouldn't it be clearer to do this part at the caller? */ - if (qkey->constr_queryno != RI_PLAN_CHECK_LOOKUPPK_FROM_PK && - expect_OK == SPI_OK_SELECT && - (SPI_processed == 0) == (qkey->constr_queryno == RI_PLAN_CHECK_LOOKUPPK)) + if (expect_OK == SPI_OK_SELECT && SPI_processed != 0) ri_ReportViolation(riinfo, pk_rel, fk_rel, newslot ? newslot : oldslot, NULL, - qkey->constr_queryno, false); + false, false); return SPI_processed != 0; } @@ -2334,9 +2489,9 @@ ri_ExtractValues(Relation rel, TupleTableSlot *slot, /* * Produce an error report * - * If the failed constraint was on insert/update to the FK table, - * we want the key names and values extracted from there, and the error - * message to look like 'key blah is not present in PK'. + * If the failed constraint was on insert/update to the FK table (on_fk is + * true), we want the key names and values extracted from there, and the + * error message to look like 'key blah is not present in PK'. * Otherwise, the attr names and values come from the PK table and the * message looks like 'key blah is still referenced from FK'. */ @@ -2344,22 +2499,20 @@ static void ri_ReportViolation(const RI_ConstraintInfo *riinfo, Relation pk_rel, Relation fk_rel, TupleTableSlot *violatorslot, TupleDesc tupdesc, - int queryno, bool partgone) + bool on_fk, bool partgone) { StringInfoData key_names; StringInfoData key_values; - bool onfk; const int16 *attnums; Oid rel_oid; AclResult aclresult; bool has_perm = true; /* - * Determine which relation to complain about. If tupdesc wasn't passed - * by caller, assume the violator tuple came from there. + * If tupdesc wasn't passed by caller, assume the violator tuple came from + * there. */ - onfk = (queryno == RI_PLAN_CHECK_LOOKUPPK); - if (onfk) + if (on_fk) { attnums = riinfo->fk_attnums; rel_oid = fk_rel->rd_id; @@ -2461,7 +2614,7 @@ ri_ReportViolation(const RI_ConstraintInfo *riinfo, key_names.data, key_values.data, RelationGetRelationName(fk_rel)), errtableconstraint(fk_rel, NameStr(riinfo->conname)))); - else if (onfk) + else if (on_fk) ereport(ERROR, (errcode(ERRCODE_FOREIGN_KEY_VIOLATION), errmsg("insert or update on table \"%s\" violates foreign key constraint \"%s\"", @@ -2768,7 +2921,10 @@ ri_AttributesEqual(Oid eq_opr, Oid typeid, * ri_HashCompareOp - * * See if we know how to compare two values, and create a new hash entry - * if not. + * if not. The entry contains the FmgrInfo of the equality operator function + * and that of the cast function, if one is needed to convert the right + * operand (whose type OID has been passed) before passing it to the equality + * function. */ static RI_CompareHashEntry * ri_HashCompareOp(Oid eq_opr, Oid typeid) @@ -2824,8 +2980,16 @@ ri_HashCompareOp(Oid eq_opr, Oid typeid) * moment since that will never be generated for implicit coercions. */ op_input_types(eq_opr, &lefttype, &righttype); - Assert(lefttype == righttype); - if (typeid == lefttype) + + /* + * Don't need to cast if the values that will be passed to the + * operator will be of expected operand type(s). The operator can be + * cross-type (such as when called by ri_ReferencedKeyExists()), in + * which case, we only need the cast if the right operand value + * doesn't match the type expected by the operator. + */ + if ((lefttype == righttype && typeid == lefttype) || + (lefttype != righttype && typeid == righttype)) castfunc = InvalidOid; /* simplest case */ else { -- 2.24.1