At Thu, 3 Dec 2020 16:41:45 +0900, Amit Langote <amitlangot...@gmail.com> wrote
in
> On Thu, Dec 3, 2020 at 2:29 PM Kyotaro Horiguchi
> <horikyota....@gmail.com> wrote:
> > At Thu, 3 Dec 2020 12:27:53 +0900, Amit Langote <amitlangot...@gmail.com>
> > wrote in
> > > On Thu, Dec 3, 2020 at 10:15 AM Kyotaro Horiguchi
> > > <horikyota....@gmail.com> wrote:
> > > For the queries on the referencing side ("check" side),
> > > type/collation/attribute name determined using the above are going to
> > > be the same for all partitions in a given tree irrespective of the
> > > attribute number, because they're logically the same column. On the
> >
> > Yes, I know that, which is what I meant by "practically" or
> > "actually", but it is not explicitly defined AFAICS.
>
> Well, I think it's great that we don't have to worry *in this part of
> the code* about partition's fk_attnums not being congruent with the
> root parent's, because ensuring that is the responsibility of the
> other parts of the system such as DDL. If we have any problems in
> this area, they should be dealt with by ensuring that there are no
> bugs in those other parts.
Agreed.
> > Thus that would be no longer an issue if we explicitly define that
> > "When conpraentid stores a valid value, each element of fk_attnums
> > points to logically the same attribute with the RI_ConstraintInfo for
> > the parent constraint." Or I'd be happy if we have such a comment
> > there instead.
>
> I saw a comment in Kuroda-san's v2 patch that is perhaps meant to
> address this point, but the placement needs to be reconsidered:
Ah, yes, that comes from my proposal.
> @@ -366,6 +368,14 @@ RI_FKey_check(TriggerData *trigdata)
> querysep = "WHERE";
> for (int i = 0; i < riinfo->nkeys; i++)
> {
> +
> + /*
> + * We share the same plan among all relations in a partition
> + * hierarchy. The plan is guaranteed to be compatible since all of
> + * the member relations are guaranteed to have the equivalent set
> + * of foreign keys in fk_attnums[].
> + */
> +
> Oid pk_type = RIAttType(pk_rel, riinfo->pk_attnums[i]);
> Oid fk_type = RIAttType(fk_rel, riinfo->fk_attnums[i]);
>
> A more appropriate place for this kind of comment would be where
> fk_attnums is defined or in ri_BuildQueryKey() that is shared by
> different RI query issuing functions.
Yeah, I wanted more appropriate place for the comment. That place
seems reasonable.
> > > referenced side ("restrict", "cascade", "set" side), as you already
> > > mentioned, fk_attnums refers to the top parent table of the
> > > referencing side, so no possibility of they being different in the
> > > various referenced partitions' RI_ConstraintInfos.
> >
> > Right. (I'm not sure I have mention that here, though:p)A
>
> Maybe I misread but I think you did in your email dated Dec 1 where you said:
>
> "After an off-list discussion, we confirmed that even in that case the
> patch works as is because fk_attnum (or contuple.conkey) always stores
> key attnums compatible to the topmost parent when conparent has a
> valid value (assuming the current usage of fk_attnum), but I still
> feel uneasy to rely on that unclear behavior."
fk_attnums *doesn't* refers to to top parent talbe of the referencing
side. it refers to attributes of the partition that is compatible with
the same element of fk_attnums of the topmost parent. Maybe I'm
misreading.
> > > On the topic of how we'd be able to share even the RI_ConstraintInfos
> > > among partitions, that would indeed look a bit more elaborate than the
> > > patch we have right now.
> >
> > Maybe just letting the hash entry for the child riinfo point to the
> > parent riinfo if all members (other than constraint_id, of course)
> > share the exactly the same values. No need to count references since
> > we don't going to remove riinfos.
>
> Ah, something maybe worth trying. Although the memory we'd save by
> sharing the RI_ConstraintInfos would not add that much to the savings
> we're having by sharing the plan, because it's the plans that are a
> memory hog AFAIK.
I agree that plans are rather large but the sharable part of the
RI_ConstraintInfos is 536 bytes, I'm not sure it is small enough
comparing to the plans. But that has somewhat large footprint.. (See
the attached)
> > > > About your patch, it calculates the root constrid at the time an
> > > > riinfo is created, but when the root-partition is further attached to
> > > > another partitioned-table after the riinfo creation,
> > > > constraint_root_id gets stale. Of course that dones't matter
> > > > practically, though.
> > >
> > > Maybe we could also store the hash value of the root constraint OID as
> > > rootHashValue and check for that one too in
> > > InvalidateConstraintCacheCallBack(). That would take care of this
> > > unless I'm missing something.
> >
> > Seems to be sound.
>
> Okay, thanks.
>
> I have attached a patch in which I've tried to merge the ideas from
> both my patch and Kuroda-san's. I liked that his patch added
> conparentid to RI_ConstraintInfo because that saves a needless
> syscache lookup for constraints that don't have a parent. I've kept
> my idea to compute the root constraint id only once in
> ri_LoadConstraint(), not on every invocation of ri_BuildQueryKey().
> Kuroda-san, anything you'd like to add to that?
regards.
--
Kyotaro Horiguchi
NTT Open Source Software Center
>From bbdd0647452a26fdf7da313967f85241c6111fcb Mon Sep 17 00:00:00 2001
From: Kyotaro Horiguchi <horikyoga....@gmail.com>
Date: Thu, 3 Dec 2020 16:15:57 +0900
Subject: [PATCH 1/2] separte riinfo
---
src/backend/utils/adt/ri_triggers.c | 281 ++++++++++++++++------------
1 file changed, 161 insertions(+), 120 deletions(-)
diff --git a/src/backend/utils/adt/ri_triggers.c b/src/backend/utils/adt/ri_triggers.c
index 02b1a3868f..3f8407c311 100644
--- a/src/backend/utils/adt/ri_triggers.c
+++ b/src/backend/utils/adt/ri_triggers.c
@@ -97,14 +97,10 @@
* Information extracted from an FK pg_constraint entry. This is cached in
* ri_constraint_cache.
*/
-typedef struct RI_ConstraintInfo
+typedef struct RI_ConstraintParam
{
- Oid constraint_id; /* OID of pg_constraint entry (hash key) */
- bool valid; /* successfully initialized? */
- uint32 oidHashValue; /* hash value of pg_constraint OID */
- NameData conname; /* name of the FK constraint */
+ Oid query_key;
Oid pk_relid; /* referenced relation */
- Oid fk_relid; /* referencing relation */
char confupdtype; /* foreign key's ON UPDATE action */
char confdeltype; /* foreign key's ON DELETE action */
char confmatchtype; /* foreign key's match type */
@@ -114,6 +110,17 @@ typedef struct RI_ConstraintInfo
Oid pf_eq_oprs[RI_MAX_NUMKEYS]; /* equality operators (PK = FK) */
Oid pp_eq_oprs[RI_MAX_NUMKEYS]; /* equality operators (PK = PK) */
Oid ff_eq_oprs[RI_MAX_NUMKEYS]; /* equality operators (FK = FK) */
+ struct RI_ConstraintInfo *ownerinfo; /* owner info of this param */
+} RI_ConstraintParam;
+
+typedef struct RI_ConstraintInfo
+{
+ Oid constraint_id;
+ bool valid; /* successfully initialized? */
+ uint32 oidHashValue; /* hash value of pg_constraint OID */
+ NameData conname; /* name of the FK constraint */
+ Oid fk_relid; /* referencing relation */
+ RI_ConstraintParam *param; /* sharable parameters */
dlist_node valid_link; /* Link in list of valid entries */
} RI_ConstraintInfo;
@@ -264,7 +271,7 @@ RI_FKey_check(TriggerData *trigdata)
* SELECT FOR KEY SHARE will get on it.
*/
fk_rel = trigdata->tg_relation;
- pk_rel = table_open(riinfo->pk_relid, RowShareLock);
+ pk_rel = table_open(riinfo->param->pk_relid, RowShareLock);
switch (ri_NullCheck(RelationGetDescr(fk_rel), newslot, riinfo, false))
{
@@ -283,7 +290,7 @@ RI_FKey_check(TriggerData *trigdata)
* This is the only case that differs between the three kinds of
* MATCH.
*/
- switch (riinfo->confmatchtype)
+ switch (riinfo->param->confmatchtype)
{
case FKCONSTR_MATCH_FULL:
@@ -364,17 +371,17 @@ RI_FKey_check(TriggerData *trigdata)
appendStringInfo(&querybuf, "SELECT 1 FROM %s%s x",
pk_only, pkrelname);
querysep = "WHERE";
- for (int i = 0; i < riinfo->nkeys; i++)
+ for (int i = 0; i < riinfo->param->nkeys; i++)
{
- Oid pk_type = RIAttType(pk_rel, riinfo->pk_attnums[i]);
- Oid fk_type = RIAttType(fk_rel, riinfo->fk_attnums[i]);
+ Oid pk_type = RIAttType(pk_rel, riinfo->param->pk_attnums[i]);
+ Oid fk_type = RIAttType(fk_rel, riinfo->param->fk_attnums[i]);
quoteOneName(attname,
- RIAttName(pk_rel, riinfo->pk_attnums[i]));
+ RIAttName(pk_rel, riinfo->param->pk_attnums[i]));
sprintf(paramname, "$%d", i + 1);
ri_GenerateQual(&querybuf, querysep,
attname, pk_type,
- riinfo->pf_eq_oprs[i],
+ riinfo->param->pf_eq_oprs[i],
paramname, fk_type);
querysep = "AND";
queryoids[i] = fk_type;
@@ -382,7 +389,7 @@ RI_FKey_check(TriggerData *trigdata)
appendStringInfoString(&querybuf, " FOR KEY SHARE OF x");
/* Prepare and save the plan */
- qplan = ri_PlanCheck(querybuf.data, riinfo->nkeys, queryoids,
+ qplan = ri_PlanCheck(querybuf.data, riinfo->param->nkeys, queryoids,
&qkey, fk_rel, pk_rel);
}
@@ -492,16 +499,16 @@ ri_Check_Pk_Match(Relation pk_rel, Relation fk_rel,
appendStringInfo(&querybuf, "SELECT 1 FROM %s%s x",
pk_only, pkrelname);
querysep = "WHERE";
- for (int i = 0; i < riinfo->nkeys; i++)
+ for (int i = 0; i < riinfo->param->nkeys; i++)
{
- Oid pk_type = RIAttType(pk_rel, riinfo->pk_attnums[i]);
+ Oid pk_type = RIAttType(pk_rel, riinfo->param->pk_attnums[i]);
quoteOneName(attname,
- RIAttName(pk_rel, riinfo->pk_attnums[i]));
+ RIAttName(pk_rel, riinfo->param->pk_attnums[i]));
sprintf(paramname, "$%d", i + 1);
ri_GenerateQual(&querybuf, querysep,
attname, pk_type,
- riinfo->pp_eq_oprs[i],
+ riinfo->param->pp_eq_oprs[i],
paramname, pk_type);
querysep = "AND";
queryoids[i] = pk_type;
@@ -509,7 +516,7 @@ ri_Check_Pk_Match(Relation pk_rel, Relation fk_rel,
appendStringInfoString(&querybuf, " FOR KEY SHARE OF x");
/* Prepare and save the plan */
- qplan = ri_PlanCheck(querybuf.data, riinfo->nkeys, queryoids,
+ qplan = ri_PlanCheck(querybuf.data, riinfo->param->nkeys, queryoids,
&qkey, fk_rel, pk_rel);
}
@@ -679,19 +686,19 @@ ri_restrict(TriggerData *trigdata, bool is_no_action)
appendStringInfo(&querybuf, "SELECT 1 FROM %s%s x",
fk_only, fkrelname);
querysep = "WHERE";
- for (int i = 0; i < riinfo->nkeys; i++)
+ for (int i = 0; i < riinfo->param->nkeys; i++)
{
- Oid pk_type = RIAttType(pk_rel, riinfo->pk_attnums[i]);
- Oid fk_type = RIAttType(fk_rel, riinfo->fk_attnums[i]);
- Oid pk_coll = RIAttCollation(pk_rel, riinfo->pk_attnums[i]);
- Oid fk_coll = RIAttCollation(fk_rel, riinfo->fk_attnums[i]);
+ Oid pk_type = RIAttType(pk_rel, riinfo->param->pk_attnums[i]);
+ Oid fk_type = RIAttType(fk_rel, riinfo->param->fk_attnums[i]);
+ Oid pk_coll = RIAttCollation(pk_rel, riinfo->param->pk_attnums[i]);
+ Oid fk_coll = RIAttCollation(fk_rel, riinfo->param->fk_attnums[i]);
quoteOneName(attname,
- RIAttName(fk_rel, riinfo->fk_attnums[i]));
+ RIAttName(fk_rel, riinfo->param->fk_attnums[i]));
sprintf(paramname, "$%d", i + 1);
ri_GenerateQual(&querybuf, querysep,
paramname, pk_type,
- riinfo->pf_eq_oprs[i],
+ riinfo->param->pf_eq_oprs[i],
attname, fk_type);
if (pk_coll != fk_coll && !get_collation_isdeterministic(pk_coll))
ri_GenerateQualCollation(&querybuf, pk_coll);
@@ -701,7 +708,7 @@ ri_restrict(TriggerData *trigdata, bool is_no_action)
appendStringInfoString(&querybuf, " FOR KEY SHARE OF x");
/* Prepare and save the plan */
- qplan = ri_PlanCheck(querybuf.data, riinfo->nkeys, queryoids,
+ qplan = ri_PlanCheck(querybuf.data, riinfo->param->nkeys, queryoids,
&qkey, fk_rel, pk_rel);
}
@@ -785,19 +792,19 @@ RI_FKey_cascade_del(PG_FUNCTION_ARGS)
appendStringInfo(&querybuf, "DELETE FROM %s%s",
fk_only, fkrelname);
querysep = "WHERE";
- for (int i = 0; i < riinfo->nkeys; i++)
+ for (int i = 0; i < riinfo->param->nkeys; i++)
{
- Oid pk_type = RIAttType(pk_rel, riinfo->pk_attnums[i]);
- Oid fk_type = RIAttType(fk_rel, riinfo->fk_attnums[i]);
- Oid pk_coll = RIAttCollation(pk_rel, riinfo->pk_attnums[i]);
- Oid fk_coll = RIAttCollation(fk_rel, riinfo->fk_attnums[i]);
+ Oid pk_type = RIAttType(pk_rel, riinfo->param->pk_attnums[i]);
+ Oid fk_type = RIAttType(fk_rel, riinfo->param->fk_attnums[i]);
+ Oid pk_coll = RIAttCollation(pk_rel, riinfo->param->pk_attnums[i]);
+ Oid fk_coll = RIAttCollation(fk_rel, riinfo->param->fk_attnums[i]);
quoteOneName(attname,
- RIAttName(fk_rel, riinfo->fk_attnums[i]));
+ RIAttName(fk_rel, riinfo->param->fk_attnums[i]));
sprintf(paramname, "$%d", i + 1);
ri_GenerateQual(&querybuf, querysep,
paramname, pk_type,
- riinfo->pf_eq_oprs[i],
+ riinfo->param->pf_eq_oprs[i],
attname, fk_type);
if (pk_coll != fk_coll && !get_collation_isdeterministic(pk_coll))
ri_GenerateQualCollation(&querybuf, pk_coll);
@@ -806,7 +813,7 @@ RI_FKey_cascade_del(PG_FUNCTION_ARGS)
}
/* Prepare and save the plan */
- qplan = ri_PlanCheck(querybuf.data, riinfo->nkeys, queryoids,
+ qplan = ri_PlanCheck(querybuf.data, riinfo->param->nkeys, queryoids,
&qkey, fk_rel, pk_rel);
}
@@ -901,22 +908,22 @@ RI_FKey_cascade_upd(PG_FUNCTION_ARGS)
fk_only, fkrelname);
querysep = "";
qualsep = "WHERE";
- for (int i = 0, j = riinfo->nkeys; i < riinfo->nkeys; i++, j++)
+ for (int i = 0, j = riinfo->param->nkeys; i < riinfo->param->nkeys; i++, j++)
{
- Oid pk_type = RIAttType(pk_rel, riinfo->pk_attnums[i]);
- Oid fk_type = RIAttType(fk_rel, riinfo->fk_attnums[i]);
- Oid pk_coll = RIAttCollation(pk_rel, riinfo->pk_attnums[i]);
- Oid fk_coll = RIAttCollation(fk_rel, riinfo->fk_attnums[i]);
+ Oid pk_type = RIAttType(pk_rel, riinfo->param->pk_attnums[i]);
+ Oid fk_type = RIAttType(fk_rel, riinfo->param->fk_attnums[i]);
+ Oid pk_coll = RIAttCollation(pk_rel, riinfo->param->pk_attnums[i]);
+ Oid fk_coll = RIAttCollation(fk_rel, riinfo->param->fk_attnums[i]);
quoteOneName(attname,
- RIAttName(fk_rel, riinfo->fk_attnums[i]));
+ RIAttName(fk_rel, riinfo->param->fk_attnums[i]));
appendStringInfo(&querybuf,
"%s %s = $%d",
querysep, attname, i + 1);
sprintf(paramname, "$%d", j + 1);
ri_GenerateQual(&qualbuf, qualsep,
paramname, pk_type,
- riinfo->pf_eq_oprs[i],
+ riinfo->param->pf_eq_oprs[i],
attname, fk_type);
if (pk_coll != fk_coll && !get_collation_isdeterministic(pk_coll))
ri_GenerateQualCollation(&querybuf, pk_coll);
@@ -928,7 +935,7 @@ RI_FKey_cascade_upd(PG_FUNCTION_ARGS)
appendBinaryStringInfo(&querybuf, qualbuf.data, qualbuf.len);
/* Prepare and save the plan */
- qplan = ri_PlanCheck(querybuf.data, riinfo->nkeys * 2, queryoids,
+ qplan = ri_PlanCheck(querybuf.data, riinfo->param->nkeys * 2, queryoids,
&qkey, fk_rel, pk_rel);
}
@@ -1080,15 +1087,15 @@ ri_set(TriggerData *trigdata, bool is_set_null)
fk_only, fkrelname);
querysep = "";
qualsep = "WHERE";
- for (int i = 0; i < riinfo->nkeys; i++)
+ for (int i = 0; i < riinfo->param->nkeys; i++)
{
- Oid pk_type = RIAttType(pk_rel, riinfo->pk_attnums[i]);
- Oid fk_type = RIAttType(fk_rel, riinfo->fk_attnums[i]);
- Oid pk_coll = RIAttCollation(pk_rel, riinfo->pk_attnums[i]);
- Oid fk_coll = RIAttCollation(fk_rel, riinfo->fk_attnums[i]);
+ Oid pk_type = RIAttType(pk_rel, riinfo->param->pk_attnums[i]);
+ Oid fk_type = RIAttType(fk_rel, riinfo->param->fk_attnums[i]);
+ Oid pk_coll = RIAttCollation(pk_rel, riinfo->param->pk_attnums[i]);
+ Oid fk_coll = RIAttCollation(fk_rel, riinfo->param->fk_attnums[i]);
quoteOneName(attname,
- RIAttName(fk_rel, riinfo->fk_attnums[i]));
+ RIAttName(fk_rel, riinfo->param->fk_attnums[i]));
appendStringInfo(&querybuf,
"%s %s = %s",
querysep, attname,
@@ -1096,7 +1103,7 @@ ri_set(TriggerData *trigdata, bool is_set_null)
sprintf(paramname, "$%d", i + 1);
ri_GenerateQual(&qualbuf, qualsep,
paramname, pk_type,
- riinfo->pf_eq_oprs[i],
+ riinfo->param->pf_eq_oprs[i],
attname, fk_type);
if (pk_coll != fk_coll && !get_collation_isdeterministic(pk_coll))
ri_GenerateQualCollation(&querybuf, pk_coll);
@@ -1107,7 +1114,7 @@ ri_set(TriggerData *trigdata, bool is_set_null)
appendBinaryStringInfo(&querybuf, qualbuf.data, qualbuf.len);
/* Prepare and save the plan */
- qplan = ri_PlanCheck(querybuf.data, riinfo->nkeys, queryoids,
+ qplan = ri_PlanCheck(querybuf.data, riinfo->param->nkeys, queryoids,
&qkey, fk_rel, pk_rel);
}
@@ -1217,7 +1224,7 @@ RI_FKey_fk_upd_check_required(Trigger *trigger, Relation fk_rel,
*/
else if (ri_nullcheck == RI_KEYS_SOME_NULL)
{
- switch (riinfo->confmatchtype)
+ switch (riinfo->param->confmatchtype)
{
case FKCONSTR_MATCH_SIMPLE:
@@ -1333,14 +1340,14 @@ RI_Initial_Check(Trigger *trigger, Relation fk_rel, Relation pk_rel)
fkrte->rellockmode = AccessShareLock;
fkrte->requiredPerms = ACL_SELECT;
- for (int i = 0; i < riinfo->nkeys; i++)
+ for (int i = 0; i < riinfo->param->nkeys; i++)
{
int attno;
- attno = riinfo->pk_attnums[i] - FirstLowInvalidHeapAttributeNumber;
+ attno = riinfo->param->pk_attnums[i] - FirstLowInvalidHeapAttributeNumber;
pkrte->selectedCols = bms_add_member(pkrte->selectedCols, attno);
- attno = riinfo->fk_attnums[i] - FirstLowInvalidHeapAttributeNumber;
+ attno = riinfo->param->fk_attnums[i] - FirstLowInvalidHeapAttributeNumber;
fkrte->selectedCols = bms_add_member(fkrte->selectedCols, attno);
}
@@ -1377,10 +1384,10 @@ RI_Initial_Check(Trigger *trigger, Relation fk_rel, Relation pk_rel)
initStringInfo(&querybuf);
appendStringInfoString(&querybuf, "SELECT ");
sep = "";
- for (int i = 0; i < riinfo->nkeys; i++)
+ for (int i = 0; i < riinfo->param->nkeys; i++)
{
quoteOneName(fkattname,
- RIAttName(fk_rel, riinfo->fk_attnums[i]));
+ RIAttName(fk_rel, riinfo->param->fk_attnums[i]));
appendStringInfo(&querybuf, "%sfk.%s", sep, fkattname);
sep = ", ";
}
@@ -1398,20 +1405,20 @@ RI_Initial_Check(Trigger *trigger, Relation fk_rel, Relation pk_rel)
strcpy(pkattname, "pk.");
strcpy(fkattname, "fk.");
sep = "(";
- for (int i = 0; i < riinfo->nkeys; i++)
+ for (int i = 0; i < riinfo->param->nkeys; i++)
{
- Oid pk_type = RIAttType(pk_rel, riinfo->pk_attnums[i]);
- Oid fk_type = RIAttType(fk_rel, riinfo->fk_attnums[i]);
- Oid pk_coll = RIAttCollation(pk_rel, riinfo->pk_attnums[i]);
- Oid fk_coll = RIAttCollation(fk_rel, riinfo->fk_attnums[i]);
+ Oid pk_type = RIAttType(pk_rel, riinfo->param->pk_attnums[i]);
+ Oid fk_type = RIAttType(fk_rel, riinfo->param->fk_attnums[i]);
+ Oid pk_coll = RIAttCollation(pk_rel, riinfo->param->pk_attnums[i]);
+ Oid fk_coll = RIAttCollation(fk_rel, riinfo->param->fk_attnums[i]);
quoteOneName(pkattname + 3,
- RIAttName(pk_rel, riinfo->pk_attnums[i]));
+ RIAttName(pk_rel, riinfo->param->pk_attnums[i]));
quoteOneName(fkattname + 3,
- RIAttName(fk_rel, riinfo->fk_attnums[i]));
+ RIAttName(fk_rel, riinfo->param->fk_attnums[i]));
ri_GenerateQual(&querybuf, sep,
pkattname, pk_type,
- riinfo->pf_eq_oprs[i],
+ riinfo->param->pf_eq_oprs[i],
fkattname, fk_type);
if (pk_coll != fk_coll)
ri_GenerateQualCollation(&querybuf, pk_coll);
@@ -1422,17 +1429,17 @@ RI_Initial_Check(Trigger *trigger, Relation fk_rel, Relation pk_rel)
* It's sufficient to test any one pk attribute for null to detect a join
* failure.
*/
- quoteOneName(pkattname, RIAttName(pk_rel, riinfo->pk_attnums[0]));
+ quoteOneName(pkattname, RIAttName(pk_rel, riinfo->param->pk_attnums[0]));
appendStringInfo(&querybuf, ") WHERE pk.%s IS NULL AND (", pkattname);
sep = "";
- for (int i = 0; i < riinfo->nkeys; i++)
+ for (int i = 0; i < riinfo->param->nkeys; i++)
{
- quoteOneName(fkattname, RIAttName(fk_rel, riinfo->fk_attnums[i]));
+ quoteOneName(fkattname, RIAttName(fk_rel, riinfo->param->fk_attnums[i]));
appendStringInfo(&querybuf,
"%sfk.%s IS NOT NULL",
sep, fkattname);
- switch (riinfo->confmatchtype)
+ switch (riinfo->param->confmatchtype)
{
case FKCONSTR_MATCH_SIMPLE:
sep = " AND ";
@@ -1505,7 +1512,10 @@ RI_Initial_Check(Trigger *trigger, Relation fk_rel, Relation pk_rel)
HeapTuple tuple = SPI_tuptable->vals[0];
TupleDesc tupdesc = SPI_tuptable->tupdesc;
RI_ConstraintInfo fake_riinfo;
+ RI_ConstraintParam fake_riparam;
+ fake_riinfo.param = &fake_riparam;
+
slot = MakeSingleTupleTableSlot(tupdesc, &TTSOpsVirtual);
heap_deform_tuple(tuple, tupdesc,
@@ -1522,15 +1532,15 @@ RI_Initial_Check(Trigger *trigger, Relation fk_rel, Relation pk_rel)
* or fk_rel's tupdesc.
*/
memcpy(&fake_riinfo, riinfo, sizeof(RI_ConstraintInfo));
- for (int i = 0; i < fake_riinfo.nkeys; i++)
- fake_riinfo.fk_attnums[i] = i + 1;
+ for (int i = 0; i < fake_riinfo.param->nkeys; i++)
+ fake_riinfo.param->fk_attnums[i] = i + 1;
/*
* If it's MATCH FULL, and there are any nulls in the FK keys,
* complain about that rather than the lack of a match. MATCH FULL
* disallows partially-null FK rows.
*/
- if (fake_riinfo.confmatchtype == FKCONSTR_MATCH_FULL &&
+ if (fake_riinfo.param->confmatchtype == FKCONSTR_MATCH_FULL &&
ri_NullCheck(tupdesc, slot, &fake_riinfo, false) != RI_KEYS_NONE_NULL)
ereport(ERROR,
(errcode(ERRCODE_FOREIGN_KEY_VIOLATION),
@@ -1615,10 +1625,10 @@ RI_PartitionRemove_Check(Trigger *trigger, Relation fk_rel, Relation pk_rel)
initStringInfo(&querybuf);
appendStringInfoString(&querybuf, "SELECT ");
sep = "";
- for (i = 0; i < riinfo->nkeys; i++)
+ for (i = 0; i < riinfo->param->nkeys; i++)
{
quoteOneName(fkattname,
- RIAttName(fk_rel, riinfo->fk_attnums[i]));
+ RIAttName(fk_rel, riinfo->param->fk_attnums[i]));
appendStringInfo(&querybuf, "%sfk.%s", sep, fkattname);
sep = ", ";
}
@@ -1633,20 +1643,20 @@ RI_PartitionRemove_Check(Trigger *trigger, Relation fk_rel, Relation pk_rel)
strcpy(pkattname, "pk.");
strcpy(fkattname, "fk.");
sep = "(";
- for (i = 0; i < riinfo->nkeys; i++)
+ for (i = 0; i < riinfo->param->nkeys; i++)
{
- Oid pk_type = RIAttType(pk_rel, riinfo->pk_attnums[i]);
- Oid fk_type = RIAttType(fk_rel, riinfo->fk_attnums[i]);
- Oid pk_coll = RIAttCollation(pk_rel, riinfo->pk_attnums[i]);
- Oid fk_coll = RIAttCollation(fk_rel, riinfo->fk_attnums[i]);
+ Oid pk_type = RIAttType(pk_rel, riinfo->param->pk_attnums[i]);
+ Oid fk_type = RIAttType(fk_rel, riinfo->param->fk_attnums[i]);
+ Oid pk_coll = RIAttCollation(pk_rel, riinfo->param->pk_attnums[i]);
+ Oid fk_coll = RIAttCollation(fk_rel, riinfo->param->fk_attnums[i]);
quoteOneName(pkattname + 3,
- RIAttName(pk_rel, riinfo->pk_attnums[i]));
+ RIAttName(pk_rel, riinfo->param->pk_attnums[i]));
quoteOneName(fkattname + 3,
- RIAttName(fk_rel, riinfo->fk_attnums[i]));
+ RIAttName(fk_rel, riinfo->param->fk_attnums[i]));
ri_GenerateQual(&querybuf, sep,
pkattname, pk_type,
- riinfo->pf_eq_oprs[i],
+ riinfo->param->pf_eq_oprs[i],
fkattname, fk_type);
if (pk_coll != fk_coll)
ri_GenerateQualCollation(&querybuf, pk_coll);
@@ -1666,13 +1676,13 @@ RI_PartitionRemove_Check(Trigger *trigger, Relation fk_rel, Relation pk_rel)
appendStringInfoString(&querybuf, ") WHERE (");
sep = "";
- for (i = 0; i < riinfo->nkeys; i++)
+ for (i = 0; i < riinfo->param->nkeys; i++)
{
- quoteOneName(fkattname, RIAttName(fk_rel, riinfo->fk_attnums[i]));
+ quoteOneName(fkattname, RIAttName(fk_rel, riinfo->param->fk_attnums[i]));
appendStringInfo(&querybuf,
"%sfk.%s IS NOT NULL",
sep, fkattname);
- switch (riinfo->confmatchtype)
+ switch (riinfo->param->confmatchtype)
{
case FKCONSTR_MATCH_SIMPLE:
sep = " AND ";
@@ -1762,8 +1772,8 @@ RI_PartitionRemove_Check(Trigger *trigger, Relation fk_rel, Relation pk_rel)
* or fk_rel's tupdesc.
*/
memcpy(&fake_riinfo, riinfo, sizeof(RI_ConstraintInfo));
- for (i = 0; i < fake_riinfo.nkeys; i++)
- fake_riinfo.pk_attnums[i] = i + 1;
+ for (i = 0; i < fake_riinfo.param->nkeys; i++)
+ fake_riinfo.param->pk_attnums[i] = i + 1;
ri_ReportViolation(&fake_riinfo, pk_rel, fk_rel,
slot, tupdesc, 0, true);
@@ -1905,7 +1915,7 @@ ri_BuildQueryKey(RI_QueryKey *key, const RI_ConstraintInfo *riinfo,
* We assume struct RI_QueryKey contains no padding bytes, else we'd need
* to use memset to clear them.
*/
- key->constr_id = riinfo->constraint_id;
+ key->constr_id = riinfo->param->query_key;
key->constr_queryno = constr_queryno;
}
@@ -1983,25 +1993,25 @@ ri_FetchConstraintInfo(Trigger *trigger, Relation trig_rel, bool rel_is_pk)
if (rel_is_pk)
{
if (riinfo->fk_relid != trigger->tgconstrrelid ||
- riinfo->pk_relid != RelationGetRelid(trig_rel))
+ riinfo->param->pk_relid != RelationGetRelid(trig_rel))
elog(ERROR, "wrong pg_constraint entry for trigger \"%s\" on table \"%s\"",
trigger->tgname, RelationGetRelationName(trig_rel));
}
else
{
if (riinfo->fk_relid != RelationGetRelid(trig_rel) ||
- riinfo->pk_relid != trigger->tgconstrrelid)
+ riinfo->param->pk_relid != trigger->tgconstrrelid)
elog(ERROR, "wrong pg_constraint entry for trigger \"%s\" on table \"%s\"",
trigger->tgname, RelationGetRelationName(trig_rel));
}
- if (riinfo->confmatchtype != FKCONSTR_MATCH_FULL &&
- riinfo->confmatchtype != FKCONSTR_MATCH_PARTIAL &&
- riinfo->confmatchtype != FKCONSTR_MATCH_SIMPLE)
+ if (riinfo->param->confmatchtype != FKCONSTR_MATCH_FULL &&
+ riinfo->param->confmatchtype != FKCONSTR_MATCH_PARTIAL &&
+ riinfo->param->confmatchtype != FKCONSTR_MATCH_SIMPLE)
elog(ERROR, "unrecognized confmatchtype: %d",
- riinfo->confmatchtype);
+ riinfo->param->confmatchtype);
- if (riinfo->confmatchtype == FKCONSTR_MATCH_PARTIAL)
+ if (riinfo->param->confmatchtype == FKCONSTR_MATCH_PARTIAL)
ereport(ERROR,
(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
errmsg("MATCH PARTIAL not yet implemented")));
@@ -2009,6 +2019,27 @@ ri_FetchConstraintInfo(Trigger *trigger, Relation trig_rel, bool rel_is_pk)
return riinfo;
}
+/*
+ * get_ri_constraint_root
+ * Returns a given RI constraint's root parent
+ */
+static Oid
+get_ri_constraint_root(Oid constrOid)
+{
+ HeapTuple tuple;
+ Oid constrParentOid;
+
+ tuple = SearchSysCache1(CONSTROID, ObjectIdGetDatum(constrOid));
+ if (!HeapTupleIsValid(tuple))
+ elog(ERROR, "cache lookup failed for constraint %u", constrOid);
+ constrParentOid = ((Form_pg_constraint) GETSTRUCT(tuple))->conparentid;
+ ReleaseSysCache(tuple);
+ if (OidIsValid(constrParentOid))
+ return get_ri_constraint_root(constrParentOid);
+
+ return constrOid;
+}
+
/*
* Fetch or create the RI_ConstraintInfo struct for an FK constraint.
*/
@@ -2032,11 +2063,20 @@ ri_LoadConstraintInfo(Oid constraintOid)
riinfo = (RI_ConstraintInfo *) hash_search(ri_constraint_cache,
(void *) &constraintOid,
HASH_ENTER, &found);
+
if (!found)
riinfo->valid = false;
else if (riinfo->valid)
return riinfo;
+ /* allocate riinfo's own param memory if needed */
+ if (!found || riinfo->param->ownerinfo != riinfo)
+ {
+ riinfo->param = (RI_ConstraintParam *)
+ MemoryContextAlloc(TopMemoryContext, sizeof(RI_ConstraintParam));
+ riinfo->param->ownerinfo = riinfo;
+ }
+
/*
* Fetch the pg_constraint row so we can fill in the entry.
*/
@@ -2051,22 +2091,23 @@ ri_LoadConstraintInfo(Oid constraintOid)
/* And extract data */
Assert(riinfo->constraint_id == constraintOid);
+ riinfo->param->query_key = get_ri_constraint_root(riinfo->constraint_id);
riinfo->oidHashValue = GetSysCacheHashValue1(CONSTROID,
ObjectIdGetDatum(constraintOid));
memcpy(&riinfo->conname, &conForm->conname, sizeof(NameData));
- riinfo->pk_relid = conForm->confrelid;
+ riinfo->param->pk_relid = conForm->confrelid;
riinfo->fk_relid = conForm->conrelid;
- riinfo->confupdtype = conForm->confupdtype;
- riinfo->confdeltype = conForm->confdeltype;
- riinfo->confmatchtype = conForm->confmatchtype;
+ riinfo->param->confupdtype = conForm->confupdtype;
+ riinfo->param->confdeltype = conForm->confdeltype;
+ riinfo->param->confmatchtype = conForm->confmatchtype;
DeconstructFkConstraintRow(tup,
- &riinfo->nkeys,
- riinfo->fk_attnums,
- riinfo->pk_attnums,
- riinfo->pf_eq_oprs,
- riinfo->pp_eq_oprs,
- riinfo->ff_eq_oprs);
+ &riinfo->param->nkeys,
+ riinfo->param->fk_attnums,
+ riinfo->param->pk_attnums,
+ riinfo->param->pf_eq_oprs,
+ riinfo->param->pp_eq_oprs,
+ riinfo->param->ff_eq_oprs);
ReleaseSysCache(tup);
@@ -2227,7 +2268,7 @@ ri_PerformCheck(const RI_ConstraintInfo *riinfo,
vals, nulls);
if (oldslot)
ri_ExtractValues(source_rel, oldslot, riinfo, source_is_pk,
- vals + riinfo->nkeys, nulls + riinfo->nkeys);
+ vals + riinfo->param->nkeys, nulls + riinfo->param->nkeys);
}
else
{
@@ -2320,11 +2361,11 @@ ri_ExtractValues(Relation rel, TupleTableSlot *slot,
bool isnull;
if (rel_is_pk)
- attnums = riinfo->pk_attnums;
+ attnums = riinfo->param->pk_attnums;
else
- attnums = riinfo->fk_attnums;
+ attnums = riinfo->param->fk_attnums;
- for (int i = 0; i < riinfo->nkeys; i++)
+ for (int i = 0; i < riinfo->param->nkeys; i++)
{
vals[i] = slot_getattr(slot, attnums[i], &isnull);
nulls[i] = isnull ? 'n' : ' ';
@@ -2361,14 +2402,14 @@ ri_ReportViolation(const RI_ConstraintInfo *riinfo,
onfk = (queryno == RI_PLAN_CHECK_LOOKUPPK);
if (onfk)
{
- attnums = riinfo->fk_attnums;
+ attnums = riinfo->param->fk_attnums;
rel_oid = fk_rel->rd_id;
if (tupdesc == NULL)
tupdesc = fk_rel->rd_att;
}
else
{
- attnums = riinfo->pk_attnums;
+ attnums = riinfo->param->pk_attnums;
rel_oid = pk_rel->rd_id;
if (tupdesc == NULL)
tupdesc = pk_rel->rd_att;
@@ -2395,7 +2436,7 @@ ri_ReportViolation(const RI_ConstraintInfo *riinfo,
if (aclresult != ACLCHECK_OK)
{
/* Try for column-level permissions */
- for (int idx = 0; idx < riinfo->nkeys; idx++)
+ for (int idx = 0; idx < riinfo->param->nkeys; idx++)
{
aclresult = pg_attribute_aclcheck(rel_oid, attnums[idx],
GetUserId(),
@@ -2418,7 +2459,7 @@ ri_ReportViolation(const RI_ConstraintInfo *riinfo,
/* Get printable versions of the keys involved */
initStringInfo(&key_names);
initStringInfo(&key_values);
- for (int idx = 0; idx < riinfo->nkeys; idx++)
+ for (int idx = 0; idx < riinfo->param->nkeys; idx++)
{
int fnum = attnums[idx];
Form_pg_attribute att = TupleDescAttr(tupdesc, fnum - 1);
@@ -2508,11 +2549,11 @@ ri_NullCheck(TupleDesc tupDesc,
bool nonenull = true;
if (rel_is_pk)
- attnums = riinfo->pk_attnums;
+ attnums = riinfo->param->pk_attnums;
else
- attnums = riinfo->fk_attnums;
+ attnums = riinfo->param->fk_attnums;
- for (int i = 0; i < riinfo->nkeys; i++)
+ for (int i = 0; i < riinfo->param->nkeys; i++)
{
if (slot_attisnull(slot, attnums[i]))
nonenull = false;
@@ -2667,12 +2708,12 @@ ri_KeysEqual(Relation rel, TupleTableSlot *oldslot, TupleTableSlot *newslot,
const int16 *attnums;
if (rel_is_pk)
- attnums = riinfo->pk_attnums;
+ attnums = riinfo->param->pk_attnums;
else
- attnums = riinfo->fk_attnums;
+ attnums = riinfo->param->fk_attnums;
/* XXX: could be worthwhile to fetch all necessary attrs at once */
- for (int i = 0; i < riinfo->nkeys; i++)
+ for (int i = 0; i < riinfo->param->nkeys; i++)
{
Datum oldvalue;
Datum newvalue;
@@ -2714,7 +2755,7 @@ ri_KeysEqual(Relation rel, TupleTableSlot *oldslot, TupleTableSlot *newslot,
* operator. Changes that compare equal will still satisfy the
* constraint after the update.
*/
- if (!ri_AttributesEqual(riinfo->ff_eq_oprs[i], RIAttType(rel, attnums[i]),
+ if (!ri_AttributesEqual(riinfo->param->ff_eq_oprs[i], RIAttType(rel, attnums[i]),
oldvalue, newvalue))
return false;
}
--
2.18.4
>From ae7f7ab7df353044b8b2878253d1bdb8adbcd054 Mon Sep 17 00:00:00 2001
From: Kyotaro Horiguchi <horikyoga....@gmail.com>
Date: Thu, 3 Dec 2020 16:51:07 +0900
Subject: [PATCH 2/2] share param part of riinfo
---
src/backend/utils/adt/ri_triggers.c | 33 +++++++++++++++++++++++++++++
1 file changed, 33 insertions(+)
diff --git a/src/backend/utils/adt/ri_triggers.c b/src/backend/utils/adt/ri_triggers.c
index 3f8407c311..080e8af1a5 100644
--- a/src/backend/utils/adt/ri_triggers.c
+++ b/src/backend/utils/adt/ri_triggers.c
@@ -110,6 +110,8 @@ typedef struct RI_ConstraintParam
Oid pf_eq_oprs[RI_MAX_NUMKEYS]; /* equality operators (PK = FK) */
Oid pp_eq_oprs[RI_MAX_NUMKEYS]; /* equality operators (PK = PK) */
Oid ff_eq_oprs[RI_MAX_NUMKEYS]; /* equality operators (FK = FK) */
+
+ /* This should follow the aboves, see ri_LoadConstraintInfo */
struct RI_ConstraintInfo *ownerinfo; /* owner info of this param */
} RI_ConstraintParam;
@@ -2111,6 +2113,37 @@ ri_LoadConstraintInfo(Oid constraintOid)
ReleaseSysCache(tup);
+ if (OidIsValid(conForm->conparentid))
+ {
+ Oid pconid = conForm->conparentid;
+ const RI_ConstraintInfo *priinfo = ri_LoadConstraintInfo(pconid);
+ RI_ConstraintParam *p = riinfo->param;
+ RI_ConstraintParam *pp = priinfo->param;
+
+ /* share the same parameters if identical */
+ if (memcmp(p, pp, offsetof(RI_ConstraintParam, pk_attnums)) == 0)
+ {
+ int i;
+
+ for (i = 0 ; i < p->nkeys ; i++)
+ {
+ if (p->pk_attnums[i] != pp->pk_attnums[i] ||
+ p->fk_attnums[i] != pp->fk_attnums[i] ||
+ p->pf_eq_oprs[i] != pp->pf_eq_oprs[i] ||
+ p->pp_eq_oprs[i] != pp->pp_eq_oprs[i] ||
+ p->ff_eq_oprs[i] != pp->ff_eq_oprs[i])
+ break;
+ }
+ if (i == p->nkeys)
+ {
+ /* this riinfo can share the parameters with parent */
+ Assert(p->ownerinfo == riinfo);
+ pfree(riinfo->param);
+ riinfo->param = pp;
+ }
+ }
+ }
+
/*
* For efficient processing of invalidation messages below, we keep a
* doubly-linked list, and a count, of all currently valid entries.
--
2.18.4
