public inbox for [email protected]help / color / mirror / Atom feed
[PATCH v10 2/2] fk_arrays_elems 4+ messages / 3 participants [nested] [flat]
* [PATCH v10 2/2] fk_arrays_elems @ 2021-03-15 15:10 Mark Rofail <[email protected]> 0 siblings, 0 replies; 4+ messages in thread From: Mark Rofail @ 2021-03-15 15:10 UTC (permalink / raw) --- doc/src/sgml/catalogs.sgml | 16 + doc/src/sgml/ddl.sgml | 105 ++++ doc/src/sgml/ref/create_table.sgml | 113 +++- src/backend/catalog/heap.c | 1 + src/backend/catalog/index.c | 1 + src/backend/catalog/pg_constraint.c | 36 +- src/backend/commands/tablecmds.c | 126 ++++- src/backend/commands/trigger.c | 1 + src/backend/commands/typecmds.c | 1 + src/backend/nodes/copyfuncs.c | 1 + src/backend/nodes/equalfuncs.c | 1 + src/backend/nodes/outfuncs.c | 1 + src/backend/parser/gram.y | 73 ++- src/backend/parser/parse_utilcmd.c | 2 + src/backend/utils/adt/ri_triggers.c | 194 +++++-- src/backend/utils/adt/ruleutils.c | 88 +++- src/backend/utils/cache/relcache.c | 2 +- src/include/catalog/pg_constraint.h | 14 +- src/include/nodes/parsenodes.h | 5 + src/include/parser/kwlist.h | 1 + src/test/regress/expected/element_fk.out | 625 +++++++++++++++++++++++ src/test/regress/parallel_schedule | 2 +- src/test/regress/serial_schedule | 1 + src/test/regress/sql/element_fk.sql | 475 +++++++++++++++++ 24 files changed, 1808 insertions(+), 77 deletions(-) create mode 100644 src/test/regress/expected/element_fk.out create mode 100644 src/test/regress/sql/element_fk.sql diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml index b1de6d0674..e8be7a0229 100644 --- a/doc/src/sgml/catalogs.sgml +++ b/doc/src/sgml/catalogs.sgml @@ -2646,6 +2646,16 @@ SCRAM-SHA-256$<replaceable><iteration count></replaceable>:<replaceable>&l </para></entry> </row> + <row> + <entry><structfield>confreftype</structfield></entry> + <entry><type>char[]</type></entry> + <entry></entry> + <entry>If a foreign key, the reference semantics for each column: + <literal>p</literal> = plain (simple equality), + <literal>e</literal> = each element of referencing array must have a match + </entry> + </row> + <row> <entry role="catalog_table_entry"><para role="column_definition"> <structfield>conpfeqop</structfield> <type>oid[]</type> @@ -2701,6 +2711,12 @@ SCRAM-SHA-256$<replaceable><iteration count></replaceable>:<replaceable>&l </tgroup> </table> + <para> + When <structfield>confreftype</structfield> indicates array to scalar + foreign key reference semantics, the equality operators listed in + <structfield>conpfeqop</structfield> etc are for the array's element type. + </para> + <para> In the case of an exclusion constraint, <structfield>conkey</structfield> is only useful for constraint elements that are simple column references. diff --git a/doc/src/sgml/ddl.sgml b/doc/src/sgml/ddl.sgml index 422c1180ba..19ddcf1c52 100644 --- a/doc/src/sgml/ddl.sgml +++ b/doc/src/sgml/ddl.sgml @@ -1079,6 +1079,111 @@ CREATE TABLE order_items ( </para> </sect2> + <sect2 id="ddl-constraints-element-fk"> + <title>Array Element Foreign Keys</title> + + <indexterm> + <primary>Array Element Foreign Keys</primary> + </indexterm> + + <indexterm> + <primary>ELEMENT foreign key</primary> + </indexterm> + + <indexterm> + <primary>constraint</primary> + <secondary>Array ELEMENT foreign key</secondary> + </indexterm> + + <indexterm> + <primary>constraint</primary> + <secondary>ELEMENT foreign key</secondary> + </indexterm> + + <indexterm> + <primary>referential integrity</primary> + </indexterm> + + <para> + Another option you have with foreign keys is to use a referencing column + which is an array of elements with the same type (or a compatible one) as + the referenced column in the related table. This feature is called + <firstterm>Array Element Foreign Keys</firstterm> and is implemented in PostgreSQL + with <firstterm>EACH ELEMENT OF foreign key constraints</firstterm>, as + described in the following example: +<programlisting> +CREATE TABLE drivers ( + driver_id integer PRIMARY KEY, + first_name text, + last_name text +); + +CREATE TABLE racing ( + race_id integer PRIMARY KEY, + title text, + race_day date, + final_positions integer[], + FOREIGN KEY <emphasis>(EACH ELEMENT OF final_positions) REFERENCES drivers</emphasis> +); +</programlisting> + The above example uses an array (<literal>final_positions</literal>) to + store the results of a race: for each of its elements a referential + integrity check is enforced on the <literal>drivers</literal> table. Note + that <literal>(EACH ELEMENT OF ...) REFERENCES</literal> is an extension of + PostgreSQL and it is not included in the SQL standard. + </para> + + <para> + We currently only support the table constraint form. + </para> + + <para> + Even though the most common use case for Array Element Foreign Keys constraint is on + a single column key, you can define an Array Element Foreign Keys constraint on a + group of columns. +<programlisting> +CREATE TABLE available_moves ( + kind text, + move text, + description text, + PRIMARY KEY (kind, move) +); + +CREATE TABLE paths ( + description text, + kind text, + moves text[], + <emphasis>FOREIGN KEY (kind, EACH ELEMENT OF moves) REFERENCES available_moves (kind, move)</emphasis> +); + +INSERT INTO available_moves VALUES ('relative', 'LN', 'look north'); +INSERT INTO available_moves VALUES ('relative', 'RL', 'rotate left'); +INSERT INTO available_moves VALUES ('relative', 'RR', 'rotate right'); +INSERT INTO available_moves VALUES ('relative', 'MF', 'move forward'); +INSERT INTO available_moves VALUES ('absolute', 'N', 'move north'); +INSERT INTO available_moves VALUES ('absolute', 'S', 'move south'); +INSERT INTO available_moves VALUES ('absolute', 'E', 'move east'); +INSERT INTO available_moves VALUES ('absolute', 'W', 'move west'); + +INSERT INTO paths VALUES ('L-shaped path', 'relative', '{LN, RL, MF, RR, MF, MF}'); +INSERT INTO paths VALUES ('L-shaped path', 'absolute', '{W, N, N}'); +</programlisting> + In addition to standard foreign key requirements, array + <literal>ELEMENT</literal> foreign key constraints require that the + referencing column is an array of a compatible type to the corresponding + referenced column. + + Note that we currently only support one array reference per foreign key. + </para> + + <para> + For more detailed information on Array Element Foreign Keys options and special + cases, please refer to the documentation for + <xref linkend="sql-createtable-foreign-key"/> and + <xref linkend="sql-createtable-element-foreign-key-constraints"/>. + </para> + </sect2> + <sect2 id="ddl-constraints-exclusion"> <title>Exclusion Constraints</title> diff --git a/doc/src/sgml/ref/create_table.sgml b/doc/src/sgml/ref/create_table.sgml index 71703da85a..ac18c53578 100644 --- a/doc/src/sgml/ref/create_table.sgml +++ b/doc/src/sgml/ref/create_table.sgml @@ -1032,10 +1032,10 @@ WITH ( MODULUS <replaceable class="parameter">numeric_literal</replaceable>, REM </listitem> </varlistentry> - <varlistentry> + <varlistentry id="sql-createtable-foreign-key" xreflabel="FOREIGN KEY"> <term><literal>REFERENCES <replaceable class="parameter">reftable</replaceable> [ ( <replaceable class="parameter">refcolumn</replaceable> ) ] [ MATCH <replaceable class="parameter">matchtype</replaceable> ] [ ON DELETE <replaceable class="parameter">referential_action</replaceable> ] [ ON UPDATE <replaceable class="parameter">referential_action</replaceable> ]</literal> (column constraint)</term> - <term><literal>FOREIGN KEY ( <replaceable class="parameter">column_name</replaceable> [, ... ] ) + <term><literal>FOREIGN KEY ( [EACH ELEMENT OF] <replaceable class="parameter">column_name</replaceable> [, ... ] ) REFERENCES <replaceable class="parameter">reftable</replaceable> [ ( <replaceable class="parameter">refcolumn</replaceable> [, ... ] ) ] [ MATCH <replaceable class="parameter">matchtype</replaceable> ] [ ON DELETE <replaceable class="parameter">referential_action</replaceable> ] @@ -1060,6 +1060,18 @@ WITH ( MODULUS <replaceable class="parameter">numeric_literal</replaceable>, REM tables and permanent tables. </para> + <para> + In case the column name <replaceable class="parameter">column</replaceable> is + prepended with the <literal>EACH ELEMENT OF</literal> keyword and <replaceable + class="parameter">column</replaceable> is an array of elements compatible with + the corresponding <replaceable class="parameter">refcolumn</replaceable> in + <replaceable class="parameter">reftable</replaceable>, an Array Element Foreign Key + constraint is put in place (see <xref + linkend="sql-createtable-element-foreign-key-constraints"/> for more + information). Multi-column keys with more than one <literal>EACH ELEMENT + OF</literal> column are currently not allowed. + </para> + <para> A value inserted into the referencing column(s) is matched against the values of the referenced table and referenced columns using the @@ -1123,7 +1135,8 @@ WITH ( MODULUS <replaceable class="parameter">numeric_literal</replaceable>, REM <para> Delete any rows referencing the deleted row, or update the values of the referencing column(s) to the new values of the - referenced columns, respectively. + referenced columns, respectively. + Currently not supported with Array Element Foreign Keys. </para> </listitem> </varlistentry> @@ -1133,6 +1146,7 @@ WITH ( MODULUS <replaceable class="parameter">numeric_literal</replaceable>, REM <listitem> <para> Set the referencing column(s) to null. + Currently not supported with Array Element Foreign Keys. </para> </listitem> </varlistentry> @@ -1144,6 +1158,7 @@ WITH ( MODULUS <replaceable class="parameter">numeric_literal</replaceable>, REM Set the referencing column(s) to their default values. (There must be a row in the referenced table matching the default values, if they are not null, or the operation will fail.) + Currently not supported with Array Element Foreign Keys. </para> </listitem> </varlistentry> @@ -1159,6 +1174,89 @@ WITH ( MODULUS <replaceable class="parameter">numeric_literal</replaceable>, REM </listitem> </varlistentry> + <varlistentry id="sql-createtable-element-foreign-key-constraints" xreflabel="Array ELEMENT Foreign Keys"> + <term><literal>EACH ELEMENT OF <replaceable class="parameter">reftable</replaceable> [ ( <replaceable class="parameter">refcolumn</replaceable> ) ] [ MATCH <replaceable class="parameter">matchtype</replaceable> ] [ ON DELETE <replaceable class="parameter">action</replaceable> ] [ ON UPDATE <replaceable class="parameter">action</replaceable> ]</literal> (column constraint)</term> + + <listitem> + <para> + The <literal>EACH ELEMENT OF REFERENCES</literal> definition specifies an + <quote>Array Element Foreign Key</quote>, a special kind of foreign key constraint + requiring the referencing column to be an array of elements of the same type (or + a compatible one) as the referenced column in the referenced table. The value of + each element of the <replaceable class="parameter">refcolumn</replaceable> array + will be matched against some row of <replaceable + class="parameter">reftable</replaceable>. + </para> + + <para> + Array Element Foreign Keys are an extension of PostgreSQL and are not included in the SQL standard. + </para> + + <para> + Even with Array Element Foreign Keys, modifications in the referenced column can trigger + actions to be performed on the referencing array. Similarly to standard foreign + keys, you can specify these actions using the <literal>ON DELETE</literal> and + <literal>ON UPDATE</literal> clauses. However, only the following actions for + each clause are currently allowed: + + <variablelist> + <varlistentry> + <term><literal>ON UPDATE/DELETE NO ACTION</literal></term> + <listitem> + <para> + Same as standard foreign key constraints. This is the default action. + </para> + </listitem> + </varlistentry> + + <varlistentry> + <term><literal>ON UPDATE/DELETE RESTRICT</literal></term> + <listitem> + <para> + Same as standard foreign key constraints. + </para> + </listitem> + </varlistentry> + </variablelist> + </para> + + <para> + It is advisable to index the referencing column using a GIN index as it + considerably enhances the performance. Also concerning coercion while using the + GIN index: + +<programlisting> +CREATE TABLE pktableforarray ( + ptest1 int2 PRIMARY KEY, + ptest2 text +); + +CREATE TABLE fktableforarray ( + ftest1 int4[], + FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES pktableforarray, + ftest2 int +); +</programlisting> + This syntax is fine since it will cast ptest1 to int4 upon RI checks, + +<programlisting> +CREATE TABLE pktableforarray ( + ptest1 int4 PRIMARY KEY, + ptest2 text +); + +CREATE TABLE fktableforarray ( + ftest1 int2[], + FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES pktableforarray, + ftest2 int +); +</programlisting> + however, this syntax will cast ftest1 to int4 upon RI checks, thus defeating the + purpose of the index. + </para> + </listitem> + </varlistentry> + <varlistentry> <term><literal>DEFERRABLE</literal></term> <term><literal>NOT DEFERRABLE</literal></term> @@ -2306,6 +2404,15 @@ CREATE TABLE cities_partdef </para> </refsect2> + <refsect2> + <title id="sql-createtable-foreign-key-arrays">Array <literal>ELEMENT</literal> Foreign Keys</title> + + <para> + Array <literal>ELEMENT</literal> Foreign Keys and the <literal>EACH ELEMENT + OF</literal> clause are a <productname>PostgreSQL</productname> extension. + </para> + </refsect2> + <refsect2> <title><literal>PARTITION BY</literal> Clause</title> diff --git a/src/backend/catalog/heap.c b/src/backend/catalog/heap.c index 9abc4a1f55..a9e2ab33b0 100644 --- a/src/backend/catalog/heap.c +++ b/src/backend/catalog/heap.c @@ -2463,6 +2463,7 @@ StoreRelCheck(Relation rel, const char *ccname, Node *expr, NULL, NULL, NULL, + NULL, 0, ' ', ' ', diff --git a/src/backend/catalog/index.c b/src/backend/catalog/index.c index 4ef61b5efd..c315142733 100644 --- a/src/backend/catalog/index.c +++ b/src/backend/catalog/index.c @@ -2110,6 +2110,7 @@ index_constraint_create(Relation heapRelation, NULL, NULL, NULL, + NULL, 0, ' ', ' ', diff --git a/src/backend/catalog/pg_constraint.c b/src/backend/catalog/pg_constraint.c index 0081558c48..be1fc54a35 100644 --- a/src/backend/catalog/pg_constraint.c +++ b/src/backend/catalog/pg_constraint.c @@ -62,6 +62,7 @@ CreateConstraintEntry(const char *constraintName, Oid indexRelId, Oid foreignRelId, const int16 *foreignKey, + const char *foreignRefType, const Oid *pfEqOp, const Oid *ppEqOp, const Oid *ffEqOp, @@ -84,6 +85,7 @@ CreateConstraintEntry(const char *constraintName, Datum values[Natts_pg_constraint]; ArrayType *conkeyArray; ArrayType *confkeyArray; + ArrayType *confreftypeArray; ArrayType *conpfeqopArray; ArrayType *conppeqopArray; ArrayType *conffeqopArray; @@ -123,7 +125,11 @@ CreateConstraintEntry(const char *constraintName, for (i = 0; i < foreignNKeys; i++) fkdatums[i] = Int16GetDatum(foreignKey[i]); confkeyArray = construct_array(fkdatums, foreignNKeys, - INT2OID, 2, true, TYPALIGN_SHORT); + INT2OID, sizeof(int16), true, TYPALIGN_SHORT); + for (i = 0; i < foreignNKeys; i++) + fkdatums[i] = CharGetDatum(foreignRefType[i]); + confreftypeArray = construct_array(fkdatums, foreignNKeys, + CHAROID, sizeof(char), true, TYPALIGN_CHAR); for (i = 0; i < foreignNKeys; i++) fkdatums[i] = ObjectIdGetDatum(pfEqOp[i]); conpfeqopArray = construct_array(fkdatums, foreignNKeys, @@ -140,6 +146,7 @@ CreateConstraintEntry(const char *constraintName, else { confkeyArray = NULL; + confreftypeArray = NULL; conpfeqopArray = NULL; conppeqopArray = NULL; conffeqopArray = NULL; @@ -196,6 +203,11 @@ CreateConstraintEntry(const char *constraintName, else nulls[Anum_pg_constraint_confkey - 1] = true; + if (confreftypeArray) + values[Anum_pg_constraint_confreftype - 1] = PointerGetDatum(confreftypeArray); + else + nulls[Anum_pg_constraint_confreftype - 1] = true; + if (conpfeqopArray) values[Anum_pg_constraint_conpfeqop - 1] = PointerGetDatum(conpfeqopArray); else @@ -1163,7 +1175,8 @@ get_primary_key_attnos(Oid relid, bool deferrableOk, Oid *constraintOid) void DeconstructFkConstraintRow(HeapTuple tuple, int *numfks, AttrNumber *conkey, AttrNumber *confkey, - Oid *pf_eq_oprs, Oid *pp_eq_oprs, Oid *ff_eq_oprs) + Oid *pf_eq_oprs, Oid *pp_eq_oprs, Oid *ff_eq_oprs, + char *fk_reftypes) { Oid constrId; Datum adatum; @@ -1208,6 +1221,25 @@ DeconstructFkConstraintRow(HeapTuple tuple, int *numfks, if ((Pointer) arr != DatumGetPointer(adatum)) pfree(arr); /* free de-toasted copy, if any */ + if (fk_reftypes) + { + adatum = SysCacheGetAttr(CONSTROID, tuple, + Anum_pg_constraint_confreftype, &isNull); + if (isNull) + elog(ERROR, "null confreftype for constraint %u", constrId); + arr = DatumGetArrayTypeP(adatum); /* ensure not toasted */ + if (ARR_NDIM(arr) != 1 || + ARR_ELEMTYPE(arr) != CHAROID) + elog(ERROR, "confreftype is not a 1-D char array"); + if (ARR_HASNULL(arr)) + elog(ERROR, "confreftype contains nulls"); + if (ARR_DIMS(arr)[0] != numkeys) + elog(ERROR, "confreftype length does not equal the number of foreign keys"); + memcpy(fk_reftypes, ARR_DATA_PTR(arr), numkeys * sizeof(char)); + if ((Pointer) arr != DatumGetPointer(adatum)) + pfree(arr); /* free de-toasted copy, if any */ + } + if (pf_eq_oprs) { adatum = SysCacheGetAttr(CONSTROID, tuple, diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c index 559fa1d2e5..91764a6a43 100644 --- a/src/backend/commands/tablecmds.c +++ b/src/backend/commands/tablecmds.c @@ -39,6 +39,7 @@ #include "catalog/pg_inherits.h" #include "catalog/pg_namespace.h" #include "catalog/pg_opclass.h" +#include "catalog/pg_operator.h" #include "catalog/pg_tablespace.h" #include "catalog/pg_trigger.h" #include "catalog/pg_type.h" @@ -447,12 +448,12 @@ static ObjectAddress ATAddForeignKeyConstraint(List **wqueue, AlteredTableInfo * LOCKMODE lockmode); static ObjectAddress addFkRecurseReferenced(List **wqueue, Constraint *fkconstraint, Relation rel, Relation pkrel, Oid indexOid, Oid parentConstr, - int numfks, int16 *pkattnum, int16 *fkattnum, + int numfks, int16 *pkattnum, int16 *fkattnum, char *fkreftypes, Oid *pfeqoperators, Oid *ppeqoperators, Oid *ffeqoperators, bool old_check_ok); -static void addFkRecurseReferencing(List **wqueue, Constraint *fkconstraint, - Relation rel, Relation pkrel, Oid indexOid, Oid parentConstr, - int numfks, int16 *pkattnum, int16 *fkattnum, +static void addFkRecurseReferencing(List **wqueue, Constraint *fkconstraint, Relation rel, + Relation pkrel, Oid indexOid, Oid parentConstr, + int numfks, int16 *pkattnum, int16 *fkattnum, char *fkreftypes, Oid *pfeqoperators, Oid *ppeqoperators, Oid *ffeqoperators, bool old_check_ok, LOCKMODE lockmode); static void CloneForeignKeyConstraints(List **wqueue, Relation parentRel, @@ -8485,6 +8486,7 @@ ATAddForeignKeyConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel, Relation pkrel; int16 pkattnum[INDEX_MAX_KEYS]; int16 fkattnum[INDEX_MAX_KEYS]; + char fkreftypes[INDEX_MAX_KEYS]; Oid pktypoid[INDEX_MAX_KEYS]; Oid fktypoid[INDEX_MAX_KEYS]; Oid opclasses[INDEX_MAX_KEYS]; @@ -8492,9 +8494,11 @@ ATAddForeignKeyConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel, Oid ppeqoperators[INDEX_MAX_KEYS]; Oid ffeqoperators[INDEX_MAX_KEYS]; int i; + ListCell *lc; int numfks, numpks; Oid indexOid; + bool has_each_element; bool old_check_ok; ObjectAddress address; ListCell *old_pfeqop_item = list_head(fkconstraint->old_conpfeqop); @@ -8583,6 +8587,7 @@ ATAddForeignKeyConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel, */ MemSet(pkattnum, 0, sizeof(pkattnum)); MemSet(fkattnum, 0, sizeof(fkattnum)); + MemSet(fkreftypes, 0, sizeof(fkreftypes)); MemSet(pktypoid, 0, sizeof(pktypoid)); MemSet(fktypoid, 0, sizeof(fktypoid)); MemSet(opclasses, 0, sizeof(opclasses)); @@ -8594,6 +8599,54 @@ ATAddForeignKeyConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel, fkconstraint->fk_attrs, fkattnum, fktypoid); + /* + * Validate the reference semantics codes, too, and convert list to array + * format to pass to CreateConstraintEntry. + */ + Assert(list_length(fkconstraint->fk_reftypes) == numfks); + has_each_element = false; + i = 0; + foreach(lc, fkconstraint->fk_reftypes) + { + char reftype = lfirst_int(lc); + + switch (reftype) + { + case FKCONSTR_REF_PLAIN: + /* OK, nothing to do */ + break; + case FKCONSTR_REF_EACH_ELEMENT: + /* At most one FK column can be an array reference */ + if (has_each_element) + ereport(ERROR, + (errcode(ERRCODE_INVALID_FOREIGN_KEY), + errmsg("foreign keys support only one array column"))); + + has_each_element = true; + break; + default: + elog(ERROR, "invalid fk_reftype: %d", (int) reftype); + break; + } + fkreftypes[i] = reftype; + i++; + } + + /* + * Array foreign keys support only UPDATE/DELETE NO ACTION, UPDATE/DELETE + * RESTRICT + */ + if (has_each_element) + { + if ((fkconstraint->fk_upd_action != FKCONSTR_ACTION_NOACTION && + fkconstraint->fk_upd_action != FKCONSTR_ACTION_RESTRICT) || + (fkconstraint->fk_del_action != FKCONSTR_ACTION_NOACTION && + fkconstraint->fk_del_action != FKCONSTR_ACTION_RESTRICT)) + ereport(ERROR, + (errcode(ERRCODE_INVALID_FOREIGN_KEY), + errmsg("Array Element Foreign Keys support only NO ACTION and RESTRICT actions"))); + } + /* * If the attribute list for the referenced table was omitted, lookup the * definition of the primary key and use it. Otherwise, validate the @@ -8707,6 +8760,37 @@ ATAddForeignKeyConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel, elog(ERROR, "only b-tree indexes are supported for foreign keys"); eqstrategy = BTEqualStrategyNumber; + /* + * If this is an array foreign key, we must look up the operators for + * the array element type, not the array type itself. + */ + if (fkreftypes[i] == FKCONSTR_REF_EACH_ELEMENT) + { + /* We check if the array element type exists and is of valid Oid */ + fktype = get_base_element_type(fktype); + if (!OidIsValid(fktype)) + ereport(ERROR, + (errcode(ERRCODE_INVALID_FOREIGN_KEY), + errmsg("foreign key constraint \"%s\" cannot be implemented", + fkconstraint->conname), + errdetail("Key column \"%s\" has type %s which is not an array type.", + strVal(list_nth(fkconstraint->fk_attrs, i)), + format_type_be(fktypoid[i])))); + + /* + * make sure the <<@ operator can work with the specified operand + * types + */ + if (fktype != pktype) + ereport(ERROR, + (errcode(ERRCODE_INVALID_FOREIGN_KEY), + errmsg("foreign key constraint \"%s\" cannot be implemented", + fkconstraint->conname), + errdetail("Unssupported relation between %s and %s.", + format_type_be(fktypoid[i]), + format_type_be(pktype)))); + } + /* * There had better be a primary equality operator for the index. * We'll use it for PK = PK comparisons. @@ -8739,6 +8823,10 @@ ATAddForeignKeyConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel, ffeqop = InvalidOid; } + // XXX: Fix logic for selecting operators + if (fkreftypes[i] == FKCONSTR_REF_EACH_ELEMENT) + ffeqop = ARRAY_EQ_OP; + if (!(OidIsValid(pfeqop) && OidIsValid(ffeqop))) { /* @@ -8806,6 +8894,13 @@ ATAddForeignKeyConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel, * We may assume that pg_constraint.conkey is not changing. */ old_fktype = attr->atttypid; + if (fkreftypes[i] == FKCONSTR_REF_EACH_ELEMENT) + { + old_fktype = get_base_element_type(old_fktype); + /* this shouldn't happen ... */ + if (!OidIsValid(old_fktype)) + elog(ERROR, "old foreign key column is not an array"); + } new_fktype = fktype; old_pathtype = findFkeyCast(pfeqop_right, old_fktype, &old_castfunc); @@ -8865,6 +8960,7 @@ ATAddForeignKeyConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel, numfks, pkattnum, fkattnum, + fkreftypes, pfeqoperators, ppeqoperators, ffeqoperators, @@ -8877,6 +8973,7 @@ ATAddForeignKeyConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel, numfks, pkattnum, fkattnum, + fkreftypes, pfeqoperators, ppeqoperators, ffeqoperators, @@ -8920,7 +9017,7 @@ static ObjectAddress addFkRecurseReferenced(List **wqueue, Constraint *fkconstraint, Relation rel, Relation pkrel, Oid indexOid, Oid parentConstr, int numfks, - int16 *pkattnum, int16 *fkattnum, Oid *pfeqoperators, + int16 *pkattnum, int16 *fkattnum, char *fkreftypes, Oid *pfeqoperators, Oid *ppeqoperators, Oid *ffeqoperators, bool old_check_ok) { ObjectAddress address; @@ -8990,6 +9087,7 @@ addFkRecurseReferenced(List **wqueue, Constraint *fkconstraint, Relation rel, indexOid, RelationGetRelid(pkrel), pkattnum, + fkreftypes, pfeqoperators, ppeqoperators, ffeqoperators, @@ -9075,7 +9173,7 @@ addFkRecurseReferenced(List **wqueue, Constraint *fkconstraint, Relation rel, indexOid, RelationGetRelationName(partRel)); addFkRecurseReferenced(wqueue, fkconstraint, rel, partRel, partIndexId, constrOid, numfks, - mapped_pkattnum, fkattnum, + mapped_pkattnum, fkattnum, fkreftypes, pfeqoperators, ppeqoperators, ffeqoperators, old_check_ok); @@ -9124,7 +9222,7 @@ addFkRecurseReferenced(List **wqueue, Constraint *fkconstraint, Relation rel, static void addFkRecurseReferencing(List **wqueue, Constraint *fkconstraint, Relation rel, Relation pkrel, Oid indexOid, Oid parentConstr, - int numfks, int16 *pkattnum, int16 *fkattnum, + int numfks, int16 *pkattnum, int16 *fkattnum, char *fkreftypes, Oid *pfeqoperators, Oid *ppeqoperators, Oid *ffeqoperators, bool old_check_ok, LOCKMODE lockmode) { @@ -9258,6 +9356,7 @@ addFkRecurseReferencing(List **wqueue, Constraint *fkconstraint, Relation rel, indexOid, RelationGetRelid(pkrel), pkattnum, + fkreftypes, pfeqoperators, ppeqoperators, ffeqoperators, @@ -9293,6 +9392,7 @@ addFkRecurseReferencing(List **wqueue, Constraint *fkconstraint, Relation rel, numfks, pkattnum, mapped_fkattnum, + fkreftypes, pfeqoperators, ppeqoperators, ffeqoperators, @@ -9401,6 +9501,7 @@ CloneFkReferenced(Relation parentRel, Relation partitionRel) Oid conpfeqop[INDEX_MAX_KEYS]; Oid conppeqop[INDEX_MAX_KEYS]; Oid conffeqop[INDEX_MAX_KEYS]; + char confreftype[INDEX_MAX_KEYS]; Constraint *fkconstraint; tuple = SearchSysCache1(CONSTROID, constrOid); @@ -9433,8 +9534,8 @@ CloneFkReferenced(Relation parentRel, Relation partitionRel) confkey, conpfeqop, conppeqop, - conffeqop); - + conffeqop, + confreftype); for (int i = 0; i < numfks; i++) mapped_confkey[i] = attmap->attnums[confkey[i] - 1]; @@ -9477,6 +9578,7 @@ CloneFkReferenced(Relation parentRel, Relation partitionRel) numfks, mapped_confkey, conkey, + confreftype, conpfeqop, conppeqop, conffeqop, @@ -9547,6 +9649,7 @@ CloneFkReferencing(List **wqueue, Relation parentRel, Relation partRel) AttrNumber conkey[INDEX_MAX_KEYS]; AttrNumber mapped_conkey[INDEX_MAX_KEYS]; AttrNumber confkey[INDEX_MAX_KEYS]; + char confreftype[INDEX_MAX_KEYS]; Oid conpfeqop[INDEX_MAX_KEYS]; Oid conppeqop[INDEX_MAX_KEYS]; Oid conffeqop[INDEX_MAX_KEYS]; @@ -9581,7 +9684,8 @@ CloneFkReferencing(List **wqueue, Relation parentRel, Relation partRel) ShareRowExclusiveLock, NULL); DeconstructFkConstraintRow(tuple, &numfks, conkey, confkey, - conpfeqop, conppeqop, conffeqop); + conpfeqop, conppeqop, conffeqop, + confreftype); for (int i = 0; i < numfks; i++) mapped_conkey[i] = attmap->attnums[conkey[i] - 1]; @@ -9660,6 +9764,7 @@ CloneFkReferencing(List **wqueue, Relation parentRel, Relation partRel) indexOid, constrForm->confrelid, /* same foreign rel */ confkey, + confreftype, conpfeqop, conppeqop, conffeqop, @@ -9698,6 +9803,7 @@ CloneFkReferencing(List **wqueue, Relation parentRel, Relation partRel) numfks, confkey, mapped_conkey, + confreftype, conpfeqop, conppeqop, conffeqop, diff --git a/src/backend/commands/trigger.c b/src/backend/commands/trigger.c index 4e4e05844c..90c7495354 100644 --- a/src/backend/commands/trigger.c +++ b/src/backend/commands/trigger.c @@ -801,6 +801,7 @@ CreateTrigger(CreateTrigStmt *stmt, const char *queryString, NULL, NULL, NULL, + NULL, 0, ' ', ' ', diff --git a/src/backend/commands/typecmds.c b/src/backend/commands/typecmds.c index 76218fb47e..3fd3f0fe57 100644 --- a/src/backend/commands/typecmds.c +++ b/src/backend/commands/typecmds.c @@ -3553,6 +3553,7 @@ domainAddConstraint(Oid domainOid, Oid domainNamespace, Oid baseTypeOid, NULL, NULL, NULL, + NULL, 0, ' ', ' ', diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c index da91cbd2b1..05875012f2 100644 --- a/src/backend/nodes/copyfuncs.c +++ b/src/backend/nodes/copyfuncs.c @@ -3033,6 +3033,7 @@ _copyConstraint(const Constraint *from) COPY_NODE_FIELD(pktable); COPY_NODE_FIELD(fk_attrs); COPY_NODE_FIELD(pk_attrs); + COPY_NODE_FIELD(fk_reftypes); COPY_SCALAR_FIELD(fk_matchtype); COPY_SCALAR_FIELD(fk_upd_action); COPY_SCALAR_FIELD(fk_del_action); diff --git a/src/backend/nodes/equalfuncs.c b/src/backend/nodes/equalfuncs.c index c2d73626fc..67e001d822 100644 --- a/src/backend/nodes/equalfuncs.c +++ b/src/backend/nodes/equalfuncs.c @@ -2642,6 +2642,7 @@ _equalConstraint(const Constraint *a, const Constraint *b) COMPARE_NODE_FIELD(pktable); COMPARE_NODE_FIELD(fk_attrs); COMPARE_NODE_FIELD(pk_attrs); + COMPARE_NODE_FIELD(fk_reftypes); COMPARE_SCALAR_FIELD(fk_matchtype); COMPARE_SCALAR_FIELD(fk_upd_action); COMPARE_SCALAR_FIELD(fk_del_action); diff --git a/src/backend/nodes/outfuncs.c b/src/backend/nodes/outfuncs.c index 6493a03ff8..82e95c1a87 100644 --- a/src/backend/nodes/outfuncs.c +++ b/src/backend/nodes/outfuncs.c @@ -3643,6 +3643,7 @@ _outConstraint(StringInfo str, const Constraint *node) WRITE_NODE_FIELD(pktable); WRITE_NODE_FIELD(fk_attrs); WRITE_NODE_FIELD(pk_attrs); + WRITE_NODE_FIELD(fk_reftypes); WRITE_CHAR_FIELD(fk_matchtype); WRITE_CHAR_FIELD(fk_upd_action); WRITE_CHAR_FIELD(fk_del_action); diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y index 652be0b96d..1264da7f6f 100644 --- a/src/backend/parser/gram.y +++ b/src/backend/parser/gram.y @@ -134,6 +134,19 @@ typedef struct SelectLimit LimitOption limitOption; } SelectLimit; +/* + * Private struct for the result of foreign_key_column_elem production contains + * reftype that denotes type of foreign key; can be one of the following: + * + * FKCONSTR_REF_PLAIN plain foreign keys + * FKCONSTR_REF_EACH_ELEMENT foreign key arrays + */ +typedef struct FKColElem +{ + Node *name; /* name of the column (a String) */ + char reftype; /* FKCONSTR_REF_xxx code */ +} FKColElem; + /* ConstraintAttributeSpec yields an integer bitmask of these flags: */ #define CAS_NOT_DEFERRABLE 0x01 #define CAS_DEFERRABLE 0x02 @@ -191,6 +204,7 @@ static RangeVar *makeRangeVarFromAnyName(List *names, int position, core_yyscan_ static void SplitColQualList(List *qualList, List **constraintList, CollateClause **collClause, core_yyscan_t yyscanner); +static void SplitFKColElems(List *fkcolelems, List **names, List **reftypes); static void processCASbits(int cas_bits, int location, const char *constrType, bool *deferrable, bool *initdeferred, bool *not_valid, bool *no_inherit, core_yyscan_t yyscanner); @@ -241,6 +255,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query); A_Indices *aind; ResTarget *target; struct PrivTarget *privtarget; + struct FKColElem *fkcolelem; AccessPriv *accesspriv; struct ImportQual *importqual; InsertStmt *istmt; @@ -375,6 +390,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query); %type <accesspriv> privilege %type <list> privileges privilege_list %type <privtarget> privilege_target +%type <fkcolelem> foreign_key_column_elem %type <objwithargs> function_with_argtypes aggregate_with_argtypes operator_with_argtypes procedure_with_argtypes function_with_argtypes_common %type <list> function_with_argtypes_list aggregate_with_argtypes_list operator_with_argtypes_list procedure_with_argtypes_list %type <ival> defacl_privilege_target @@ -412,7 +428,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query); execute_param_clause using_clause returning_clause opt_enum_val_list enum_val_list table_func_column_list create_generic_options alter_generic_options - relation_expr_list dostmt_opt_list + relation_expr_list dostmt_opt_list foreign_key_column_list transform_element_list transform_type_list TriggerTransitions TriggerReferencing vacuum_relation_list opt_vacuum_relation_list @@ -642,7 +658,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query); DETACH DICTIONARY DISABLE_P DISCARD DISTINCT DO DOCUMENT_P DOMAIN_P DOUBLE_P DROP - EACH ELSE ENABLE_P ENCODING ENCRYPTED END_P ENUM_P ESCAPE EVENT EXCEPT + EACH ELEMENT ELSE ENABLE_P ENCODING ENCRYPTED END_P ENUM_P ESCAPE EVENT EXCEPT EXCLUDE EXCLUDING EXCLUSIVE EXECUTE EXISTS EXPLAIN EXPRESSION EXTENSION EXTERNAL EXTRACT @@ -3621,8 +3637,10 @@ ColConstraintElem: n->contype = CONSTR_FOREIGN; n->location = @1; n->pktable = $2; + /* fk_attrs will be filled in by parse analysis */ n->fk_attrs = NIL; n->pk_attrs = $3; + n->fk_reftypes = list_make1_int(FKCONSTR_REF_PLAIN); n->fk_matchtype = $4; n->fk_upd_action = (char) ($5 >> 8); n->fk_del_action = (char) ($5 & 0xFF); @@ -3824,14 +3842,15 @@ ConstraintElem: NULL, yyscanner); $$ = (Node *)n; } - | FOREIGN KEY '(' columnList ')' REFERENCES qualified_name - opt_column_list key_match key_actions ConstraintAttributeSpec + | FOREIGN KEY '(' foreign_key_column_list ')' REFERENCES + qualified_name opt_column_list key_match key_actions + ConstraintAttributeSpec { Constraint *n = makeNode(Constraint); n->contype = CONSTR_FOREIGN; n->location = @1; + SplitFKColElems($4, &n->fk_attrs, &n->fk_reftypes); n->pktable = $7; - n->fk_attrs = $4; n->pk_attrs = $8; n->fk_matchtype = $9; n->fk_upd_action = (char) ($10 >> 8); @@ -3865,6 +3884,30 @@ columnElem: ColId } ; +foreign_key_column_list: + foreign_key_column_elem + { $$ = list_make1($1); } + | foreign_key_column_list ',' foreign_key_column_elem + { $$ = lappend($1, $3); } + ; + +foreign_key_column_elem: + ColId + { + FKColElem *n = (FKColElem *) palloc(sizeof(FKColElem)); + n->name = (Node *) makeString($1); + n->reftype = FKCONSTR_REF_PLAIN; + $$ = n; + } + | EACH ELEMENT OF ColId + { + FKColElem *n = (FKColElem *) palloc(sizeof(FKColElem)); + n->name = (Node *) makeString($4); + n->reftype = FKCONSTR_REF_EACH_ELEMENT; + $$ = n; + } + ; + opt_c_include: INCLUDE '(' columnList ')' { $$ = $3; } | /* EMPTY */ { $$ = NIL; } ; @@ -15332,6 +15375,7 @@ unreserved_keyword: | DOUBLE_P | DROP | EACH + | ELEMENT | ENABLE_P | ENCODING | ENCRYPTED @@ -15868,6 +15912,7 @@ bare_label_keyword: | DOUBLE_P | DROP | EACH + | ELEMENT | ELSE | ENABLE_P | ENCODING @@ -16887,6 +16932,24 @@ SplitColQualList(List *qualList, *constraintList = qualList; } +/* Split a list of FKColElem structs into separate name and reftype lists */ +static void +SplitFKColElems(List *fkcolelems, List **names, List **reftypes) +{ + ListCell *lc; + + *names = NIL; + *reftypes = NIL; + + foreach(lc, fkcolelems) + { + FKColElem *fkcolelem = (FKColElem *) lfirst(lc); + + *names = lappend(*names, fkcolelem->name); + *reftypes = lappend_int(*reftypes, fkcolelem->reftype); + } +} + /* * Process result of ConstraintAttributeSpec, and set appropriate bool flags * in the output command node. Pass NULL for any flags the particular diff --git a/src/backend/parser/parse_utilcmd.c b/src/backend/parser/parse_utilcmd.c index d56f81c79f..c31138bbb5 100644 --- a/src/backend/parser/parse_utilcmd.c +++ b/src/backend/parser/parse_utilcmd.c @@ -799,6 +799,8 @@ transformColumnDefinition(CreateStmtContext *cxt, ColumnDef *column) * list of FK constraints to be processed later. */ constraint->fk_attrs = list_make1(makeString(column->colname)); + /* grammar should have set fk_reftypes */ + Assert(list_length(constraint->fk_reftypes) == 1); cxt->fkconstraints = lappend(cxt->fkconstraints, constraint); break; diff --git a/src/backend/utils/adt/ri_triggers.c b/src/backend/utils/adt/ri_triggers.c index 09a2ad2881..5c61a07866 100644 --- a/src/backend/utils/adt/ri_triggers.c +++ b/src/backend/utils/adt/ri_triggers.c @@ -114,6 +114,7 @@ typedef struct RI_ConstraintInfo int nkeys; /* number of key columns */ int16 pk_attnums[RI_MAX_NUMKEYS]; /* attnums of referenced cols */ int16 fk_attnums[RI_MAX_NUMKEYS]; /* attnums of referencing cols */ + char fk_reftypes[RI_MAX_NUMKEYS]; /* reference semantics */ 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) */ @@ -352,42 +353,102 @@ RI_FKey_check(TriggerData *trigdata) const char *querysep; Oid queryoids[RI_MAX_NUMKEYS]; const char *pk_only; + int each_elem; - /* ---------- - * 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++) + for (each_elem = 0; each_elem < riinfo->nkeys; each_elem++) + if (riinfo->fk_reftypes[each_elem] == FKCONSTR_REF_EACH_ELEMENT) + break; + + if (each_elem < riinfo->nkeys) { - Oid pk_type = RIAttType(pk_rel, riinfo->pk_attnums[i]); - Oid fk_type = RIAttType(fk_rel, riinfo->fk_attnums[i]); + /* + * The query string built is: + * + * SELECT 1 WHERE NOT EXISTS + * ( + * SELECT 1 FROM pg_catalog.unnest($1) + * WHERE unnest IS NOT NULL AND NOT EXISTS + * ( + * SELECT 1 FROM [ONLY] pktable x WHERE pkatt1 = [$2 | unnest] [AND ...] + * FOR KEY SHARE OF x + * ) + * ) + * + * The type ids 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 WHERE NOT EXISTS (" + "SELECT 1 FROM pg_catalog.unnest($%d) WHERE unnest IS NOT NULL" + " AND NOT EXISTS (SELECT 1 FROM %s%s x", + each_elem + 1, 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])); + if (riinfo->fk_reftypes[i] == FKCONSTR_REF_EACH_ELEMENT) + sprintf(paramname, "unnest"); + else + sprintf(paramname, "$%d", i + 1); + ri_GenerateQual(&querybuf, querysep, + attname, pk_type, + riinfo->pf_eq_oprs[i], + paramname, fk_type); + querysep = "AND"; + queryoids[i] = fk_type; + } + appendStringInfoString(&querybuf, " FOR KEY SHARE OF x))"); - quoteOneName(attname, - RIAttName(pk_rel, riinfo->pk_attnums[i])); - sprintf(paramname, "$%d", i + 1); - ri_GenerateQual(&querybuf, querysep, - attname, pk_type, - riinfo->pf_eq_oprs[i], - paramname, fk_type); - querysep = "AND"; - queryoids[i] = fk_type; + /* Prepare and save the plan */ + qplan = ri_PlanCheck(querybuf.data, riinfo->nkeys, queryoids, + &qkey, fk_rel, pk_rel); } - appendStringInfoString(&querybuf, " FOR KEY SHARE OF x"); + else + { + /* + * The query string built is: + * + * SELECT 1 FROM [ONLY] <pktable> x WHERE pkatt1 = $1 [AND ...] + * FOR KEY SHARE OF x + * + * The type ids for the $ parameters are those of the + * corresponding FK attributes. + */ + initStringInfo(&querybuf); + pk_only = pk_rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE ? + "" : "ONLY "; + quoteRelationName(pkrelname, pk_rel); + appendStringInfo(&querybuf, "SELECT 1 FROM %s%s x", + pk_only, pkrelname); + querysep = "WHERE"; + for (int i = 0; i < riinfo->nkeys; i++) + { + Oid pk_type = RIAttType(pk_rel, riinfo->pk_attnums[i]); + Oid fk_type = RIAttType(fk_rel, riinfo->fk_attnums[i]); + + quoteOneName(attname, + RIAttName(pk_rel, riinfo->pk_attnums[i])); + sprintf(paramname, "$%d", i + 1); + ri_GenerateQual(&querybuf, querysep, + attname, pk_type, + riinfo->pf_eq_oprs[i], + paramname, fk_type); + querysep = "AND"; + queryoids[i] = fk_type; + } + appendStringInfoString(&querybuf, " FOR KEY SHARE OF x"); - /* Prepare and save the plan */ - qplan = ri_PlanCheck(querybuf.data, riinfo->nkeys, queryoids, - &qkey, fk_rel, pk_rel); + /* Prepare and save the plan */ + qplan = ri_PlanCheck(querybuf.data, riinfo->nkeys, queryoids, + &qkey, fk_rel, pk_rel); + } } /* @@ -695,7 +756,7 @@ ri_restrict(TriggerData *trigdata, bool is_no_action) sprintf(paramname, "$%d", i + 1); ri_GenerateQual(&querybuf, querysep, paramname, pk_type, - riinfo->pf_eq_oprs[i], + riinfo->fk_reftypes[i] == FKCONSTR_REF_EACH_ELEMENT ? OID_ARRAY_ELEMCONTAINED_OP : riinfo->pf_eq_oprs[i], // XXX attname, fk_type); if (pk_coll != fk_coll && !get_collation_isdeterministic(pk_coll)) ri_GenerateQualCollation(&querybuf, pk_coll); @@ -801,7 +862,7 @@ RI_FKey_cascade_del(PG_FUNCTION_ARGS) sprintf(paramname, "$%d", i + 1); ri_GenerateQual(&querybuf, querysep, paramname, pk_type, - riinfo->pf_eq_oprs[i], + riinfo->fk_reftypes[i] == FKCONSTR_REF_EACH_ELEMENT ? OID_ARRAY_ELEMCONTAINED_OP : riinfo->pf_eq_oprs[i], // XXX attname, fk_type); if (pk_coll != fk_coll && !get_collation_isdeterministic(pk_coll)) ri_GenerateQualCollation(&querybuf, pk_coll); @@ -920,7 +981,7 @@ RI_FKey_cascade_upd(PG_FUNCTION_ARGS) sprintf(paramname, "$%d", j + 1); ri_GenerateQual(&qualbuf, qualsep, paramname, pk_type, - riinfo->pf_eq_oprs[i], + riinfo->fk_reftypes[i] == FKCONSTR_REF_EACH_ELEMENT ? OID_ARRAY_ELEMCONTAINED_OP : riinfo->pf_eq_oprs[i], // XXX attname, fk_type); if (pk_coll != fk_coll && !get_collation_isdeterministic(pk_coll)) ri_GenerateQualCollation(&querybuf, pk_coll); @@ -1100,7 +1161,7 @@ ri_set(TriggerData *trigdata, bool is_set_null) sprintf(paramname, "$%d", i + 1); ri_GenerateQual(&qualbuf, qualsep, paramname, pk_type, - riinfo->pf_eq_oprs[i], + riinfo->fk_reftypes[i] == FKCONSTR_REF_EACH_ELEMENT ? OID_ARRAY_ELEMCONTAINED_OP : riinfo->pf_eq_oprs[i], // XXX attname, fk_type); if (pk_coll != fk_coll && !get_collation_isdeterministic(pk_coll)) ri_GenerateQualCollation(&querybuf, pk_coll); @@ -1313,6 +1374,7 @@ RI_Initial_Check(Trigger *trigger, Relation fk_rel, Relation pk_rel) char workmembuf[32]; int spi_result; SPIPlanPtr qplan; + int each_elem; riinfo = ri_FetchConstraintInfo(trigger, fk_rel, false); @@ -1374,6 +1436,14 @@ RI_Initial_Check(Trigger *trigger, Relation fk_rel, Relation pk_rel) * For MATCH FULL: * (fk.keycol1 IS NOT NULL [OR ...]) * + * In case of an array ELEMENT column, relname is replaced with the + * following subquery: + * + * SELECT unnest("keycol1") k1, "keycol1" ak1 [, ...] + * FROM ONLY "public"."fk" + * + * where all the columns are renamed in order to prevent name collisions. + * * We attach COLLATE clauses to the operators when comparing columns * that have different collations. *---------- @@ -1391,13 +1461,30 @@ RI_Initial_Check(Trigger *trigger, Relation fk_rel, Relation pk_rel) quoteRelationName(pkrelname, pk_rel); quoteRelationName(fkrelname, fk_rel); + fk_only = fk_rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE ? "" : "ONLY "; pk_only = pk_rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE ? "" : "ONLY "; - appendStringInfo(&querybuf, - " FROM %s%s fk LEFT OUTER JOIN %s%s pk ON", - fk_only, fkrelname, pk_only, pkrelname); + + for (each_elem = 0; each_elem < riinfo->nkeys; each_elem++) + if (riinfo->fk_reftypes[each_elem] == FKCONSTR_REF_EACH_ELEMENT) + break; + + if (each_elem < riinfo->nkeys) + { + quoteOneName(fkattname, RIAttName(fk_rel, riinfo->fk_attnums[each_elem])); + appendStringInfo(&querybuf, " FROM %s%s fk" + " CROSS JOIN LATERAL pg_catalog.unnest(fk.%s)" + " LEFT OUTER JOIN %s%s pk ON", + fk_only, fkrelname, fkattname, pk_only, pkrelname); + } + else + { + appendStringInfo(&querybuf, + " FROM %s%s fk LEFT OUTER JOIN %s%s pk ON", + fk_only, fkrelname, pk_only, pkrelname); + } strcpy(pkattname, "pk."); strcpy(fkattname, "fk."); @@ -1408,15 +1495,22 @@ RI_Initial_Check(Trigger *trigger, Relation fk_rel, Relation pk_rel) 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 *tmp; quoteOneName(pkattname + 3, RIAttName(pk_rel, riinfo->pk_attnums[i])); - quoteOneName(fkattname + 3, - RIAttName(fk_rel, riinfo->fk_attnums[i])); + if (riinfo->fk_reftypes[i] == FKCONSTR_REF_EACH_ELEMENT) + tmp = "unnest"; + else + { + quoteOneName(fkattname + 3, + RIAttName(fk_rel, riinfo->fk_attnums[i])); + tmp = fkattname; + } ri_GenerateQual(&querybuf, sep, pkattname, pk_type, riinfo->pf_eq_oprs[i], - fkattname, fk_type); + tmp, fk_type); if (pk_coll != fk_coll) ri_GenerateQualCollation(&querybuf, pk_coll); sep = "AND"; @@ -1432,10 +1526,15 @@ RI_Initial_Check(Trigger *trigger, Relation fk_rel, Relation pk_rel) sep = ""; for (int i = 0; i < riinfo->nkeys; i++) { - quoteOneName(fkattname, RIAttName(fk_rel, riinfo->fk_attnums[i])); - appendStringInfo(&querybuf, - "%sfk.%s IS NOT NULL", - sep, fkattname); + if (riinfo->fk_reftypes[i] == FKCONSTR_REF_EACH_ELEMENT) + appendStringInfo(&querybuf, "%sunnest IS NOT NULL", sep); + else + { + quoteOneName(fkattname, RIAttName(fk_rel, riinfo->fk_attnums[i])); + appendStringInfo(&querybuf, + "%sfk.%s IS NOT NULL", + sep, fkattname); + } switch (riinfo->confmatchtype) { case FKCONSTR_MATCH_SIMPLE: @@ -1650,7 +1749,7 @@ RI_PartitionRemove_Check(Trigger *trigger, Relation fk_rel, Relation pk_rel) RIAttName(fk_rel, riinfo->fk_attnums[i])); ri_GenerateQual(&querybuf, sep, pkattname, pk_type, - riinfo->pf_eq_oprs[i], + riinfo->fk_reftypes[i] == FKCONSTR_REF_EACH_ELEMENT ? OID_ARRAY_ELEMCONTAINED_OP : riinfo->pf_eq_oprs[i], // XXX fkattname, fk_type); if (pk_coll != fk_coll) ri_GenerateQualCollation(&querybuf, pk_coll); @@ -2094,7 +2193,8 @@ ri_LoadConstraintInfo(Oid constraintOid) riinfo->pk_attnums, riinfo->pf_eq_oprs, riinfo->pp_eq_oprs, - riinfo->ff_eq_oprs); + riinfo->ff_eq_oprs, + riinfo->fk_reftypes); ReleaseSysCache(tup); diff --git a/src/backend/utils/adt/ruleutils.c b/src/backend/utils/adt/ruleutils.c index 879288c139..534920cd1b 100644 --- a/src/backend/utils/adt/ruleutils.c +++ b/src/backend/utils/adt/ruleutils.c @@ -330,6 +330,9 @@ static char *pg_get_viewdef_worker(Oid viewoid, static char *pg_get_triggerdef_worker(Oid trigid, bool pretty); static int decompile_column_index_array(Datum column_index_array, Oid relId, StringInfo buf); +static void decompile_fk_column_index_array(Datum column_index_array, + Datum fk_reftype_array, + Oid relId, StringInfo buf); static char *pg_get_ruledef_worker(Oid ruleoid, int prettyFlags); static char *pg_get_indexdef_worker(Oid indexrelid, int colno, const Oid *excludeOps, @@ -2006,7 +2009,8 @@ pg_get_constraintdef_worker(Oid constraintId, bool fullCommand, { case CONSTRAINT_FOREIGN: { - Datum val; + Datum colindexes; + Datum reftypes; bool isnull; const char *string; @@ -2014,13 +2018,21 @@ pg_get_constraintdef_worker(Oid constraintId, bool fullCommand, appendStringInfoString(&buf, "FOREIGN KEY ("); /* Fetch and build referencing-column list */ - val = SysCacheGetAttr(CONSTROID, tup, - Anum_pg_constraint_conkey, &isnull); + colindexes = SysCacheGetAttr(CONSTROID, tup, + Anum_pg_constraint_conkey, + &isnull); if (isnull) elog(ERROR, "null conkey for constraint %u", constraintId); + reftypes = SysCacheGetAttr(CONSTROID, tup, + Anum_pg_constraint_confreftype, + &isnull); + if (isnull) + elog(ERROR, "null confreftype for constraint %u", + constraintId); - decompile_column_index_array(val, conForm->conrelid, &buf); + decompile_fk_column_index_array(colindexes, reftypes, + conForm->conrelid, &buf); /* add foreign relation name */ appendStringInfo(&buf, ") REFERENCES %s(", @@ -2028,13 +2040,15 @@ pg_get_constraintdef_worker(Oid constraintId, bool fullCommand, NIL)); /* Fetch and build referenced-column list */ - val = SysCacheGetAttr(CONSTROID, tup, - Anum_pg_constraint_confkey, &isnull); + colindexes = SysCacheGetAttr(CONSTROID, tup, + Anum_pg_constraint_confkey, + &isnull); if (isnull) elog(ERROR, "null confkey for constraint %u", constraintId); - decompile_column_index_array(val, conForm->confrelid, &buf); + decompile_column_index_array(colindexes, + conForm->confrelid, &buf); appendStringInfoChar(&buf, ')'); @@ -2363,6 +2377,66 @@ decompile_column_index_array(Datum column_index_array, Oid relId, return nKeys; } +/* + * Convert an int16[] Datum and a char[] Datum into a comma-separated list of + * column names for the indicated relation, prefixed by appropriate keywords + * depending on the foreign key reference semantics indicated by the char[] + * entries. Append the text to buf. + */ +static void +decompile_fk_column_index_array(Datum column_index_array, + Datum fk_reftype_array, + Oid relId, StringInfo buf) +{ + Datum *keys; + int nKeys; + Datum *reftypes; + int nReftypes; + int j; + + /* Extract data from array of int16 */ + deconstruct_array(DatumGetArrayTypeP(column_index_array), + INT2OID, sizeof(int16), true, 's', + &keys, NULL, &nKeys); + + /* Extract data from array of char */ + deconstruct_array(DatumGetArrayTypeP(fk_reftype_array), + CHAROID, sizeof(char), true, 'c', + &reftypes, NULL, &nReftypes); + + if (nKeys != nReftypes) + elog(ERROR, "wrong confreftype cardinality"); + + for (j = 0; j < nKeys; j++) + { + char *colName; + const char *prefix; + + colName = get_attname(relId, DatumGetInt16(keys[j]), false); + + switch (DatumGetChar(reftypes[j])) + { + case FKCONSTR_REF_PLAIN: + prefix = ""; + break; + case FKCONSTR_REF_EACH_ELEMENT: + prefix = "EACH ELEMENT OF "; + break; + default: + elog(ERROR, "invalid fk_reftype: %d", + (int) DatumGetChar(reftypes[j])); + prefix = NULL; /* keep compiler quiet */ + break; + } + + if (j == 0) + appendStringInfo(buf, "%s%s", prefix, + quote_identifier(colName)); + else + appendStringInfo(buf, ", %s%s", prefix, + quote_identifier(colName)); + } +} /* ---------- * pg_get_expr - Decompile an expression tree diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c index 7ef510cd01..b43ca0e927 100644 --- a/src/backend/utils/cache/relcache.c +++ b/src/backend/utils/cache/relcache.c @@ -4469,7 +4469,7 @@ RelationGetFKeyList(Relation relation) info->conkey, info->confkey, info->conpfeqop, - NULL, NULL); + NULL, NULL, NULL); /* Add FK's node to the result list */ result = lappend(result, info); diff --git a/src/include/catalog/pg_constraint.h b/src/include/catalog/pg_constraint.h index 63f0f8bf41..901371db71 100644 --- a/src/include/catalog/pg_constraint.h +++ b/src/include/catalog/pg_constraint.h @@ -120,9 +120,19 @@ CATALOG(pg_constraint,2606,ConstraintRelationId) */ int16 confkey[1]; + /* + * If a foreign key, the reference semantics for each column. + * 'p' to signify plain foreign key + * 'e' to signify each element foreign key array + */ + char confreftype[1]; + /* * If a foreign key, the OIDs of the PK = FK equality operators for each * column of the constraint + * + * Note: for Array Element Foreign Keys, all these operators are for the + * array's element type. */ Oid conpfeqop[1] BKI_LOOKUP(pg_operator); @@ -219,6 +229,7 @@ extern Oid CreateConstraintEntry(const char *constraintName, Oid indexRelId, Oid foreignRelId, const int16 *foreignKey, + const char *foreignRefType, const Oid *pfEqOp, const Oid *ppEqOp, const Oid *ffEqOp, @@ -259,7 +270,8 @@ extern Bitmapset *get_primary_key_attnos(Oid relid, bool deferrableOk, Oid *constraintOid); extern void DeconstructFkConstraintRow(HeapTuple tuple, int *numfks, AttrNumber *conkey, AttrNumber *confkey, - Oid *pf_eq_oprs, Oid *pp_eq_oprs, Oid *ff_eq_oprs); + Oid *pf_eq_oprs, Oid *pp_eq_oprs, Oid *ff_eq_oprs, + char *fk_reftypes); extern bool check_functional_grouping(Oid relid, Index varno, Index varlevelsup, diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h index 236832a2ca..8fd49aacd4 100644 --- a/src/include/nodes/parsenodes.h +++ b/src/include/nodes/parsenodes.h @@ -2189,6 +2189,10 @@ typedef enum ConstrType /* types of constraints */ #define FKCONSTR_MATCH_PARTIAL 'p' #define FKCONSTR_MATCH_SIMPLE 's' + /* Foreign key column reference semantics codes */ +#define FKCONSTR_REF_PLAIN 'p' +#define FKCONSTR_REF_EACH_ELEMENT 'e' + typedef struct Constraint { NodeTag type; @@ -2229,6 +2233,7 @@ typedef struct Constraint RangeVar *pktable; /* Primary key table */ List *fk_attrs; /* Attributes of foreign key */ List *pk_attrs; /* Corresponding attrs in PK table */ + List *fk_reftypes; /* Per-column reference semantics (int List) */ char fk_matchtype; /* FULL, PARTIAL, SIMPLE */ char fk_upd_action; /* ON UPDATE action */ char fk_del_action; /* ON DELETE action */ diff --git a/src/include/parser/kwlist.h b/src/include/parser/kwlist.h index 28083aaac9..58b937ddfa 100644 --- a/src/include/parser/kwlist.h +++ b/src/include/parser/kwlist.h @@ -142,6 +142,7 @@ PG_KEYWORD("domain", DOMAIN_P, UNRESERVED_KEYWORD, BARE_LABEL) PG_KEYWORD("double", DOUBLE_P, UNRESERVED_KEYWORD, BARE_LABEL) PG_KEYWORD("drop", DROP, UNRESERVED_KEYWORD, BARE_LABEL) PG_KEYWORD("each", EACH, UNRESERVED_KEYWORD, BARE_LABEL) +PG_KEYWORD("element", ELEMENT, UNRESERVED_KEYWORD, BARE_LABEL) PG_KEYWORD("else", ELSE, RESERVED_KEYWORD, BARE_LABEL) PG_KEYWORD("enable", ENABLE_P, UNRESERVED_KEYWORD, BARE_LABEL) PG_KEYWORD("encoding", ENCODING, UNRESERVED_KEYWORD, BARE_LABEL) diff --git a/src/test/regress/expected/element_fk.out b/src/test/regress/expected/element_fk.out new file mode 100644 index 0000000000..d83040dbc3 --- /dev/null +++ b/src/test/regress/expected/element_fk.out @@ -0,0 +1,625 @@ +-- EACH-ELEMENT FK CONSTRAINTS +CREATE TABLE PKTABLEFORARRAY ( ptest1 int PRIMARY KEY, ptest2 text ); +-- Insert test data into PKTABLEFORARRAY +INSERT INTO PKTABLEFORARRAY VALUES (1, 'Test1'); +INSERT INTO PKTABLEFORARRAY VALUES (2, 'Test2'); +INSERT INTO PKTABLEFORARRAY VALUES (3, 'Test3'); +INSERT INTO PKTABLEFORARRAY VALUES (4, 'Test4'); +INSERT INTO PKTABLEFORARRAY VALUES (5, 'Test5'); +-- Check alter table +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], ftest2 int ); +ALTER TABLE FKTABLEFORARRAY ADD CONSTRAINT FKARRAY FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY; +DROP TABLE FKTABLEFORARRAY; +-- Check alter table with rows +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], ftest2 int ); +INSERT INTO FKTABLEFORARRAY VALUES ('{1}', 1); +ALTER TABLE FKTABLEFORARRAY ADD CONSTRAINT FKARRAY FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY; +DROP TABLE FKTABLEFORARRAY; +-- Check alter table with failing rows +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], ftest2 int ); +INSERT INTO FKTABLEFORARRAY VALUES ('{10,1}', 2); +ALTER TABLE FKTABLEFORARRAY ADD CONSTRAINT FKARRAY FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY; +ERROR: insert or update on table "fktableforarray" violates foreign key constraint "fkarray" +DETAIL: Key (ftest1)=({10,1}) is not present in table "pktableforarray". +DROP TABLE FKTABLEFORARRAY; +-- Check create table +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY, ftest2 int ); +CREATE TABLE FKTABLEFORARRAYMDIM ( ftest1 int[][], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY, ftest2 int ); +CREATE TABLE FKTABLEFORARRAYNOTNULL ( ftest1 int[] NOT NULL, FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY, ftest2 int ); +-- Insert successful rows into FK TABLE +INSERT INTO FKTABLEFORARRAY VALUES ('{1}', 3); +INSERT INTO FKTABLEFORARRAY VALUES ('{2}', 4); +INSERT INTO FKTABLEFORARRAY VALUES ('{1}', 5); +INSERT INTO FKTABLEFORARRAY VALUES ('{3}', 6); +INSERT INTO FKTABLEFORARRAY VALUES ('{1}', 7); +INSERT INTO FKTABLEFORARRAY VALUES ('{4,5}', 8); +INSERT INTO FKTABLEFORARRAY VALUES ('{4,4}', 9); +INSERT INTO FKTABLEFORARRAY VALUES (NULL, 10); +INSERT INTO FKTABLEFORARRAY VALUES ('{}', 11); +INSERT INTO FKTABLEFORARRAY VALUES ('{1,NULL}', 12); +INSERT INTO FKTABLEFORARRAY VALUES ('{NULL}', 13); +INSERT INTO FKTABLEFORARRAYMDIM VALUES ('{{4,5},{1,2},{1,3}}', 14); +INSERT INTO FKTABLEFORARRAYMDIM VALUES ('{{4,5},{NULL,2},{NULL,3}}', 15); +-- Insert failed rows into FK TABLE +INSERT INTO FKTABLEFORARRAY VALUES ('{6}', 16); +ERROR: insert or update on table "fktableforarray" violates foreign key constraint "fktableforarray_ftest1_fkey" +DETAIL: Key (ftest1)=({6}) is not present in table "pktableforarray". +INSERT INTO FKTABLEFORARRAY VALUES ('{4,6}', 17); +ERROR: insert or update on table "fktableforarray" violates foreign key constraint "fktableforarray_ftest1_fkey" +DETAIL: Key (ftest1)=({4,6}) is not present in table "pktableforarray". +INSERT INTO FKTABLEFORARRAY VALUES ('{6,NULL}', 18); +ERROR: insert or update on table "fktableforarray" violates foreign key constraint "fktableforarray_ftest1_fkey" +DETAIL: Key (ftest1)=({6,NULL}) is not present in table "pktableforarray". +INSERT INTO FKTABLEFORARRAY VALUES ('{6,NULL,4,NULL}', 19); +ERROR: insert or update on table "fktableforarray" violates foreign key constraint "fktableforarray_ftest1_fkey" +DETAIL: Key (ftest1)=({6,NULL,4,NULL}) is not present in table "pktableforarray". +INSERT INTO FKTABLEFORARRAYMDIM VALUES ('{{1,2},{6,NULL}}', 20); +ERROR: insert or update on table "fktableforarraymdim" violates foreign key constraint "fktableforarraymdim_ftest1_fkey" +DETAIL: Key (ftest1)=({{1,2},{6,NULL}}) is not present in table "pktableforarray". +INSERT INTO FKTABLEFORARRAYNOTNULL VALUES (NULL, 21); +ERROR: null value in column "ftest1" of relation "fktableforarraynotnull" violates not-null constraint +DETAIL: Failing row contains (null, 21). +-- Check FKTABLE +SELECT * FROM FKTABLEFORARRAY; + ftest1 | ftest2 +----------+-------- + {1} | 3 + {2} | 4 + {1} | 5 + {3} | 6 + {1} | 7 + {4,5} | 8 + {4,4} | 9 + | 10 + {} | 11 + {1,NULL} | 12 + {NULL} | 13 +(11 rows) + +-- Delete a row from PK TABLE (must fail due to ON DELETE NO ACTION) +DELETE FROM PKTABLEFORARRAY WHERE ptest1=1; +ERROR: update or delete on table "pktableforarray" violates foreign key constraint "fktableforarray_ftest1_fkey" on table "fktableforarray" +DETAIL: Key (ptest1)=(1) is still referenced from table "fktableforarray". +-- Check FKTABLE for removal of matched row +SELECT * FROM FKTABLEFORARRAY; + ftest1 | ftest2 +----------+-------- + {1} | 3 + {2} | 4 + {1} | 5 + {3} | 6 + {1} | 7 + {4,5} | 8 + {4,4} | 9 + | 10 + {} | 11 + {1,NULL} | 12 + {NULL} | 13 +(11 rows) + +-- Update a row from PK TABLE (must fail due to ON UPDATE NO ACTION) +UPDATE PKTABLEFORARRAY SET ptest1=7 WHERE ptest1=1; +ERROR: update or delete on table "pktableforarray" violates foreign key constraint "fktableforarray_ftest1_fkey" on table "fktableforarray" +DETAIL: Key (ptest1)=(1) is still referenced from table "fktableforarray". +-- Check FKTABLE for update of matched row +SELECT * FROM FKTABLEFORARRAY; + ftest1 | ftest2 +----------+-------- + {1} | 3 + {2} | 4 + {1} | 5 + {3} | 6 + {1} | 7 + {4,5} | 8 + {4,4} | 9 + | 10 + {} | 11 + {1,NULL} | 12 + {NULL} | 13 +(11 rows) + +-- Check UPDATE on FKTABLE +UPDATE FKTABLEFORARRAY SET ftest1=ARRAY[1] WHERE ftest2=4; +-- Check FKTABLE for update of matched row +SELECT * FROM FKTABLEFORARRAY; + ftest1 | ftest2 +----------+-------- + {1} | 3 + {1} | 5 + {3} | 6 + {1} | 7 + {4,5} | 8 + {4,4} | 9 + | 10 + {} | 11 + {1,NULL} | 12 + {NULL} | 13 + {1} | 4 +(11 rows) + +DROP TABLE FKTABLEFORARRAY; +DROP TABLE FKTABLEFORARRAYNOTNULL; +DROP TABLE FKTABLEFORARRAYMDIM; +-- Allowed references with actions (NO ACTION, RESTRICT) +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE NO ACTION ON UPDATE NO ACTION, ftest2 int ); +DROP TABLE FKTABLEFORARRAY; +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE NO ACTION ON UPDATE RESTRICT, ftest2 int ); +DROP TABLE FKTABLEFORARRAY; +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE RESTRICT ON UPDATE NO ACTION, ftest2 int ); +DROP TABLE FKTABLEFORARRAY; +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE RESTRICT ON UPDATE RESTRICT, ftest2 int ); +DROP TABLE FKTABLEFORARRAY; +-- Not allowed references (SET NULL, SET DEFAULT, CASCADE) +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE NO ACTION ON UPDATE SET DEFAULT, ftest2 int ); +ERROR: Array Element Foreign Keys support only NO ACTION and RESTRICT actions +DROP TABLE IF EXISTS FKTABLEFORARRAY; +NOTICE: table "fktableforarray" does not exist, skipping +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE NO ACTION ON UPDATE SET NULL, ftest2 int ); +ERROR: Array Element Foreign Keys support only NO ACTION and RESTRICT actions +DROP TABLE IF EXISTS FKTABLEFORARRAY; +NOTICE: table "fktableforarray" does not exist, skipping +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE RESTRICT ON UPDATE SET DEFAULT, ftest2 int ); +ERROR: Array Element Foreign Keys support only NO ACTION and RESTRICT actions +DROP TABLE IF EXISTS FKTABLEFORARRAY; +NOTICE: table "fktableforarray" does not exist, skipping +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE RESTRICT ON UPDATE SET NULL, ftest2 int ); +ERROR: Array Element Foreign Keys support only NO ACTION and RESTRICT actions +DROP TABLE IF EXISTS FKTABLEFORARRAY; +NOTICE: table "fktableforarray" does not exist, skipping +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE CASCADE ON UPDATE NO ACTION, ftest2 int ); +ERROR: Array Element Foreign Keys support only NO ACTION and RESTRICT actions +DROP TABLE FKTABLEFORARRAY; +ERROR: table "fktableforarray" does not exist +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE CASCADE ON UPDATE RESTRICT, ftest2 int ); +ERROR: Array Element Foreign Keys support only NO ACTION and RESTRICT actions +DROP TABLE FKTABLEFORARRAY; +ERROR: table "fktableforarray" does not exist +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE CASCADE ON UPDATE SET DEFAULT, ftest2 int ); +ERROR: Array Element Foreign Keys support only NO ACTION and RESTRICT actions +DROP TABLE IF EXISTS FKTABLEFORARRAY; +NOTICE: table "fktableforarray" does not exist, skipping +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE CASCADE ON UPDATE SET NULL, ftest2 int ); +ERROR: Array Element Foreign Keys support only NO ACTION and RESTRICT actions +DROP TABLE IF EXISTS FKTABLEFORARRAY; +NOTICE: table "fktableforarray" does not exist, skipping +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE SET DEFAULT ON UPDATE NO ACTION, ftest2 int ); +ERROR: Array Element Foreign Keys support only NO ACTION and RESTRICT actions +DROP TABLE IF EXISTS FKTABLEFORARRAY; +NOTICE: table "fktableforarray" does not exist, skipping +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE SET DEFAULT ON UPDATE RESTRICT, ftest2 int ); +ERROR: Array Element Foreign Keys support only NO ACTION and RESTRICT actions +DROP TABLE IF EXISTS FKTABLEFORARRAY; +NOTICE: table "fktableforarray" does not exist, skipping +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE SET DEFAULT ON UPDATE SET DEFAULT, ftest2 int ); +ERROR: Array Element Foreign Keys support only NO ACTION and RESTRICT actions +DROP TABLE IF EXISTS FKTABLEFORARRAY; +NOTICE: table "fktableforarray" does not exist, skipping +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE SET DEFAULT ON UPDATE SET NULL, ftest2 int ); +ERROR: Array Element Foreign Keys support only NO ACTION and RESTRICT actions +DROP TABLE IF EXISTS FKTABLEFORARRAY; +NOTICE: table "fktableforarray" does not exist, skipping +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE SET NULL ON UPDATE NO ACTION, ftest2 int ); +ERROR: Array Element Foreign Keys support only NO ACTION and RESTRICT actions +DROP TABLE IF EXISTS FKTABLEFORARRAY; +NOTICE: table "fktableforarray" does not exist, skipping +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE SET NULL ON UPDATE RESTRICT, ftest2 int ); +ERROR: Array Element Foreign Keys support only NO ACTION and RESTRICT actions +DROP TABLE IF EXISTS FKTABLEFORARRAY; +NOTICE: table "fktableforarray" does not exist, skipping +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE SET NULL ON UPDATE SET DEFAULT, ftest2 int ); +ERROR: Array Element Foreign Keys support only NO ACTION and RESTRICT actions +DROP TABLE IF EXISTS FKTABLEFORARRAY; +NOTICE: table "fktableforarray" does not exist, skipping +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE SET NULL ON UPDATE SET NULL, ftest2 int ); +ERROR: Array Element Foreign Keys support only NO ACTION and RESTRICT actions +DROP TABLE IF EXISTS FKTABLEFORARRAY; +NOTICE: table "fktableforarray" does not exist, skipping +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE NO ACTION ON UPDATE CASCADE, ftest2 int ); +ERROR: Array Element Foreign Keys support only NO ACTION and RESTRICT actions +DROP TABLE IF EXISTS FKTABLEFORARRAY; +NOTICE: table "fktableforarray" does not exist, skipping +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE RESTRICT ON UPDATE CASCADE, ftest2 int ); +ERROR: Array Element Foreign Keys support only NO ACTION and RESTRICT actions +DROP TABLE IF EXISTS FKTABLEFORARRAY; +NOTICE: table "fktableforarray" does not exist, skipping +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE CASCADE ON UPDATE CASCADE, ftest2 int ); +ERROR: Array Element Foreign Keys support only NO ACTION and RESTRICT actions +DROP TABLE IF EXISTS FKTABLEFORARRAY; +NOTICE: table "fktableforarray" does not exist, skipping +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE SET DEFAULT ON UPDATE CASCADE, ftest2 int ); +ERROR: Array Element Foreign Keys support only NO ACTION and RESTRICT actions +DROP TABLE IF EXISTS FKTABLEFORARRAY; +NOTICE: table "fktableforarray" does not exist, skipping +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE SET NULL ON UPDATE CASCADE, ftest2 int ); +ERROR: Array Element Foreign Keys support only NO ACTION and RESTRICT actions +DROP TABLE IF EXISTS FKTABLEFORARRAY; +NOTICE: table "fktableforarray" does not exist, skipping +-- Cleanup +DROP TABLE PKTABLEFORARRAY; +-- Check reference on empty table +CREATE TABLE PKTABLEFORARRAY (ptest1 int PRIMARY KEY); +CREATE TABLE FKTABLEFORARRAY (ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY); +INSERT INTO FKTABLEFORARRAY VALUES ('{}'); +DROP TABLE FKTABLEFORARRAY; +DROP TABLE PKTABLEFORARRAY; +-- Repeat a similar test using CHAR(1) keys rather than INTEGER +CREATE TABLE PKTABLEFORARRAY ( ptest1 CHAR(1) PRIMARY KEY, ptest2 text ); +-- Populate the primary table +INSERT INTO PKTABLEFORARRAY VALUES ('A', 'Test A'); +INSERT INTO PKTABLEFORARRAY VALUES ('B', 'Test B'); +INSERT INTO PKTABLEFORARRAY VALUES ('C', 'Test C'); +-- Create the referencing table +CREATE TABLE FKTABLEFORARRAY ( ftest1 char(1)[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON UPDATE RESTRICT ON DELETE RESTRICT, ftest2 int ); +-- Insert valid rows into FK TABLE +INSERT INTO FKTABLEFORARRAY VALUES ('{"A"}', 1); +INSERT INTO FKTABLEFORARRAY VALUES ('{"B"}', 2); +INSERT INTO FKTABLEFORARRAY VALUES ('{"C"}', 3); +INSERT INTO FKTABLEFORARRAY VALUES ('{"A","B","C"}', 4); +-- Insert invalid rows into FK TABLE +INSERT INTO FKTABLEFORARRAY VALUES ('{"D"}', 5); +ERROR: insert or update on table "fktableforarray" violates foreign key constraint "fktableforarray_ftest1_fkey" +DETAIL: Key (ftest1)=({D}) is not present in table "pktableforarray". +INSERT INTO FKTABLEFORARRAY VALUES ('{"A","B","D"}', 6); +ERROR: insert or update on table "fktableforarray" violates foreign key constraint "fktableforarray_ftest1_fkey" +DETAIL: Key (ftest1)=({A,B,D}) is not present in table "pktableforarray". +-- Check FKTABLE +SELECT * FROM FKTABLEFORARRAY; + ftest1 | ftest2 +---------+-------- + {A} | 1 + {B} | 2 + {C} | 3 + {A,B,C} | 4 +(4 rows) + +-- Delete a row from PK TABLE (must fail due to ON DELETE RESTRICT) +DELETE FROM PKTABLEFORARRAY WHERE ptest1='A'; +ERROR: update or delete on table "pktableforarray" violates foreign key constraint "fktableforarray_ftest1_fkey" on table "fktableforarray" +DETAIL: Key (ptest1)=(A) is still referenced from table "fktableforarray". +-- Check FKTABLE for removal of matched row +SELECT * FROM FKTABLEFORARRAY; + ftest1 | ftest2 +---------+-------- + {A} | 1 + {B} | 2 + {C} | 3 + {A,B,C} | 4 +(4 rows) + +-- Update a row from PK TABLE (must fail due to ON UPDATE RESTRICT) +UPDATE PKTABLEFORARRAY SET ptest1='D' WHERE ptest1='B'; +ERROR: update or delete on table "pktableforarray" violates foreign key constraint "fktableforarray_ftest1_fkey" on table "fktableforarray" +DETAIL: Key (ptest1)=(B) is still referenced from table "fktableforarray". +-- Check FKTABLE for update of matched row +SELECT * FROM FKTABLEFORARRAY; + ftest1 | ftest2 +---------+-------- + {A} | 1 + {B} | 2 + {C} | 3 + {A,B,C} | 4 +(4 rows) + +-- Cleanup +DROP TABLE FKTABLEFORARRAY; +DROP TABLE PKTABLEFORARRAY; +-- Composite primary keys +CREATE TABLE PKTABLEFORARRAY ( id1 CHAR(1), id2 CHAR(1), ptest2 text, PRIMARY KEY (id1, id2) ); +-- Populate the primary table +INSERT INTO PKTABLEFORARRAY VALUES ('A', 'A', 'Test A'); +INSERT INTO PKTABLEFORARRAY VALUES ('A', 'B', 'Test B'); +INSERT INTO PKTABLEFORARRAY VALUES ('B', 'C', 'Test B'); +-- Create the referencing table +CREATE TABLE FKTABLEFORARRAY ( fid1 CHAR(1), fid2 CHAR(1)[], ftest2 text, FOREIGN KEY (fid1, EACH ELEMENT OF fid2) REFERENCES PKTABLEFORARRAY); +-- Insert valid rows into FK TABLE +INSERT INTO FKTABLEFORARRAY VALUES ('A', ARRAY['A','B'], '1'); +INSERT INTO FKTABLEFORARRAY VALUES ('B', ARRAY['C'], '2'); +-- Insert invalid rows into FK TABLE +INSERT INTO FKTABLEFORARRAY VALUES ('A', ARRAY['A','B', 'C'], '3'); +ERROR: insert or update on table "fktableforarray" violates foreign key constraint "fktableforarray_fid1_fid2_fkey" +DETAIL: Key (fid1, fid2)=(A, {A,B,C}) is not present in table "pktableforarray". +INSERT INTO FKTABLEFORARRAY VALUES ('B', ARRAY['A'], '4'); +ERROR: insert or update on table "fktableforarray" violates foreign key constraint "fktableforarray_fid1_fid2_fkey" +DETAIL: Key (fid1, fid2)=(B, {A}) is not present in table "pktableforarray". +-- Cleanup +DROP TABLE FKTABLEFORARRAY; +DROP TABLE PKTABLEFORARRAY; +-- Test Array Element Foreign Keys with composite type +CREATE TYPE INVOICEID AS (year_part INTEGER, progressive_part INTEGER); +CREATE TABLE PKTABLEFORARRAY ( id INVOICEID PRIMARY KEY, ptest2 text); +-- Populate the primary table +INSERT INTO PKTABLEFORARRAY VALUES (ROW(2010, 99), 'Last invoice for 2010'); +INSERT INTO PKTABLEFORARRAY VALUES (ROW(2011, 1), 'First invoice for 2011'); +INSERT INTO PKTABLEFORARRAY VALUES (ROW(2011, 2), 'Second invoice for 2011'); +-- Create the referencing table +CREATE TABLE FKTABLEFORARRAY ( id SERIAL PRIMARY KEY, invoice_ids INVOICEID[], FOREIGN KEY (EACH ELEMENT OF invoice_ids) REFERENCES PKTABLEFORARRAY, ftest2 TEXT ); +-- Insert valid rows into FK TABLE +INSERT INTO FKTABLEFORARRAY(invoice_ids, ftest2) VALUES (ARRAY['(2010,99)']::INVOICEID[], 'Product A'); +INSERT INTO FKTABLEFORARRAY(invoice_ids, ftest2) VALUES (ARRAY['(2011,1)','(2011,2)']::INVOICEID[], 'Product B'); +INSERT INTO FKTABLEFORARRAY(invoice_ids, ftest2) VALUES (ARRAY['(2011,2)']::INVOICEID[], 'Product C'); +-- Insert invalid rows into FK TABLE +INSERT INTO FKTABLEFORARRAY(invoice_ids, ftest2) VALUES (ARRAY['(2011,99)']::INVOICEID[], 'Product A'); +ERROR: insert or update on table "fktableforarray" violates foreign key constraint "fktableforarray_invoice_ids_fkey" +DETAIL: Key (invoice_ids)=({"(2011,99)"}) is not present in table "pktableforarray". +INSERT INTO FKTABLEFORARRAY(invoice_ids, ftest2) VALUES (ARRAY['(2011,1)','(2010,1)']::INVOICEID[], 'Product B'); +ERROR: insert or update on table "fktableforarray" violates foreign key constraint "fktableforarray_invoice_ids_fkey" +DETAIL: Key (invoice_ids)=({"(2011,1)","(2010,1)"}) is not present in table "pktableforarray". +-- Check FKTABLE +SELECT * FROM FKTABLEFORARRAY; + id | invoice_ids | ftest2 +----+-------------------------+----------- + 1 | {"(2010,99)"} | Product A + 2 | {"(2011,1)","(2011,2)"} | Product B + 3 | {"(2011,2)"} | Product C +(3 rows) + +-- Delete a row from PK TABLE +DELETE FROM PKTABLEFORARRAY WHERE id=ROW(2010,99); +ERROR: update or delete on table "pktableforarray" violates foreign key constraint "fktableforarray_invoice_ids_fkey" on table "fktableforarray" +DETAIL: Key (id)=((2010,99)) is still referenced from table "fktableforarray". +-- Check FKTABLE for removal of matched row +SELECT * FROM FKTABLEFORARRAY; + id | invoice_ids | ftest2 +----+-------------------------+----------- + 1 | {"(2010,99)"} | Product A + 2 | {"(2011,1)","(2011,2)"} | Product B + 3 | {"(2011,2)"} | Product C +(3 rows) + +-- Update a row from PK TABLE +UPDATE PKTABLEFORARRAY SET id=ROW(2011,99) WHERE id=ROW(2011,1); +ERROR: update or delete on table "pktableforarray" violates foreign key constraint "fktableforarray_invoice_ids_fkey" on table "fktableforarray" +DETAIL: Key (id)=((2011,1)) is still referenced from table "fktableforarray". +-- Check FKTABLE for update of matched row +SELECT * FROM FKTABLEFORARRAY; + id | invoice_ids | ftest2 +----+-------------------------+----------- + 1 | {"(2010,99)"} | Product A + 2 | {"(2011,1)","(2011,2)"} | Product B + 3 | {"(2011,2)"} | Product C +(3 rows) + +-- Cleanup +DROP TABLE FKTABLEFORARRAY; +DROP TABLE PKTABLEFORARRAY; +DROP TYPE INVOICEID; +-- Check for an array column referencing another array column (NOT ELEMENT FOREIGN KEY) +-- Create primary table with an array primary key +CREATE TABLE PKTABLEFORARRAY ( id INT[] PRIMARY KEY, ptest2 text); +-- Create the referencing table +CREATE TABLE FKTABLEFORARRAY ( id SERIAL PRIMARY KEY, fids INT[] REFERENCES PKTABLEFORARRAY, ftest2 TEXT ); +-- Populate the primary table +INSERT INTO PKTABLEFORARRAY VALUES ('{1,1}', 'A'); +INSERT INTO PKTABLEFORARRAY VALUES ('{1,2}', 'B'); +-- Insert valid rows into FK TABLE +INSERT INTO FKTABLEFORARRAY (fids, ftest2) VALUES ('{1,1}', 'Product A'); +INSERT INTO FKTABLEFORARRAY (fids, ftest2) VALUES ('{1,2}', 'Product B'); +-- Insert invalid rows into FK TABLE +INSERT INTO FKTABLEFORARRAY (fids, ftest2) VALUES ('{0,1}', 'Product C'); +ERROR: insert or update on table "fktableforarray" violates foreign key constraint "fktableforarray_fids_fkey" +DETAIL: Key (fids)=({0,1}) is not present in table "pktableforarray". +INSERT INTO FKTABLEFORARRAY (fids, ftest2) VALUES ('{2,1}', 'Product D'); +ERROR: insert or update on table "fktableforarray" violates foreign key constraint "fktableforarray_fids_fkey" +DETAIL: Key (fids)=({2,1}) is not present in table "pktableforarray". +-- Cleanup +DROP TABLE FKTABLEFORARRAY; +DROP TABLE PKTABLEFORARRAY; +-- --------------------------------------- +-- Multi-column "ELEMENT" foreign key tests +-- --------------------------------------- +-- Create DIM1 table with two-column primary key +CREATE TABLE DIM1 (X INTEGER NOT NULL, Y INTEGER NOT NULL, PRIMARY KEY (X, Y)); +-- Populate DIM1 table pairs +INSERT INTO DIM1 SELECT x.t, x.t * y.t + FROM (SELECT generate_series(1, 10) AS t) x, + (SELECT generate_series(0, 10) AS t) y; +-- Test with TABLE declaration of an element foreign key constraint (NO ACTION) +CREATE TABLE F1 ( + x INTEGER PRIMARY KEY, y INTEGER[], + FOREIGN KEY (x, EACH ELEMENT OF y) REFERENCES DIM1(x, y) +); +-- Insert facts +INSERT INTO F1 VALUES (1, '{0,1,2,3,4,5}'); -- OK +INSERT INTO F1 VALUES (2, '{0,2,4,6}'); -- OK +INSERT INTO F1 VALUES (3, '{0,3,6,9,0,3,6,9,0,0,0,0,9,9}'); -- OK (multiple occurrences) +INSERT INTO F1 VALUES (4, '{0,2,4}'); -- FAILS (2 is not present) +ERROR: insert or update on table "f1" violates foreign key constraint "f1_x_y_fkey" +DETAIL: Key (x, y)=(4, {0,2,4}) is not present in table "dim1". +INSERT INTO F1 VALUES (4, '{0,NULL,4}'); -- OK +INSERT INTO F1 VALUES (5, '{0,NULL,5}'); -- OK +-- Try updates +UPDATE F1 SET y = '{0,2,4,6}' WHERE x = 2; -- OK +UPDATE F1 SET y = '{0,2,3,4,6}' WHERE x = 2; -- FAILS +ERROR: insert or update on table "f1" violates foreign key constraint "f1_x_y_fkey" +DETAIL: Key (x, y)=(2, {0,2,3,4,6}) is not present in table "dim1". +UPDATE F1 SET x = 20, y = '{0,2,3,4,6}' WHERE x = 2; -- FAILS (20 does not exist) +ERROR: insert or update on table "f1" violates foreign key constraint "f1_x_y_fkey" +DETAIL: Key (x, y)=(20, {0,2,3,4,6}) is not present in table "dim1". +UPDATE F1 SET y = '{0,4,8}' WHERE x = 4; -- OK +UPDATE F1 SET y = '{0,5,NULL,10}' WHERE x = 5; -- OK +DROP TABLE F1; +-- Test with FOREIGN KEY after TABLE population +CREATE TABLE F1 ( + x INTEGER PRIMARY KEY, y INTEGER[] +); +-- Insert facts +INSERT INTO F1 VALUES (1, '{0,1,2,3,4,5}'); -- OK +INSERT INTO F1 VALUES (2, '{0,2,4,6}'); -- OK +INSERT INTO F1 VALUES (3, '{0,3,6,9,0,3,6,9,0,0,0,0,9,9}'); -- OK (multiple occurrences) +INSERT INTO F1 VALUES (4, '{0,2,4}'); -- OK (2 is not present) +INSERT INTO F1 VALUES (5, '{0,NULL,5}'); -- OK +-- Add foreign key (FAILS) +ALTER TABLE F1 ADD FOREIGN KEY (x, EACH ELEMENT OF y) REFERENCES DIM1(x, y); +ERROR: insert or update on table "f1" violates foreign key constraint "f1_x_y_fkey" +DETAIL: Key (x, y)=(4, {0,2,4}) is not present in table "dim1". +DROP TABLE F1; +-- Test with TABLE declaration of a two-dim ELEMENT foreign key constraint (FAILS) +CREATE TABLE F1 ( + x INTEGER[] PRIMARY KEY, y INTEGER[], + FOREIGN KEY (EACH ELEMENT OF x, EACH ELEMENT OF y) REFERENCES DIM1(x, y) +); +ERROR: foreign keys support only one array column +-- Test with two-dim ELEMENT foreign key after TABLE population +CREATE TABLE F1 ( + x INTEGER[] PRIMARY KEY, y INTEGER[] +); +INSERT INTO F1 VALUES ('{1}', '{0,1,2,3,4,5}'); -- OK +INSERT INTO F1 VALUES ('{1,2}', '{0,2,4,6}'); -- OK +-- Add foreign key (FAILS) +ALTER TABLE F1 ADD FOREIGN KEY (EACH ELEMENT OF x, EACH ELEMENT OF y) REFERENCES DIM1(x, y); +ERROR: foreign keys support only one array column +DROP TABLE F1; +-- Cleanup +DROP TABLE DIM1; +-- Check for potential name conflicts (with internal integrity checks) +CREATE TABLE x1(x1 int, x2 int, PRIMARY KEY(x1,x2)); +INSERT INTO x1 VALUES + (1,4), + (1,5), + (2,4), + (2,5), + (3,6), + (3,7) +; +CREATE TABLE x2(x1 int[], x2 int, FOREIGN KEY(EACH ELEMENT OF x1, x2) REFERENCES x1); +INSERT INTO x2 VALUES ('{1,2}',4); +INSERT INTO x2 VALUES ('{1,3}',6); -- FAILS +ERROR: insert or update on table "x2" violates foreign key constraint "x2_x1_x2_fkey" +DETAIL: Key (x1, x2)=({1,3}, 6) is not present in table "x1". +DROP TABLE x2; +CREATE TABLE x2(x1 int[], x2 int); +INSERT INTO x2 VALUES ('{1,2}',4); +INSERT INTO x2 VALUES ('{1,3}',6); +ALTER TABLE x2 ADD CONSTRAINT fk_const FOREIGN KEY(EACH ELEMENT OF x1, x2) REFERENCES x1; -- FAILS +ERROR: insert or update on table "x2" violates foreign key constraint "fk_const" +DETAIL: Key (x1, x2)=({1,3}, 6) is not present in table "x1". +DROP TABLE x2; +DROP TABLE x1; +-- --------------------------------------- +-- Multi-dimensional "ELEMENT" foreign key tests +-- --------------------------------------- +-- Create DIM1 table with two-column primary key +CREATE TABLE DIM1 (X INTEGER NOT NULL PRIMARY KEY, + CODE TEXT NOT NULL UNIQUE); +-- Populate DIM1 table pairs +INSERT INTO DIM1 SELECT t, 'DIM1-' || lpad(t::TEXT, 2, '0') + FROM (SELECT generate_series(1, 10)) x(t); +-- Test with TABLE declaration of an element foreign key constraint (NO ACTION) +CREATE TABLE F1 ( + ID SERIAL PRIMARY KEY, + SLOTS INTEGER[3][3], FOREIGN KEY (EACH ELEMENT OF SLOTS) REFERENCES DIM1 +); +INSERT INTO F1(SLOTS) VALUES ('{{NULL, 1, NULL}, {NULL, NULL, 3}, {NULL, NULL, 6}}'); -- OK +INSERT INTO F1(SLOTS) VALUES ('{{NULL, 1, NULL}, {NULL, NULL, 11}, {NULL, NULL, 6}}'); -- FAILS +ERROR: insert or update on table "f1" violates foreign key constraint "f1_slots_fkey" +DETAIL: Key (slots)=({{NULL,1,NULL},{NULL,NULL,11},{NULL,NULL,6}}) is not present in table "dim1". +INSERT INTO F1(SLOTS) VALUES ('{{1, 2, 3}, {4, 5, 6}, {7, 8, 9}}'); -- OK +INSERT INTO F1(SLOTS) VALUES ('{1, 2, 3, 4, 5, 6, 7, 8, 9}'); -- OK +UPDATE F1 SET SLOTS = '{{NULL, 1, NULL}, {NULL, NULL, 3}, {7, 8, 10}}' WHERE ID = 1; -- OK +UPDATE F1 SET SLOTS = '{{100, 100, 100}, {NULL, NULL, 20}, {7, 8, 10}}' WHERE ID = 1; -- FAILS +ERROR: insert or update on table "f1" violates foreign key constraint "f1_slots_fkey" +DETAIL: Key (slots)=({{100,100,100},{NULL,NULL,20},{7,8,10}}) is not present in table "dim1". +DROP TABLE F1; +-- Test with postponed foreign key +CREATE TABLE F1 ( + ID SERIAL PRIMARY KEY, + SLOTS INTEGER[3][3] +); +INSERT INTO F1(SLOTS) VALUES ('{{NULL, 1, NULL}, {NULL, NULL, 3}, {NULL, NULL, 6}}'); -- OK +INSERT INTO F1(SLOTS) VALUES ('{{NULL, 1, NULL}, {NULL, NULL, 11}, {NULL, NULL, 6}}'); -- OK +INSERT INTO F1(SLOTS) VALUES ('{{1, 2, 3}, {4, 5, 6}, {7, 8, 9}}'); -- OK +INSERT INTO F1(SLOTS) VALUES ('{1, 2, 3, 4, 5, 6, 7, 8, 9}'); -- OK +ALTER TABLE F1 ADD FOREIGN KEY (EACH ELEMENT OF SLOTS) REFERENCES DIM1; -- FAILS +ERROR: insert or update on table "f1" violates foreign key constraint "f1_slots_fkey" +DETAIL: Key (slots)=({{NULL,1,NULL},{NULL,NULL,11},{NULL,NULL,6}}) is not present in table "dim1". +DELETE FROM F1 WHERE ID = 2; -- REMOVE ISSUE +ALTER TABLE F1 ADD FOREIGN KEY (EACH ELEMENT OF SLOTS) REFERENCES DIM1; -- NOW OK +INSERT INTO F1(SLOTS) VALUES ('{{NULL, 1, NULL}, {NULL, NULL, 11}, {NULL, NULL, 6}}'); -- FAILS +ERROR: insert or update on table "f1" violates foreign key constraint "f1_slots_fkey" +DETAIL: Key (slots)=({{NULL,1,NULL},{NULL,NULL,11},{NULL,NULL,6}}) is not present in table "dim1". +DROP TABLE F1; +-- Leave tables in the database +CREATE TABLE PKTABLEFORELEMENTFK ( ptest1 int PRIMARY KEY, ptest2 text ); +CREATE TABLE FKTABLEFORELEMENTFK ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORELEMENTFK, ftest2 int ); +-- Check ALTER TABLE ALTER TYPE +ALTER TABLE FKTABLEFORELEMENTFK ALTER FTEST1 TYPE INT[]; +-- Check GIN index +-- Define PKTABLEFORARRAYGIN +CREATE TABLE PKTABLEFORARRAYGIN ( ptest1 int PRIMARY KEY, ptest2 text ); +-- Insert test data into PKTABLEFORARRAYGIN +INSERT INTO PKTABLEFORARRAYGIN VALUES (1, 'Test1'); +INSERT INTO PKTABLEFORARRAYGIN VALUES (2, 'Test2'); +INSERT INTO PKTABLEFORARRAYGIN VALUES (3, 'Test3'); +INSERT INTO PKTABLEFORARRAYGIN VALUES (4, 'Test4'); +INSERT INTO PKTABLEFORARRAYGIN VALUES (5, 'Test5'); +INSERT INTO PKTABLEFORARRAYGIN VALUES (6, 'Test6'); +-- Define FKTABLEFORARRAYGIN +CREATE TABLE FKTABLEFORARRAYGIN ( ftest1 int[], + ftest2 int PRIMARY KEY, + FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAYGIN + ON DELETE NO ACTION ON UPDATE NO ACTION); +-- -- Create index on FKTABLEFORARRAYGIN +CREATE INDEX ON FKTABLEFORARRAYGIN USING gin (ftest1 array_ops); +-- Populate Table +INSERT INTO FKTABLEFORARRAYGIN VALUES ('{5}', 1); +INSERT INTO FKTABLEFORARRAYGIN VALUES ('{3,2}', 2); +INSERT INTO FKTABLEFORARRAYGIN VALUES ('{3,5,2,5}', 3); +INSERT INTO FKTABLEFORARRAYGIN VALUES ('{3,4,4}', 4); +INSERT INTO FKTABLEFORARRAYGIN VALUES ('{3,5,4,1,3}', 5); +INSERT INTO FKTABLEFORARRAYGIN VALUES ('{1}', 6); +INSERT INTO FKTABLEFORARRAYGIN VALUES ('{5,1}', 7); +INSERT INTO FKTABLEFORARRAYGIN VALUES ('{2,1,2,4,1}', 8); +INSERT INTO FKTABLEFORARRAYGIN VALUES ('{4,2}', 9); +INSERT INTO FKTABLEFORARRAYGIN VALUES ('{3,4,5,3}', 10); +-- Try UPDATE +UPDATE PKTABLEFORARRAYGIN SET ptest1=7 WHERE ptest1=6; +--- Try using the indexable operators +SELECT COUNT(*) FROM FKTABLEFORARRAYGIN; + count +------- + 10 +(1 row) + +SELECT COUNT(*) FROM FKTABLEFORARRAYGIN WHERE ftest1 @> ARRAY[5]; + count +------- + 5 +(1 row) + +SELECT COUNT(*) FROM FKTABLEFORARRAYGIN WHERE ftest1 @>> 5; + count +------- + 5 +(1 row) + +-- Cleanup +DROP TABLE FKTABLEFORARRAYGIN; +DROP TABLE PKTABLEFORARRAYGIN; +-- --------------------------------------- +-- Invalid referencing key tests +-- --------------------------------------- +CREATE TABLE PKTABLEVIOLATING ( ptest1 int PRIMARY KEY, ptest2 text ); +-- Attempt fk constraint between int <-> int +CREATE TABLE FKTABLEVIOLATING ( ftest1 int, FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORELEMENTFK, ftest2 int ); +ERROR: foreign key constraint "fktableviolating_ftest1_fkey" cannot be implemented +DETAIL: Key column "ftest1" has type integer which is not an array type. +-- Attempt fk constraint between int <-> char[] +CREATE TABLE FKTABLEVIOLATING ( ftest1 char[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORELEMENTFK, ftest2 int ); +ERROR: foreign key constraint "fktableviolating_ftest1_fkey" cannot be implemented +DETAIL: Unssupported relation between character[] and integer. +-- Attempt fk constraint between int <-> char +CREATE TABLE FKTABLEVIOLATING ( ftest1 char, FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORELEMENTFK, ftest2 int ); +ERROR: foreign key constraint "fktableviolating_ftest1_fkey" cannot be implemented +DETAIL: Key column "ftest1" has type character which is not an array type. +-- Attempt fk constraint between int <-> smallint[] +CREATE TABLE FKTABLEVIOLATING ( ftest1 smallint[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORELEMENTFK, ftest2 int ); +ERROR: foreign key constraint "fktableviolating_ftest1_fkey" cannot be implemented +DETAIL: Unssupported relation between smallint[] and integer. +-- Attempt fk constraint between int <-> bigint[] +CREATE TABLE FKTABLEVIOLATING ( ftest1 bigint[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORELEMENTFK, ftest2 int ); +ERROR: foreign key constraint "fktableviolating_ftest1_fkey" cannot be implemented +DETAIL: Unssupported relation between bigint[] and integer. +-- Attempt fk constraint between int <-> int2vector +CREATE TABLE FKTABLEVIOLATING ( ftest1 int2vector, FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORELEMENTFK, ftest2 int ); +ERROR: foreign key constraint "fktableviolating_ftest1_fkey" cannot be implemented +DETAIL: Unssupported relation between int2vector and integer. diff --git a/src/test/regress/parallel_schedule b/src/test/regress/parallel_schedule index e280198b17..ec01fffc97 100644 --- a/src/test/regress/parallel_schedule +++ b/src/test/regress/parallel_schedule @@ -80,7 +80,7 @@ test: brin gin gist spgist privileges init_privs security_label collate matview # ---------- # Another group of parallel tests # ---------- -test: create_table_like alter_generic alter_operator misc async dbsize misc_functions sysviews tsrf tid tidscan tidrangescan collate.icu.utf8 incremental_sort +test: create_table_like alter_generic alter_operator misc async dbsize misc_functions sysviews tsrf tid tidscan tidrangescan collate.icu.utf8 incremental_sort element_fk # rules cannot run concurrently with any test that creates # a view or rule in the public schema diff --git a/src/test/regress/serial_schedule b/src/test/regress/serial_schedule index 6a57e889a1..35c7eaff5a 100644 --- a/src/test/regress/serial_schedule +++ b/src/test/regress/serial_schedule @@ -140,6 +140,7 @@ test: tid test: tidscan test: tidrangescan test: collate.icu.utf8 +test: element_fk test: rules test: psql test: psql_crosstab diff --git a/src/test/regress/sql/element_fk.sql b/src/test/regress/sql/element_fk.sql new file mode 100644 index 0000000000..2b3ec36113 --- /dev/null +++ b/src/test/regress/sql/element_fk.sql @@ -0,0 +1,475 @@ +-- EACH-ELEMENT FK CONSTRAINTS + +CREATE TABLE PKTABLEFORARRAY ( ptest1 int PRIMARY KEY, ptest2 text ); + +-- Insert test data into PKTABLEFORARRAY +INSERT INTO PKTABLEFORARRAY VALUES (1, 'Test1'); +INSERT INTO PKTABLEFORARRAY VALUES (2, 'Test2'); +INSERT INTO PKTABLEFORARRAY VALUES (3, 'Test3'); +INSERT INTO PKTABLEFORARRAY VALUES (4, 'Test4'); +INSERT INTO PKTABLEFORARRAY VALUES (5, 'Test5'); + +-- Check alter table +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], ftest2 int ); +ALTER TABLE FKTABLEFORARRAY ADD CONSTRAINT FKARRAY FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY; +DROP TABLE FKTABLEFORARRAY; + +-- Check alter table with rows +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], ftest2 int ); +INSERT INTO FKTABLEFORARRAY VALUES ('{1}', 1); +ALTER TABLE FKTABLEFORARRAY ADD CONSTRAINT FKARRAY FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY; +DROP TABLE FKTABLEFORARRAY; + +-- Check alter table with failing rows +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], ftest2 int ); +INSERT INTO FKTABLEFORARRAY VALUES ('{10,1}', 2); +ALTER TABLE FKTABLEFORARRAY ADD CONSTRAINT FKARRAY FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY; +DROP TABLE FKTABLEFORARRAY; + +-- Check create table +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY, ftest2 int ); +CREATE TABLE FKTABLEFORARRAYMDIM ( ftest1 int[][], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY, ftest2 int ); +CREATE TABLE FKTABLEFORARRAYNOTNULL ( ftest1 int[] NOT NULL, FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY, ftest2 int ); + +-- Insert successful rows into FK TABLE +INSERT INTO FKTABLEFORARRAY VALUES ('{1}', 3); +INSERT INTO FKTABLEFORARRAY VALUES ('{2}', 4); +INSERT INTO FKTABLEFORARRAY VALUES ('{1}', 5); +INSERT INTO FKTABLEFORARRAY VALUES ('{3}', 6); +INSERT INTO FKTABLEFORARRAY VALUES ('{1}', 7); +INSERT INTO FKTABLEFORARRAY VALUES ('{4,5}', 8); +INSERT INTO FKTABLEFORARRAY VALUES ('{4,4}', 9); +INSERT INTO FKTABLEFORARRAY VALUES (NULL, 10); +INSERT INTO FKTABLEFORARRAY VALUES ('{}', 11); +INSERT INTO FKTABLEFORARRAY VALUES ('{1,NULL}', 12); +INSERT INTO FKTABLEFORARRAY VALUES ('{NULL}', 13); +INSERT INTO FKTABLEFORARRAYMDIM VALUES ('{{4,5},{1,2},{1,3}}', 14); +INSERT INTO FKTABLEFORARRAYMDIM VALUES ('{{4,5},{NULL,2},{NULL,3}}', 15); + +-- Insert failed rows into FK TABLE +INSERT INTO FKTABLEFORARRAY VALUES ('{6}', 16); +INSERT INTO FKTABLEFORARRAY VALUES ('{4,6}', 17); +INSERT INTO FKTABLEFORARRAY VALUES ('{6,NULL}', 18); +INSERT INTO FKTABLEFORARRAY VALUES ('{6,NULL,4,NULL}', 19); +INSERT INTO FKTABLEFORARRAYMDIM VALUES ('{{1,2},{6,NULL}}', 20); +INSERT INTO FKTABLEFORARRAYNOTNULL VALUES (NULL, 21); + +-- Check FKTABLE +SELECT * FROM FKTABLEFORARRAY; + +-- Delete a row from PK TABLE (must fail due to ON DELETE NO ACTION) +DELETE FROM PKTABLEFORARRAY WHERE ptest1=1; + +-- Check FKTABLE for removal of matched row +SELECT * FROM FKTABLEFORARRAY; + +-- Update a row from PK TABLE (must fail due to ON UPDATE NO ACTION) +UPDATE PKTABLEFORARRAY SET ptest1=7 WHERE ptest1=1; + +-- Check FKTABLE for update of matched row +SELECT * FROM FKTABLEFORARRAY; + +-- Check UPDATE on FKTABLE +UPDATE FKTABLEFORARRAY SET ftest1=ARRAY[1] WHERE ftest2=4; + +-- Check FKTABLE for update of matched row +SELECT * FROM FKTABLEFORARRAY; + +DROP TABLE FKTABLEFORARRAY; +DROP TABLE FKTABLEFORARRAYNOTNULL; +DROP TABLE FKTABLEFORARRAYMDIM; + +-- Allowed references with actions (NO ACTION, RESTRICT) +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE NO ACTION ON UPDATE NO ACTION, ftest2 int ); +DROP TABLE FKTABLEFORARRAY; +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE NO ACTION ON UPDATE RESTRICT, ftest2 int ); +DROP TABLE FKTABLEFORARRAY; +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE RESTRICT ON UPDATE NO ACTION, ftest2 int ); +DROP TABLE FKTABLEFORARRAY; +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE RESTRICT ON UPDATE RESTRICT, ftest2 int ); +DROP TABLE FKTABLEFORARRAY; +-- Not allowed references (SET NULL, SET DEFAULT, CASCADE) +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE NO ACTION ON UPDATE SET DEFAULT, ftest2 int ); +DROP TABLE IF EXISTS FKTABLEFORARRAY; +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE NO ACTION ON UPDATE SET NULL, ftest2 int ); +DROP TABLE IF EXISTS FKTABLEFORARRAY; +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE RESTRICT ON UPDATE SET DEFAULT, ftest2 int ); +DROP TABLE IF EXISTS FKTABLEFORARRAY; +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE RESTRICT ON UPDATE SET NULL, ftest2 int ); +DROP TABLE IF EXISTS FKTABLEFORARRAY; +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE CASCADE ON UPDATE NO ACTION, ftest2 int ); +DROP TABLE FKTABLEFORARRAY; +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE CASCADE ON UPDATE RESTRICT, ftest2 int ); +DROP TABLE FKTABLEFORARRAY; +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE CASCADE ON UPDATE SET DEFAULT, ftest2 int ); +DROP TABLE IF EXISTS FKTABLEFORARRAY; +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE CASCADE ON UPDATE SET NULL, ftest2 int ); +DROP TABLE IF EXISTS FKTABLEFORARRAY; +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE SET DEFAULT ON UPDATE NO ACTION, ftest2 int ); +DROP TABLE IF EXISTS FKTABLEFORARRAY; +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE SET DEFAULT ON UPDATE RESTRICT, ftest2 int ); +DROP TABLE IF EXISTS FKTABLEFORARRAY; +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE SET DEFAULT ON UPDATE SET DEFAULT, ftest2 int ); +DROP TABLE IF EXISTS FKTABLEFORARRAY; +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE SET DEFAULT ON UPDATE SET NULL, ftest2 int ); +DROP TABLE IF EXISTS FKTABLEFORARRAY; +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE SET NULL ON UPDATE NO ACTION, ftest2 int ); +DROP TABLE IF EXISTS FKTABLEFORARRAY; +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE SET NULL ON UPDATE RESTRICT, ftest2 int ); +DROP TABLE IF EXISTS FKTABLEFORARRAY; +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE SET NULL ON UPDATE SET DEFAULT, ftest2 int ); +DROP TABLE IF EXISTS FKTABLEFORARRAY; +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE SET NULL ON UPDATE SET NULL, ftest2 int ); +DROP TABLE IF EXISTS FKTABLEFORARRAY; +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE NO ACTION ON UPDATE CASCADE, ftest2 int ); +DROP TABLE IF EXISTS FKTABLEFORARRAY; +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE RESTRICT ON UPDATE CASCADE, ftest2 int ); +DROP TABLE IF EXISTS FKTABLEFORARRAY; +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE CASCADE ON UPDATE CASCADE, ftest2 int ); +DROP TABLE IF EXISTS FKTABLEFORARRAY; +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE SET DEFAULT ON UPDATE CASCADE, ftest2 int ); +DROP TABLE IF EXISTS FKTABLEFORARRAY; +CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON DELETE SET NULL ON UPDATE CASCADE, ftest2 int ); +DROP TABLE IF EXISTS FKTABLEFORARRAY; + +-- Cleanup +DROP TABLE PKTABLEFORARRAY; + +-- Check reference on empty table +CREATE TABLE PKTABLEFORARRAY (ptest1 int PRIMARY KEY); +CREATE TABLE FKTABLEFORARRAY (ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY); +INSERT INTO FKTABLEFORARRAY VALUES ('{}'); +DROP TABLE FKTABLEFORARRAY; +DROP TABLE PKTABLEFORARRAY; + +-- Repeat a similar test using CHAR(1) keys rather than INTEGER +CREATE TABLE PKTABLEFORARRAY ( ptest1 CHAR(1) PRIMARY KEY, ptest2 text ); + +-- Populate the primary table +INSERT INTO PKTABLEFORARRAY VALUES ('A', 'Test A'); +INSERT INTO PKTABLEFORARRAY VALUES ('B', 'Test B'); +INSERT INTO PKTABLEFORARRAY VALUES ('C', 'Test C'); + +-- Create the referencing table +CREATE TABLE FKTABLEFORARRAY ( ftest1 char(1)[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAY ON UPDATE RESTRICT ON DELETE RESTRICT, ftest2 int ); + +-- Insert valid rows into FK TABLE +INSERT INTO FKTABLEFORARRAY VALUES ('{"A"}', 1); +INSERT INTO FKTABLEFORARRAY VALUES ('{"B"}', 2); +INSERT INTO FKTABLEFORARRAY VALUES ('{"C"}', 3); +INSERT INTO FKTABLEFORARRAY VALUES ('{"A","B","C"}', 4); + +-- Insert invalid rows into FK TABLE +INSERT INTO FKTABLEFORARRAY VALUES ('{"D"}', 5); +INSERT INTO FKTABLEFORARRAY VALUES ('{"A","B","D"}', 6); + +-- Check FKTABLE +SELECT * FROM FKTABLEFORARRAY; + +-- Delete a row from PK TABLE (must fail due to ON DELETE RESTRICT) +DELETE FROM PKTABLEFORARRAY WHERE ptest1='A'; + +-- Check FKTABLE for removal of matched row +SELECT * FROM FKTABLEFORARRAY; + +-- Update a row from PK TABLE (must fail due to ON UPDATE RESTRICT) +UPDATE PKTABLEFORARRAY SET ptest1='D' WHERE ptest1='B'; + +-- Check FKTABLE for update of matched row +SELECT * FROM FKTABLEFORARRAY; + +-- Cleanup +DROP TABLE FKTABLEFORARRAY; +DROP TABLE PKTABLEFORARRAY; + +-- Composite primary keys +CREATE TABLE PKTABLEFORARRAY ( id1 CHAR(1), id2 CHAR(1), ptest2 text, PRIMARY KEY (id1, id2) ); + +-- Populate the primary table +INSERT INTO PKTABLEFORARRAY VALUES ('A', 'A', 'Test A'); +INSERT INTO PKTABLEFORARRAY VALUES ('A', 'B', 'Test B'); +INSERT INTO PKTABLEFORARRAY VALUES ('B', 'C', 'Test B'); + +-- Create the referencing table +CREATE TABLE FKTABLEFORARRAY ( fid1 CHAR(1), fid2 CHAR(1)[], ftest2 text, FOREIGN KEY (fid1, EACH ELEMENT OF fid2) REFERENCES PKTABLEFORARRAY); + +-- Insert valid rows into FK TABLE +INSERT INTO FKTABLEFORARRAY VALUES ('A', ARRAY['A','B'], '1'); +INSERT INTO FKTABLEFORARRAY VALUES ('B', ARRAY['C'], '2'); + +-- Insert invalid rows into FK TABLE +INSERT INTO FKTABLEFORARRAY VALUES ('A', ARRAY['A','B', 'C'], '3'); +INSERT INTO FKTABLEFORARRAY VALUES ('B', ARRAY['A'], '4'); + +-- Cleanup +DROP TABLE FKTABLEFORARRAY; +DROP TABLE PKTABLEFORARRAY; + +-- Test Array Element Foreign Keys with composite type +CREATE TYPE INVOICEID AS (year_part INTEGER, progressive_part INTEGER); +CREATE TABLE PKTABLEFORARRAY ( id INVOICEID PRIMARY KEY, ptest2 text); + +-- Populate the primary table +INSERT INTO PKTABLEFORARRAY VALUES (ROW(2010, 99), 'Last invoice for 2010'); +INSERT INTO PKTABLEFORARRAY VALUES (ROW(2011, 1), 'First invoice for 2011'); +INSERT INTO PKTABLEFORARRAY VALUES (ROW(2011, 2), 'Second invoice for 2011'); + +-- Create the referencing table +CREATE TABLE FKTABLEFORARRAY ( id SERIAL PRIMARY KEY, invoice_ids INVOICEID[], FOREIGN KEY (EACH ELEMENT OF invoice_ids) REFERENCES PKTABLEFORARRAY, ftest2 TEXT ); + +-- Insert valid rows into FK TABLE +INSERT INTO FKTABLEFORARRAY(invoice_ids, ftest2) VALUES (ARRAY['(2010,99)']::INVOICEID[], 'Product A'); +INSERT INTO FKTABLEFORARRAY(invoice_ids, ftest2) VALUES (ARRAY['(2011,1)','(2011,2)']::INVOICEID[], 'Product B'); +INSERT INTO FKTABLEFORARRAY(invoice_ids, ftest2) VALUES (ARRAY['(2011,2)']::INVOICEID[], 'Product C'); + +-- Insert invalid rows into FK TABLE +INSERT INTO FKTABLEFORARRAY(invoice_ids, ftest2) VALUES (ARRAY['(2011,99)']::INVOICEID[], 'Product A'); +INSERT INTO FKTABLEFORARRAY(invoice_ids, ftest2) VALUES (ARRAY['(2011,1)','(2010,1)']::INVOICEID[], 'Product B'); + +-- Check FKTABLE +SELECT * FROM FKTABLEFORARRAY; + +-- Delete a row from PK TABLE +DELETE FROM PKTABLEFORARRAY WHERE id=ROW(2010,99); + +-- Check FKTABLE for removal of matched row +SELECT * FROM FKTABLEFORARRAY; + +-- Update a row from PK TABLE +UPDATE PKTABLEFORARRAY SET id=ROW(2011,99) WHERE id=ROW(2011,1); + +-- Check FKTABLE for update of matched row +SELECT * FROM FKTABLEFORARRAY; + +-- Cleanup +DROP TABLE FKTABLEFORARRAY; +DROP TABLE PKTABLEFORARRAY; +DROP TYPE INVOICEID; + +-- Check for an array column referencing another array column (NOT ELEMENT FOREIGN KEY) +-- Create primary table with an array primary key +CREATE TABLE PKTABLEFORARRAY ( id INT[] PRIMARY KEY, ptest2 text); + +-- Create the referencing table +CREATE TABLE FKTABLEFORARRAY ( id SERIAL PRIMARY KEY, fids INT[] REFERENCES PKTABLEFORARRAY, ftest2 TEXT ); + +-- Populate the primary table +INSERT INTO PKTABLEFORARRAY VALUES ('{1,1}', 'A'); +INSERT INTO PKTABLEFORARRAY VALUES ('{1,2}', 'B'); + +-- Insert valid rows into FK TABLE +INSERT INTO FKTABLEFORARRAY (fids, ftest2) VALUES ('{1,1}', 'Product A'); +INSERT INTO FKTABLEFORARRAY (fids, ftest2) VALUES ('{1,2}', 'Product B'); + +-- Insert invalid rows into FK TABLE +INSERT INTO FKTABLEFORARRAY (fids, ftest2) VALUES ('{0,1}', 'Product C'); +INSERT INTO FKTABLEFORARRAY (fids, ftest2) VALUES ('{2,1}', 'Product D'); + +-- Cleanup +DROP TABLE FKTABLEFORARRAY; +DROP TABLE PKTABLEFORARRAY; + +-- --------------------------------------- +-- Multi-column "ELEMENT" foreign key tests +-- --------------------------------------- + +-- Create DIM1 table with two-column primary key +CREATE TABLE DIM1 (X INTEGER NOT NULL, Y INTEGER NOT NULL, PRIMARY KEY (X, Y)); +-- Populate DIM1 table pairs +INSERT INTO DIM1 SELECT x.t, x.t * y.t + FROM (SELECT generate_series(1, 10) AS t) x, + (SELECT generate_series(0, 10) AS t) y; + + +-- Test with TABLE declaration of an element foreign key constraint (NO ACTION) +CREATE TABLE F1 ( + x INTEGER PRIMARY KEY, y INTEGER[], + FOREIGN KEY (x, EACH ELEMENT OF y) REFERENCES DIM1(x, y) +); +-- Insert facts +INSERT INTO F1 VALUES (1, '{0,1,2,3,4,5}'); -- OK +INSERT INTO F1 VALUES (2, '{0,2,4,6}'); -- OK +INSERT INTO F1 VALUES (3, '{0,3,6,9,0,3,6,9,0,0,0,0,9,9}'); -- OK (multiple occurrences) +INSERT INTO F1 VALUES (4, '{0,2,4}'); -- FAILS (2 is not present) +INSERT INTO F1 VALUES (4, '{0,NULL,4}'); -- OK +INSERT INTO F1 VALUES (5, '{0,NULL,5}'); -- OK +-- Try updates +UPDATE F1 SET y = '{0,2,4,6}' WHERE x = 2; -- OK +UPDATE F1 SET y = '{0,2,3,4,6}' WHERE x = 2; -- FAILS +UPDATE F1 SET x = 20, y = '{0,2,3,4,6}' WHERE x = 2; -- FAILS (20 does not exist) +UPDATE F1 SET y = '{0,4,8}' WHERE x = 4; -- OK +UPDATE F1 SET y = '{0,5,NULL,10}' WHERE x = 5; -- OK +DROP TABLE F1; + + +-- Test with FOREIGN KEY after TABLE population +CREATE TABLE F1 ( + x INTEGER PRIMARY KEY, y INTEGER[] +); +-- Insert facts +INSERT INTO F1 VALUES (1, '{0,1,2,3,4,5}'); -- OK +INSERT INTO F1 VALUES (2, '{0,2,4,6}'); -- OK +INSERT INTO F1 VALUES (3, '{0,3,6,9,0,3,6,9,0,0,0,0,9,9}'); -- OK (multiple occurrences) +INSERT INTO F1 VALUES (4, '{0,2,4}'); -- OK (2 is not present) +INSERT INTO F1 VALUES (5, '{0,NULL,5}'); -- OK +-- Add foreign key (FAILS) +ALTER TABLE F1 ADD FOREIGN KEY (x, EACH ELEMENT OF y) REFERENCES DIM1(x, y); +DROP TABLE F1; + + +-- Test with TABLE declaration of a two-dim ELEMENT foreign key constraint (FAILS) +CREATE TABLE F1 ( + x INTEGER[] PRIMARY KEY, y INTEGER[], + FOREIGN KEY (EACH ELEMENT OF x, EACH ELEMENT OF y) REFERENCES DIM1(x, y) +); + + +-- Test with two-dim ELEMENT foreign key after TABLE population +CREATE TABLE F1 ( + x INTEGER[] PRIMARY KEY, y INTEGER[] +); +INSERT INTO F1 VALUES ('{1}', '{0,1,2,3,4,5}'); -- OK +INSERT INTO F1 VALUES ('{1,2}', '{0,2,4,6}'); -- OK +-- Add foreign key (FAILS) +ALTER TABLE F1 ADD FOREIGN KEY (EACH ELEMENT OF x, EACH ELEMENT OF y) REFERENCES DIM1(x, y); +DROP TABLE F1; + +-- Cleanup +DROP TABLE DIM1; + + +-- Check for potential name conflicts (with internal integrity checks) +CREATE TABLE x1(x1 int, x2 int, PRIMARY KEY(x1,x2)); +INSERT INTO x1 VALUES + (1,4), + (1,5), + (2,4), + (2,5), + (3,6), + (3,7) +; +CREATE TABLE x2(x1 int[], x2 int, FOREIGN KEY(EACH ELEMENT OF x1, x2) REFERENCES x1); +INSERT INTO x2 VALUES ('{1,2}',4); +INSERT INTO x2 VALUES ('{1,3}',6); -- FAILS +DROP TABLE x2; +CREATE TABLE x2(x1 int[], x2 int); +INSERT INTO x2 VALUES ('{1,2}',4); +INSERT INTO x2 VALUES ('{1,3}',6); +ALTER TABLE x2 ADD CONSTRAINT fk_const FOREIGN KEY(EACH ELEMENT OF x1, x2) REFERENCES x1; -- FAILS +DROP TABLE x2; +DROP TABLE x1; + + +-- --------------------------------------- +-- Multi-dimensional "ELEMENT" foreign key tests +-- --------------------------------------- + +-- Create DIM1 table with two-column primary key +CREATE TABLE DIM1 (X INTEGER NOT NULL PRIMARY KEY, + CODE TEXT NOT NULL UNIQUE); +-- Populate DIM1 table pairs +INSERT INTO DIM1 SELECT t, 'DIM1-' || lpad(t::TEXT, 2, '0') + FROM (SELECT generate_series(1, 10)) x(t); + +-- Test with TABLE declaration of an element foreign key constraint (NO ACTION) +CREATE TABLE F1 ( + ID SERIAL PRIMARY KEY, + SLOTS INTEGER[3][3], FOREIGN KEY (EACH ELEMENT OF SLOTS) REFERENCES DIM1 +); +INSERT INTO F1(SLOTS) VALUES ('{{NULL, 1, NULL}, {NULL, NULL, 3}, {NULL, NULL, 6}}'); -- OK +INSERT INTO F1(SLOTS) VALUES ('{{NULL, 1, NULL}, {NULL, NULL, 11}, {NULL, NULL, 6}}'); -- FAILS +INSERT INTO F1(SLOTS) VALUES ('{{1, 2, 3}, {4, 5, 6}, {7, 8, 9}}'); -- OK +INSERT INTO F1(SLOTS) VALUES ('{1, 2, 3, 4, 5, 6, 7, 8, 9}'); -- OK +UPDATE F1 SET SLOTS = '{{NULL, 1, NULL}, {NULL, NULL, 3}, {7, 8, 10}}' WHERE ID = 1; -- OK +UPDATE F1 SET SLOTS = '{{100, 100, 100}, {NULL, NULL, 20}, {7, 8, 10}}' WHERE ID = 1; -- FAILS +DROP TABLE F1; + +-- Test with postponed foreign key +CREATE TABLE F1 ( + ID SERIAL PRIMARY KEY, + SLOTS INTEGER[3][3] +); +INSERT INTO F1(SLOTS) VALUES ('{{NULL, 1, NULL}, {NULL, NULL, 3}, {NULL, NULL, 6}}'); -- OK +INSERT INTO F1(SLOTS) VALUES ('{{NULL, 1, NULL}, {NULL, NULL, 11}, {NULL, NULL, 6}}'); -- OK +INSERT INTO F1(SLOTS) VALUES ('{{1, 2, 3}, {4, 5, 6}, {7, 8, 9}}'); -- OK +INSERT INTO F1(SLOTS) VALUES ('{1, 2, 3, 4, 5, 6, 7, 8, 9}'); -- OK +ALTER TABLE F1 ADD FOREIGN KEY (EACH ELEMENT OF SLOTS) REFERENCES DIM1; -- FAILS +DELETE FROM F1 WHERE ID = 2; -- REMOVE ISSUE +ALTER TABLE F1 ADD FOREIGN KEY (EACH ELEMENT OF SLOTS) REFERENCES DIM1; -- NOW OK +INSERT INTO F1(SLOTS) VALUES ('{{NULL, 1, NULL}, {NULL, NULL, 11}, {NULL, NULL, 6}}'); -- FAILS +DROP TABLE F1; + +-- Leave tables in the database +CREATE TABLE PKTABLEFORELEMENTFK ( ptest1 int PRIMARY KEY, ptest2 text ); +CREATE TABLE FKTABLEFORELEMENTFK ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORELEMENTFK, ftest2 int ); + +-- Check ALTER TABLE ALTER TYPE +ALTER TABLE FKTABLEFORELEMENTFK ALTER FTEST1 TYPE INT[]; + +-- Check GIN index +-- Define PKTABLEFORARRAYGIN +CREATE TABLE PKTABLEFORARRAYGIN ( ptest1 int PRIMARY KEY, ptest2 text ); + +-- Insert test data into PKTABLEFORARRAYGIN +INSERT INTO PKTABLEFORARRAYGIN VALUES (1, 'Test1'); +INSERT INTO PKTABLEFORARRAYGIN VALUES (2, 'Test2'); +INSERT INTO PKTABLEFORARRAYGIN VALUES (3, 'Test3'); +INSERT INTO PKTABLEFORARRAYGIN VALUES (4, 'Test4'); +INSERT INTO PKTABLEFORARRAYGIN VALUES (5, 'Test5'); +INSERT INTO PKTABLEFORARRAYGIN VALUES (6, 'Test6'); + +-- Define FKTABLEFORARRAYGIN +CREATE TABLE FKTABLEFORARRAYGIN ( ftest1 int[], + ftest2 int PRIMARY KEY, + FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORARRAYGIN + ON DELETE NO ACTION ON UPDATE NO ACTION); + +-- -- Create index on FKTABLEFORARRAYGIN +CREATE INDEX ON FKTABLEFORARRAYGIN USING gin (ftest1 array_ops); + +-- Populate Table +INSERT INTO FKTABLEFORARRAYGIN VALUES ('{5}', 1); +INSERT INTO FKTABLEFORARRAYGIN VALUES ('{3,2}', 2); +INSERT INTO FKTABLEFORARRAYGIN VALUES ('{3,5,2,5}', 3); +INSERT INTO FKTABLEFORARRAYGIN VALUES ('{3,4,4}', 4); +INSERT INTO FKTABLEFORARRAYGIN VALUES ('{3,5,4,1,3}', 5); +INSERT INTO FKTABLEFORARRAYGIN VALUES ('{1}', 6); +INSERT INTO FKTABLEFORARRAYGIN VALUES ('{5,1}', 7); +INSERT INTO FKTABLEFORARRAYGIN VALUES ('{2,1,2,4,1}', 8); +INSERT INTO FKTABLEFORARRAYGIN VALUES ('{4,2}', 9); +INSERT INTO FKTABLEFORARRAYGIN VALUES ('{3,4,5,3}', 10); + +-- Try UPDATE +UPDATE PKTABLEFORARRAYGIN SET ptest1=7 WHERE ptest1=6; + +--- Try using the indexable operators +SELECT COUNT(*) FROM FKTABLEFORARRAYGIN; +SELECT COUNT(*) FROM FKTABLEFORARRAYGIN WHERE ftest1 @> ARRAY[5]; +SELECT COUNT(*) FROM FKTABLEFORARRAYGIN WHERE ftest1 @>> 5; + +-- Cleanup +DROP TABLE FKTABLEFORARRAYGIN; +DROP TABLE PKTABLEFORARRAYGIN; + +-- --------------------------------------- +-- Invalid referencing key tests +-- --------------------------------------- +CREATE TABLE PKTABLEVIOLATING ( ptest1 int PRIMARY KEY, ptest2 text ); + +-- Attempt fk constraint between int <-> int +CREATE TABLE FKTABLEVIOLATING ( ftest1 int, FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORELEMENTFK, ftest2 int ); + +-- Attempt fk constraint between int <-> char[] +CREATE TABLE FKTABLEVIOLATING ( ftest1 char[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORELEMENTFK, ftest2 int ); + +-- Attempt fk constraint between int <-> char +CREATE TABLE FKTABLEVIOLATING ( ftest1 char, FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORELEMENTFK, ftest2 int ); + +-- Attempt fk constraint between int <-> smallint[] +CREATE TABLE FKTABLEVIOLATING ( ftest1 smallint[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORELEMENTFK, ftest2 int ); + +-- Attempt fk constraint between int <-> bigint[] +CREATE TABLE FKTABLEVIOLATING ( ftest1 bigint[], FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORELEMENTFK, ftest2 int ); + +-- Attempt fk constraint between int <-> int2vector +CREATE TABLE FKTABLEVIOLATING ( ftest1 int2vector, FOREIGN KEY (EACH ELEMENT OF ftest1) REFERENCES PKTABLEFORELEMENTFK, ftest2 int ); -- 2.30.1 --------------3A927D779D0A0B91F132AD46 Content-Type: text/x-patch; charset=UTF-8; name="v10-0001-anyarray_anyelement_operators.patch" Content-Transfer-Encoding: 7bit Content-Disposition: attachment; filename="v10-0001-anyarray_anyelement_operators.patch" ^ permalink raw reply [nested|flat] 4+ messages in thread
* Re: The documentation for READ COMMITTED may be incomplete or wrong @ 2023-05-18 22:26 Tom Lane <[email protected]> 0 siblings, 1 reply; 4+ messages in thread From: Tom Lane @ 2023-05-18 22:26 UTC (permalink / raw) To: Nathan Bossart <[email protected]>; +Cc: Aleksander Alekseev <[email protected]>; Amit Langote <[email protected]>; PostgreSQL Hackers <[email protected]> Nathan Bossart <[email protected]> writes: > On Thu, May 18, 2023 at 04:03:36PM -0400, Tom Lane wrote: >> What we need to do, I think, is set epqstate->relsubs_done[] for >> all target relations except the one we are stuffing a tuple into. > This seems generally reasonable to me. Here's a draft patch for this. I think it's OK for HEAD, but we are going to need some different details for the back branches, because adding a field to struct EPQState would create an ABI break. Our standard trick of shoving the field to the end in the back branches won't help, because struct EPQState is actually embedded in struct ModifyTableState, meaning that all the subsequent fields therein will move. Maybe we can get away with that, but I bet not. I think what the back branches will have to do is reinitialize the relsubs_done array every time through EvalPlanQual(), which is a bit sad but probably doesn't amount to anything compared to the startup overhead of the sub-executor. Debian Code Search doesn't know of any outside code touching relsubs_done, so I think we are safe in dropping that code in ExecScanReScan. It seems quite pointless anyway considering that up to now, EvalPlanQualBegin has always zeroed the whole array. regards, tom lane Attachments: [text/x-diff] v1-fix-EPQ-HEAD-only.patch (15.7K, ../../[email protected]/2-v1-fix-EPQ-HEAD-only.patch) download | inline diff: diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c index 0186be452c..c518daa7d0 100644 --- a/src/backend/executor/execMain.c +++ b/src/backend/executor/execMain.c @@ -2469,7 +2469,7 @@ ExecBuildAuxRowMark(ExecRowMark *erm, List *targetlist) * relation - table containing tuple * rti - rangetable index of table containing tuple * inputslot - tuple for processing - this can be the slot from - * EvalPlanQualSlot(), for the increased efficiency. + * EvalPlanQualSlot() for this rel, for increased efficiency. * * This tests whether the tuple in inputslot still matches the relevant * quals. For that result to be useful, typically the input tuple has to be @@ -2503,6 +2503,13 @@ EvalPlanQual(EPQState *epqstate, Relation relation, if (testslot != inputslot) ExecCopySlot(testslot, inputslot); + /* + * Mark that an EPQ tuple is available for this relation. (If there is + * more than one result relation, the others remain marked as having no + * tuple available.) + */ + epqstate->relsubs_done[rti - 1] = false; + /* * Run the EPQ query. We assume it will return at most one tuple. */ @@ -2519,11 +2526,12 @@ EvalPlanQual(EPQState *epqstate, Relation relation, ExecMaterializeSlot(slot); /* - * Clear out the test tuple. This is needed in case the EPQ query is - * re-used to test a tuple for a different relation. (Not clear that can - * really happen, but let's be safe.) + * Clear out the test tuple, and mark that no tuple is available here. + * This is needed in case the EPQ state is re-used to test a tuple for a + * different target relation. */ ExecClearTuple(testslot); + epqstate->relsubs_done[rti - 1] = true; return slot; } @@ -2532,18 +2540,26 @@ EvalPlanQual(EPQState *epqstate, Relation relation, * EvalPlanQualInit -- initialize during creation of a plan state node * that might need to invoke EPQ processing. * + * If the caller intends to use EvalPlanQual(), resultRelations should be + * a list of RT indexes of potential target relations for EvalPlanQual(), + * and we will arrange that the other listed relations don't return any + * tuple during an EvalPlanQual() call. Otherwise resultRelations + * should be NIL. + * * Note: subplan/auxrowmarks can be NULL/NIL if they will be set later * with EvalPlanQualSetPlan. */ void EvalPlanQualInit(EPQState *epqstate, EState *parentestate, - Plan *subplan, List *auxrowmarks, int epqParam) + Plan *subplan, List *auxrowmarks, + int epqParam, List *resultRelations) { Index rtsize = parentestate->es_range_table_size; /* initialize data not changing over EPQState's lifetime */ epqstate->parentestate = parentestate; epqstate->epqParam = epqParam; + epqstate->resultRelations = resultRelations; /* * Allocate space to reference a slot for each potential rti - do so now @@ -2760,10 +2776,20 @@ EvalPlanQualBegin(EPQState *epqstate) /* * We already have a suitable child EPQ tree, so just reset it. */ - Index rtsize = parentestate->es_range_table_size; PlanState *rcplanstate = epqstate->recheckplanstate; - MemSet(epqstate->relsubs_done, 0, rtsize * sizeof(bool)); + /* + * We reset the relsubs_done[] flags only when the caller of + * EvalPlanQualInit provided no result relations. Otherwise, we keep + * the current state (which should have all result relations marked as + * "done", and others not). + */ + if (epqstate->resultRelations == NIL) + { + Index rtsize = parentestate->es_range_table_size; + + memset(epqstate->relsubs_done, 0, rtsize * sizeof(bool)); + } /* Recopy current values of parent parameters */ if (parentestate->es_plannedstmt->paramExecTypes != NIL) @@ -2931,11 +2957,20 @@ EvalPlanQualStart(EPQState *epqstate, Plan *planTree) } /* - * Initialize per-relation EPQ tuple states to not-fetched. + * Initialize per-relation EPQ tuple states. Result relations, if any, + * get marked as already-fetched; others as not-fetched. */ epqstate->relsubs_done = (bool *) palloc0(rtsize * sizeof(bool)); + foreach(l, epqstate->resultRelations) + { + int rtindex = lfirst_int(l); + + Assert(rtindex > 0 && rtindex <= rtsize); + epqstate->relsubs_done[rtindex - 1] = true; + } + /* * Initialize the private state information for all the nodes in the part * of the plan tree we need to run. This opens files, allocates storage diff --git a/src/backend/executor/execScan.c b/src/backend/executor/execScan.c index cf1871b0f5..ea8c5ab149 100644 --- a/src/backend/executor/execScan.c +++ b/src/backend/executor/execScan.c @@ -69,13 +69,12 @@ ExecScanFetch(ScanState *node, else if (epqstate->relsubs_done[scanrelid - 1]) { /* - * Return empty slot, as we already performed an EPQ substitution - * for this relation. + * Return empty slot, as either there is no EPQ tuple for this rel + * or we already returned it. */ TupleTableSlot *slot = node->ss_ScanTupleSlot; - /* Return empty slot, as we already returned a tuple */ return ExecClearTuple(slot); } else if (epqstate->relsubs_slot[scanrelid - 1] != NULL) @@ -88,7 +87,7 @@ ExecScanFetch(ScanState *node, Assert(epqstate->relsubs_rowmark[scanrelid - 1] == NULL); - /* Mark to remember that we shouldn't return more */ + /* Mark to remember that we shouldn't return it again */ epqstate->relsubs_done[scanrelid - 1] = true; /* Return empty slot if we haven't got a test tuple */ @@ -298,45 +297,9 @@ ExecAssignScanProjectionInfoWithVarno(ScanState *node, int varno) void ExecScanReScan(ScanState *node) { - EState *estate = node->ps.state; - /* * We must clear the scan tuple so that observers (e.g., execCurrent.c) * can tell that this plan node is not positioned on a tuple. */ ExecClearTuple(node->ss_ScanTupleSlot); - - /* Rescan EvalPlanQual tuple if we're inside an EvalPlanQual recheck */ - if (estate->es_epq_active != NULL) - { - EPQState *epqstate = estate->es_epq_active; - Index scanrelid = ((Scan *) node->ps.plan)->scanrelid; - - if (scanrelid > 0) - epqstate->relsubs_done[scanrelid - 1] = false; - else - { - Bitmapset *relids; - int rtindex = -1; - - /* - * If an FDW or custom scan provider has replaced the join with a - * scan, there are multiple RTIs; reset the epqScanDone flag for - * all of them. - */ - if (IsA(node->ps.plan, ForeignScan)) - relids = ((ForeignScan *) node->ps.plan)->fs_base_relids; - else if (IsA(node->ps.plan, CustomScan)) - relids = ((CustomScan *) node->ps.plan)->custom_relids; - else - elog(ERROR, "unexpected scan node: %d", - (int) nodeTag(node->ps.plan)); - - while ((rtindex = bms_next_member(relids, rtindex)) >= 0) - { - Assert(rtindex > 0); - epqstate->relsubs_done[rtindex - 1] = false; - } - } - } } diff --git a/src/backend/executor/nodeLockRows.c b/src/backend/executor/nodeLockRows.c index 407414fc0c..e459971d32 100644 --- a/src/backend/executor/nodeLockRows.c +++ b/src/backend/executor/nodeLockRows.c @@ -108,7 +108,6 @@ lnext: /* this child is inactive right now */ erm->ermActive = false; ItemPointerSetInvalid(&(erm->curCtid)); - ExecClearTuple(markSlot); continue; } } @@ -370,7 +369,7 @@ ExecInitLockRows(LockRows *node, EState *estate, int eflags) /* Now we have the info needed to set up EPQ state */ EvalPlanQualInit(&lrstate->lr_epqstate, estate, - outerPlan, epq_arowmarks, node->epqParam); + outerPlan, epq_arowmarks, node->epqParam, NIL); return lrstate; } diff --git a/src/backend/executor/nodeModifyTable.c b/src/backend/executor/nodeModifyTable.c index dc1a2ec551..7f5002527f 100644 --- a/src/backend/executor/nodeModifyTable.c +++ b/src/backend/executor/nodeModifyTable.c @@ -3985,7 +3985,8 @@ ExecInitModifyTable(ModifyTable *node, EState *estate, int eflags) } /* set up epqstate with dummy subplan data for the moment */ - EvalPlanQualInit(&mtstate->mt_epqstate, estate, NULL, NIL, node->epqParam); + EvalPlanQualInit(&mtstate->mt_epqstate, estate, NULL, NIL, + node->epqParam, node->resultRelations); mtstate->fireBSTriggers = true; /* diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index 879309b316..4b67098814 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -2674,7 +2674,7 @@ apply_handle_update_internal(ApplyExecutionData *edata, bool found; MemoryContext oldctx; - EvalPlanQualInit(&epqstate, estate, NULL, NIL, -1); + EvalPlanQualInit(&epqstate, estate, NULL, NIL, -1, NIL); ExecOpenIndices(relinfo, false); found = FindReplTupleInLocalRel(estate, localrel, @@ -2827,7 +2827,7 @@ apply_handle_delete_internal(ApplyExecutionData *edata, TupleTableSlot *localslot; bool found; - EvalPlanQualInit(&epqstate, estate, NULL, NIL, -1); + EvalPlanQualInit(&epqstate, estate, NULL, NIL, -1, NIL); ExecOpenIndices(relinfo, false); found = FindReplTupleInLocalRel(estate, localrel, remoterel, localindexoid, @@ -3054,7 +3054,7 @@ apply_handle_tuple_routing(ApplyExecutionData *edata, */ EPQState epqstate; - EvalPlanQualInit(&epqstate, estate, NULL, NIL, -1); + EvalPlanQualInit(&epqstate, estate, NULL, NIL, -1, NIL); ExecOpenIndices(partrelinfo, false); EvalPlanQualSetSlot(&epqstate, remoteslot_part); diff --git a/src/include/executor/executor.h b/src/include/executor/executor.h index f9e6bf3d4a..ac02247947 100644 --- a/src/include/executor/executor.h +++ b/src/include/executor/executor.h @@ -233,7 +233,8 @@ extern ExecAuxRowMark *ExecBuildAuxRowMark(ExecRowMark *erm, List *targetlist); extern TupleTableSlot *EvalPlanQual(EPQState *epqstate, Relation relation, Index rti, TupleTableSlot *inputslot); extern void EvalPlanQualInit(EPQState *epqstate, EState *parentestate, - Plan *subplan, List *auxrowmarks, int epqParam); + Plan *subplan, List *auxrowmarks, + int epqParam, List *resultRelations); extern void EvalPlanQualSetPlan(EPQState *epqstate, Plan *subplan, List *auxrowmarks); extern TupleTableSlot *EvalPlanQualSlot(EPQState *epqstate, diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h index 61b3517906..155186ac1f 100644 --- a/src/include/nodes/execnodes.h +++ b/src/include/nodes/execnodes.h @@ -1169,15 +1169,16 @@ typedef struct PlanState */ typedef struct EPQState { - /* Initialized at EvalPlanQualInit() time: */ - + /* These are initialized by EvalPlanQualInit() and do not change later: */ EState *parentestate; /* main query's EState */ int epqParam; /* ID of Param to force scan node re-eval */ + List *resultRelations; /* integer list of RT indexes, or NIL */ /* - * Tuples to be substituted by scan nodes. They need to set up, before - * calling EvalPlanQual()/EvalPlanQualNext(), into the slot returned by - * EvalPlanQualSlot(scanrelid). The array is indexed by scanrelid - 1. + * relsubs_slot[scanrelid - 1] holds the EPQ test tuple to be returned by + * the scan node for the scanrelid'th RT index, in place of performing an + * actual table scan. Callers should use EvalPlanQualSlot() to fetch + * these slots. */ List *tuple_table; /* tuple table for relsubs_slot */ TupleTableSlot **relsubs_slot; @@ -1211,8 +1212,8 @@ typedef struct EPQState ExecAuxRowMark **relsubs_rowmark; /* - * True if a relation's EPQ tuple has been fetched for relation, indexed - * by scanrelid - 1. + * relsubs_done[scanrelid - 1] is true if there is no EPQ tuple for this + * target relation or it has already been fetched in the current EPQ run. */ bool *relsubs_done; diff --git a/src/test/isolation/expected/eval-plan-qual.out b/src/test/isolation/expected/eval-plan-qual.out index 6af262ec5d..f2a50f46a3 100644 --- a/src/test/isolation/expected/eval-plan-qual.out +++ b/src/test/isolation/expected/eval-plan-qual.out @@ -842,6 +842,90 @@ step c1: COMMIT; step writep3b: <... completed> step c2: COMMIT; +starting permutation: writep4a writep4b c1 c2 readp +step writep4a: UPDATE p SET c = 4 WHERE c = 0; +step writep4b: UPDATE p SET b = -4 WHERE c = 0; <waiting ...> +step c1: COMMIT; +step writep4b: <... completed> +step c2: COMMIT; +step readp: SELECT tableoid::regclass, ctid, * FROM p; +tableoid|ctid |a|b|c +--------+------+-+-+- +c1 |(0,2) |0|0|1 +c1 |(0,3) |0|0|2 +c1 |(0,5) |0|1|1 +c1 |(0,6) |0|1|2 +c1 |(0,8) |0|2|1 +c1 |(0,9) |0|2|2 +c1 |(0,11)|0|0|4 +c1 |(0,12)|0|1|4 +c1 |(0,13)|0|2|4 +c1 |(0,14)|0|3|4 +c2 |(0,2) |1|0|1 +c2 |(0,3) |1|0|2 +c2 |(0,5) |1|1|1 +c2 |(0,6) |1|1|2 +c2 |(0,8) |1|2|1 +c2 |(0,9) |1|2|2 +c2 |(0,11)|1|0|4 +c2 |(0,12)|1|1|4 +c2 |(0,13)|1|2|4 +c2 |(0,14)|1|3|4 +c3 |(0,2) |2|0|1 +c3 |(0,3) |2|0|2 +c3 |(0,5) |2|1|1 +c3 |(0,6) |2|1|2 +c3 |(0,8) |2|2|1 +c3 |(0,9) |2|2|2 +c3 |(0,11)|2|0|4 +c3 |(0,12)|2|1|4 +c3 |(0,13)|2|2|4 +c3 |(0,14)|2|3|4 +(30 rows) + + +starting permutation: writep4a deletep4 c1 c2 readp +step writep4a: UPDATE p SET c = 4 WHERE c = 0; +step deletep4: DELETE FROM p WHERE c = 0; <waiting ...> +step c1: COMMIT; +step deletep4: <... completed> +step c2: COMMIT; +step readp: SELECT tableoid::regclass, ctid, * FROM p; +tableoid|ctid |a|b|c +--------+------+-+-+- +c1 |(0,2) |0|0|1 +c1 |(0,3) |0|0|2 +c1 |(0,5) |0|1|1 +c1 |(0,6) |0|1|2 +c1 |(0,8) |0|2|1 +c1 |(0,9) |0|2|2 +c1 |(0,11)|0|0|4 +c1 |(0,12)|0|1|4 +c1 |(0,13)|0|2|4 +c1 |(0,14)|0|3|4 +c2 |(0,2) |1|0|1 +c2 |(0,3) |1|0|2 +c2 |(0,5) |1|1|1 +c2 |(0,6) |1|1|2 +c2 |(0,8) |1|2|1 +c2 |(0,9) |1|2|2 +c2 |(0,11)|1|0|4 +c2 |(0,12)|1|1|4 +c2 |(0,13)|1|2|4 +c2 |(0,14)|1|3|4 +c3 |(0,2) |2|0|1 +c3 |(0,3) |2|0|2 +c3 |(0,5) |2|1|1 +c3 |(0,6) |2|1|2 +c3 |(0,8) |2|2|1 +c3 |(0,9) |2|2|2 +c3 |(0,11)|2|0|4 +c3 |(0,12)|2|1|4 +c3 |(0,13)|2|2|4 +c3 |(0,14)|2|3|4 +(30 rows) + + starting permutation: wx2 partiallock c2 c1 read step wx2: UPDATE accounts SET balance = balance + 450 WHERE accountid = 'checking' RETURNING balance; balance diff --git a/src/test/isolation/specs/eval-plan-qual.spec b/src/test/isolation/specs/eval-plan-qual.spec index 768f7098b9..cb56578208 100644 --- a/src/test/isolation/specs/eval-plan-qual.spec +++ b/src/test/isolation/specs/eval-plan-qual.spec @@ -102,11 +102,15 @@ step upsert1 { # when the first updated tuple was in a non-first child table. # writep2/returningp1 tests a memory allocation issue # writep3a/writep3b tests updates touching more than one table +# writep4a/writep4b tests a case where matches in another table confused EPQ +# writep4a/deletep4 tests the same case in the DELETE path +step readp { SELECT tableoid::regclass, ctid, * FROM p; } step readp1 { SELECT tableoid::regclass, ctid, * FROM p WHERE b IN (0, 1) AND c = 0 FOR UPDATE; } step writep1 { UPDATE p SET b = -1 WHERE a = 1 AND b = 1 AND c = 0; } step writep2 { UPDATE p SET b = -b WHERE a = 1 AND c = 0; } step writep3a { UPDATE p SET b = -b WHERE c = 0; } +step writep4a { UPDATE p SET c = 4 WHERE c = 0; } step c1 { COMMIT; } step r1 { ROLLBACK; } @@ -210,6 +214,8 @@ step returningp1 { SELECT * FROM u; } step writep3b { UPDATE p SET b = -b WHERE c = 0; } +step writep4b { UPDATE p SET b = -4 WHERE c = 0; } +step deletep4 { DELETE FROM p WHERE c = 0; } step readforss { SELECT ta.id AS ta_id, ta.value AS ta_value, (SELECT ROW(tb.id, tb.value) @@ -347,6 +353,8 @@ permutation upsert1 upsert2 c1 c2 read permutation readp1 writep1 readp2 c1 c2 permutation writep2 returningp1 c1 c2 permutation writep3a writep3b c1 c2 +permutation writep4a writep4b c1 c2 readp +permutation writep4a deletep4 c1 c2 readp permutation wx2 partiallock c2 c1 read permutation wx2 lockwithvalues c2 c1 read permutation wx2_ext partiallock_ext c2 c1 read_ext ^ permalink raw reply [nested|flat] 4+ messages in thread
* Re: The documentation for READ COMMITTED may be incomplete or wrong @ 2023-05-18 23:57 Tom Lane <[email protected]> parent: Tom Lane <[email protected]> 0 siblings, 1 reply; 4+ messages in thread From: Tom Lane @ 2023-05-18 23:57 UTC (permalink / raw) To: Nathan Bossart <[email protected]>; +Cc: Aleksander Alekseev <[email protected]>; Amit Langote <[email protected]>; PostgreSQL Hackers <[email protected]> I wrote: > Debian Code Search doesn't know of any outside code touching > relsubs_done, so I think we are safe in dropping that code in > ExecScanReScan. It seems quite pointless anyway considering > that up to now, EvalPlanQualBegin has always zeroed the whole > array. Oh, belay that. What I'd forgotten is that it's possible that the target relation is on the inside of a nestloop, meaning that we might need to fetch the EPQ substitute tuple more than once. So there are three possible states: blocked (never return a tuple), ready to return a tuple, and done returning a tuple for this scan. ExecScanReScan needs to reset "done" to "ready", but not touch the "blocked" state. The attached v2 mechanizes that using two bool arrays. What I'm thinking about doing to back-patch this is to replace one of the pointer fields in EPQState with a pointer to a subsidiary palloc'd structure, where we can put the new fields along with the cannibalized old one. We've done something similar before, and it seems a lot safer than having basically different logic in v16 than earlier branches. regards, tom lane Attachments: [text/x-diff] v2-fix-EPQ-HEAD-only.patch (15.9K, ../../[email protected]/2-v2-fix-EPQ-HEAD-only.patch) download | inline diff: diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c index 0186be452c..c76fdf59ec 100644 --- a/src/backend/executor/execMain.c +++ b/src/backend/executor/execMain.c @@ -2469,7 +2469,7 @@ ExecBuildAuxRowMark(ExecRowMark *erm, List *targetlist) * relation - table containing tuple * rti - rangetable index of table containing tuple * inputslot - tuple for processing - this can be the slot from - * EvalPlanQualSlot(), for the increased efficiency. + * EvalPlanQualSlot() for this rel, for increased efficiency. * * This tests whether the tuple in inputslot still matches the relevant * quals. For that result to be useful, typically the input tuple has to be @@ -2503,6 +2503,14 @@ EvalPlanQual(EPQState *epqstate, Relation relation, if (testslot != inputslot) ExecCopySlot(testslot, inputslot); + /* + * Mark that an EPQ tuple is available for this relation. (If there is + * more than one result relation, the others remain marked as having no + * tuple available.) + */ + epqstate->relsubs_done[rti - 1] = false; + epqstate->relsubs_blocked[rti - 1] = false; + /* * Run the EPQ query. We assume it will return at most one tuple. */ @@ -2519,11 +2527,12 @@ EvalPlanQual(EPQState *epqstate, Relation relation, ExecMaterializeSlot(slot); /* - * Clear out the test tuple. This is needed in case the EPQ query is - * re-used to test a tuple for a different relation. (Not clear that can - * really happen, but let's be safe.) + * Clear out the test tuple, and mark that no tuple is available here. + * This is needed in case the EPQ state is re-used to test a tuple for a + * different target relation. */ ExecClearTuple(testslot); + epqstate->relsubs_blocked[rti - 1] = true; return slot; } @@ -2532,18 +2541,26 @@ EvalPlanQual(EPQState *epqstate, Relation relation, * EvalPlanQualInit -- initialize during creation of a plan state node * that might need to invoke EPQ processing. * + * If the caller intends to use EvalPlanQual(), resultRelations should be + * a list of RT indexes of potential target relations for EvalPlanQual(), + * and we will arrange that the other listed relations don't return any + * tuple during an EvalPlanQual() call. Otherwise resultRelations + * should be NIL. + * * Note: subplan/auxrowmarks can be NULL/NIL if they will be set later * with EvalPlanQualSetPlan. */ void EvalPlanQualInit(EPQState *epqstate, EState *parentestate, - Plan *subplan, List *auxrowmarks, int epqParam) + Plan *subplan, List *auxrowmarks, + int epqParam, List *resultRelations) { Index rtsize = parentestate->es_range_table_size; /* initialize data not changing over EPQState's lifetime */ epqstate->parentestate = parentestate; epqstate->epqParam = epqParam; + epqstate->resultRelations = resultRelations; /* * Allocate space to reference a slot for each potential rti - do so now @@ -2566,6 +2583,7 @@ EvalPlanQualInit(EPQState *epqstate, EState *parentestate, epqstate->recheckplanstate = NULL; epqstate->relsubs_rowmark = NULL; epqstate->relsubs_done = NULL; + epqstate->relsubs_blocked = NULL; } /* @@ -2763,7 +2781,13 @@ EvalPlanQualBegin(EPQState *epqstate) Index rtsize = parentestate->es_range_table_size; PlanState *rcplanstate = epqstate->recheckplanstate; - MemSet(epqstate->relsubs_done, 0, rtsize * sizeof(bool)); + /* + * Reset the relsubs_done[] flags to equal relsubs_blocked[], so that + * the EPQ run will never attempt to fetch tuples from blocked target + * relations. + */ + memcpy(epqstate->relsubs_done, epqstate->relsubs_blocked, + rtsize * sizeof(bool)); /* Recopy current values of parent parameters */ if (parentestate->es_plannedstmt->paramExecTypes != NIL) @@ -2931,10 +2955,22 @@ EvalPlanQualStart(EPQState *epqstate, Plan *planTree) } /* - * Initialize per-relation EPQ tuple states to not-fetched. + * Initialize per-relation EPQ tuple states. Result relations, if any, + * get marked as blocked; others as not-fetched. */ - epqstate->relsubs_done = (bool *) - palloc0(rtsize * sizeof(bool)); + epqstate->relsubs_done = palloc_array(bool, rtsize); + epqstate->relsubs_blocked = palloc0_array(bool, rtsize); + + foreach(l, epqstate->resultRelations) + { + int rtindex = lfirst_int(l); + + Assert(rtindex > 0 && rtindex <= rtsize); + epqstate->relsubs_blocked[rtindex - 1] = true; + } + + memcpy(epqstate->relsubs_done, epqstate->relsubs_blocked, + rtsize * sizeof(bool)); /* * Initialize the private state information for all the nodes in the part @@ -3010,4 +3046,5 @@ EvalPlanQualEnd(EPQState *epqstate) epqstate->recheckplanstate = NULL; epqstate->relsubs_rowmark = NULL; epqstate->relsubs_done = NULL; + epqstate->relsubs_blocked = NULL; } diff --git a/src/backend/executor/execScan.c b/src/backend/executor/execScan.c index cf1871b0f5..a47c8f5f71 100644 --- a/src/backend/executor/execScan.c +++ b/src/backend/executor/execScan.c @@ -69,13 +69,12 @@ ExecScanFetch(ScanState *node, else if (epqstate->relsubs_done[scanrelid - 1]) { /* - * Return empty slot, as we already performed an EPQ substitution - * for this relation. + * Return empty slot, as either there is no EPQ tuple for this rel + * or we already returned it. */ TupleTableSlot *slot = node->ss_ScanTupleSlot; - /* Return empty slot, as we already returned a tuple */ return ExecClearTuple(slot); } else if (epqstate->relsubs_slot[scanrelid - 1] != NULL) @@ -88,7 +87,7 @@ ExecScanFetch(ScanState *node, Assert(epqstate->relsubs_rowmark[scanrelid - 1] == NULL); - /* Mark to remember that we shouldn't return more */ + /* Mark to remember that we shouldn't return it again */ epqstate->relsubs_done[scanrelid - 1] = true; /* Return empty slot if we haven't got a test tuple */ @@ -306,14 +305,18 @@ ExecScanReScan(ScanState *node) */ ExecClearTuple(node->ss_ScanTupleSlot); - /* Rescan EvalPlanQual tuple if we're inside an EvalPlanQual recheck */ + /* + * Rescan EvalPlanQual tuple(s) if we're inside an EvalPlanQual recheck. + * But don't lose the "blocked" status of blocked target relations. + */ if (estate->es_epq_active != NULL) { EPQState *epqstate = estate->es_epq_active; Index scanrelid = ((Scan *) node->ps.plan)->scanrelid; if (scanrelid > 0) - epqstate->relsubs_done[scanrelid - 1] = false; + epqstate->relsubs_done[scanrelid - 1] = + epqstate->relsubs_blocked[scanrelid - 1]; else { Bitmapset *relids; @@ -335,7 +338,8 @@ ExecScanReScan(ScanState *node) while ((rtindex = bms_next_member(relids, rtindex)) >= 0) { Assert(rtindex > 0); - epqstate->relsubs_done[rtindex - 1] = false; + epqstate->relsubs_done[rtindex - 1] = + epqstate->relsubs_blocked[rtindex - 1]; } } } diff --git a/src/backend/executor/nodeLockRows.c b/src/backend/executor/nodeLockRows.c index 407414fc0c..e459971d32 100644 --- a/src/backend/executor/nodeLockRows.c +++ b/src/backend/executor/nodeLockRows.c @@ -108,7 +108,6 @@ lnext: /* this child is inactive right now */ erm->ermActive = false; ItemPointerSetInvalid(&(erm->curCtid)); - ExecClearTuple(markSlot); continue; } } @@ -370,7 +369,7 @@ ExecInitLockRows(LockRows *node, EState *estate, int eflags) /* Now we have the info needed to set up EPQ state */ EvalPlanQualInit(&lrstate->lr_epqstate, estate, - outerPlan, epq_arowmarks, node->epqParam); + outerPlan, epq_arowmarks, node->epqParam, NIL); return lrstate; } diff --git a/src/backend/executor/nodeModifyTable.c b/src/backend/executor/nodeModifyTable.c index dc1a2ec551..7f5002527f 100644 --- a/src/backend/executor/nodeModifyTable.c +++ b/src/backend/executor/nodeModifyTable.c @@ -3985,7 +3985,8 @@ ExecInitModifyTable(ModifyTable *node, EState *estate, int eflags) } /* set up epqstate with dummy subplan data for the moment */ - EvalPlanQualInit(&mtstate->mt_epqstate, estate, NULL, NIL, node->epqParam); + EvalPlanQualInit(&mtstate->mt_epqstate, estate, NULL, NIL, + node->epqParam, node->resultRelations); mtstate->fireBSTriggers = true; /* diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index 879309b316..4b67098814 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -2674,7 +2674,7 @@ apply_handle_update_internal(ApplyExecutionData *edata, bool found; MemoryContext oldctx; - EvalPlanQualInit(&epqstate, estate, NULL, NIL, -1); + EvalPlanQualInit(&epqstate, estate, NULL, NIL, -1, NIL); ExecOpenIndices(relinfo, false); found = FindReplTupleInLocalRel(estate, localrel, @@ -2827,7 +2827,7 @@ apply_handle_delete_internal(ApplyExecutionData *edata, TupleTableSlot *localslot; bool found; - EvalPlanQualInit(&epqstate, estate, NULL, NIL, -1); + EvalPlanQualInit(&epqstate, estate, NULL, NIL, -1, NIL); ExecOpenIndices(relinfo, false); found = FindReplTupleInLocalRel(estate, localrel, remoterel, localindexoid, @@ -3054,7 +3054,7 @@ apply_handle_tuple_routing(ApplyExecutionData *edata, */ EPQState epqstate; - EvalPlanQualInit(&epqstate, estate, NULL, NIL, -1); + EvalPlanQualInit(&epqstate, estate, NULL, NIL, -1, NIL); ExecOpenIndices(partrelinfo, false); EvalPlanQualSetSlot(&epqstate, remoteslot_part); diff --git a/src/include/executor/executor.h b/src/include/executor/executor.h index f9e6bf3d4a..ac02247947 100644 --- a/src/include/executor/executor.h +++ b/src/include/executor/executor.h @@ -233,7 +233,8 @@ extern ExecAuxRowMark *ExecBuildAuxRowMark(ExecRowMark *erm, List *targetlist); extern TupleTableSlot *EvalPlanQual(EPQState *epqstate, Relation relation, Index rti, TupleTableSlot *inputslot); extern void EvalPlanQualInit(EPQState *epqstate, EState *parentestate, - Plan *subplan, List *auxrowmarks, int epqParam); + Plan *subplan, List *auxrowmarks, + int epqParam, List *resultRelations); extern void EvalPlanQualSetPlan(EPQState *epqstate, Plan *subplan, List *auxrowmarks); extern TupleTableSlot *EvalPlanQualSlot(EPQState *epqstate, diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h index 61b3517906..68ddcd6794 100644 --- a/src/include/nodes/execnodes.h +++ b/src/include/nodes/execnodes.h @@ -1169,15 +1169,16 @@ typedef struct PlanState */ typedef struct EPQState { - /* Initialized at EvalPlanQualInit() time: */ - + /* These are initialized by EvalPlanQualInit() and do not change later: */ EState *parentestate; /* main query's EState */ int epqParam; /* ID of Param to force scan node re-eval */ + List *resultRelations; /* integer list of RT indexes, or NIL */ /* - * Tuples to be substituted by scan nodes. They need to set up, before - * calling EvalPlanQual()/EvalPlanQualNext(), into the slot returned by - * EvalPlanQualSlot(scanrelid). The array is indexed by scanrelid - 1. + * relsubs_slot[scanrelid - 1] holds the EPQ test tuple to be returned by + * the scan node for the scanrelid'th RT index, in place of performing an + * actual table scan. Callers should use EvalPlanQualSlot() to fetch + * these slots. */ List *tuple_table; /* tuple table for relsubs_slot */ TupleTableSlot **relsubs_slot; @@ -1211,11 +1212,17 @@ typedef struct EPQState ExecAuxRowMark **relsubs_rowmark; /* - * True if a relation's EPQ tuple has been fetched for relation, indexed - * by scanrelid - 1. + * relsubs_done[scanrelid - 1] is true if there is no EPQ tuple for this + * target relation or it has already been fetched in the current EPQ run. */ bool *relsubs_done; + /* + * relsubs_blocked[scanrelid - 1] is true if there is no EPQ tuple for + * this target relation. + */ + bool *relsubs_blocked; + PlanState *recheckplanstate; /* EPQ specific exec nodes, for ->plan */ } EPQState; diff --git a/src/test/isolation/expected/eval-plan-qual.out b/src/test/isolation/expected/eval-plan-qual.out index 6af262ec5d..f2a50f46a3 100644 --- a/src/test/isolation/expected/eval-plan-qual.out +++ b/src/test/isolation/expected/eval-plan-qual.out @@ -842,6 +842,90 @@ step c1: COMMIT; step writep3b: <... completed> step c2: COMMIT; +starting permutation: writep4a writep4b c1 c2 readp +step writep4a: UPDATE p SET c = 4 WHERE c = 0; +step writep4b: UPDATE p SET b = -4 WHERE c = 0; <waiting ...> +step c1: COMMIT; +step writep4b: <... completed> +step c2: COMMIT; +step readp: SELECT tableoid::regclass, ctid, * FROM p; +tableoid|ctid |a|b|c +--------+------+-+-+- +c1 |(0,2) |0|0|1 +c1 |(0,3) |0|0|2 +c1 |(0,5) |0|1|1 +c1 |(0,6) |0|1|2 +c1 |(0,8) |0|2|1 +c1 |(0,9) |0|2|2 +c1 |(0,11)|0|0|4 +c1 |(0,12)|0|1|4 +c1 |(0,13)|0|2|4 +c1 |(0,14)|0|3|4 +c2 |(0,2) |1|0|1 +c2 |(0,3) |1|0|2 +c2 |(0,5) |1|1|1 +c2 |(0,6) |1|1|2 +c2 |(0,8) |1|2|1 +c2 |(0,9) |1|2|2 +c2 |(0,11)|1|0|4 +c2 |(0,12)|1|1|4 +c2 |(0,13)|1|2|4 +c2 |(0,14)|1|3|4 +c3 |(0,2) |2|0|1 +c3 |(0,3) |2|0|2 +c3 |(0,5) |2|1|1 +c3 |(0,6) |2|1|2 +c3 |(0,8) |2|2|1 +c3 |(0,9) |2|2|2 +c3 |(0,11)|2|0|4 +c3 |(0,12)|2|1|4 +c3 |(0,13)|2|2|4 +c3 |(0,14)|2|3|4 +(30 rows) + + +starting permutation: writep4a deletep4 c1 c2 readp +step writep4a: UPDATE p SET c = 4 WHERE c = 0; +step deletep4: DELETE FROM p WHERE c = 0; <waiting ...> +step c1: COMMIT; +step deletep4: <... completed> +step c2: COMMIT; +step readp: SELECT tableoid::regclass, ctid, * FROM p; +tableoid|ctid |a|b|c +--------+------+-+-+- +c1 |(0,2) |0|0|1 +c1 |(0,3) |0|0|2 +c1 |(0,5) |0|1|1 +c1 |(0,6) |0|1|2 +c1 |(0,8) |0|2|1 +c1 |(0,9) |0|2|2 +c1 |(0,11)|0|0|4 +c1 |(0,12)|0|1|4 +c1 |(0,13)|0|2|4 +c1 |(0,14)|0|3|4 +c2 |(0,2) |1|0|1 +c2 |(0,3) |1|0|2 +c2 |(0,5) |1|1|1 +c2 |(0,6) |1|1|2 +c2 |(0,8) |1|2|1 +c2 |(0,9) |1|2|2 +c2 |(0,11)|1|0|4 +c2 |(0,12)|1|1|4 +c2 |(0,13)|1|2|4 +c2 |(0,14)|1|3|4 +c3 |(0,2) |2|0|1 +c3 |(0,3) |2|0|2 +c3 |(0,5) |2|1|1 +c3 |(0,6) |2|1|2 +c3 |(0,8) |2|2|1 +c3 |(0,9) |2|2|2 +c3 |(0,11)|2|0|4 +c3 |(0,12)|2|1|4 +c3 |(0,13)|2|2|4 +c3 |(0,14)|2|3|4 +(30 rows) + + starting permutation: wx2 partiallock c2 c1 read step wx2: UPDATE accounts SET balance = balance + 450 WHERE accountid = 'checking' RETURNING balance; balance diff --git a/src/test/isolation/specs/eval-plan-qual.spec b/src/test/isolation/specs/eval-plan-qual.spec index 768f7098b9..cb56578208 100644 --- a/src/test/isolation/specs/eval-plan-qual.spec +++ b/src/test/isolation/specs/eval-plan-qual.spec @@ -102,11 +102,15 @@ step upsert1 { # when the first updated tuple was in a non-first child table. # writep2/returningp1 tests a memory allocation issue # writep3a/writep3b tests updates touching more than one table +# writep4a/writep4b tests a case where matches in another table confused EPQ +# writep4a/deletep4 tests the same case in the DELETE path +step readp { SELECT tableoid::regclass, ctid, * FROM p; } step readp1 { SELECT tableoid::regclass, ctid, * FROM p WHERE b IN (0, 1) AND c = 0 FOR UPDATE; } step writep1 { UPDATE p SET b = -1 WHERE a = 1 AND b = 1 AND c = 0; } step writep2 { UPDATE p SET b = -b WHERE a = 1 AND c = 0; } step writep3a { UPDATE p SET b = -b WHERE c = 0; } +step writep4a { UPDATE p SET c = 4 WHERE c = 0; } step c1 { COMMIT; } step r1 { ROLLBACK; } @@ -210,6 +214,8 @@ step returningp1 { SELECT * FROM u; } step writep3b { UPDATE p SET b = -b WHERE c = 0; } +step writep4b { UPDATE p SET b = -4 WHERE c = 0; } +step deletep4 { DELETE FROM p WHERE c = 0; } step readforss { SELECT ta.id AS ta_id, ta.value AS ta_value, (SELECT ROW(tb.id, tb.value) @@ -347,6 +353,8 @@ permutation upsert1 upsert2 c1 c2 read permutation readp1 writep1 readp2 c1 c2 permutation writep2 returningp1 c1 c2 permutation writep3a writep3b c1 c2 +permutation writep4a writep4b c1 c2 readp +permutation writep4a deletep4 c1 c2 readp permutation wx2 partiallock c2 c1 read permutation wx2 lockwithvalues c2 c1 read permutation wx2_ext partiallock_ext c2 c1 read_ext ^ permalink raw reply [nested|flat] 4+ messages in thread
* Re: The documentation for READ COMMITTED may be incomplete or wrong @ 2023-05-19 11:02 Aleksander Alekseev <[email protected]> parent: Tom Lane <[email protected]> 0 siblings, 0 replies; 4+ messages in thread From: Aleksander Alekseev @ 2023-05-19 11:02 UTC (permalink / raw) To: PostgreSQL Hackers <[email protected]>; +Cc: Tom Lane <[email protected]>; Nathan Bossart <[email protected]>; Amit Langote <[email protected]> Hi, > The attached v2 mechanizes that using two bool arrays. I tested the patch on several combinations of operating systems (LInux, MacOS) and architectures (x64, RISC-V) available to me at the moment, with both Meson and Autotools. Also I made sure eval-plan-qual.spec fails when the C code is untouched. The patch passed all the checks I could come up with. -- Best regards, Aleksander Alekseev ^ permalink raw reply [nested|flat] 4+ messages in thread
end of thread, other threads:[~2023-05-19 11:02 UTC | newest] Thread overview: 4+ messages (download: mbox mbox.gz follow: Atom feed) -- links below jump to the message on this page -- 2021-03-15 15:10 [PATCH v10 2/2] fk_arrays_elems Mark Rofail <[email protected]> 2023-05-18 22:26 Re: The documentation for READ COMMITTED may be incomplete or wrong Tom Lane <[email protected]> 2023-05-18 23:57 ` Re: The documentation for READ COMMITTED may be incomplete or wrong Tom Lane <[email protected]> 2023-05-19 11:02 ` Re: The documentation for READ COMMITTED may be incomplete or wrong Aleksander Alekseev <[email protected]>
This inbox is served by agora; see mirroring instructions for how to clone and mirror all data and code used for this inbox