After having reviewed [1] more than a year ago (the problem I found was that
the transient table is not available for deferred constraints), I've tried to
implement the same in an alternative way. The RI triggers still work as row
level triggers, but if multiple events of the same kind appear in the queue,
they are all passed to the trigger function at once. Thus the check query does
not have to be executed that frequently.
Some performance comparisons are below. (Besides the execution time, please
note the difference in the number of trigger function executions.) In general,
the checks are significantly faster if there are many rows to process, and a
bit slower when we only need to check a single row. However I'm not sure about
the accuracy if only a single row is measured (if a single row check is
performed several times, the execution time appears to fluctuate).
Comments are welcome.
Setup
=====
CREATE TABLE p(i int primary key);
INSERT INTO p SELECT x FROM generate_series(1, 16384) g(x);
CREATE TABLE f(i int REFERENCES p);
Insert many rows into the FK table
==================================
master:
EXPLAIN ANALYZE INSERT INTO f SELECT i FROM generate_series(1, 16384) g(i);
QUERY PLAN
--------------------------------------------------------------------------------------------------------------------------------
Insert on f (cost=0.00..163.84 rows=16384 width=4) (actual
time=32.741..32.741 rows=0 loops=1)
-> Function Scan on generate_series g (cost=0.00..163.84 rows=16384
width=4) (actual time=2.403..4.802 rows=16384 loops=1)
Planning Time: 0.050 ms
Trigger for constraint f_i_fkey: time=448.986 calls=16384
Execution Time: 485.444 ms
(5 rows)
patched:
EXPLAIN ANALYZE INSERT INTO f SELECT i FROM generate_series(1, 16384) g(i);
QUERY PLAN
--------------------------------------------------------------------------------------------------------------------------------
Insert on f (cost=0.00..163.84 rows=16384 width=4) (actual
time=34.053..34.053 rows=0 loops=1)
-> Function Scan on generate_series g (cost=0.00..163.84 rows=16384
width=4) (actual time=2.223..4.448 rows=16384 loops=1)
Planning Time: 0.047 ms
Trigger for constraint f_i_fkey: time=105.164 calls=8
Execution Time: 141.201 ms
Insert a single row into the FK table
=====================================
master:
EXPLAIN ANALYZE INSERT INTO f VALUES (1);
QUERY PLAN
------------------------------------------------------------------------------------------
Insert on f (cost=0.00..0.01 rows=1 width=4) (actual time=0.060..0.060 rows=0
loops=1)
-> Result (cost=0.00..0.01 rows=1 width=4) (actual time=0.002..0.002
rows=1 loops=1)
Planning Time: 0.026 ms
Trigger for constraint f_i_fkey: time=0.435 calls=1
Execution Time: 0.517 ms
(5 rows)
patched:
EXPLAIN ANALYZE INSERT INTO f VALUES (1);
QUERY PLAN
------------------------------------------------------------------------------------------
Insert on f (cost=0.00..0.01 rows=1 width=4) (actual time=0.066..0.066 rows=0
loops=1)
-> Result (cost=0.00..0.01 rows=1 width=4) (actual time=0.002..0.002
rows=1 loops=1)
Planning Time: 0.025 ms
Trigger for constraint f_i_fkey: time=0.578 calls=1
Execution Time: 0.670 ms
Check if FK row exists during deletion from the PK
==================================================
master:
DELETE FROM p WHERE i=16384;
ERROR: update or delete on table "p" violates foreign key constraint
"f_i_fkey" on table "f"
DETAIL: Key (i)=(16384) is still referenced from table "f".
Time: 3.381 ms
patched:
DELETE FROM p WHERE i=16384;
ERROR: update or delete on table "p" violates foreign key constraint
"f_i_fkey" on table "f"
DETAIL: Key (i)=(16384) is still referenced from table "f".
Time: 5.561 ms
Cascaded DELETE --- many PK rows
================================
DROP TABLE f;
CREATE TABLE f(i int REFERENCES p ON UPDATE CASCADE ON DELETE CASCADE);
INSERT INTO f SELECT i FROM generate_series(1, 16384) g(i);
master:
EXPLAIN ANALYZE DELETE FROM p;
QUERY PLAN
-----------------------------------------------------------------------------------------------------------
Delete on p (cost=0.00..236.84 rows=16384 width=6) (actual
time=38.334..38.334 rows=0 loops=1)
-> Seq Scan on p (cost=0.00..236.84 rows=16384 width=6) (actual
time=0.019..3.925 rows=16384 loops=1)
Planning Time: 0.049 ms
Trigger for constraint f_i_fkey: time=31348.756 calls=16384
Execution Time: 31390.784 ms
patched:
EXPLAIN ANALYZE DELETE FROM p;
QUERY PLAN
-----------------------------------------------------------------------------------------------------------
Delete on p (cost=0.00..236.84 rows=16384 width=6) (actual
time=33.360..33.360 rows=0 loops=1)
-> Seq Scan on p (cost=0.00..236.84 rows=16384 width=6) (actual
time=0.012..3.183 rows=16384 loops=1)
Planning Time: 0.094 ms
Trigger for constraint f_i_fkey: time=9.580 calls=8
Execution Time: 43.941 ms
Cascaded DELETE --- a single PK row
===================================
INSERT INTO p SELECT x FROM generate_series(1, 16384) g(x);
INSERT INTO f SELECT i FROM generate_series(1, 16384) g(i);
master:
DELETE FROM p WHERE i=16384;
DELETE 1
Time: 5.754 ms
patched:
DELETE FROM p WHERE i=16384;
DELETE 1
Time: 8.098 ms
Cascaded UPDATE - many rows
===========================
master:
EXPLAIN ANALYZE UPDATE p SET i = i + 16384;
QUERY PLAN
------------------------------------------------------------------------------------------------------------
Update on p (cost=0.00..277.80 rows=16384 width=10) (actual
time=166.954..166.954 rows=0 loops=1)
-> Seq Scan on p (cost=0.00..277.80 rows=16384 width=10) (actual
time=0.013..7.780 rows=16384 loops=1)
Planning Time: 0.177 ms
Trigger for constraint f_i_fkey on p: time=60405.362 calls=16384
Trigger for constraint f_i_fkey on f: time=455.874 calls=16384
Execution Time: 61036.996 ms
patched:
EXPLAIN ANALYZE UPDATE p SET i = i + 16384;
QUERY PLAN
------------------------------------------------------------------------------------------------------------
Update on p (cost=0.00..277.77 rows=16382 width=10) (actual
time=159.512..159.512 rows=0 loops=1)
-> Seq Scan on p (cost=0.00..277.77 rows=16382 width=10) (actual
time=0.014..7.783 rows=16382 loops=1)
Planning Time: 0.146 ms
Trigger for constraint f_i_fkey on p: time=169.628 calls=9
Trigger for constraint f_i_fkey on f: time=124.079 calls=2
Execution Time: 456.072 ms
Cascaded UPDATE - a single row
==============================
master:
UPDATE p SET i = i - 16384 WHERE i=32767;
UPDATE 1
Time: 4.858 ms
patched:
UPDATE p SET i = i - 16384 WHERE i=32767;
UPDATE 1
Time: 11.955 ms
[1] https://commitfest.postgresql.org/22/1975/
--
Antonin Houska
Web: https://www.cybertec-postgresql.com
>From 35606a7e4e66f3279f52be941b0b9bce29d73de3 Mon Sep 17 00:00:00 2001
From: Antonin Houska <a...@cybertec.at>
Date: Wed, 8 Apr 2020 15:03:20 +0200
Subject: [PATCH 1/4] Check for RI violation outside ri_PerformCheck().
---
src/backend/utils/adt/ri_triggers.c | 40 ++++++++++++++---------------
1 file changed, 20 insertions(+), 20 deletions(-)
diff --git a/src/backend/utils/adt/ri_triggers.c b/src/backend/utils/adt/ri_triggers.c
index bb49e80d16..6220872126 100644
--- a/src/backend/utils/adt/ri_triggers.c
+++ b/src/backend/utils/adt/ri_triggers.c
@@ -389,11 +389,16 @@ RI_FKey_check(TriggerData *trigdata)
/*
* 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 (!ri_PerformCheck(riinfo, &qkey, qplan,
+ fk_rel, pk_rel,
+ NULL, newslot,
+ false,
+ SPI_OK_SELECT))
+ ri_ReportViolation(riinfo,
+ pk_rel, fk_rel,
+ newslot,
+ NULL,
+ qkey.constr_queryno, false);
if (SPI_finish() != SPI_OK_FINISH)
elog(ERROR, "SPI_finish failed");
@@ -708,11 +713,16 @@ ri_restrict(TriggerData *trigdata, bool is_no_action)
/*
* We have a plan now. Run it to check for existing references.
*/
- ri_PerformCheck(riinfo, &qkey, qplan,
- fk_rel, pk_rel,
- oldslot, NULL,
- true, /* must detect new rows */
- SPI_OK_SELECT);
+ if (ri_PerformCheck(riinfo, &qkey, qplan,
+ fk_rel, pk_rel,
+ oldslot, NULL,
+ true, /* must detect new rows */
+ SPI_OK_SELECT))
+ ri_ReportViolation(riinfo,
+ pk_rel, fk_rel,
+ oldslot,
+ NULL,
+ qkey.constr_queryno, false);
if (SPI_finish() != SPI_OK_FINISH)
elog(ERROR, "SPI_finish failed");
@@ -2288,16 +2298,6 @@ ri_PerformCheck(const RI_ConstraintInfo *riinfo,
RelationGetRelationName(fk_rel)),
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))
- ri_ReportViolation(riinfo,
- pk_rel, fk_rel,
- newslot ? newslot : oldslot,
- NULL,
- qkey->constr_queryno, false);
-
return SPI_processed != 0;
}
--
2.20.1
>From 58926a4546b3918af8f6e6691956731d8c902701 Mon Sep 17 00:00:00 2001
From: Antonin Houska <a...@cybertec.at>
Date: Wed, 8 Apr 2020 15:03:20 +0200
Subject: [PATCH 2/4] Changed ri_GenerateQual() so it generates the whole
qualifier.
This way we can use the function to reduce the amount of copy&pasted code a
bit.
---
src/backend/utils/adt/ri_triggers.c | 288 +++++++++++++++-------------
1 file changed, 159 insertions(+), 129 deletions(-)
diff --git a/src/backend/utils/adt/ri_triggers.c b/src/backend/utils/adt/ri_triggers.c
index 6220872126..3bedb75846 100644
--- a/src/backend/utils/adt/ri_triggers.c
+++ b/src/backend/utils/adt/ri_triggers.c
@@ -180,11 +180,17 @@ static Datum ri_restrict(TriggerData *trigdata, bool is_no_action);
static Datum ri_set(TriggerData *trigdata, bool is_set_null);
static void quoteOneName(char *buffer, const char *name);
static void quoteRelationName(char *buffer, Relation rel);
-static void ri_GenerateQual(StringInfo buf,
- const char *sep,
- const char *leftop, Oid leftoptype,
- Oid opoid,
- const char *rightop, Oid rightoptype);
+static char *ri_ColNameQuoted(const char *tabname, const char *attname);
+static void ri_GenerateQual(StringInfo buf, char *sep, int nkeys,
+ const char *ltabname, Relation lrel,
+ const int16 *lattnums,
+ const char *rtabname, Relation rrel,
+ const int16 *rattnums, const Oid *eq_oprs);
+static void ri_GenerateQualComponent(StringInfo buf,
+ const char *sep,
+ const char *leftop, Oid leftoptype,
+ Oid opoid,
+ const char *rightop, Oid rightoptype);
static void ri_GenerateQualCollation(StringInfo buf, Oid collation);
static int ri_NullCheck(TupleDesc tupdesc, TupleTableSlot *slot,
const RI_ConstraintInfo *riinfo, bool rel_is_pk);
@@ -372,10 +378,10 @@ RI_FKey_check(TriggerData *trigdata)
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);
+ ri_GenerateQualComponent(&querybuf, querysep,
+ attname, pk_type,
+ riinfo->pf_eq_oprs[i],
+ paramname, fk_type);
querysep = "AND";
queryoids[i] = fk_type;
}
@@ -504,10 +510,10 @@ ri_Check_Pk_Match(Relation pk_rel, Relation fk_rel,
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);
+ ri_GenerateQualComponent(&querybuf, querysep,
+ attname, pk_type,
+ riinfo->pp_eq_oprs[i],
+ paramname, pk_type);
querysep = "AND";
queryoids[i] = pk_type;
}
@@ -694,10 +700,10 @@ ri_restrict(TriggerData *trigdata, bool is_no_action)
quoteOneName(attname,
RIAttName(fk_rel, riinfo->fk_attnums[i]));
sprintf(paramname, "$%d", i + 1);
- ri_GenerateQual(&querybuf, querysep,
- paramname, pk_type,
- riinfo->pf_eq_oprs[i],
- attname, fk_type);
+ ri_GenerateQualComponent(&querybuf, querysep,
+ paramname, pk_type,
+ riinfo->pf_eq_oprs[i],
+ attname, fk_type);
if (pk_coll != fk_coll && !get_collation_isdeterministic(pk_coll))
ri_GenerateQualCollation(&querybuf, pk_coll);
querysep = "AND";
@@ -805,10 +811,10 @@ RI_FKey_cascade_del(PG_FUNCTION_ARGS)
quoteOneName(attname,
RIAttName(fk_rel, riinfo->fk_attnums[i]));
sprintf(paramname, "$%d", i + 1);
- ri_GenerateQual(&querybuf, querysep,
- paramname, pk_type,
- riinfo->pf_eq_oprs[i],
- attname, fk_type);
+ ri_GenerateQualComponent(&querybuf, querysep,
+ paramname, pk_type,
+ riinfo->pf_eq_oprs[i],
+ attname, fk_type);
if (pk_coll != fk_coll && !get_collation_isdeterministic(pk_coll))
ri_GenerateQualCollation(&querybuf, pk_coll);
querysep = "AND";
@@ -924,10 +930,10 @@ RI_FKey_cascade_upd(PG_FUNCTION_ARGS)
"%s %s = $%d",
querysep, attname, i + 1);
sprintf(paramname, "$%d", j + 1);
- ri_GenerateQual(&qualbuf, qualsep,
- paramname, pk_type,
- riinfo->pf_eq_oprs[i],
- attname, fk_type);
+ ri_GenerateQualComponent(&qualbuf, qualsep,
+ paramname, pk_type,
+ riinfo->pf_eq_oprs[i],
+ attname, fk_type);
if (pk_coll != fk_coll && !get_collation_isdeterministic(pk_coll))
ri_GenerateQualCollation(&querybuf, pk_coll);
querysep = ",";
@@ -1104,10 +1110,10 @@ ri_set(TriggerData *trigdata, bool is_set_null)
querysep, attname,
is_set_null ? "NULL" : "DEFAULT");
sprintf(paramname, "$%d", i + 1);
- ri_GenerateQual(&qualbuf, qualsep,
- paramname, pk_type,
- riinfo->pf_eq_oprs[i],
- attname, fk_type);
+ ri_GenerateQualComponent(&qualbuf, qualsep,
+ paramname, pk_type,
+ riinfo->pf_eq_oprs[i],
+ attname, fk_type);
if (pk_coll != fk_coll && !get_collation_isdeterministic(pk_coll))
ri_GenerateQualCollation(&querybuf, pk_coll);
querysep = ",";
@@ -1402,31 +1408,13 @@ RI_Initial_Check(Trigger *trigger, Relation fk_rel, Relation pk_rel)
pk_only = pk_rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE ?
"" : "ONLY ";
appendStringInfo(&querybuf,
- " FROM %s%s fk LEFT OUTER JOIN %s%s pk ON",
+ " FROM %s%s fk LEFT OUTER JOIN %s%s pk ON (",
fk_only, fkrelname, pk_only, pkrelname);
- strcpy(pkattname, "pk.");
- strcpy(fkattname, "fk.");
- sep = "(";
- 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]);
- Oid pk_coll = RIAttCollation(pk_rel, riinfo->pk_attnums[i]);
- Oid fk_coll = RIAttCollation(fk_rel, riinfo->fk_attnums[i]);
-
- quoteOneName(pkattname + 3,
- RIAttName(pk_rel, riinfo->pk_attnums[i]));
- quoteOneName(fkattname + 3,
- RIAttName(fk_rel, riinfo->fk_attnums[i]));
- ri_GenerateQual(&querybuf, sep,
- pkattname, pk_type,
- riinfo->pf_eq_oprs[i],
- fkattname, fk_type);
- if (pk_coll != fk_coll)
- ri_GenerateQualCollation(&querybuf, pk_coll);
- sep = "AND";
- }
+ ri_GenerateQual(&querybuf, "AND", riinfo->nkeys,
+ "pk", pk_rel, riinfo->pk_attnums,
+ "fk", fk_rel, riinfo->fk_attnums,
+ riinfo->pf_eq_oprs);
/*
* It's sufficient to test any one pk attribute for null to detect a join
@@ -1584,7 +1572,6 @@ RI_PartitionRemove_Check(Trigger *trigger, Relation fk_rel, Relation pk_rel)
char *constraintDef;
char pkrelname[MAX_QUOTED_REL_NAME_LEN];
char fkrelname[MAX_QUOTED_REL_NAME_LEN];
- char pkattname[MAX_QUOTED_NAME_LEN + 3];
char fkattname[MAX_QUOTED_NAME_LEN + 3];
const char *sep;
const char *fk_only;
@@ -1633,30 +1620,13 @@ RI_PartitionRemove_Check(Trigger *trigger, Relation fk_rel, Relation pk_rel)
fk_only = fk_rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE ?
"" : "ONLY ";
appendStringInfo(&querybuf,
- " FROM %s%s fk JOIN %s pk ON",
+ " FROM %s%s fk JOIN %s pk ON (",
fk_only, fkrelname, pkrelname);
- strcpy(pkattname, "pk.");
- strcpy(fkattname, "fk.");
- sep = "(";
- for (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]);
- Oid pk_coll = RIAttCollation(pk_rel, riinfo->pk_attnums[i]);
- Oid fk_coll = RIAttCollation(fk_rel, riinfo->fk_attnums[i]);
-
- quoteOneName(pkattname + 3,
- RIAttName(pk_rel, riinfo->pk_attnums[i]));
- quoteOneName(fkattname + 3,
- RIAttName(fk_rel, riinfo->fk_attnums[i]));
- ri_GenerateQual(&querybuf, sep,
- pkattname, pk_type,
- riinfo->pf_eq_oprs[i],
- fkattname, fk_type);
- if (pk_coll != fk_coll)
- ri_GenerateQualCollation(&querybuf, pk_coll);
- sep = "AND";
- }
+
+ ri_GenerateQual(&querybuf, "AND", riinfo->nkeys,
+ "pk", pk_rel, riinfo->pk_attnums,
+ "fk", fk_rel, riinfo->fk_attnums,
+ riinfo->pf_eq_oprs);
/*
* Start the WHERE clause with the partition constraint (except if this is
@@ -1820,7 +1790,44 @@ quoteRelationName(char *buffer, Relation rel)
}
/*
- * ri_GenerateQual --- generate a WHERE clause equating two variables
+ * ri_GenerateQual --- generate WHERE/ON clause.
+ *
+ * Note: to avoid unnecessary explicit casts, make sure that the left and
+ * right operands match eq_oprs expect (ie don't swap the left and right
+ * operands accidentally).
+ */
+static void
+ri_GenerateQual(StringInfo buf, char *sep, int nkeys,
+ const char *ltabname, Relation lrel,
+ const int16 *lattnums,
+ const char *rtabname, Relation rrel,
+ const int16 *rattnums,
+ const Oid *eq_oprs)
+{
+ for (int i = 0; i < nkeys; i++)
+ {
+ Oid ltype = RIAttType(lrel, lattnums[i]);
+ Oid rtype = RIAttType(rrel, rattnums[i]);
+ Oid lcoll = RIAttCollation(lrel, lattnums[i]);
+ Oid rcoll = RIAttCollation(rrel, rattnums[i]);
+ char *latt,
+ *ratt;
+ char *sep_current = i > 0 ? sep : NULL;
+
+ latt = ri_ColNameQuoted(ltabname, RIAttName(lrel, lattnums[i]));
+ ratt = ri_ColNameQuoted(rtabname, RIAttName(rrel, rattnums[i]));
+
+ ri_GenerateQualComponent(buf, sep_current, latt, ltype, eq_oprs[i],
+ ratt, rtype);
+
+ if (lcoll != rcoll)
+ ri_GenerateQualCollation(buf, lcoll);
+ }
+}
+
+/*
+ * ri_GenerateQual --- generate a component of WHERE/ON clause equating two
+ * variables, to be AND-ed to the other components.
*
* This basically appends " sep leftop op rightop" to buf, adding casts
* and schema qualification as needed to ensure that the parser will select
@@ -1828,17 +1835,86 @@ quoteRelationName(char *buffer, Relation rel)
* if they aren't variables or parameters.
*/
static void
-ri_GenerateQual(StringInfo buf,
- const char *sep,
- const char *leftop, Oid leftoptype,
- Oid opoid,
- const char *rightop, Oid rightoptype)
+ri_GenerateQualComponent(StringInfo buf,
+ const char *sep,
+ const char *leftop, Oid leftoptype,
+ Oid opoid,
+ const char *rightop, Oid rightoptype)
{
- appendStringInfo(buf, " %s ", sep);
+ if (sep)
+ appendStringInfo(buf, " %s ", sep);
generate_operator_clause(buf, leftop, leftoptype, opoid,
rightop, rightoptype);
}
+/*
+ * ri_ColNameQuoted() --- return column name, with both table and column name
+ * quoted.
+ */
+static char *
+ri_ColNameQuoted(const char *tabname, const char *attname)
+{
+ char quoted[MAX_QUOTED_NAME_LEN];
+ StringInfo result = makeStringInfo();
+
+ if (tabname && strlen(tabname) > 0)
+ {
+ quoteOneName(quoted, tabname);
+ appendStringInfo(result, "%s.", quoted);
+ }
+
+ quoteOneName(quoted, attname);
+ appendStringInfoString(result, quoted);
+
+ return result->data;
+}
+
+/*
+ * Check that RI trigger function was called in expected context
+ */
+static void
+ri_CheckTrigger(FunctionCallInfo fcinfo, const char *funcname, int tgkind)
+{
+ TriggerData *trigdata = (TriggerData *) fcinfo->context;
+
+ if (!CALLED_AS_TRIGGER(fcinfo))
+ ereport(ERROR,
+ (errcode(ERRCODE_E_R_I_E_TRIGGER_PROTOCOL_VIOLATED),
+ errmsg("function \"%s\" was not called by trigger manager", funcname)));
+
+ /*
+ * Check proper event
+ */
+ if (!TRIGGER_FIRED_AFTER(trigdata->tg_event) ||
+ !TRIGGER_FIRED_FOR_ROW(trigdata->tg_event))
+ ereport(ERROR,
+ (errcode(ERRCODE_E_R_I_E_TRIGGER_PROTOCOL_VIOLATED),
+ errmsg("function \"%s\" must be fired AFTER ROW", funcname)));
+
+ switch (tgkind)
+ {
+ case RI_TRIGTYPE_INSERT:
+ if (!TRIGGER_FIRED_BY_INSERT(trigdata->tg_event))
+ ereport(ERROR,
+ (errcode(ERRCODE_E_R_I_E_TRIGGER_PROTOCOL_VIOLATED),
+ errmsg("function \"%s\" must be fired for INSERT", funcname)));
+ break;
+ case RI_TRIGTYPE_UPDATE:
+ if (!TRIGGER_FIRED_BY_UPDATE(trigdata->tg_event))
+ ereport(ERROR,
+ (errcode(ERRCODE_E_R_I_E_TRIGGER_PROTOCOL_VIOLATED),
+ errmsg("function \"%s\" must be fired for UPDATE", funcname)));
+ break;
+
+ case RI_TRIGTYPE_DELETE:
+ if (!TRIGGER_FIRED_BY_DELETE(trigdata->tg_event))
+ ereport(ERROR,
+ (errcode(ERRCODE_E_R_I_E_TRIGGER_PROTOCOL_VIOLATED),
+ errmsg("function \"%s\" must be fired for DELETE", funcname)));
+ break;
+ }
+}
+
/*
* ri_GenerateQualCollation --- add a COLLATE spec to a WHERE clause
*
@@ -1909,52 +1985,6 @@ ri_BuildQueryKey(RI_QueryKey *key, const RI_ConstraintInfo *riinfo,
key->constr_queryno = constr_queryno;
}
-/*
- * Check that RI trigger function was called in expected context
- */
-static void
-ri_CheckTrigger(FunctionCallInfo fcinfo, const char *funcname, int tgkind)
-{
- TriggerData *trigdata = (TriggerData *) fcinfo->context;
-
- if (!CALLED_AS_TRIGGER(fcinfo))
- ereport(ERROR,
- (errcode(ERRCODE_E_R_I_E_TRIGGER_PROTOCOL_VIOLATED),
- errmsg("function \"%s\" was not called by trigger manager", funcname)));
-
- /*
- * Check proper event
- */
- if (!TRIGGER_FIRED_AFTER(trigdata->tg_event) ||
- !TRIGGER_FIRED_FOR_ROW(trigdata->tg_event))
- ereport(ERROR,
- (errcode(ERRCODE_E_R_I_E_TRIGGER_PROTOCOL_VIOLATED),
- errmsg("function \"%s\" must be fired AFTER ROW", funcname)));
-
- switch (tgkind)
- {
- case RI_TRIGTYPE_INSERT:
- if (!TRIGGER_FIRED_BY_INSERT(trigdata->tg_event))
- ereport(ERROR,
- (errcode(ERRCODE_E_R_I_E_TRIGGER_PROTOCOL_VIOLATED),
- errmsg("function \"%s\" must be fired for INSERT", funcname)));
- break;
- case RI_TRIGTYPE_UPDATE:
- if (!TRIGGER_FIRED_BY_UPDATE(trigdata->tg_event))
- ereport(ERROR,
- (errcode(ERRCODE_E_R_I_E_TRIGGER_PROTOCOL_VIOLATED),
- errmsg("function \"%s\" must be fired for UPDATE", funcname)));
- break;
- case RI_TRIGTYPE_DELETE:
- if (!TRIGGER_FIRED_BY_DELETE(trigdata->tg_event))
- ereport(ERROR,
- (errcode(ERRCODE_E_R_I_E_TRIGGER_PROTOCOL_VIOLATED),
- errmsg("function \"%s\" must be fired for DELETE", funcname)));
- break;
- }
-}
-
-
/*
* Fetch the RI_ConstraintInfo struct for the trigger's FK constraint.
*/
--
2.20.1
>From 8046a7dc0782a6ce95e808853a839ad529d76743 Mon Sep 17 00:00:00 2001
From: Antonin Houska <a...@cybertec.at>
Date: Wed, 8 Apr 2020 15:03:20 +0200
Subject: [PATCH 3/4] Return early from ri_NullCheck() if possible.
---
src/backend/utils/adt/ri_triggers.c | 10 +++++++++-
1 file changed, 9 insertions(+), 1 deletion(-)
diff --git a/src/backend/utils/adt/ri_triggers.c b/src/backend/utils/adt/ri_triggers.c
index 3bedb75846..93e46ddf7a 100644
--- a/src/backend/utils/adt/ri_triggers.c
+++ b/src/backend/utils/adt/ri_triggers.c
@@ -2541,6 +2541,13 @@ ri_NullCheck(TupleDesc tupDesc,
nonenull = false;
else
allnull = false;
+
+ /*
+ * If seen both NULL and non-NULL, the next attributes cannot change
+ * the result.
+ */
+ if (!nonenull && !allnull)
+ return RI_KEYS_SOME_NULL;
}
if (allnull)
@@ -2549,7 +2556,8 @@ ri_NullCheck(TupleDesc tupDesc,
if (nonenull)
return RI_KEYS_NONE_NULL;
- return RI_KEYS_SOME_NULL;
+ /* Should not happen. */
+ Assert(false);
}
--
2.20.1
>From 08f181a6f411ddc79a1d0ecabc26d430cf83221e Mon Sep 17 00:00:00 2001
From: Antonin Houska <a...@cybertec.at>
Date: Wed, 8 Apr 2020 15:03:20 +0200
Subject: [PATCH 4/4] Process multiple RI trigger events at a time.
It should be more efficient to execute the check query once per multiple rows
inserted / updated / deleted than to run the query for every single row again.
Separate storage is used for the RI trigger events because the "transient
table" that we provide to statement triggers would not be available for
deferred constraints. So the RI triggers still work at row level, although the
rows are processed in batches.
---
src/backend/commands/tablecmds.c | 53 +-
src/backend/commands/trigger.c | 406 +++++++--
src/backend/executor/spi.c | 16 +-
src/backend/utils/adt/ri_triggers.c | 1270 +++++++++++++++++----------
src/include/commands/trigger.h | 23 +
5 files changed, 1200 insertions(+), 568 deletions(-)
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index c8c88be2c9..feccb93b18 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -10225,6 +10225,11 @@ validateForeignKeyConstraint(char *conname,
MemoryContext oldcxt;
MemoryContext perTupCxt;
+ LOCAL_FCINFO(fcinfo, 0);
+ TriggerData trigdata = {0};
+ ResourceOwner saveResourceOwner;
+ Tuplestorestate *table;
+
ereport(DEBUG1,
(errmsg("validating foreign key constraint \"%s\"", conname)));
@@ -10259,6 +10264,11 @@ validateForeignKeyConstraint(char *conname,
slot = table_slot_create(rel, NULL);
scan = table_beginscan(rel, snapshot, 0, NULL);
+ saveResourceOwner = CurrentResourceOwner;
+ CurrentResourceOwner = CurTransactionResourceOwner;
+ table = tuplestore_begin_heap(false, false, work_mem);
+ CurrentResourceOwner = saveResourceOwner;
+
perTupCxt = AllocSetContextCreate(CurrentMemoryContext,
"validateForeignKeyConstraint",
ALLOCSET_SMALL_SIZES);
@@ -10266,34 +10276,33 @@ validateForeignKeyConstraint(char *conname,
while (table_scan_getnextslot(scan, ForwardScanDirection, slot))
{
- LOCAL_FCINFO(fcinfo, 0);
- TriggerData trigdata = {0};
-
CHECK_FOR_INTERRUPTS();
- /*
- * Make a call to the trigger function
- *
- * No parameters are passed, but we do set a context
- */
- MemSet(fcinfo, 0, SizeForFunctionCallInfo(0));
+ tuplestore_puttupleslot(table, slot);
- /*
- * We assume RI_FKey_check_ins won't look at flinfo...
- */
- trigdata.type = T_TriggerData;
- trigdata.tg_event = TRIGGER_EVENT_INSERT | TRIGGER_EVENT_ROW;
- trigdata.tg_relation = rel;
- trigdata.tg_trigtuple = ExecFetchSlotHeapTuple(slot, false, NULL);
- trigdata.tg_trigslot = slot;
- trigdata.tg_trigger = &trig;
+ MemoryContextReset(perTupCxt);
+ }
- fcinfo->context = (Node *) &trigdata;
+ /*
+ * Make a call to the trigger function
+ *
+ * No parameters are passed, but we do set a context
+ */
+ MemSet(fcinfo, 0, SizeForFunctionCallInfo(0));
- RI_FKey_check_ins(fcinfo);
+ /*
+ * We assume RI_FKey_check_ins won't look at flinfo...
+ */
+ trigdata.type = T_TriggerData;
+ trigdata.tg_event = TRIGGER_EVENT_INSERT | TRIGGER_EVENT_ROW;
+ trigdata.tg_relation = rel;
+ trigdata.tg_trigslot = slot;
+ trigdata.tg_trigger = &trig;
+ trigdata.tg_oldtable = table;
- MemoryContextReset(perTupCxt);
- }
+ fcinfo->context = (Node *) &trigdata;
+
+ RI_FKey_check_ins(fcinfo);
MemoryContextSwitchTo(oldcxt);
MemoryContextDelete(perTupCxt);
diff --git a/src/backend/commands/trigger.c b/src/backend/commands/trigger.c
index ed551ab73a..cc18167fb4 100644
--- a/src/backend/commands/trigger.c
+++ b/src/backend/commands/trigger.c
@@ -105,6 +105,8 @@ static void AfterTriggerSaveEvent(EState *estate, ResultRelInfo *relinfo,
static void AfterTriggerEnlargeQueryState(void);
static bool before_stmt_triggers_fired(Oid relid, CmdType cmdType);
+static TIDArray *alloc_tid_array(void);
+static void add_tid(TIDArray *ta, ItemPointer item);
/*
* Create a trigger. Returns the address of the created trigger.
@@ -3337,10 +3339,14 @@ typedef struct AfterTriggerEventList
/* Macros to help in iterating over a list of events */
#define for_each_chunk(cptr, evtlist) \
for (cptr = (evtlist).head; cptr != NULL; cptr = cptr->next)
+#define next_event_in_chunk(eptr, cptr) \
+ (AfterTriggerEvent) (((char *) eptr) + SizeofTriggerEvent(eptr))
#define for_each_event(eptr, cptr) \
for (eptr = (AfterTriggerEvent) CHUNK_DATA_START(cptr); \
(char *) eptr < (cptr)->freeptr; \
- eptr = (AfterTriggerEvent) (((char *) eptr) + SizeofTriggerEvent(eptr)))
+ eptr = next_event_in_chunk(eptr, cptr))
+#define is_last_event_in_chunk(eptr, cptr) \
+ ((((char *) eptr) + SizeofTriggerEvent(eptr)) >= (cptr)->freeptr)
/* Use this if no special per-chunk processing is needed */
#define for_each_event_chunk(eptr, cptr, evtlist) \
for_each_chunk(cptr, evtlist) for_each_event(eptr, cptr)
@@ -3488,9 +3494,17 @@ static void AfterTriggerExecute(EState *estate,
TriggerDesc *trigdesc,
FmgrInfo *finfo,
Instrumentation *instr,
+ TriggerData *trig_last,
MemoryContext per_tuple_context,
+ MemoryContext batch_context,
TupleTableSlot *trig_tuple_slot1,
TupleTableSlot *trig_tuple_slot2);
+static void AfterTriggerExecuteRI(EState *estate,
+ ResultRelInfo *relInfo,
+ FmgrInfo *finfo,
+ Instrumentation *instr,
+ TriggerData *trig_last,
+ MemoryContext batch_context);
static AfterTriggersTableData *GetAfterTriggersTableData(Oid relid,
CmdType cmdType);
static void AfterTriggerFreeQuery(AfterTriggersQueryData *qs);
@@ -3807,13 +3821,16 @@ afterTriggerDeleteHeadEventChunk(AfterTriggersQueryData *qs)
* fmgr lookup cache space at the caller level. (For triggers fired at
* the end of a query, we can even piggyback on the executor's state.)
*
- * event: event currently being fired.
+ * event: event currently being fired. Pass NULL if the current batch of RI
+ * trigger events should be processed.
* rel: open relation for event.
* trigdesc: working copy of rel's trigger info.
* finfo: array of fmgr lookup cache entries (one per trigger in trigdesc).
* instr: array of EXPLAIN ANALYZE instrumentation nodes (one per trigger),
* or NULL if no instrumentation is wanted.
+ * trig_last: trigger info used for the last trigger execution.
* per_tuple_context: memory context to call trigger function in.
+ * batch_context: memory context to store tuples for RI triggers.
* trig_tuple_slot1: scratch slot for tg_trigtuple (foreign tables only)
* trig_tuple_slot2: scratch slot for tg_newtuple (foreign tables only)
* ----------
@@ -3824,39 +3841,55 @@ AfterTriggerExecute(EState *estate,
ResultRelInfo *relInfo,
TriggerDesc *trigdesc,
FmgrInfo *finfo, Instrumentation *instr,
+ TriggerData *trig_last,
MemoryContext per_tuple_context,
+ MemoryContext batch_context,
TupleTableSlot *trig_tuple_slot1,
TupleTableSlot *trig_tuple_slot2)
{
Relation rel = relInfo->ri_RelationDesc;
AfterTriggerShared evtshared = GetTriggerSharedData(event);
Oid tgoid = evtshared->ats_tgoid;
- TriggerData LocTriggerData = {0};
HeapTuple rettuple;
- int tgindx;
bool should_free_trig = false;
bool should_free_new = false;
+ bool is_new = false;
- /*
- * Locate trigger in trigdesc.
- */
- for (tgindx = 0; tgindx < trigdesc->numtriggers; tgindx++)
+ if (trig_last->tg_trigger == NULL)
{
- if (trigdesc->triggers[tgindx].tgoid == tgoid)
+ int tgindx;
+
+ /*
+ * Locate trigger in trigdesc.
+ */
+ for (tgindx = 0; tgindx < trigdesc->numtriggers; tgindx++)
{
- LocTriggerData.tg_trigger = &(trigdesc->triggers[tgindx]);
- break;
+ if (trigdesc->triggers[tgindx].tgoid == tgoid)
+ {
+ trig_last->tg_trigger = &(trigdesc->triggers[tgindx]);
+ trig_last->tgindx = tgindx;
+ break;
+ }
}
+ if (trig_last->tg_trigger == NULL)
+ elog(ERROR, "could not find trigger %u", tgoid);
+
+ if (RI_FKey_trigger_type(trig_last->tg_trigger->tgfoid) !=
+ RI_TRIGGER_NONE)
+ trig_last->is_ri_trigger = true;
+
+ is_new = true;
}
- if (LocTriggerData.tg_trigger == NULL)
- elog(ERROR, "could not find trigger %u", tgoid);
+
+ /* trig_last for non-RI trigger should always be initialized again. */
+ Assert(trig_last->is_ri_trigger || is_new);
/*
* If doing EXPLAIN ANALYZE, start charging time to this trigger. We want
* to include time spent re-fetching tuples in the trigger cost.
*/
- if (instr)
- InstrStartNode(instr + tgindx);
+ if (instr && !trig_last->is_ri_trigger)
+ InstrStartNode(instr + trig_last->tgindx);
/*
* Fetch the required tuple(s).
@@ -3864,6 +3897,9 @@ AfterTriggerExecute(EState *estate,
switch (event->ate_flags & AFTER_TRIGGER_TUP_BITS)
{
case AFTER_TRIGGER_FDW_FETCH:
+ /* Foreign keys are not supported on foreign tables. */
+ Assert(!trig_last->is_ri_trigger);
+
{
Tuplestorestate *fdw_tuplestore = GetCurrentFDWTuplestore();
@@ -3879,6 +3915,8 @@ AfterTriggerExecute(EState *estate,
}
/* fall through */
case AFTER_TRIGGER_FDW_REUSE:
+ /* Foreign keys are not supported on foreign tables. */
+ Assert(!trig_last->is_ri_trigger);
/*
* Store tuple in the slot so that tg_trigtuple does not reference
@@ -3889,38 +3927,56 @@ AfterTriggerExecute(EState *estate,
* that is stored as a heap tuple, constructed in different memory
* context, in the slot anyway.
*/
- LocTriggerData.tg_trigslot = trig_tuple_slot1;
- LocTriggerData.tg_trigtuple =
+ trig_last->tg_trigslot = trig_tuple_slot1;
+ trig_last->tg_trigtuple =
ExecFetchSlotHeapTuple(trig_tuple_slot1, true, &should_free_trig);
if ((evtshared->ats_event & TRIGGER_EVENT_OPMASK) ==
TRIGGER_EVENT_UPDATE)
{
- LocTriggerData.tg_newslot = trig_tuple_slot2;
- LocTriggerData.tg_newtuple =
+ trig_last->tg_newslot = trig_tuple_slot2;
+ trig_last->tg_newtuple =
ExecFetchSlotHeapTuple(trig_tuple_slot2, true, &should_free_new);
}
else
{
- LocTriggerData.tg_newtuple = NULL;
+ trig_last->tg_newtuple = NULL;
}
break;
default:
if (ItemPointerIsValid(&(event->ate_ctid1)))
{
- LocTriggerData.tg_trigslot = ExecGetTriggerOldSlot(estate, relInfo);
+ if (!trig_last->is_ri_trigger)
+ {
+ trig_last->tg_trigslot = ExecGetTriggerOldSlot(estate,
+ relInfo);
- if (!table_tuple_fetch_row_version(rel, &(event->ate_ctid1),
- SnapshotAny,
- LocTriggerData.tg_trigslot))
- elog(ERROR, "failed to fetch tuple1 for AFTER trigger");
- LocTriggerData.tg_trigtuple =
- ExecFetchSlotHeapTuple(LocTriggerData.tg_trigslot, false, &should_free_trig);
+ if (!table_tuple_fetch_row_version(rel, &(event->ate_ctid1),
+ SnapshotAny,
+ trig_last->tg_trigslot))
+ elog(ERROR, "failed to fetch tuple1 for AFTER trigger");
+
+ trig_last->tg_trigtuple =
+ ExecFetchSlotHeapTuple(trig_last->tg_trigslot, false,
+ &should_free_trig);
+ }
+ else
+ {
+ if (trig_last->ri_tids_old == NULL)
+ {
+ MemoryContext oldcxt;
+
+ oldcxt = MemoryContextSwitchTo(batch_context);
+ trig_last->ri_tids_old = alloc_tid_array();
+ MemoryContextSwitchTo(oldcxt);
+ }
+ add_tid(trig_last->ri_tids_old, &(event->ate_ctid1));
+ }
}
else
{
- LocTriggerData.tg_trigtuple = NULL;
+ trig_last->tg_trigtuple = NULL;
}
/* don't touch ctid2 if not there */
@@ -3928,18 +3984,36 @@ AfterTriggerExecute(EState *estate,
AFTER_TRIGGER_2CTID &&
ItemPointerIsValid(&(event->ate_ctid2)))
{
- LocTriggerData.tg_newslot = ExecGetTriggerNewSlot(estate, relInfo);
+ if (!trig_last->is_ri_trigger)
+ {
+ trig_last->tg_newslot = ExecGetTriggerNewSlot(estate,
+ relInfo);
- if (!table_tuple_fetch_row_version(rel, &(event->ate_ctid2),
- SnapshotAny,
- LocTriggerData.tg_newslot))
- elog(ERROR, "failed to fetch tuple2 for AFTER trigger");
- LocTriggerData.tg_newtuple =
- ExecFetchSlotHeapTuple(LocTriggerData.tg_newslot, false, &should_free_new);
+ if (!table_tuple_fetch_row_version(rel, &(event->ate_ctid2),
+ SnapshotAny,
+ trig_last->tg_newslot))
+ elog(ERROR, "failed to fetch tuple2 for AFTER trigger");
+
+ trig_last->tg_newtuple =
+ ExecFetchSlotHeapTuple(trig_last->tg_newslot, false,
+ &should_free_new);
+ }
+ else
+ {
+ if (trig_last->ri_tids_new == NULL)
+ {
+ MemoryContext oldcxt;
+
+ oldcxt = MemoryContextSwitchTo(batch_context);
+ trig_last->ri_tids_new = alloc_tid_array();
+ MemoryContextSwitchTo(oldcxt);
+ }
+ add_tid(trig_last->ri_tids_new, &(event->ate_ctid2));
+ }
}
else
{
- LocTriggerData.tg_newtuple = NULL;
+ trig_last->tg_newtuple = NULL;
}
}
@@ -3949,19 +4023,26 @@ AfterTriggerExecute(EState *estate,
* a trigger, mark it "closed" so that it cannot change anymore. If any
* additional events of the same type get queued in the current trigger
* query level, they'll go into new transition tables.
+ *
+ * RI triggers treat the tuplestores specially, see above.
*/
- LocTriggerData.tg_oldtable = LocTriggerData.tg_newtable = NULL;
+ if (!trig_last->is_ri_trigger)
+ trig_last->tg_oldtable = trig_last->tg_newtable = NULL;
+
if (evtshared->ats_table)
{
- if (LocTriggerData.tg_trigger->tgoldtable)
+ /* There shouldn't be any transition table for an RI trigger event. */
+ Assert(!trig_last->is_ri_trigger);
+
+ if (trig_last->tg_trigger->tgoldtable)
{
- LocTriggerData.tg_oldtable = evtshared->ats_table->old_tuplestore;
+ trig_last->tg_oldtable = evtshared->ats_table->old_tuplestore;
evtshared->ats_table->closed = true;
}
- if (LocTriggerData.tg_trigger->tgnewtable)
+ if (trig_last->tg_trigger->tgnewtable)
{
- LocTriggerData.tg_newtable = evtshared->ats_table->new_tuplestore;
+ trig_last->tg_newtable = evtshared->ats_table->new_tuplestore;
evtshared->ats_table->closed = true;
}
}
@@ -3969,54 +4050,139 @@ AfterTriggerExecute(EState *estate,
/*
* Setup the remaining trigger information
*/
- LocTriggerData.type = T_TriggerData;
- LocTriggerData.tg_event =
- evtshared->ats_event & (TRIGGER_EVENT_OPMASK | TRIGGER_EVENT_ROW);
- LocTriggerData.tg_relation = rel;
- if (TRIGGER_FOR_UPDATE(LocTriggerData.tg_trigger->tgtype))
- LocTriggerData.tg_updatedcols = evtshared->ats_modifiedcols;
-
- MemoryContextReset(per_tuple_context);
+ if (is_new)
+ {
+ trig_last->type = T_TriggerData;
+ trig_last->tg_event =
+ evtshared->ats_event & (TRIGGER_EVENT_OPMASK | TRIGGER_EVENT_ROW);
+ trig_last->tg_relation = rel;
+ if (TRIGGER_FOR_UPDATE(trig_last->tg_trigger->tgtype))
+ trig_last->tg_updatedcols = evtshared->ats_modifiedcols;
+ }
/*
- * Call the trigger and throw away any possibly returned updated tuple.
- * (Don't let ExecCallTriggerFunc measure EXPLAIN time.)
+ * RI triggers are executed in batches, see the top of the function.
*/
- rettuple = ExecCallTriggerFunc(&LocTriggerData,
- tgindx,
- finfo,
- NULL,
- per_tuple_context);
- if (rettuple != NULL &&
- rettuple != LocTriggerData.tg_trigtuple &&
- rettuple != LocTriggerData.tg_newtuple)
- heap_freetuple(rettuple);
+ if (!trig_last->is_ri_trigger)
+ {
+ MemoryContextReset(per_tuple_context);
+
+ /*
+ * Call the trigger and throw away any possibly returned updated
+ * tuple. (Don't let ExecCallTriggerFunc measure EXPLAIN time.)
+ */
+ rettuple = ExecCallTriggerFunc(trig_last,
+ trig_last->tgindx,
+ finfo,
+ NULL,
+ per_tuple_context);
+ if (rettuple != NULL &&
+ rettuple != trig_last->tg_trigtuple &&
+ rettuple != trig_last->tg_newtuple)
+ heap_freetuple(rettuple);
+ }
/*
* Release resources
*/
if (should_free_trig)
- heap_freetuple(LocTriggerData.tg_trigtuple);
+ heap_freetuple(trig_last->tg_trigtuple);
if (should_free_new)
- heap_freetuple(LocTriggerData.tg_newtuple);
+ heap_freetuple(trig_last->tg_newtuple);
- /* don't clear slots' contents if foreign table */
- if (trig_tuple_slot1 == NULL)
+ /*
+ * Don't clear slots' contents if foreign table.
+ *
+ * For for RI trigger we manage these slots separately, see
+ * AfterTriggerExecuteRI().
+ */
+ if (trig_tuple_slot1 == NULL && !trig_last->is_ri_trigger)
{
- if (LocTriggerData.tg_trigslot)
- ExecClearTuple(LocTriggerData.tg_trigslot);
- if (LocTriggerData.tg_newslot)
- ExecClearTuple(LocTriggerData.tg_newslot);
+ if (trig_last->tg_trigslot)
+ ExecClearTuple(trig_last->tg_trigslot);
+ if (trig_last->tg_newslot)
+ ExecClearTuple(trig_last->tg_newslot);
}
/*
* If doing EXPLAIN ANALYZE, stop charging time to this trigger, and count
* one "tuple returned" (really the number of firings).
*/
- if (instr)
- InstrStopNode(instr + tgindx, 1);
+ if (instr && !trig_last->is_ri_trigger)
+ InstrStopNode(instr + trig_last->tgindx, 1);
+
+ /* RI triggers use trig_last across calls. */
+ if (!trig_last->is_ri_trigger)
+ memset(trig_last, 0, sizeof(TriggerData));
}
+/*
+ * AfterTriggerExecuteRI()
+ *
+ * Execute an RI trigger. It's assumed that AfterTriggerExecute() recognized
+ * RI trigger events and only added them to the batch instead of executing
+ * them. The actual processing of the batch is done by this function.
+ */
+static void
+AfterTriggerExecuteRI(EState *estate,
+ ResultRelInfo *relInfo,
+ FmgrInfo *finfo,
+ Instrumentation *instr,
+ TriggerData *trig_last,
+ MemoryContext batch_context)
+{
+ HeapTuple rettuple;
+
+ /*
+ * AfterTriggerExecute() must have been called for this trigger already.
+ */
+ Assert(trig_last->tg_trigger);
+ Assert(trig_last->is_ri_trigger);
+
+ /*
+ * RI trigger constructs a local tuplestore when it needs it. The point is
+ * that it might need to check visibility first. If we put the tuples into
+ * a tuplestore now, it'd be hard to keep pins of the containing buffers,
+ * and so table_tuple_satisfies_snapshot check wouldn't work.
+ */
+ Assert(trig_last->tg_oldtable == NULL);
+ Assert(trig_last->tg_newtable == NULL);
+
+ /* Initialize the slots to retrieve the rows by TID. */
+ trig_last->tg_trigslot = ExecGetTriggerOldSlot(estate, relInfo);
+ trig_last->tg_newslot = ExecGetTriggerNewSlot(estate, relInfo);
+
+ if (instr)
+ InstrStartNode(instr + trig_last->tgindx);
+
+ /*
+ * Call the trigger and throw away any possibly returned updated tuple.
+ * (Don't let ExecCallTriggerFunc measure EXPLAIN time.)
+ *
+ * batch_context already contains the TIDs of the affected rows. The RI
+ * trigger should also use this context to create the tuplestore for them.
+ */
+ rettuple = ExecCallTriggerFunc(trig_last,
+ trig_last->tgindx,
+ finfo,
+ NULL,
+ batch_context);
+ if (rettuple != NULL &&
+ rettuple != trig_last->tg_trigtuple &&
+ rettuple != trig_last->tg_newtuple)
+ heap_freetuple(rettuple);
+
+ if (instr)
+ InstrStopNode(instr + trig_last->tgindx, 1);
+
+ ExecClearTuple(trig_last->tg_trigslot);
+ ExecClearTuple(trig_last->tg_newslot);
+
+ MemoryContextReset(batch_context);
+
+ memset(trig_last, 0, sizeof(TriggerData));
+ return;
+}
/*
* afterTriggerMarkEvents()
@@ -4112,7 +4278,8 @@ afterTriggerInvokeEvents(AfterTriggerEventList *events,
{
bool all_fired = true;
AfterTriggerEventChunk *chunk;
- MemoryContext per_tuple_context;
+ MemoryContext per_tuple_context,
+ batch_context;
bool local_estate = false;
ResultRelInfo *rInfo = NULL;
Relation rel = NULL;
@@ -4121,6 +4288,7 @@ afterTriggerInvokeEvents(AfterTriggerEventList *events,
Instrumentation *instr = NULL;
TupleTableSlot *slot1 = NULL,
*slot2 = NULL;
+ TriggerData trig_last;
/* Make a local EState if need be */
if (estate == NULL)
@@ -4134,6 +4302,14 @@ afterTriggerInvokeEvents(AfterTriggerEventList *events,
AllocSetContextCreate(CurrentMemoryContext,
"AfterTriggerTupleContext",
ALLOCSET_DEFAULT_SIZES);
+ /* Separate context for a batch of RI trigger events. */
+ batch_context =
+ AllocSetContextCreate(CurrentMemoryContext,
+ "AfterTriggerBatchContext",
+ ALLOCSET_DEFAULT_SIZES);
+
+ /* No trigger executed yet in this batch. */
+ memset(&trig_last, 0, sizeof(TriggerData));
for_each_chunk(chunk, *events)
{
@@ -4150,6 +4326,8 @@ afterTriggerInvokeEvents(AfterTriggerEventList *events,
if ((event->ate_flags & AFTER_TRIGGER_IN_PROGRESS) &&
evtshared->ats_firing_id == firing_id)
{
+ bool fire_ri_batch = false;
+
/*
* So let's fire it... but first, find the correct relation if
* this is not the same relation as before.
@@ -4180,12 +4358,60 @@ afterTriggerInvokeEvents(AfterTriggerEventList *events,
}
/*
- * Fire it. Note that the AFTER_TRIGGER_IN_PROGRESS flag is
- * still set, so recursive examinations of the event list
- * won't try to re-fire it.
+ * Fire it (or add the corresponding tuple(s) to the current
+ * batch if it's RI trigger).
+ *
+ * Note that the AFTER_TRIGGER_IN_PROGRESS flag is still set,
+ * so recursive examinations of the event list won't try to
+ * re-fire it.
*/
- AfterTriggerExecute(estate, event, rInfo, trigdesc, finfo, instr,
- per_tuple_context, slot1, slot2);
+ AfterTriggerExecute(estate, event, rInfo, trigdesc, finfo,
+ instr, &trig_last,
+ per_tuple_context, batch_context,
+ slot1, slot2);
+
+ /*
+ * RI trigger events are processed in batches, so extra work
+ * might be needed to finish the current batch. It's important
+ * to do this before the chunk iteration ends because the
+ * trigger execution may generate other events.
+ *
+ * XXX Implement maximum batch size so that constraint
+ * violations are reported as soon as possible?
+ */
+ if (trig_last.tg_trigger && trig_last.is_ri_trigger)
+ {
+ if (is_last_event_in_chunk(event, chunk))
+ fire_ri_batch = true;
+ else
+ {
+ AfterTriggerEvent evtnext;
+ AfterTriggerShared evtshnext;
+
+ /*
+ * We even need to look ahead because the next event
+ * might be affected by execution of the current one.
+ * For example if the next event is an AS trigger
+ * event to be cancelled (cancel_prior_stmt_triggers)
+ * because the current event, during its execution,
+ * generates a new AS event for the same trigger.
+ */
+ evtnext = next_event_in_chunk(event, chunk);
+ evtshnext = GetTriggerSharedData(evtnext);
+
+ if (evtshnext != evtshared)
+ fire_ri_batch = true;
+ }
+ }
+
+ if (fire_ri_batch)
+ AfterTriggerExecuteRI(estate,
+ rInfo,
+ finfo,
+ instr,
+ &trig_last,
+ batch_context);
+
/*
* Mark the event as done.
@@ -4216,6 +4442,7 @@ afterTriggerInvokeEvents(AfterTriggerEventList *events,
events->tailfree = chunk->freeptr;
}
}
+
if (slot1 != NULL)
{
ExecDropSingleTupleTableSlot(slot1);
@@ -4224,6 +4451,7 @@ afterTriggerInvokeEvents(AfterTriggerEventList *events,
/* Release working resources */
MemoryContextDelete(per_tuple_context);
+ MemoryContextDelete(batch_context);
if (local_estate)
{
@@ -5812,3 +6040,29 @@ pg_trigger_depth(PG_FUNCTION_ARGS)
{
PG_RETURN_INT32(MyTriggerDepth);
}
+
+static TIDArray *
+alloc_tid_array(void)
+{
+ TIDArray *result = (TIDArray *) palloc(sizeof(TIDArray));
+
+ /* XXX Tune the chunk size. */
+ result->nmax = 1024;
+ result->tids = (ItemPointer) palloc(result->nmax *
+ sizeof(ItemPointerData));
+ result->n = 0;
+ return result;
+}
+
+static void
+add_tid(TIDArray *ta, ItemPointer item)
+{
+ if (ta->n == ta->nmax)
+ {
+ ta->nmax += 1024;
+ ta->tids = (ItemPointer) repalloc(ta->tids,
+ ta->nmax * sizeof(ItemPointerData));
+ }
+ memcpy(ta->tids + ta->n, item, sizeof(ItemPointerData));
+ ta->n++;
+}
diff --git a/src/backend/executor/spi.c b/src/backend/executor/spi.c
index b108168821..37026219b6 100644
--- a/src/backend/executor/spi.c
+++ b/src/backend/executor/spi.c
@@ -2929,12 +2929,14 @@ SPI_register_trigger_data(TriggerData *tdata)
if (tdata->tg_newtable)
{
EphemeralNamedRelation enr =
- palloc(sizeof(EphemeralNamedRelationData));
+ palloc0(sizeof(EphemeralNamedRelationData));
int rc;
enr->md.name = tdata->tg_trigger->tgnewtable;
- enr->md.reliddesc = tdata->tg_relation->rd_id;
- enr->md.tupdesc = NULL;
+ if (tdata->desc)
+ enr->md.tupdesc = tdata->desc;
+ else
+ enr->md.reliddesc = tdata->tg_relation->rd_id;
enr->md.enrtype = ENR_NAMED_TUPLESTORE;
enr->md.enrtuples = tuplestore_tuple_count(tdata->tg_newtable);
enr->reldata = tdata->tg_newtable;
@@ -2946,12 +2948,14 @@ SPI_register_trigger_data(TriggerData *tdata)
if (tdata->tg_oldtable)
{
EphemeralNamedRelation enr =
- palloc(sizeof(EphemeralNamedRelationData));
+ palloc0(sizeof(EphemeralNamedRelationData));
int rc;
enr->md.name = tdata->tg_trigger->tgoldtable;
- enr->md.reliddesc = tdata->tg_relation->rd_id;
- enr->md.tupdesc = NULL;
+ if (tdata->desc)
+ enr->md.tupdesc = tdata->desc;
+ else
+ enr->md.reliddesc = tdata->tg_relation->rd_id;
enr->md.enrtype = ENR_NAMED_TUPLESTORE;
enr->md.enrtuples = tuplestore_tuple_count(tdata->tg_oldtable);
enr->reldata = tdata->tg_oldtable;
diff --git a/src/backend/utils/adt/ri_triggers.c b/src/backend/utils/adt/ri_triggers.c
index 93e46ddf7a..8d952e0866 100644
--- a/src/backend/utils/adt/ri_triggers.c
+++ b/src/backend/utils/adt/ri_triggers.c
@@ -69,15 +69,17 @@
/* 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_CHECK_LOOKUPPK_INS 1
+#define RI_PLAN_CHECK_LOOKUPPK_UPD 2
+#define RI_PLAN_CHECK_LOOKUPPK_FROM_PK 3
#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 4
+#define RI_PLAN_CASCADE_UPD_DOUPDATE 5
+#define RI_PLAN_RESTRICT_CHECKREF 6
+#define RI_PLAN_RESTRICT_CHECKREF_NO_ACTION 7
+#define RI_PLAN_SETNULL_DOUPDATE 8
+#define RI_PLAN_SETDEFAULT_DOUPDATE 9
#define MAX_QUOTED_NAME_LEN (NAMEDATALEN*2+3)
#define MAX_QUOTED_REL_NAME_LEN (MAX_QUOTED_NAME_LEN*2)
@@ -114,6 +116,7 @@ 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) */
+ TupleDesc desc_pk_both; /* Both OLD an NEW version of PK table row. */
dlist_node valid_link; /* Link in list of valid entries */
} RI_ConstraintInfo;
@@ -173,10 +176,15 @@ static int ri_constraint_cache_valid_count = 0;
/*
* Local function prototypes
*/
-static bool ri_Check_Pk_Match(Relation pk_rel, Relation fk_rel,
- TupleTableSlot *oldslot,
- const RI_ConstraintInfo *riinfo);
+static char *RI_FKey_check_query(const RI_ConstraintInfo *riinfo,
+ Relation fk_rel, Relation pk_rel,
+ bool insert);
+static bool RI_FKey_check_query_required(Trigger *trigger, Relation fk_rel,
+ TupleTableSlot *newslot);
static Datum ri_restrict(TriggerData *trigdata, bool is_no_action);
+static char *ri_restrict_query(const RI_ConstraintInfo *riinfo,
+ Relation fk_rel, Relation pk_rel,
+ bool no_action);
static Datum ri_set(TriggerData *trigdata, bool is_set_null);
static void quoteOneName(char *buffer, const char *name);
static void quoteRelationName(char *buffer, Relation rel);
@@ -186,6 +194,9 @@ static void ri_GenerateQual(StringInfo buf, char *sep, int nkeys,
const int16 *lattnums,
const char *rtabname, Relation rrel,
const int16 *rattnums, const Oid *eq_oprs);
+static void ri_GenerateKeyList(StringInfo buf, int nkeys,
+ const char *tabname, Relation rel,
+ const int16 *attnums);
static void ri_GenerateQualComponent(StringInfo buf,
const char *sep,
const char *leftop, Oid leftoptype,
@@ -212,21 +223,37 @@ static void ri_CheckTrigger(FunctionCallInfo fcinfo, const char *funcname,
int tgkind);
static const RI_ConstraintInfo *ri_FetchConstraintInfo(Trigger *trigger,
Relation trig_rel, bool rel_is_pk);
-static const RI_ConstraintInfo *ri_LoadConstraintInfo(Oid constraintOid);
-static SPIPlanPtr ri_PlanCheck(const char *querystr, int nargs, Oid *argtypes,
- RI_QueryKey *qkey, Relation fk_rel, Relation pk_rel);
+static const RI_ConstraintInfo *ri_LoadConstraintInfo(Oid constraintOid,
+ Relation trig_rel,
+ bool rel_is_pk);
+static SPIPlanPtr ri_PlanCheck(const char *querystr, RI_QueryKey *qkey,
+ Relation fk_rel, Relation pk_rel);
static bool ri_PerformCheck(const RI_ConstraintInfo *riinfo,
RI_QueryKey *qkey, SPIPlanPtr qplan,
Relation fk_rel, Relation pk_rel,
- TupleTableSlot *oldslot, TupleTableSlot *newslot,
bool detectNewRows, int expect_OK);
-static void ri_ExtractValues(Relation rel, TupleTableSlot *slot,
- const RI_ConstraintInfo *riinfo, bool rel_is_pk,
- Datum *vals, char *nulls);
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();
+static int ri_register_trigger_data(TriggerData *tdata,
+ Tuplestorestate *oldtable,
+ Tuplestorestate *newtable,
+ TupleDesc desc);
+static Tuplestorestate *get_event_tuplestore(TriggerData *trigdata, bool old,
+ Snapshot snapshot);
+static Tuplestorestate *get_event_tuplestore_for_cascade_update(TriggerData *trigdata,
+ const RI_ConstraintInfo *riinfo);
+static void add_key_attrs_to_tupdesc(TupleDesc tupdesc,
+ Relation rel,
+ const RI_ConstraintInfo *riinfo,
+ bool old);
+static void add_key_values(TupleTableSlot *slot,
+ const RI_ConstraintInfo *riinfo,
+ Relation rel, ItemPointer ip,
+ Datum *key_values, bool *key_nulls,
+ Datum *values, bool *nulls, bool old);
+static TupleTableSlot *get_violator_tuple(Relation rel);
/*
@@ -240,37 +267,212 @@ RI_FKey_check(TriggerData *trigdata)
const RI_ConstraintInfo *riinfo;
Relation fk_rel;
Relation pk_rel;
- TupleTableSlot *newslot;
RI_QueryKey qkey;
SPIPlanPtr qplan;
+ Tuplestorestate *oldtable = NULL;
+ Tuplestorestate *newtable = NULL;
riinfo = ri_FetchConstraintInfo(trigdata->tg_trigger,
trigdata->tg_relation, false);
- if (TRIGGER_FIRED_BY_UPDATE(trigdata->tg_event))
- newslot = trigdata->tg_newslot;
- else
- newslot = trigdata->tg_trigslot;
+ /*
+ * Get the relation descriptors of the FK and PK tables.
+ *
+ * pk_rel is opened in RowShareLock mode since that's what our eventual
+ * SELECT FOR KEY SHARE will get on it.
+ */
+ fk_rel = trigdata->tg_relation;
+ pk_rel = table_open(riinfo->pk_relid, RowShareLock);
/*
+ * Retrieve the changed rows and put them into the appropriate tuplestore.
+ *
* We should not even consider checking the row if it is no longer valid,
* since it was either deleted (so the deferred check should be skipped)
* or updated (in which case only the latest version of the row should be
- * checked). Test its liveness according to SnapshotSelf. We need pin
- * and lock on the buffer to call HeapTupleSatisfiesVisibility. Caller
+ * checked). Test its liveness according to SnapshotSelf. We need pin
+ * and lock on the buffer to call HeapTupleSatisfiesVisibility. Caller
* should be holding pin, but not lock.
*/
- if (!table_tuple_satisfies_snapshot(trigdata->tg_relation, newslot, SnapshotSelf))
- return PointerGetDatum(NULL);
+ if (TRIGGER_FIRED_BY_INSERT(trigdata->tg_event))
+ {
+ if (trigdata->ri_tids_old)
+ oldtable = get_event_tuplestore(trigdata, true, SnapshotSelf);
+ else
+ {
+ /* The whole table is passed if not called from trigger.c */
+ oldtable = trigdata->tg_oldtable;
+ }
+ }
+ else
+ {
+ if (trigdata->ri_tids_new)
+ newtable = get_event_tuplestore(trigdata, false, SnapshotSelf);
+ else
+ {
+ /* The whole table is passed if not called from trigger.c */
+ newtable = trigdata->tg_newtable;
+ }
+ }
+
+ if (SPI_connect() != SPI_OK_CONNECT)
+ elog(ERROR, "SPI_connect failed");
+
+ if (ri_register_trigger_data(trigdata, oldtable, newtable, NULL) !=
+ SPI_OK_TD_REGISTER)
+ elog(ERROR, "ri_register_trigger_data failed");
+
+ if (TRIGGER_FIRED_BY_INSERT(trigdata->tg_event))
+ {
+ /* Fetch or prepare a saved plan for the real check */
+ ri_BuildQueryKey(&qkey, riinfo, RI_PLAN_CHECK_LOOKUPPK_INS);
+
+ if ((qplan = ri_FetchPreparedPlan(&qkey)) == NULL)
+ {
+ char *query;
+
+ query = RI_FKey_check_query(riinfo, fk_rel, pk_rel, true);
+
+ /* Prepare and save the plan */
+ qplan = ri_PlanCheck(query, &qkey, fk_rel, pk_rel);
+ }
+ }
+ else
+ {
+ Assert((TRIGGER_FIRED_BY_UPDATE(trigdata->tg_event)));
+
+ /* Fetch or prepare a saved plan for the real check */
+ ri_BuildQueryKey(&qkey, riinfo, RI_PLAN_CHECK_LOOKUPPK_UPD);
+
+ if ((qplan = ri_FetchPreparedPlan(&qkey)) == NULL)
+ {
+ char *query;
+
+ query = RI_FKey_check_query(riinfo, fk_rel, pk_rel, false);
+
+ /* Prepare and save the plan */
+ qplan = ri_PlanCheck(query, &qkey, fk_rel, pk_rel);
+ }
+ }
/*
- * Get the relation descriptors of the FK and PK tables.
- *
- * pk_rel is opened in RowShareLock mode since that's what our eventual
- * SELECT FOR KEY SHARE will get on it.
+ * Now check that foreign key exists in PK table
*/
- fk_rel = trigdata->tg_relation;
- pk_rel = table_open(riinfo->pk_relid, RowShareLock);
+ if (ri_PerformCheck(riinfo, &qkey, qplan,
+ fk_rel, pk_rel,
+ false,
+ SPI_OK_SELECT))
+ {
+ TupleTableSlot *violatorslot = get_violator_tuple(fk_rel);
+
+ ri_ReportViolation(riinfo,
+ pk_rel, fk_rel,
+ violatorslot,
+ NULL,
+ qkey.constr_queryno, false);
+ }
+
+ if (SPI_finish() != SPI_OK_FINISH)
+ elog(ERROR, "SPI_finish failed");
+
+ table_close(pk_rel, RowShareLock);
+
+ return PointerGetDatum(NULL);
+}
+
+/* ----------
+ * Construct the query to check inserted/updated rows of the FK table.
+ *
+ * If "insert" is true, the rows are inserted, otherwise they are updated.
+ *
+ * The query string built is
+ * SELECT t.fkatt1 [, ...]
+ * FROM <tgtable> t LEFT JOIN LATERAL
+ * (SELECT t.fkatt1 [, ...]
+ * FROM [ONLY] <pktable> p
+ * WHERE t.fkatt1 = p.pkatt1 [AND ...]
+ * FOR KEY SHARE OF p) AS m
+ * ON t.fkatt1 = m.fkatt1 [AND ...]
+ * WHERE m.fkatt1 ISNULL
+ * LIMIT 1
+ *
+ * where <tgtable> is "tgoldtable" for INSERT and "tgnewtable" for UPDATE
+ * events.
+ *
+ * It returns the first row that violates the constraint.
+ *
+ * "m" returns the new rows that do have matching PK row. It is a subquery
+ * because the FOR KEY SHARE clause cannot reference the nullable side of an
+ * outer join.
+ *
+ * XXX "tgoldtable" looks confusing for insert, but that's where
+ * AfterTriggerExecute() stores tuples whose events don't have
+ * AFTER_TRIGGER_2CTID set. For a non-RI trigger, the inserted tuple would
+ * fall into tg_trigtuple as opposed to tg_newtuple, which seems a similar
+ * problem. It doesn't seem worth any renaming or adding extra tuplestores to
+ * TriggerData.
+ * ----------
+ */
+static char *
+RI_FKey_check_query(const RI_ConstraintInfo *riinfo, Relation fk_rel,
+ Relation pk_rel, bool insert)
+{
+ StringInfo querybuf = makeStringInfo();
+ StringInfo subquerybuf = makeStringInfo();
+ char pkrelname[MAX_QUOTED_REL_NAME_LEN];
+ const char *pk_only;
+ const char *tgtable;
+ char *col_test;
+
+ tgtable = insert ? "tgoldtable" : "tgnewtable";
+
+ quoteRelationName(pkrelname, pk_rel);
+
+ /* Construct the subquery. */
+ pk_only = pk_rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE ?
+ "" : "ONLY ";
+ appendStringInfoString(subquerybuf,
+ "(SELECT ");
+ ri_GenerateKeyList(subquerybuf, riinfo->nkeys, "t", fk_rel,
+ riinfo->fk_attnums);
+ appendStringInfo(subquerybuf,
+ " FROM %s%s p WHERE ",
+ pk_only, pkrelname);
+ ri_GenerateQual(subquerybuf, "AND", riinfo->nkeys,
+ "p", pk_rel, riinfo->pk_attnums,
+ "t", fk_rel, riinfo->fk_attnums,
+ riinfo->pf_eq_oprs);
+ appendStringInfoString(subquerybuf, " FOR KEY SHARE OF p) AS m");
+
+ /* Now the main query. */
+ appendStringInfoString(querybuf, "SELECT ");
+ ri_GenerateKeyList(querybuf, riinfo->nkeys, "t", fk_rel,
+ riinfo->fk_attnums);
+ appendStringInfo(querybuf,
+ " FROM %s t LEFT JOIN LATERAL %s ON ",
+ tgtable, subquerybuf->data);
+ ri_GenerateQual(querybuf, "AND", riinfo->nkeys,
+ "t", fk_rel, riinfo->fk_attnums,
+ "m", fk_rel, riinfo->fk_attnums,
+ riinfo->ff_eq_oprs);
+ col_test = ri_ColNameQuoted("m", RIAttName(fk_rel, riinfo->fk_attnums[0]));
+ appendStringInfo(querybuf, " WHERE %s ISNULL ", col_test);
+ appendStringInfoString(querybuf, " LIMIT 1");
+
+ return querybuf->data;
+}
+
+/*
+ * Check if the PK table needs to be queried (using the query generated by
+ * RI_FKey_check_query).
+ */
+static bool
+RI_FKey_check_query_required(Trigger *trigger, Relation fk_rel,
+ TupleTableSlot *newslot)
+{
+ const RI_ConstraintInfo *riinfo;
+
+ riinfo = ri_FetchConstraintInfo(trigger, fk_rel, false);
switch (ri_NullCheck(RelationGetDescr(fk_rel), newslot, riinfo, false))
{
@@ -280,8 +482,7 @@ RI_FKey_check(TriggerData *trigdata)
* No further check needed - an all-NULL key passes every type of
* foreign key constraint.
*/
- table_close(pk_rel, RowShareLock);
- return PointerGetDatum(NULL);
+ return false;
case RI_KEYS_SOME_NULL:
@@ -305,8 +506,7 @@ RI_FKey_check(TriggerData *trigdata)
errdetail("MATCH FULL does not allow mixing of null and nonnull key values."),
errtableconstraint(fk_rel,
NameStr(riinfo->conname))));
- table_close(pk_rel, RowShareLock);
- return PointerGetDatum(NULL);
+ break;
case FKCONSTR_MATCH_SIMPLE:
@@ -314,17 +514,16 @@ RI_FKey_check(TriggerData *trigdata)
* MATCH SIMPLE - if ANY column is null, the key passes
* the constraint.
*/
- table_close(pk_rel, RowShareLock);
- return PointerGetDatum(NULL);
+ return false;
#ifdef NOT_USED
case FKCONSTR_MATCH_PARTIAL:
/*
* 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 the query to only
+ * include non-null columns, or by writing a special
+ * version)
*/
break;
#endif
@@ -333,85 +532,12 @@ RI_FKey_check(TriggerData *trigdata)
case RI_KEYS_NONE_NULL:
/*
- * Have a full qualified key - continue below for all three kinds
- * of MATCH.
+ * Have a full qualified key - regular check is needed.
*/
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] <pktable> 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_GenerateQualComponent(&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
- */
- if (!ri_PerformCheck(riinfo, &qkey, qplan,
- fk_rel, pk_rel,
- NULL, newslot,
- false,
- SPI_OK_SELECT))
- ri_ReportViolation(riinfo,
- pk_rel, fk_rel,
- newslot,
- NULL,
- qkey.constr_queryno, false);
-
- if (SPI_finish() != SPI_OK_FINISH)
- elog(ERROR, "SPI_finish failed");
-
- table_close(pk_rel, RowShareLock);
-
- return PointerGetDatum(NULL);
+ return true;
}
@@ -447,99 +573,6 @@ RI_FKey_check_upd(PG_FUNCTION_ARGS)
}
-/*
- * ri_Check_Pk_Match
- *
- * Check to see if another PK row has been created that provides the same
- * key values as the "oldslot" that's been modified or deleted in our trigger
- * event. Returns true if a match is found in the PK table.
- *
- * We assume the caller checked that the oldslot contains no NULL key values,
- * since otherwise a match is impossible.
- */
-static bool
-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] <pktable> 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_GenerateQualComponent(&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;
-}
-
-
/*
* RI_FKey_noaction_del -
*
@@ -626,9 +659,9 @@ ri_restrict(TriggerData *trigdata, bool is_no_action)
const RI_ConstraintInfo *riinfo;
Relation fk_rel;
Relation pk_rel;
- TupleTableSlot *oldslot;
RI_QueryKey qkey;
SPIPlanPtr qplan;
+ Tuplestorestate *oldtable;
riinfo = ri_FetchConstraintInfo(trigdata->tg_trigger,
trigdata->tg_relation, true);
@@ -641,79 +674,54 @@ ri_restrict(TriggerData *trigdata, bool is_no_action)
*/
fk_rel = table_open(riinfo->fk_relid, RowShareLock);
pk_rel = trigdata->tg_relation;
- oldslot = trigdata->tg_trigslot;
- /*
- * If another PK row now exists providing the old key values, we should
- * not do anything. However, this check should only be made in the NO
- * ACTION case; in RESTRICT cases we don't wish to allow another row to be
- * substituted.
- */
- if (is_no_action &&
- ri_Check_Pk_Match(pk_rel, fk_rel, oldslot, riinfo))
- {
- table_close(fk_rel, RowShareLock);
- return PointerGetDatum(NULL);
- }
+ oldtable = get_event_tuplestore(trigdata, true, NULL);
if (SPI_connect() != SPI_OK_CONNECT)
elog(ERROR, "SPI_connect failed");
- /*
- * Fetch or prepare a saved plan for the restrict lookup (it's the same
- * query for delete and update cases)
- */
- ri_BuildQueryKey(&qkey, riinfo, RI_PLAN_RESTRICT_CHECKREF);
+ if (ri_register_trigger_data(trigdata, oldtable, NULL, NULL) !=
+ SPI_OK_TD_REGISTER)
+ elog(ERROR, "ri_register_trigger_data failed");
- if ((qplan = ri_FetchPreparedPlan(&qkey)) == NULL)
+ if (!is_no_action)
{
- StringInfoData querybuf;
- char fkrelname[MAX_QUOTED_REL_NAME_LEN];
- char attname[MAX_QUOTED_NAME_LEN];
- char paramname[16];
- const char *querysep;
- Oid queryoids[RI_MAX_NUMKEYS];
- const char *fk_only;
-
- /* ----------
- * The query string built is
- * SELECT 1 FROM [ONLY] <fktable> x WHERE $1 = fkatt1 [AND ...]
- * FOR KEY SHARE OF x
- * The type id's for the $ parameters are those of the
- * corresponding PK attributes.
- * ----------
+ /*
+ * Fetch or prepare a saved plan for the restrict lookup (it's the
+ * same query for delete and update cases)
*/
- initStringInfo(&querybuf);
- fk_only = fk_rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE ?
- "" : "ONLY ";
- quoteRelationName(fkrelname, fk_rel);
- appendStringInfo(&querybuf, "SELECT 1 FROM %s%s x",
- fk_only, fkrelname);
- querysep = "WHERE";
- for (int i = 0; i < riinfo->nkeys; i++)
+ ri_BuildQueryKey(&qkey, riinfo, RI_PLAN_RESTRICT_CHECKREF);
+
+ if ((qplan = ri_FetchPreparedPlan(&qkey)) == NULL)
{
- 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]);
+ char *query;
- quoteOneName(attname,
- RIAttName(fk_rel, riinfo->fk_attnums[i]));
- sprintf(paramname, "$%d", i + 1);
- ri_GenerateQualComponent(&querybuf, querysep,
- paramname, pk_type,
- riinfo->pf_eq_oprs[i],
- attname, fk_type);
- if (pk_coll != fk_coll && !get_collation_isdeterministic(pk_coll))
- ri_GenerateQualCollation(&querybuf, pk_coll);
- querysep = "AND";
- queryoids[i] = pk_type;
+ query = ri_restrict_query(riinfo, fk_rel, pk_rel, false);
+
+ /* Prepare and save the plan */
+ qplan = ri_PlanCheck(query, &qkey, fk_rel, pk_rel);
}
- appendStringInfoString(&querybuf, " FOR KEY SHARE OF x");
+ }
+ else
+ {
+ /*
+ * If another PK row now exists providing the old key values, we
+ * should not do anything. However, this check should only be made in
+ * the NO ACTION case; in RESTRICT cases we don't wish to allow
+ * another row to be substituted.
+ */
+ ri_BuildQueryKey(&qkey, riinfo, RI_PLAN_RESTRICT_CHECKREF_NO_ACTION);
- /* Prepare and save the plan */
- qplan = ri_PlanCheck(querybuf.data, riinfo->nkeys, queryoids,
- &qkey, fk_rel, pk_rel);
+ if ((qplan = ri_FetchPreparedPlan(&qkey)) == NULL)
+ {
+
+ char *query;
+
+ query = ri_restrict_query(riinfo, fk_rel, pk_rel, true);
+
+ /* Prepare and save the plan */
+ qplan = ri_PlanCheck(query, &qkey, fk_rel, pk_rel);
+ }
}
/*
@@ -721,14 +729,17 @@ ri_restrict(TriggerData *trigdata, bool is_no_action)
*/
if (ri_PerformCheck(riinfo, &qkey, qplan,
fk_rel, pk_rel,
- oldslot, NULL,
true, /* must detect new rows */
SPI_OK_SELECT))
+ {
+ TupleTableSlot *violatorslot = get_violator_tuple(pk_rel);
+
ri_ReportViolation(riinfo,
pk_rel, fk_rel,
- oldslot,
+ violatorslot,
NULL,
qkey.constr_queryno, false);
+ }
if (SPI_finish() != SPI_OK_FINISH)
elog(ERROR, "SPI_finish failed");
@@ -738,6 +749,76 @@ ri_restrict(TriggerData *trigdata, bool is_no_action)
return PointerGetDatum(NULL);
}
+/* ----------
+ * Construct the query to check whether deleted row of the PK table is still
+ * referenced by the FK table.
+ *
+ * If "pk_rel" is NULL, the query string built is
+ * SELECT o.*
+ * FROM [ONLY] <fktable> f, tgoldtable o
+ * WHERE f.fkatt1 = o.pkatt1 [AND ...]
+ * FOR KEY SHARE OF f
+ * LIMIT 1
+ *
+ * If no_action is true,also check if the row being deleted was re-inserted
+ * into the PK table (or in case of UPDATE, if row with the old key is there
+ * again):
+ *
+ * SELECT o.pkatt1 [, ...]
+ * FROM [ONLY] <fktable> f, tgoldtable o
+ * WHERE f.fkatt1 = o.pkatt1 [AND ...] AND NOT EXISTS
+ * (SELECT 1
+ * FROM <pktable> p
+ * WHERE p.pkatt1 = o.pkatt1 [, ...]
+ * FOR KEY SHARE OF p)
+ * FOR KEY SHARE OF f
+ * LIMIT 1
+ *
+ * TODO Is ONLY needed for the the PK table?
+ * ----------
+ */
+static char *
+ri_restrict_query(const RI_ConstraintInfo *riinfo, Relation fk_rel,
+ Relation pk_rel, bool no_action)
+{
+ StringInfo querybuf = makeStringInfo();
+ StringInfo subquerybuf = NULL;
+ char fkrelname[MAX_QUOTED_REL_NAME_LEN];
+ const char *fk_only;
+
+ if (no_action)
+ {
+ char pkrelname[MAX_QUOTED_REL_NAME_LEN];
+
+ subquerybuf = makeStringInfo();
+ quoteRelationName(pkrelname, pk_rel);
+ appendStringInfo(subquerybuf,
+ "(SELECT 1 FROM %s p WHERE ", pkrelname);
+ ri_GenerateQual(subquerybuf, "AND", riinfo->nkeys,
+ "p", pk_rel, riinfo->pk_attnums,
+ "o", pk_rel, riinfo->pk_attnums,
+ riinfo->pp_eq_oprs);
+ appendStringInfoString(subquerybuf, " FOR KEY SHARE OF p)");
+ }
+
+ fk_only = fk_rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE ?
+ "" : "ONLY ";
+ quoteRelationName(fkrelname, fk_rel);
+ appendStringInfoString(querybuf, "SELECT ");
+ ri_GenerateKeyList(querybuf, riinfo->nkeys, "o", pk_rel,
+ riinfo->pk_attnums);
+ appendStringInfo(querybuf, " FROM %s%s f, tgoldtable o WHERE ",
+ fk_only, fkrelname);
+ ri_GenerateQual(querybuf, "AND", riinfo->nkeys,
+ "o", pk_rel, riinfo->pk_attnums,
+ "f", fk_rel, riinfo->fk_attnums,
+ riinfo->pf_eq_oprs);
+ if (no_action)
+ appendStringInfo(querybuf, " AND NOT EXISTS %s", subquerybuf->data);
+ appendStringInfoString(querybuf, " FOR KEY SHARE OF f LIMIT 1");
+
+ return querybuf->data;
+}
/*
* RI_FKey_cascade_del -
@@ -751,9 +832,9 @@ RI_FKey_cascade_del(PG_FUNCTION_ARGS)
const RI_ConstraintInfo *riinfo;
Relation fk_rel;
Relation pk_rel;
- TupleTableSlot *oldslot;
RI_QueryKey qkey;
SPIPlanPtr qplan;
+ Tuplestorestate *oldtable;
/* Check that this is a valid trigger call on the right time and event. */
ri_CheckTrigger(fcinfo, "RI_FKey_cascade_del", RI_TRIGTYPE_DELETE);
@@ -769,61 +850,52 @@ RI_FKey_cascade_del(PG_FUNCTION_ARGS)
*/
fk_rel = table_open(riinfo->fk_relid, RowExclusiveLock);
pk_rel = trigdata->tg_relation;
- oldslot = trigdata->tg_trigslot;
+
+ oldtable = get_event_tuplestore(trigdata, true, NULL);
if (SPI_connect() != SPI_OK_CONNECT)
elog(ERROR, "SPI_connect failed");
+ if (ri_register_trigger_data(trigdata, oldtable, NULL, NULL) !=
+ SPI_OK_TD_REGISTER)
+ elog(ERROR, "ri_register_trigger_data failed");
+
/* Fetch or prepare a saved plan for the cascaded delete */
ri_BuildQueryKey(&qkey, riinfo, RI_PLAN_CASCADE_DEL_DODELETE);
if ((qplan = ri_FetchPreparedPlan(&qkey)) == NULL)
{
- StringInfoData querybuf;
+ StringInfo querybuf = makeStringInfo();
+ StringInfo subquerybuf = makeStringInfo();
char fkrelname[MAX_QUOTED_REL_NAME_LEN];
- char attname[MAX_QUOTED_NAME_LEN];
- char paramname[16];
- const char *querysep;
- Oid queryoids[RI_MAX_NUMKEYS];
const char *fk_only;
/* ----------
* The query string built is
- * DELETE FROM [ONLY] <fktable> WHERE $1 = fkatt1 [AND ...]
- * The type id's for the $ parameters are those of the
- * corresponding PK attributes.
+ *
+ * DELETE FROM [ONLY] <fktable> f
+ * WHERE EXISTS
+ * (SELECT 1
+ * FROM tgoldtable o
+ * WHERE o.pkatt1 = f.fkatt1 [AND ...])
* ----------
*/
- initStringInfo(&querybuf);
+ appendStringInfoString(subquerybuf,
+ "SELECT 1 FROM tgoldtable o WHERE ");
+ ri_GenerateQual(subquerybuf, "AND", riinfo->nkeys,
+ "o", pk_rel, riinfo->pk_attnums,
+ "f", fk_rel, riinfo->fk_attnums,
+ riinfo->pf_eq_oprs);
+
fk_only = fk_rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE ?
"" : "ONLY ";
quoteRelationName(fkrelname, fk_rel);
- appendStringInfo(&querybuf, "DELETE FROM %s%s",
- fk_only, fkrelname);
- 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]);
- Oid pk_coll = RIAttCollation(pk_rel, riinfo->pk_attnums[i]);
- Oid fk_coll = RIAttCollation(fk_rel, riinfo->fk_attnums[i]);
-
- quoteOneName(attname,
- RIAttName(fk_rel, riinfo->fk_attnums[i]));
- sprintf(paramname, "$%d", i + 1);
- ri_GenerateQualComponent(&querybuf, querysep,
- paramname, pk_type,
- riinfo->pf_eq_oprs[i],
- attname, fk_type);
- if (pk_coll != fk_coll && !get_collation_isdeterministic(pk_coll))
- ri_GenerateQualCollation(&querybuf, pk_coll);
- querysep = "AND";
- queryoids[i] = pk_type;
- }
+ appendStringInfo(querybuf,
+ "DELETE FROM %s%s f WHERE EXISTS (%s) ",
+ fk_only, fkrelname, subquerybuf->data);
/* Prepare and save the plan */
- qplan = ri_PlanCheck(querybuf.data, riinfo->nkeys, queryoids,
- &qkey, fk_rel, pk_rel);
+ qplan = ri_PlanCheck(querybuf->data, &qkey, fk_rel, pk_rel);
}
/*
@@ -832,7 +904,6 @@ RI_FKey_cascade_del(PG_FUNCTION_ARGS)
*/
ri_PerformCheck(riinfo, &qkey, qplan,
fk_rel, pk_rel,
- oldslot, NULL,
true, /* must detect new rows */
SPI_OK_DELETE);
@@ -857,10 +928,9 @@ RI_FKey_cascade_upd(PG_FUNCTION_ARGS)
const RI_ConstraintInfo *riinfo;
Relation fk_rel;
Relation pk_rel;
- TupleTableSlot *newslot;
- TupleTableSlot *oldslot;
RI_QueryKey qkey;
SPIPlanPtr qplan;
+ Tuplestorestate *table;
/* Check that this is a valid trigger call on the right time and event. */
ri_CheckTrigger(fcinfo, "RI_FKey_cascade_upd", RI_TRIGTYPE_UPDATE);
@@ -877,12 +947,22 @@ RI_FKey_cascade_upd(PG_FUNCTION_ARGS)
*/
fk_rel = table_open(riinfo->fk_relid, RowExclusiveLock);
pk_rel = trigdata->tg_relation;
- newslot = trigdata->tg_newslot;
- oldslot = trigdata->tg_trigslot;
+
+ /*
+ * In this case, both old and new values should be in the same tuplestore
+ * because there's no useful join column.
+ */
+ table = get_event_tuplestore_for_cascade_update(trigdata, riinfo);
if (SPI_connect() != SPI_OK_CONNECT)
elog(ERROR, "SPI_connect failed");
+ /* Here it doesn't matter whether we call the table "old" or "new". */
+ if (ri_register_trigger_data(trigdata, NULL, table,
+ riinfo->desc_pk_both) !=
+ SPI_OK_TD_REGISTER)
+ elog(ERROR, "ri_register_trigger_data failed");
+
/* Fetch or prepare a saved plan for the cascaded update */
ri_BuildQueryKey(&qkey, riinfo, RI_PLAN_CASCADE_UPD_DOUPDATE);
@@ -891,21 +971,20 @@ RI_FKey_cascade_upd(PG_FUNCTION_ARGS)
StringInfoData querybuf;
StringInfoData qualbuf;
char fkrelname[MAX_QUOTED_REL_NAME_LEN];
- char attname[MAX_QUOTED_NAME_LEN];
- char paramname[16];
- const char *querysep;
- const char *qualsep;
- Oid queryoids[RI_MAX_NUMKEYS * 2];
const char *fk_only;
+ int i;
/* ----------
* The query string built is
- * UPDATE [ONLY] <fktable> SET fkatt1 = $1 [, ...]
- * WHERE $n = fkatt1 [AND ...]
- * The type id's for the $ parameters are those of the
- * corresponding PK attributes. Note that we are assuming
- * there is an assignment cast from the PK to the FK type;
- * else the parser will fail.
+ *
+ * UPDATE [ONLY] <fktable> f
+ * SET fkatt1 = n.pkatt1_new [, ...]
+ * FROM tgnewtable n
+ * WHERE
+ * f.fkatt1 = n.pkatt1_old [AND ...]
+ *
+ * Note that we are assuming there is an assignment cast from the PK
+ * to the FK type; else the parser will fail.
* ----------
*/
initStringInfo(&querybuf);
@@ -913,39 +992,43 @@ RI_FKey_cascade_upd(PG_FUNCTION_ARGS)
fk_only = fk_rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE ?
"" : "ONLY ";
quoteRelationName(fkrelname, fk_rel);
- appendStringInfo(&querybuf, "UPDATE %s%s SET",
- fk_only, fkrelname);
- querysep = "";
- qualsep = "WHERE";
- for (int i = 0, j = riinfo->nkeys; i < riinfo->nkeys; i++, j++)
+ appendStringInfo(&querybuf, "UPDATE %s%s f SET ", fk_only, fkrelname);
+
+ for (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]);
- Oid pk_coll = RIAttCollation(pk_rel, riinfo->pk_attnums[i]);
- Oid fk_coll = RIAttCollation(fk_rel, riinfo->fk_attnums[i]);
+ char *latt = ri_ColNameQuoted("", RIAttName(fk_rel, riinfo->fk_attnums[i]));
+ Oid lcoll = RIAttCollation(fk_rel, riinfo->fk_attnums[i]);
+ char ratt[NAMEDATALEN];
+ Oid rcoll = RIAttCollation(pk_rel, riinfo->pk_attnums[i]);
- quoteOneName(attname,
- RIAttName(fk_rel, riinfo->fk_attnums[i]));
- appendStringInfo(&querybuf,
- "%s %s = $%d",
- querysep, attname, i + 1);
- sprintf(paramname, "$%d", j + 1);
- ri_GenerateQualComponent(&qualbuf, qualsep,
- paramname, pk_type,
- riinfo->pf_eq_oprs[i],
- attname, fk_type);
- if (pk_coll != fk_coll && !get_collation_isdeterministic(pk_coll))
- ri_GenerateQualCollation(&querybuf, pk_coll);
- querysep = ",";
- qualsep = "AND";
- queryoids[i] = pk_type;
- queryoids[j] = pk_type;
+ snprintf(ratt, NAMEDATALEN, "n.pkatt%d_new", i + 1);
+
+ if (i > 0)
+ appendStringInfoString(&querybuf, ", ");
+
+ appendStringInfo(&querybuf, "%s = %s", latt, ratt);
+
+ if (lcoll != rcoll)
+ ri_GenerateQualCollation(&querybuf, lcoll);
+ }
+
+ appendStringInfo(&querybuf, " FROM tgnewtable n WHERE");
+
+ for (i = 0; i < riinfo->nkeys; i++)
+ {
+ char *fattname;
+
+ if (i > 0)
+ appendStringInfoString(&querybuf, " AND");
+
+ fattname =
+ ri_ColNameQuoted("f",
+ RIAttName(fk_rel, riinfo->fk_attnums[i]));
+ appendStringInfo(&querybuf, " %s = n.pkatt%d_old", fattname, i + 1);
}
- appendBinaryStringInfo(&querybuf, qualbuf.data, qualbuf.len);
/* Prepare and save the plan */
- qplan = ri_PlanCheck(querybuf.data, riinfo->nkeys * 2, queryoids,
- &qkey, fk_rel, pk_rel);
+ qplan = ri_PlanCheck(querybuf.data, &qkey, fk_rel, pk_rel);
}
/*
@@ -953,7 +1036,6 @@ RI_FKey_cascade_upd(PG_FUNCTION_ARGS)
*/
ri_PerformCheck(riinfo, &qkey, qplan,
fk_rel, pk_rel,
- oldslot, newslot,
true, /* must detect new rows */
SPI_OK_UPDATE);
@@ -1038,9 +1120,9 @@ ri_set(TriggerData *trigdata, bool is_set_null)
const RI_ConstraintInfo *riinfo;
Relation fk_rel;
Relation pk_rel;
- TupleTableSlot *oldslot;
RI_QueryKey qkey;
SPIPlanPtr qplan;
+ Tuplestorestate *oldtable;
riinfo = ri_FetchConstraintInfo(trigdata->tg_trigger,
trigdata->tg_relation, true);
@@ -1053,11 +1135,16 @@ ri_set(TriggerData *trigdata, bool is_set_null)
*/
fk_rel = table_open(riinfo->fk_relid, RowExclusiveLock);
pk_rel = trigdata->tg_relation;
- oldslot = trigdata->tg_trigslot;
+
+ oldtable = get_event_tuplestore(trigdata, true, NULL);
if (SPI_connect() != SPI_OK_CONNECT)
elog(ERROR, "SPI_connect failed");
+ if (ri_register_trigger_data(trigdata, oldtable, NULL, NULL) !=
+ SPI_OK_TD_REGISTER)
+ elog(ERROR, "ri_register_trigger_data failed");
+
/*
* Fetch or prepare a saved plan for the set null/default operation (it's
* the same query for delete and update cases)
@@ -1070,38 +1157,28 @@ ri_set(TriggerData *trigdata, bool is_set_null)
if ((qplan = ri_FetchPreparedPlan(&qkey)) == NULL)
{
StringInfoData querybuf;
- StringInfoData qualbuf;
char fkrelname[MAX_QUOTED_REL_NAME_LEN];
- char attname[MAX_QUOTED_NAME_LEN];
- char paramname[16];
const char *querysep;
- const char *qualsep;
- Oid queryoids[RI_MAX_NUMKEYS];
const char *fk_only;
/* ----------
* The query string built is
- * UPDATE [ONLY] <fktable> SET fkatt1 = {NULL|DEFAULT} [, ...]
- * WHERE $1 = fkatt1 [AND ...]
- * The type id's for the $ parameters are those of the
- * corresponding PK attributes.
+ * UPDATE [ONLY] <fktable> f
+ * SET fkatt1 = {NULL|DEFAULT} [, ...]
+ * FROM tgoldtable o
+ * WHERE o.pkatt1 = f.fkatt1 [AND ...]
* ----------
*/
initStringInfo(&querybuf);
- initStringInfo(&qualbuf);
fk_only = fk_rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE ?
"" : "ONLY ";
quoteRelationName(fkrelname, fk_rel);
- appendStringInfo(&querybuf, "UPDATE %s%s SET",
+ appendStringInfo(&querybuf, "UPDATE %s%s f SET",
fk_only, fkrelname);
querysep = "";
- qualsep = "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]);
- Oid pk_coll = RIAttCollation(pk_rel, riinfo->pk_attnums[i]);
- Oid fk_coll = RIAttCollation(fk_rel, riinfo->fk_attnums[i]);
+ char attname[MAX_QUOTED_NAME_LEN];
quoteOneName(attname,
RIAttName(fk_rel, riinfo->fk_attnums[i]));
@@ -1109,22 +1186,17 @@ ri_set(TriggerData *trigdata, bool is_set_null)
"%s %s = %s",
querysep, attname,
is_set_null ? "NULL" : "DEFAULT");
- sprintf(paramname, "$%d", i + 1);
- ri_GenerateQualComponent(&qualbuf, qualsep,
- paramname, pk_type,
- riinfo->pf_eq_oprs[i],
- attname, fk_type);
- if (pk_coll != fk_coll && !get_collation_isdeterministic(pk_coll))
- ri_GenerateQualCollation(&querybuf, pk_coll);
querysep = ",";
- qualsep = "AND";
- queryoids[i] = pk_type;
}
- appendBinaryStringInfo(&querybuf, qualbuf.data, qualbuf.len);
+
+ appendStringInfoString(&querybuf, " FROM tgoldtable o WHERE ");
+ ri_GenerateQual(&querybuf, "AND", riinfo->nkeys,
+ "o", pk_rel, riinfo->pk_attnums,
+ "f", fk_rel, riinfo->fk_attnums,
+ riinfo->pf_eq_oprs);
/* Prepare and save the plan */
- qplan = ri_PlanCheck(querybuf.data, riinfo->nkeys, queryoids,
- &qkey, fk_rel, pk_rel);
+ qplan = ri_PlanCheck(querybuf.data, &qkey, fk_rel, pk_rel);
}
/*
@@ -1132,7 +1204,6 @@ ri_set(TriggerData *trigdata, bool is_set_null)
*/
ri_PerformCheck(riinfo, &qkey, qplan,
fk_rel, pk_rel,
- oldslot, NULL,
true, /* must detect new rows */
SPI_OK_UPDATE);
@@ -1542,7 +1613,7 @@ RI_Initial_Check(Trigger *trigger, Relation fk_rel, Relation pk_rel)
ri_ReportViolation(&fake_riinfo,
pk_rel, fk_rel,
slot, tupdesc,
- RI_PLAN_CHECK_LOOKUPPK, false);
+ RI_PLAN_CHECK_LOOKUPPK_INS, false);
ExecDropSingleTupleTableSlot(slot);
}
@@ -1847,6 +1918,25 @@ ri_GenerateQualComponent(StringInfo buf,
rightop, rightoptype);
}
+/*
+ * ri_GenerateKeyList --- generate comma-separated list of key attributes.
+ */
+static void
+ri_GenerateKeyList(StringInfo buf, int nkeys,
+ const char *tabname, Relation rel,
+ const int16 *attnums)
+{
+ for (int i = 0; i < nkeys; i++)
+ {
+ char *att = ri_ColNameQuoted(tabname, RIAttName(rel, attnums[i]));
+
+ if (i > 0)
+ appendStringInfoString(buf, ", ");
+
+ appendStringInfoString(buf, att);
+ }
+}
+
/*
* ri_ColNameQuoted() --- return column name, with both table and column name
* quoted.
@@ -2007,7 +2097,7 @@ ri_FetchConstraintInfo(Trigger *trigger, Relation trig_rel, bool rel_is_pk)
errhint("Remove this referential integrity trigger and its mates, then do ALTER TABLE ADD CONSTRAINT.")));
/* Find or create a hashtable entry for the constraint */
- riinfo = ri_LoadConstraintInfo(constraintOid);
+ riinfo = ri_LoadConstraintInfo(constraintOid, trig_rel, rel_is_pk);
/* Do some easy cross-checks against the trigger call data */
if (rel_is_pk)
@@ -2043,12 +2133,15 @@ ri_FetchConstraintInfo(Trigger *trigger, Relation trig_rel, bool rel_is_pk)
* Fetch or create the RI_ConstraintInfo struct for an FK constraint.
*/
static const RI_ConstraintInfo *
-ri_LoadConstraintInfo(Oid constraintOid)
+ri_LoadConstraintInfo(Oid constraintOid, Relation trig_rel, bool rel_is_pk)
{
RI_ConstraintInfo *riinfo;
bool found;
HeapTuple tup;
Form_pg_constraint conForm;
+ MemoryContext oldcxt;
+ TupleDesc tupdesc_both;
+ Relation pk_rel;
/*
* On the first call initialize the hashtable
@@ -2100,6 +2193,24 @@ ri_LoadConstraintInfo(Oid constraintOid)
ReleaseSysCache(tup);
+ /*
+ * Construct the descriptor to store both OLD and NEW tuple into when
+ * processing ON UPDATE CASCADE. Only key columns are needed for that.
+ */
+ if (rel_is_pk)
+ pk_rel = trig_rel;
+ else
+ pk_rel = table_open(riinfo->pk_relid, AccessShareLock);
+ oldcxt = MemoryContextSwitchTo(TopMemoryContext);
+ tupdesc_both = CreateTemplateTupleDesc(2 * riinfo->nkeys);
+ /* Add the key attributes for both OLD and NEW. */
+ add_key_attrs_to_tupdesc(tupdesc_both, pk_rel, riinfo, true);
+ add_key_attrs_to_tupdesc(tupdesc_both, pk_rel, riinfo, false);
+ MemoryContextSwitchTo(oldcxt);
+ riinfo->desc_pk_both = tupdesc_both;
+ if (!rel_is_pk)
+ table_close(pk_rel, AccessShareLock);
+
/*
* For efficient processing of invalidation messages below, we keep a
* doubly-linked list, and a count, of all currently valid entries.
@@ -2165,8 +2276,8 @@ InvalidateConstraintCacheCallBack(Datum arg, int cacheid, uint32 hashvalue)
* so that we don't need to plan it again.
*/
static SPIPlanPtr
-ri_PlanCheck(const char *querystr, int nargs, Oid *argtypes,
- RI_QueryKey *qkey, Relation fk_rel, Relation pk_rel)
+ri_PlanCheck(const char *querystr, RI_QueryKey *qkey, Relation fk_rel,
+ Relation pk_rel)
{
SPIPlanPtr qplan;
Relation query_rel;
@@ -2189,7 +2300,7 @@ ri_PlanCheck(const char *querystr, int nargs, Oid *argtypes,
SECURITY_NOFORCE_RLS);
/* Create the plan */
- qplan = SPI_prepare(querystr, nargs, argtypes);
+ qplan = SPI_prepare(querystr, 0, NULL);
if (qplan == NULL)
elog(ERROR, "SPI_prepare returned %s for %s", SPI_result_code_string(SPI_result), querystr);
@@ -2211,20 +2322,15 @@ static bool
ri_PerformCheck(const RI_ConstraintInfo *riinfo,
RI_QueryKey *qkey, SPIPlanPtr qplan,
Relation fk_rel, Relation pk_rel,
- TupleTableSlot *oldslot, TupleTableSlot *newslot,
bool detectNewRows, int expect_OK)
{
- Relation query_rel,
- source_rel;
- bool source_is_pk;
+ Relation query_rel;
Snapshot test_snapshot;
Snapshot crosscheck_snapshot;
int limit;
int spi_result;
Oid save_userid;
int save_sec_context;
- 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
@@ -2235,39 +2341,6 @@ ri_PerformCheck(const RI_ConstraintInfo *riinfo,
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)
- {
- ri_ExtractValues(source_rel, newslot, riinfo, source_is_pk,
- vals, nulls);
- if (oldslot)
- ri_ExtractValues(source_rel, oldslot, riinfo, source_is_pk,
- vals + riinfo->nkeys, nulls + riinfo->nkeys);
- }
- else
- {
- ri_ExtractValues(source_rel, oldslot, riinfo, source_is_pk,
- vals, nulls);
- }
-
/*
* In READ COMMITTED mode, we just need to use an up-to-date regular
* snapshot, and we will see all rows that could be interesting. But in
@@ -2308,7 +2381,7 @@ ri_PerformCheck(const RI_ConstraintInfo *riinfo,
/* Finally we can run the query. */
spi_result = SPI_execute_snapshot(qplan,
- vals, nulls,
+ NULL, NULL,
test_snapshot, crosscheck_snapshot,
false, false, limit);
@@ -2328,30 +2401,7 @@ ri_PerformCheck(const RI_ConstraintInfo *riinfo,
RelationGetRelationName(fk_rel)),
errhint("This is most likely due to a rule having rewritten the query.")));
- return SPI_processed != 0;
-}
-
-/*
- * Extract fields from a tuple into Datum/nulls arrays
- */
-static void
-ri_ExtractValues(Relation rel, TupleTableSlot *slot,
- const RI_ConstraintInfo *riinfo, bool rel_is_pk,
- Datum *vals, char *nulls)
-{
- const int16 *attnums;
- bool isnull;
-
- if (rel_is_pk)
- attnums = riinfo->pk_attnums;
- else
- attnums = riinfo->fk_attnums;
-
- for (int i = 0; i < riinfo->nkeys; i++)
- {
- vals[i] = slot_getattr(slot, attnums[i], &isnull);
- nulls[i] = isnull ? 'n' : ' ';
- }
+ return SPI_processed > 0;
}
/*
@@ -2378,25 +2428,28 @@ ri_ReportViolation(const RI_ConstraintInfo *riinfo,
bool has_perm = true;
/*
- * Determine which relation to complain about. If tupdesc wasn't passed
- * by caller, assume the violator tuple came from there.
+ * Determine which relation to complain about.
*/
- onfk = (queryno == RI_PLAN_CHECK_LOOKUPPK);
+ onfk = (queryno == RI_PLAN_CHECK_LOOKUPPK_INS ||
+ queryno == RI_PLAN_CHECK_LOOKUPPK_UPD);
if (onfk)
{
attnums = riinfo->fk_attnums;
rel_oid = fk_rel->rd_id;
- if (tupdesc == NULL)
- tupdesc = fk_rel->rd_att;
}
else
{
attnums = riinfo->pk_attnums;
rel_oid = pk_rel->rd_id;
- if (tupdesc == NULL)
- tupdesc = pk_rel->rd_att;
}
+ /*
+ * If tupdesc wasn't passed by caller, assume the violator tuple matches
+ * the descriptor of the violatorslot.
+ */
+ if (tupdesc == NULL)
+ tupdesc = violatorslot->tts_tupleDescriptor;
+
/*
* Check permissions- if the user does not have access to view the data in
* any of the key columns then we don't include the errdetail() below.
@@ -2443,8 +2496,7 @@ ri_ReportViolation(const RI_ConstraintInfo *riinfo,
initStringInfo(&key_values);
for (int idx = 0; idx < riinfo->nkeys; idx++)
{
- int fnum = attnums[idx];
- Form_pg_attribute att = TupleDescAttr(tupdesc, fnum - 1);
+ Form_pg_attribute att = TupleDescAttr(tupdesc, idx);
char *name,
*val;
Datum datum;
@@ -2452,7 +2504,7 @@ ri_ReportViolation(const RI_ConstraintInfo *riinfo,
name = NameStr(att->attname);
- datum = slot_getattr(violatorslot, fnum, &isnull);
+ datum = slot_getattr(violatorslot, idx + 1, &isnull);
if (!isnull)
{
Oid foutoid;
@@ -2921,3 +2973,293 @@ RI_FKey_trigger_type(Oid tgfoid)
return RI_TRIGGER_NONE;
}
+
+/*
+ * Wrapper around SPI_register_trigger_data() that lets us register the RI
+ * trigger tuplestores w/o having to set tg_oldtable/tg_newtable and also w/o
+ * having to set tgoldtable/tgnewtable in pg_trigger.
+ *
+ * XXX This is rather a hack, try to invent something better.
+ */
+static int
+ri_register_trigger_data(TriggerData *tdata, Tuplestorestate *oldtable,
+ Tuplestorestate *newtable, TupleDesc desc)
+{
+ TriggerData *td = (TriggerData *) palloc(sizeof(TriggerData));
+ Trigger *tg = (Trigger *) palloc(sizeof(Trigger));
+ int result;
+
+ Assert(tdata->tg_trigger->tgoldtable == NULL &&
+ tdata->tg_trigger->tgnewtable == NULL);
+
+ *td = *tdata;
+
+ td->tg_oldtable = oldtable;
+ td->tg_newtable = newtable;
+
+ *tg = *tdata->tg_trigger;
+ tg->tgoldtable = pstrdup("tgoldtable");
+ tg->tgnewtable = pstrdup("tgnewtable");
+ td->tg_trigger = tg;
+ td->desc = desc;
+
+ result = SPI_register_trigger_data(td);
+
+ return result;
+}
+
+/*
+ * Turn TID array into a tuplestore. If snapshot is passed, only use tuples
+ * visible by this snapshot.
+ */
+static Tuplestorestate *
+get_event_tuplestore(TriggerData *trigdata, bool old, Snapshot snapshot)
+{
+ ResourceOwner saveResourceOwner;
+ Tuplestorestate *result;
+ TIDArray *ta;
+ ItemPointer it;
+ TupleTableSlot *slot;
+ int i;
+
+ saveResourceOwner = CurrentResourceOwner;
+ CurrentResourceOwner = CurTransactionResourceOwner;
+ result = tuplestore_begin_heap(false, false, work_mem);
+ CurrentResourceOwner = saveResourceOwner;
+
+ if (old)
+ {
+ ta = trigdata->ri_tids_old;
+ slot = trigdata->tg_trigslot;
+ }
+ else
+ {
+ ta = trigdata->ri_tids_new;
+ slot = trigdata->tg_newslot;
+ }
+
+ it = ta->tids;
+ for (i = 0; i < ta->n; i++)
+ {
+ ExecClearTuple(slot);
+
+ if (!table_tuple_fetch_row_version(trigdata->tg_relation, it,
+ SnapshotAny, slot))
+ {
+ const char *tuple_kind = old ? "tuple1" : "tuple2";
+
+ elog(ERROR, "failed to fetch %s for AFTER trigger", tuple_kind);
+ }
+
+ if (snapshot)
+ {
+ if (!table_tuple_satisfies_snapshot(trigdata->tg_relation, slot,
+ snapshot))
+ continue;
+
+ /*
+ * In fact the snapshot is passed iff the slot contains a tuple of
+ * the FK table being inserted / updated, so perform one more test
+ * before we add the tuple to the tuplestore. Otherwise we might
+ * need to remove the tuple later, which would effectively mean to
+ * create a new tuplestore and put only a subset of tuples into
+ * it.
+ */
+ if (!RI_FKey_check_query_required(trigdata->tg_trigger,
+ trigdata->tg_relation, slot))
+ continue;
+ }
+
+ /*
+ * TODO Only store the key attributes.
+ */
+ tuplestore_puttupleslot(result, slot);
+ it++;
+ }
+
+ return result;
+}
+
+/*
+ * Like get_event_tuplestore(), but put both old and new key values into the
+ * same tuple. If the query (see RI_FKey_cascade_upd) used two tuplestores, it
+ * whould have to join them somehow, but there's not suitable join column.
+ */
+static Tuplestorestate *
+get_event_tuplestore_for_cascade_update(TriggerData *trigdata,
+ const RI_ConstraintInfo *riinfo)
+{
+ ResourceOwner saveResourceOwner;
+ Tuplestorestate *result;
+ TIDArray *ta_old,
+ *ta_new;
+ ItemPointer it_old,
+ it_new;
+ TupleTableSlot *slot_old,
+ *slot_new;
+ int i;
+ Datum *values,
+ *key_values;
+ bool *nulls,
+ *key_nulls;
+ MemoryContext tuple_context;
+ Relation rel = trigdata->tg_relation;
+ TupleDesc desc_rel = RelationGetDescr(rel);
+
+ saveResourceOwner = CurrentResourceOwner;
+ CurrentResourceOwner = CurTransactionResourceOwner;
+ result = tuplestore_begin_heap(false, false, work_mem);
+ CurrentResourceOwner = saveResourceOwner;
+
+ /*
+ * This context will be used for the contents of "values".
+ *
+ * CurrentMemoryContext should be the "batch context", as passed to
+ * AfterTriggerExecuteRI().
+ */
+ tuple_context =
+ AllocSetContextCreate(CurrentMemoryContext,
+ "AfterTriggerCascadeUpdateContext",
+ ALLOCSET_DEFAULT_SIZES);
+
+ ta_old = trigdata->ri_tids_old;
+ ta_new = trigdata->ri_tids_new;
+ Assert(ta_old->n == ta_new->n);
+
+ slot_old = trigdata->tg_trigslot;
+ slot_new = trigdata->tg_newslot;
+
+ key_values = (Datum *) palloc(riinfo->nkeys * 2 * sizeof(Datum));
+ key_nulls = (bool *) palloc(riinfo->nkeys * 2 * sizeof(bool));
+ values = (Datum *) palloc(desc_rel->natts * sizeof(Datum));
+ nulls = (bool *) palloc(desc_rel->natts * sizeof(bool));
+
+ it_old = ta_old->tids;
+ it_new = ta_new->tids;
+ for (i = 0; i < ta_old->n; i++)
+ {
+ MemoryContext oldcxt;
+
+ MemoryContextReset(tuple_context);
+ oldcxt = MemoryContextSwitchTo(tuple_context);
+
+ /* Add values of the PK table, followed by the FK ones. */
+ add_key_values(slot_old, riinfo, trigdata->tg_relation, it_old,
+ key_values, key_nulls, values, nulls, true);
+ add_key_values(slot_new, riinfo, trigdata->tg_relation, it_new,
+ key_values, key_nulls, values, nulls, false);
+ MemoryContextSwitchTo(oldcxt);
+
+ tuplestore_putvalues(result, riinfo->desc_pk_both, key_values,
+ key_nulls);
+
+ it_old++;
+ it_new++;
+ }
+ MemoryContextDelete(tuple_context);
+
+ return result;
+}
+
+/*
+ * Subroutine of get_event_tuplestore_for_cascade_update(), to add key
+ * attributes of the OLD or the NEW table to tuple descriptor.
+ *
+ */
+static void
+add_key_attrs_to_tupdesc(TupleDesc tupdesc, Relation rel,
+ const RI_ConstraintInfo *riinfo, bool old)
+{
+ int i,
+ first;
+ const char *kind;
+
+ first = old ? 1 : riinfo->nkeys + 1;
+ kind = old ? "old" : "new";
+
+ for (i = 0; i < riinfo->nkeys; i++)
+ {
+ int16 attnum;
+ Oid atttypid;
+ char attname[NAMEDATALEN];
+ Form_pg_attribute att;
+
+ attnum = riinfo->pk_attnums[i];
+ atttypid = RIAttType(rel, attnum);
+
+ /*
+ * Generate unique names instead of e.g. using prefix to distinguish
+ * the old values from new ones. The prefix might be a problem due to
+ * the limited attribute name length.
+ */
+ snprintf(attname, NAMEDATALEN, "pkatt%d_%s", i + 1, kind);
+
+ att = tupdesc->attrs;
+ TupleDescInitEntry(tupdesc, first + i, attname, atttypid,
+ att->atttypmod, att->attndims);
+ att++;
+ }
+}
+
+/*
+ * Retrieve tuple using given slot, deform it and add the attribute values to
+ * "key_values" and "key_null" arrays. "values" and "nulls" is a workspace to
+ * deform the tuple into. "old" tells whether the slot contains data for the
+ * OLD table or for the NEW one.
+ */
+static void
+add_key_values(TupleTableSlot *slot, const RI_ConstraintInfo *riinfo,
+ Relation rel, ItemPointer ip,
+ Datum *key_values, bool *key_nulls,
+ Datum *values, bool *nulls, bool old)
+{
+ HeapTuple tuple;
+ bool shouldfree;
+ int i,
+ n;
+
+ ExecClearTuple(slot);
+ if (!table_tuple_fetch_row_version(rel, ip, SnapshotAny, slot))
+ {
+ const char *tuple_kind = old ? "tuple1" : "tuple2";
+
+ elog(ERROR, "failed to fetch %s for AFTER trigger", tuple_kind);
+ }
+ tuple = ExecFetchSlotHeapTuple(slot, false, &shouldfree);
+
+ heap_deform_tuple(tuple, slot->tts_tupleDescriptor, values, nulls);
+
+ /* Where to start in the output arrays? */
+ n = old ? 0 : riinfo->nkeys;
+
+ /* Pick the key values and store them in the output arrays. */
+ for (i = 0; i < riinfo->nkeys; i++)
+ {
+ int16 attnum = riinfo->pk_attnums[i];
+
+ key_values[n] = values[attnum - 1];
+ key_nulls[n] = nulls[attnum - 1];
+ n++;
+ }
+
+ if (shouldfree)
+ pfree(tuple);
+}
+
+
+/*
+ * Retrieve the row that violates RI constraint and return it in a tuple slot.
+ */
+static TupleTableSlot *
+get_violator_tuple(Relation rel)
+{
+ HeapTuple tuple;
+ TupleTableSlot *slot;
+
+ Assert(SPI_tuptable && SPI_tuptable->numvals == 1);
+
+ tuple = SPI_tuptable->vals[0];
+ slot = MakeSingleTupleTableSlot(SPI_tuptable->tupdesc, &TTSOpsHeapTuple);
+ ExecStoreHeapTuple(tuple, slot, false);
+ return slot;
+}
diff --git a/src/include/commands/trigger.h b/src/include/commands/trigger.h
index a40ddf5db5..70c214f069 100644
--- a/src/include/commands/trigger.h
+++ b/src/include/commands/trigger.h
@@ -27,19 +27,42 @@
typedef uint32 TriggerEvent;
+/*
+ * An intermediate storage for TIDs, in order to process multiple events by a
+ * single call of RI trigger.
+ *
+ * XXX Introduce a size limit and make caller of add_tid() aware of it?
+ */
+typedef struct TIDArray
+{
+ ItemPointerData *tids;
+ uint64 n;
+ uint64 nmax;
+} TIDArray;
+
typedef struct TriggerData
{
NodeTag type;
+ int tgindx;
TriggerEvent tg_event;
Relation tg_relation;
HeapTuple tg_trigtuple;
HeapTuple tg_newtuple;
Trigger *tg_trigger;
+ bool is_ri_trigger;
TupleTableSlot *tg_trigslot;
TupleTableSlot *tg_newslot;
Tuplestorestate *tg_oldtable;
Tuplestorestate *tg_newtable;
const Bitmapset *tg_updatedcols;
+
+ TupleDesc desc;
+
+ /*
+ * RI triggers receive TIDs and retrieve the tuples before they're needed.
+ */
+ TIDArray *ri_tids_old;
+ TIDArray *ri_tids_new;
} TriggerData;
/*
--
2.20.1
