Here's a rebased version of this patch. I have done nothing other than
fix the conflicts and verify that it passes existing regression tests.
I intend to go over the reviews sometime later and hopefully get it all
fixed for inclusion in pg11.
--
Álvaro Herrera https://www.2ndQuadrant.com/
PostgreSQL Development, 24x7 Support, Remote DBA, Training & Services
>From d513b449722b6e848b958584d0c81ca19b289d22 Mon Sep 17 00:00:00 2001
From: Alvaro Herrera <alvhe...@alvh.no-ip.org>
Date: Fri, 12 Jan 2018 16:28:51 -0300
Subject: [PATCH v6] Support foreign keys for array elements
---
doc/src/sgml/catalogs.sgml | 16 +
doc/src/sgml/ddl.sgml | 110 ++++
doc/src/sgml/ref/create_table.sgml | 116 +++-
src/backend/catalog/heap.c | 1 +
src/backend/catalog/index.c | 1 +
src/backend/catalog/pg_constraint.c | 14 +-
src/backend/commands/tablecmds.c | 138 ++++-
src/backend/commands/trigger.c | 6 +
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 | 66 ++-
src/backend/parser/parse_coerce.c | 84 +++
src/backend/parser/parse_utilcmd.c | 2 +
src/backend/utils/adt/ri_triggers.c | 245 +++++++--
src/backend/utils/adt/ruleutils.c | 88 ++-
src/include/catalog/pg_constraint.h | 27 +-
src/include/catalog/pg_constraint_fn.h | 1 +
src/include/nodes/parsenodes.h | 5 +
src/include/parser/kwlist.h | 1 +
src/include/parser/parse_coerce.h | 1 +
src/test/regress/expected/element_foreign_key.out | 639 ++++++++++++++++++++++
src/test/regress/parallel_schedule | 7 +-
src/test/regress/serial_schedule | 1 +
src/test/regress/sql/element_foreign_key.sql | 503 +++++++++++++++++
26 files changed, 2000 insertions(+), 76 deletions(-)
create mode 100644 src/test/regress/expected/element_foreign_key.out
create mode 100644 src/test/regress/sql/element_foreign_key.sql
diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml
index 3f02202caf..aa5f9fe21f 100644
--- a/doc/src/sgml/catalogs.sgml
+++ b/doc/src/sgml/catalogs.sgml
@@ -2342,6 +2342,16 @@ SCRAM-SHA-256$<replaceable><iteration
count></replaceable>:<replaceable>&l
</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><structfield>conpfeqop</structfield></entry>
<entry><type>oid[]</type></entry>
<entry><literal><link
linkend="catalog-pg-operator"><structname>pg_operator</structname></link>.oid</literal></entry>
@@ -2387,6 +2397,12 @@ SCRAM-SHA-256$<replaceable><iteration
count></replaceable>:<replaceable>&l
</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.
For other cases, a zero appears in <structfield>conkey</structfield>
diff --git a/doc/src/sgml/ddl.sgml b/doc/src/sgml/ddl.sgml
index b1167a40e6..5ca685b991 100644
--- a/doc/src/sgml/ddl.sgml
+++ b/doc/src/sgml/ddl.sgml
@@ -882,6 +882,116 @@ CREATE TABLE order_items (
</para>
</sect2>
+ <sect2 id="ddl-constraints-element-fk">
+ <title>Foreign Key Arrays</title>
+
+ <indexterm>
+ <primary>Foreign Key Arrays</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>Foreign Key Arrays</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 races (
+ 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 Foreign Key Array constraint is on
+ a single column key, you can define a Foreign Key Array 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>
+
+ On top of standard foreign key requirements, array
+ <literal>ELEMENT</literal> foreign key constraints require that the
+ referencing column is an array of a compatible type of the corresponding
+ referenced column.
+
+ However, it is very important to note that, currently, we only support
+ one array reference per foreign key.
+ </para>
+
+ <para>
+ For more detailed information on Foreign Key Array 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 a0c9a6d257..39b0012b1a 100644
--- a/doc/src/sgml/ref/create_table.sgml
+++ b/doc/src/sgml/ref/create_table.sgml
@@ -65,7 +65,7 @@ CREATE [ [ GLOBAL | LOCAL ] { TEMPORARY | TEMP } | UNLOGGED ]
TABLE [ IF NOT EXI
GENERATED { ALWAYS | BY DEFAULT } AS IDENTITY [ (
<replaceable>sequence_options</replaceable> ) ] |
UNIQUE <replaceable class="parameter">index_parameters</replaceable> |
PRIMARY KEY <replaceable class="parameter">index_parameters</replaceable> |
- REFERENCES <replaceable class="parameter">reftable</replaceable> [ (
<replaceable class="parameter">refcolumn</replaceable> ) ] [ MATCH FULL | MATCH
PARTIAL | MATCH SIMPLE ]
+ [ EACH ELEMENT OF ] REFERENCES <replaceable
class="parameter">reftable</replaceable> [ ( <replaceable
class="parameter">refcolumn</replaceable> ) ] [ MATCH FULL | MATCH PARTIAL |
MATCH SIMPLE ]
[ ON DELETE <replaceable class="parameter">action</replaceable> ] [ ON
UPDATE <replaceable class="parameter">action</replaceable> ] }
[ DEFERRABLE | NOT DEFERRABLE ] [ INITIALLY DEFERRED | INITIALLY IMMEDIATE ]
@@ -867,10 +867,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">action</replaceable> ] [ ON UPDATE <replaceable
class="parameter">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">action</replaceable> ]
@@ -894,6 +894,19 @@ WITH ( MODULUS <replaceable
class="parameter">numeric_literal</replaceable>, REM
</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>, a Foreign Key
Array 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
given match type. There are three match types: <literal>MATCH
@@ -956,7 +969,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 <literal>ELEMENT</literal> foreign keys.
</para>
</listitem>
</varlistentry>
@@ -965,7 +979,8 @@ WITH ( MODULUS <replaceable
class="parameter">numeric_literal</replaceable>, REM
<term><literal>SET NULL</literal></term>
<listitem>
<para>
- Set the referencing column(s) to null.
+ Set the referencing column(s) to null. Currently not supported
+ with array <literal>ELEMENT</literal> foreign keys.
</para>
</listitem>
</varlistentry>
@@ -977,6 +992,8 @@ 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 <literal>ELEMENT</literal>
+ foreign keys.
</para>
</listitem>
</varlistentry>
@@ -992,6 +1009,85 @@ WITH ( MODULUS <replaceable
class="parameter">numeric_literal</replaceable>, REM
</listitem>
</varlistentry>
+ <varlistentry id="sql-createtable-element-foreign-key-constraints"
xreflabel="FOREIGN KEY ARRAYS">
+ <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 a
+ <quote>Foreign Key Array</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>
+ Foreign Key Arrays are an extension
+ of PostgreSQL and are not included in the SQL standard.
+ </para>
+
+ <para>
+ Even with Foreign Key Arrays, 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>
+
+ <varlistentry>
+ <term><literal>ON DELETE CASCADE</literal></term>
+ <listitem>
+ <para>
+ Same as standard foreign key constraints. Deletes the entire array.
+ </para>
+ </listitem>
+ </varlistentry>
+ </variablelist>
+ </para>
+
+ <para>
+ It is advisable to index the refrencing column using 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>
@@ -1982,6 +2078,16 @@ CREATE TABLE cities_partdef
</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>ELEMENT
+ REFERENCES</literal> clause are a <productname>PostgreSQL</productname>
+ extension.
+ </para>
+ </refsect2>
+
+ <refsect2>
<title><literal>PARTITION BY</literal> Clause</title>
<para>
diff --git a/src/backend/catalog/heap.c b/src/backend/catalog/heap.c
index 089b7965f2..21836ab680 100644
--- a/src/backend/catalog/heap.c
+++ b/src/backend/catalog/heap.c
@@ -2126,6 +2126,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 330488b96f..33b40ffcb1 100644
--- a/src/backend/catalog/index.c
+++ b/src/backend/catalog/index.c
@@ -1215,6 +1215,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 442ae7e23d..37b7a3e275 100644
--- a/src/backend/catalog/pg_constraint.c
+++ b/src/backend/catalog/pg_constraint.c
@@ -59,6 +59,7 @@ CreateConstraintEntry(const char *constraintName,
Oid indexRelId,
Oid foreignRelId,
const int16 *foreignKey,
+ const char *foreignRefType,
const Oid *pfEqOp,
const Oid *ppEqOp,
const Oid *ffEqOp,
@@ -82,6 +83,7 @@ CreateConstraintEntry(const char *constraintName,
Datum values[Natts_pg_constraint];
ArrayType *conkeyArray;
ArrayType *confkeyArray;
+ ArrayType *confreftypeArray;
ArrayType *conpfeqopArray;
ArrayType *conppeqopArray;
ArrayType *conffeqopArray;
@@ -119,7 +121,11 @@ CreateConstraintEntry(const char *constraintName,
for (i = 0; i < foreignNKeys; i++)
fkdatums[i] = Int16GetDatum(foreignKey[i]);
confkeyArray = construct_array(fkdatums, foreignNKeys,
-
INT2OID, 2, true, 's');
+
INT2OID, sizeof(int16), true, 's');
+ for (i = 0; i < foreignNKeys; i++)
+ fkdatums[i] = CharGetDatum(foreignRefType[i]);
+ confreftypeArray = construct_array(fkdatums, foreignNKeys,
+
CHAROID, sizeof(char), true, 'c');
for (i = 0; i < foreignNKeys; i++)
fkdatums[i] = ObjectIdGetDatum(pfEqOp[i]);
conpfeqopArray = construct_array(fkdatums, foreignNKeys,
@@ -136,6 +142,7 @@ CreateConstraintEntry(const char *constraintName,
else
{
confkeyArray = NULL;
+ confreftypeArray = NULL;
conpfeqopArray = NULL;
conppeqopArray = NULL;
conffeqopArray = NULL;
@@ -188,6 +195,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
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index f2a928b823..58f53495e1 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -7058,6 +7058,7 @@ ATAddForeignKeyConstraint(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];
@@ -7065,10 +7066,12 @@ ATAddForeignKeyConstraint(AlteredTableInfo *tab,
Relation rel,
Oid ppeqoperators[INDEX_MAX_KEYS];
Oid ffeqoperators[INDEX_MAX_KEYS];
int i;
+ ListCell *lc;
int numfks,
numpks;
Oid indexOid;
Oid constrOid;
+ bool has_array;
bool old_check_ok;
ObjectAddress address;
ListCell *old_pfeqop_item = list_head(fkconstraint->old_conpfeqop);
@@ -7145,6 +7148,7 @@ ATAddForeignKeyConstraint(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));
@@ -7157,6 +7161,50 @@ ATAddForeignKeyConstraint(AlteredTableInfo *tab,
Relation rel,
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_array = 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_array)
+ ereport(ERROR,
+
(errcode(ERRCODE_INVALID_FOREIGN_KEY),
+ errmsg("foreign keys support only one
array column")));
+ has_array = 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 amd DELETE CASCADE actions */
+ if (has_array)
+ {
+ 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("Foreign Key Arrays 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
* supplied attribute list. In either case, discover the index OID and
@@ -7242,6 +7290,65 @@ ATAddForeignKeyConstraint(AlteredTableInfo *tab,
Relation rel,
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)
+ {
+ Oid elemopclass;
+
+ /* 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]))));
+
+ /*
+ * For the moment, we must also insist that the array's
element
+ * type have a default btree opclass that is in the
index's
+ * opfamily. This is necessary because ri_triggers.c
relies on
+ * COUNT(DISTINCT x) on the element type, as well as on
array_eq()
+ * on the array type, and we need those operations to
have the
+ * same notion of equality that we're using otherwise.
+ *
+ * XXX this restriction is pretty annoying, considering
the effort
+ * that's been put into the rest of the RI mechanisms
to make them
+ * work with nondefault equality operators. In
particular, it
+ * means that the cast-to-PK-datatype code path isn't
useful for
+ * array-to-scalar references.
+ */
+ elemopclass = GetDefaultOpClass(fktype, BTREE_AM_OID);
+ if (!OidIsValid(elemopclass) ||
+ get_opclass_family(elemopclass) != opfamily)
+ {
+ /* Get the index opclass's name for the error
message. */
+ char *opcname;
+
+ cla_ht = SearchSysCache1(CLAOID,
+
ObjectIdGetDatum(opclasses[i]));
+ if (!HeapTupleIsValid(cla_ht))
+ elog(ERROR, "cache lookup failed for
opclass %u",
+ opclasses[i]);
+ cla_tup = (Form_pg_opclass) GETSTRUCT(cla_ht);
+ opcname = pstrdup(NameStr(cla_tup->opcname));
+ ReleaseSysCache(cla_ht);
+ ereport(ERROR,
+
(errcode(ERRCODE_DATATYPE_MISMATCH),
+ errmsg("foreign key constraint \"%s\" cannot be
implemented",
+ fkconstraint->conname),
+ errdetail("Key column \"%s\"
has element type %s which does not have a default btree operator class that's
compatible with class \"%s\".",
+
strVal(list_nth(fkconstraint->fk_attrs, i)),
+
format_type_be(fktype),
+ opcname)));
+ }
+ }
+
+ /*
* There had better be a primary equality operator for the
index.
* We'll use it for PK = PK comparisons.
*/
@@ -7300,17 +7407,17 @@ ATAddForeignKeyConstraint(AlteredTableInfo *tab,
Relation rel,
}
if (!(OidIsValid(pfeqop) && OidIsValid(ffeqop)))
- ereport(ERROR,
- (errcode(ERRCODE_DATATYPE_MISMATCH),
- errmsg("foreign key constraint \"%s\" "
- "cannot be implemented",
- fkconstraint->conname),
- errdetail("Key columns \"%s\" and
\"%s\" "
- "are of incompatible
types: %s and %s.",
-
strVal(list_nth(fkconstraint->fk_attrs, i)),
-
strVal(list_nth(fkconstraint->pk_attrs, i)),
-
format_type_be(fktype),
-
format_type_be(pktype))));
+ ereport(ERROR,
+ (errcode(ERRCODE_DATATYPE_MISMATCH),
+ errmsg("foreign key constraint \"%s\" "
+ "cannot be implemented",
+ fkconstraint->conname),
+ errdetail("Key columns \"%s\" and \"%s\" "
+ "are of incompatible types:
%s and %s.",
+
strVal(list_nth(fkconstraint->fk_attrs, i)),
+
strVal(list_nth(fkconstraint->pk_attrs, i)),
+ format_type_be(fktype),
+ format_type_be(pktype))));
if (old_check_ok)
{
@@ -7340,6 +7447,13 @@ ATAddForeignKeyConstraint(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);
@@ -7382,7 +7496,6 @@ ATAddForeignKeyConstraint(AlteredTableInfo *tab, Relation
rel,
new_castfunc ==
old_castfunc &&
(!IsPolymorphicType(pfeqop_right) ||
new_fktype ==
old_fktype));
-
}
pfeqoperators[i] = pfeqop;
@@ -7406,6 +7519,7 @@ ATAddForeignKeyConstraint(AlteredTableInfo *tab, Relation
rel,
indexOid,
RelationGetRelid(pkrel),
pkattnum,
+
fkreftypes,
pfeqoperators,
ppeqoperators,
ffeqoperators,
diff --git a/src/backend/commands/trigger.c b/src/backend/commands/trigger.c
index 1c488c338a..510d6c91b0 100644
--- a/src/backend/commands/trigger.c
+++ b/src/backend/commands/trigger.c
@@ -661,6 +661,7 @@ CreateTrigger(CreateTrigStmt *stmt, const char *queryString,
NULL,
NULL,
NULL,
+
NULL,
0,
' ',
' ',
@@ -1032,6 +1033,7 @@ ConvertTriggerToFK(CreateTrigStmt *stmt, Oid funcoid)
char fk_matchtype = FKCONSTR_MATCH_SIMPLE;
List *fk_attrs = NIL;
List *pk_attrs = NIL;
+ List *fk_reftypes = NIL;
StringInfoData buf;
int funcnum;
OldTriggerInfo *info = NULL;
@@ -1061,7 +1063,10 @@ ConvertTriggerToFK(CreateTrigStmt *stmt, Oid funcoid)
if (i % 2)
fk_attrs = lappend(fk_attrs, arg);
else
+ {
pk_attrs = lappend(pk_attrs, arg);
+ fk_reftypes = lappend_int(fk_reftypes,
FKCONSTR_REF_PLAIN);
+ }
}
/* Prepare description of constraint for use in messages */
@@ -1200,6 +1205,7 @@ ConvertTriggerToFK(CreateTrigStmt *stmt, Oid funcoid)
fkcon->conname = constr_name;
fkcon->fk_attrs = fk_attrs;
fkcon->pk_attrs = pk_attrs;
+ fkcon->fk_reftypes = fk_reftypes;
fkcon->fk_matchtype = fk_matchtype;
switch (info->funcoids[0])
{
diff --git a/src/backend/commands/typecmds.c b/src/backend/commands/typecmds.c
index a40b3cf752..f10757b122 100644
--- a/src/backend/commands/typecmds.c
+++ b/src/backend/commands/typecmds.c
@@ -3164,6 +3164,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 ddbbc79823..51e971d555 100644
--- a/src/backend/nodes/copyfuncs.c
+++ b/src/backend/nodes/copyfuncs.c
@@ -2850,6 +2850,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 30ccc9c5ae..18d14b59f8 100644
--- a/src/backend/nodes/equalfuncs.c
+++ b/src/backend/nodes/equalfuncs.c
@@ -2594,6 +2594,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 5e72df137e..3a4d9535d6 100644
--- a/src/backend/nodes/outfuncs.c
+++ b/src/backend/nodes/outfuncs.c
@@ -3501,6 +3501,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 e42b7caff6..5ac121d91d 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -126,6 +126,13 @@ typedef struct ImportQual
List *table_names;
} ImportQual;
+/* Private struct for the result of foreign_key_column_elem production */
+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
@@ -183,6 +190,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);
@@ -233,6 +241,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;
@@ -358,6 +367,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
%type <list> function_with_argtypes_list aggregate_with_argtypes_list
operator_with_argtypes_list
%type <ival> defacl_privilege_target
@@ -393,7 +403,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
publication_name_list
@@ -625,8 +635,8 @@ 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
- EXCLUDE EXCLUDING EXCLUSIVE EXECUTE EXISTS EXPLAIN
+ EACH ELEMENT ELSE ENABLE_P ENCODING ENCRYPTED END_P ENUM_P ESCAPE
+ EVENT EXCEPT EXCLUDE EXCLUDING EXCLUSIVE EXECUTE EXISTS EXPLAIN
EXTENSION EXTERNAL EXTRACT
FALSE_P FAMILY FETCH FILTER FIRST_P FLOAT_P FOLLOWING FOR
@@ -3527,8 +3537,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);
@@ -3722,14 +3734,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);
@@ -3762,6 +3775,29 @@ columnElem: ColId
$$ = (Node *) makeString($1);
}
;
+ 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;
+ }
+ ;
key_match: MATCH FULL
{
@@ -14976,6 +15012,7 @@ unreserved_keyword:
| DOUBLE_P
| DROP
| EACH
+ | ELEMENT
| ENABLE_P
| ENCODING
| ENCRYPTED
@@ -16096,6 +16133,23 @@ 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_coerce.c
b/src/backend/parser/parse_coerce.c
index 085aa8766c..7fa070a3c1 100644
--- a/src/backend/parser/parse_coerce.c
+++ b/src/backend/parser/parse_coerce.c
@@ -1386,6 +1386,90 @@ select_common_type(ParseState *pstate, List *exprs,
const char *context,
}
/*
+ * look at select_common_type(); in
parse_coerce.c(src/backend/parser/parse_coerce.c)
+ *
+ * select_common_type_2args()
+ * Determine the common supertype of a list of input expressions.
+ * 'leftOid': left operand's type
+ * 'rightOid': left operand's type
+ */
+Oid select_common_type_2args(Oid leftOid, Oid rightOid)
+{
+ TYPCATEGORY leftcategory;
+ TYPCATEGORY rightcategory;
+ bool leftispreferred;
+ bool rightispreferred;
+
+ Assert(leftOid != InvalidOid);
+ Assert(rightOid != InvalidOid);
+
+ /*
+ * If input types are valid and exactly the same, just pick that type.
+ */
+ if (leftOid != UNKNOWNOID)
+ {
+ if (rightOid == leftOid)
+ return leftOid;
+ }
+
+ /*
+ * Nope, so set up for the full algorithm.
+ */
+ leftOid = getBaseType(leftOid);
+ rightOid = getBaseType(rightOid);
+ get_type_category_preferred(leftOid, &leftcategory, &leftispreferred);
+ get_type_category_preferred(rightOid, &rightcategory,
&rightispreferred);
+
+ if (rightOid != UNKNOWNOID && rightOid != leftOid)
+ {
+ if (rightcategory != leftcategory)
+ {
+ /* both types in different categories? then not much
hope... */
+ return InvalidOid;
+ }
+
+ if (can_coerce_type(1, &rightOid, &leftOid, COERCION_IMPLICIT))
+ {
+ /* can coerce to it implicitly right to left */
+ if (!can_coerce_type(1, &leftOid, &rightOid,
COERCION_IMPLICIT))
+ {
+ /* can not coerce to it implicitly right to
left, thus coercion only works one way */
+ return leftOid;
+ }
+ else
+ {
+ /* coercion works both ways, then
decide depending on preferred flag */
+ if (leftispreferred)
+ return leftOid;
+ else
+ return rightOid;
+ }
+ }
+ else{
+ /* can not coerce to it implicitly right to left */
+ if (can_coerce_type(1, &leftOid, &rightOid,
COERCION_IMPLICIT))
+ {
+ /* can coerce to it implicitly right to left,
thus coercion only works one way */
+ return rightOid;
+ }
+ else{
+ /* can not coerce either way */
+ return InvalidOid;
+ }
+ }
+ }
+
+ /*
+ * If all the inputs were UNKNOWN type --- ie, unknown-type literals ---
+ * then resolve as type TEXT.
+ */
+ if (leftOid == UNKNOWNOID)
+ leftOid = TEXTOID;
+
+ return leftOid;
+}
+
+/*
* coerce_to_common_type()
* Coerce an expression to the given type.
*
diff --git a/src/backend/parser/parse_utilcmd.c
b/src/backend/parser/parse_utilcmd.c
index 128f1679c6..4f41827d4d 100644
--- a/src/backend/parser/parse_utilcmd.c
+++ b/src/backend/parser/parse_utilcmd.c
@@ -757,6 +757,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 8faae1d069..d577df3512 100644
--- a/src/backend/utils/adt/ri_triggers.c
+++ b/src/backend/utils/adt/ri_triggers.c
@@ -115,9 +115,11 @@ typedef struct RI_ConstraintInfo
char confupdtype; /* foreign key's ON UPDATE action */
char confdeltype; /* foreign key's ON DELETE action */
char confmatchtype; /* foreign key's match type */
+ bool has_array; /* true if any reftype is
EACH_ELEMENT */
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) */
@@ -202,7 +204,8 @@ static void ri_GenerateQual(StringInfo buf,
const char *sep,
const char *leftop, Oid leftoptype,
Oid opoid,
- const char *rightop, Oid rightoptype);
+ const char *rightop, Oid rightoptype,
+ char fkreftype);
static void ri_add_cast_to(StringInfo buf, Oid typid);
static void ri_GenerateQualCollation(StringInfo buf, Oid collation);
static int ri_NullCheck(HeapTuple tup,
@@ -393,6 +396,7 @@ RI_FKey_check(TriggerData *trigdata)
if ((qplan = ri_FetchPreparedPlan(&qkey)) == NULL)
{
StringInfoData querybuf;
+ StringInfoData countbuf;
char pkrelname[MAX_QUOTED_REL_NAME_LEN];
char attname[MAX_QUOTED_NAME_LEN];
char paramname[16];
@@ -405,12 +409,27 @@ RI_FKey_check(TriggerData *trigdata)
* FOR KEY SHARE OF x
* The type id's for the $ parameters are those of the
* corresponding FK attributes.
+ *
+ * In case of an array ELEMENT foreign key, the previous query
is used
+ * to count the number of matching rows and see if every
combination
+ * is actually referenced.
+ * The wrapping query is
+ * SELECT 1 WHERE
+ * (SELECT count(DISTINCT y) FROM unnest($1) y)
+ * = (SELECT count(*) FROM (<QUERY>) z)
* ----------
*/
initStringInfo(&querybuf);
quoteRelationName(pkrelname, pk_rel);
appendStringInfo(&querybuf, "SELECT 1 FROM ONLY %s x",
pkrelname);
querysep = "WHERE";
+
+ if (riinfo->has_array)
+ {
+ initStringInfo(&countbuf);
+ appendStringInfo(&countbuf, "SELECT 1 WHERE ");
+ }
+
for (i = 0; i < riinfo->nkeys; i++)
{
Oid pk_type = RIAttType(pk_rel,
riinfo->pk_attnums[i]);
@@ -419,18 +438,41 @@ RI_FKey_check(TriggerData *trigdata)
quoteOneName(attname,
RIAttName(pk_rel,
riinfo->pk_attnums[i]));
sprintf(paramname, "$%d", i + 1);
+
+ /*
+ * In case of an array ELEMENT foreign key, we check
that each
+ * distinct non-null value in the array is present in
the PK
+ * table.
+ */
+ if (riinfo->fk_reftypes[i] == FKCONSTR_REF_EACH_ELEMENT)
+ appendStringInfo(&countbuf,
+ "(SELECT
pg_catalog.count(DISTINCT y) FROM pg_catalog.unnest(%s) y)",
+ paramname);
+
ri_GenerateQual(&querybuf, querysep,
attname, pk_type,
riinfo->pf_eq_oprs[i],
- paramname, fk_type);
+ paramname, fk_type,
+ riinfo->fk_reftypes[i]);
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,
true);
+ if (riinfo->has_array)
+ {
+ appendStringInfo(&countbuf,
+ "
OPERATOR(pg_catalog.=) (SELECT pg_catalog.count(*) FROM (%s) z)",
+ querybuf.data);
+
+ /* Prepare and save the plan for Foreign Key Arrays */
+ qplan = ri_PlanCheck(countbuf.data, riinfo->nkeys,
queryoids,
+ &qkey, fk_rel,
pk_rel, true);
+ }
+ else
+ /* Prepare and save the plan */
+ qplan = ri_PlanCheck(querybuf.data, riinfo->nkeys,
queryoids,
+ &qkey, fk_rel,
pk_rel, true);
}
/*
@@ -557,7 +599,8 @@ ri_Check_Pk_Match(Relation pk_rel, Relation fk_rel,
ri_GenerateQual(&querybuf, querysep,
attname, pk_type,
riinfo->pp_eq_oprs[i],
- paramname, pk_type);
+ paramname, pk_type,
+ FKCONSTR_REF_PLAIN);
querysep = "AND";
queryoids[i] = pk_type;
}
@@ -815,7 +858,8 @@ ri_restrict(TriggerData *trigdata, bool is_no_action)
ri_GenerateQual(&querybuf, querysep,
paramname, pk_type,
riinfo->pf_eq_oprs[i],
-
attname, fk_type);
+
attname, fk_type,
+
riinfo->fk_reftypes[i]);
querysep = "AND";
queryoids[i] = pk_type;
}
@@ -971,7 +1015,8 @@ RI_FKey_cascade_del(PG_FUNCTION_ARGS)
ri_GenerateQual(&querybuf, querysep,
paramname, pk_type,
riinfo->pf_eq_oprs[i],
-
attname, fk_type);
+
attname, fk_type,
+
riinfo->fk_reftypes[i]);
querysep = "AND";
queryoids[i] = pk_type;
}
@@ -1150,7 +1195,8 @@ RI_FKey_cascade_upd(PG_FUNCTION_ARGS)
ri_GenerateQual(&qualbuf, qualsep,
paramname, pk_type,
riinfo->pf_eq_oprs[i],
-
attname, fk_type);
+
attname, fk_type,
+
riinfo->fk_reftypes[i]);
querysep = ",";
qualsep = "AND";
queryoids[i] = pk_type;
@@ -1368,7 +1414,8 @@ ri_setnull(TriggerData *trigdata)
ri_GenerateQual(&qualbuf, qualsep,
paramname, pk_type,
riinfo->pf_eq_oprs[i],
-
attname, fk_type);
+
attname, fk_type,
+
riinfo->fk_reftypes[i]);
querysep = ",";
qualsep = "AND";
queryoids[i] = pk_type;
@@ -1585,7 +1632,8 @@ ri_setdefault(TriggerData *trigdata)
ri_GenerateQual(&qualbuf, qualsep,
paramname, pk_type,
riinfo->pf_eq_oprs[i],
-
attname, fk_type);
+
attname, fk_type,
+
riinfo->fk_reftypes[i]);
querysep = ",";
qualsep = "AND";
queryoids[i] = pk_type;
@@ -1904,6 +1952,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.
*----------
@@ -1915,15 +1971,46 @@ RI_Initial_Check(Trigger *trigger, Relation fk_rel,
Relation pk_rel)
{
quoteOneName(fkattname,
RIAttName(fk_rel,
riinfo->fk_attnums[i]));
- appendStringInfo(&querybuf, "%sfk.%s", sep, fkattname);
+ if (riinfo->has_array)
+ if (riinfo->fk_reftypes[i] == FKCONSTR_REF_EACH_ELEMENT)
+ appendStringInfo(&querybuf, "%sfk.ak%d %s",
sep, i + 1,
+ fkattname);
+ else
+ appendStringInfo(&querybuf, "%sfk.k%d %s", sep,
i + 1,
+ fkattname);
+ else
+ appendStringInfo(&querybuf, "%sfk.%s", sep, fkattname);
sep = ", ";
}
quoteRelationName(pkrelname, pk_rel);
quoteRelationName(fkrelname, fk_rel);
- appendStringInfo(&querybuf,
- " FROM ONLY %s fk LEFT OUTER JOIN ONLY
%s pk ON",
- fkrelname, pkrelname);
+
+ if (riinfo->has_array)
+ {
+ sep = "";
+ appendStringInfo(&querybuf,
+ " FROM (SELECT ");
+ for (i = 0; i < riinfo->nkeys; i++)
+ {
+ quoteOneName(fkattname,
+ RIAttName(fk_rel,
riinfo->fk_attnums[i]));
+ if (riinfo->fk_reftypes[i] == FKCONSTR_REF_EACH_ELEMENT)
+ appendStringInfo(&querybuf,
"%spg_catalog.unnest(%s) k%d, %s ak%d",
+ sep,
fkattname, i + 1, fkattname, i + 1);
+ else
+ appendStringInfo(&querybuf, "%s%s k%d", sep,
fkattname,
+ i + 1);
+ sep = ", ";
+ }
+ appendStringInfo(&querybuf,
+ " FROM ONLY %s) fk LEFT OUTER
JOIN ONLY %s pk ON",
+ fkrelname, pkrelname);
+ }
+ else
+ appendStringInfo(&querybuf,
+ " FROM ONLY %s fk LEFT OUTER
JOIN ONLY %s pk ON",
+ fkrelname, pkrelname);
strcpy(pkattname, "pk.");
strcpy(fkattname, "fk.");
@@ -1937,12 +2024,16 @@ RI_Initial_Check(Trigger *trigger, Relation fk_rel,
Relation pk_rel)
quoteOneName(pkattname + 3,
RIAttName(pk_rel,
riinfo->pk_attnums[i]));
- quoteOneName(fkattname + 3,
- RIAttName(fk_rel,
riinfo->fk_attnums[i]));
+ if (riinfo->has_array)
+ sprintf(fkattname + 3, "k%d", i + 1);
+ else
+ quoteOneName(fkattname + 3,
+ RIAttName(fk_rel,
riinfo->fk_attnums[i]));
ri_GenerateQual(&querybuf, sep,
pkattname, pk_type,
riinfo->pf_eq_oprs[i],
- fkattname, fk_type);
+ fkattname, fk_type,
+ FKCONSTR_REF_PLAIN);
if (pk_coll != fk_coll)
ri_GenerateQualCollation(&querybuf, pk_coll);
sep = "AND";
@@ -1958,7 +2049,10 @@ RI_Initial_Check(Trigger *trigger, Relation fk_rel,
Relation pk_rel)
sep = "";
for (i = 0; i < riinfo->nkeys; i++)
{
- quoteOneName(fkattname, RIAttName(fk_rel,
riinfo->fk_attnums[i]));
+ if (riinfo->has_array)
+ sprintf(fkattname, "k%d", i + 1);
+ else
+ quoteOneName(fkattname, RIAttName(fk_rel,
riinfo->fk_attnums[i]));
appendStringInfo(&querybuf,
"%sfk.%s IS NOT NULL",
sep, fkattname);
@@ -2134,20 +2228,23 @@ quoteRelationName(char *buffer, Relation rel)
/*
* ri_GenerateQual --- generate a WHERE clause equating two variables
*
- * The idea is to append " sep leftop op rightop" to buf. The complexity
- * comes from needing to be sure that the parser will select the desired
- * operator. We always name the operator using OPERATOR(schema.op) syntax
- * (readability isn't a big priority here), so as to avoid search-path
- * uncertainties. We have to emit casts too, if either input isn't already
- * the input type of the operator; else we are at the mercy of the parser's
- * heuristics for ambiguous-operator resolution.
- */
+ * The idea is to append " sep leftop op rightop" to buf, or if fkreftype is
+ * FKCONSTR_REF_EACH_ELEMENT, append " sep rightop @> leftop" to buf.
+ *
+ * The complexity comes from needing to be sure that the parser will select
+ * the desired operator. We always name the operator using
+ * OPERATOR(schema.op) syntax (readability isn't a big priority here), so as
+ * to avoid search-path uncertainties. We have to emit casts too, if either
+ * input isn't already the input type of the operator; else we are at the
+ * mercy of the parser's heuristics for ambiguous-operator resolution.
+ */
static void
ri_GenerateQual(StringInfo buf,
const char *sep,
const char *leftop, Oid leftoptype,
Oid opoid,
- const char *rightop, Oid rightoptype)
+ const char *rightop, Oid rightoptype,
+ char fkreftype)
{
HeapTuple opertup;
Form_pg_operator operform;
@@ -2163,14 +2260,57 @@ ri_GenerateQual(StringInfo buf,
nspname = get_namespace_name(operform->oprnamespace);
- appendStringInfo(buf, " %s %s", sep, leftop);
- if (leftoptype != operform->oprleft)
- ri_add_cast_to(buf, operform->oprleft);
- appendStringInfo(buf, " OPERATOR(%s.", quote_identifier(nspname));
- appendStringInfoString(buf, oprname);
- appendStringInfo(buf, ") %s", rightop);
- if (rightoptype != operform->oprright)
- ri_add_cast_to(buf, operform->oprright);
+ if (fkreftype == FKCONSTR_REF_EACH_ELEMENT)
+ {
+ Oid oprright;
+ Oid oprleft;
+ Oid oprcommon;
+ /*
+ * we first need to get the array types and decide
+ * the more appropriate one
+ */
+
+ /* get array type of refrenced element*/
+ oprright = get_array_type(operform->oprleft);
+ if (!OidIsValid(oprright))
+ ereport(ERROR,
+ (errcode(ERRCODE_UNDEFINED_OBJECT),
+ errmsg("could not find array type for data type %s",
+ format_type_be(operform->oprleft))));
+ /* get array type of refrencing element*/
+ oprleft = get_array_type(operform->oprright);
+ if (!OidIsValid(oprleft))
+ ereport(ERROR,
+ (errcode(ERRCODE_UNDEFINED_OBJECT),
+ errmsg("could not find array type for data type %s",
+ format_type_be(operform->oprright))));
+ /* compare the two array types and try to find
+ a common supertype */
+ oprcommon = select_common_type_2args(oprleft, oprright);
+ if (!OidIsValid(oprcommon))
+ ereport(ERROR,
+ (errcode(ERRCODE_UNDEFINED_OBJECT),
+ errmsg("could not find common type between %s and %s",
+ format_type_be(oprleft), format_type_be(oprright))));
+
+ appendStringInfo(buf, " %s %s", sep, rightop);
+ if (oprleft != oprcommon)
+ ri_add_cast_to(buf, oprcommon);
+ appendStringInfo(buf, " OPERATOR(pg_catalog.@>) ARRAY[%s]",
leftop);
+ if (oprright != oprcommon)
+ ri_add_cast_to(buf, oprcommon);
+ }
+ else
+ {
+ appendStringInfo(buf, " %s %s", sep, leftop);
+ if (leftoptype != operform->oprleft)
+ ri_add_cast_to(buf, operform->oprleft);
+ appendStringInfo(buf, " OPERATOR(%s.",
quote_identifier(nspname));
+ appendStringInfoString(buf, oprname);
+ appendStringInfo(buf, ") %s", rightop);
+ if (rightoptype != operform->oprright)
+ ri_add_cast_to(buf, operform->oprright);
+ }
ReleaseSysCache(opertup);
}
@@ -2378,6 +2518,7 @@ ri_LoadConstraintInfo(Oid constraintOid)
bool isNull;
ArrayType *arr;
int numkeys;
+ int i;
/*
* On the first call initialize the hashtable
@@ -2456,6 +2597,20 @@ ri_LoadConstraintInfo(Oid constraintOid)
pfree(arr); /* free de-toasted
copy, if any */
adatum = SysCacheGetAttr(CONSTROID, tup,
+
Anum_pg_constraint_confreftype, &isNull);
+ if (isNull)
+ elog(ERROR, "null confreftype for constraint %u",
constraintOid);
+ arr = DatumGetArrayTypeP(adatum); /* ensure not toasted */
+ if (ARR_NDIM(arr) != 1 ||
+ ARR_DIMS(arr)[0] != numkeys ||
+ ARR_HASNULL(arr) ||
+ ARR_ELEMTYPE(arr) != CHAROID)
+ elog(ERROR, "confreftype is not a 1-D char array");
+ memcpy(riinfo->fk_reftypes, ARR_DATA_PTR(arr), numkeys * sizeof(char));
+ if ((Pointer) arr != DatumGetPointer(adatum))
+ pfree(arr); /* free de-toasted
copy, if any */
+
+ adatum = SysCacheGetAttr(CONSTROID, tup,
Anum_pg_constraint_conpfeqop, &isNull);
if (isNull)
elog(ERROR, "null conpfeqop for constraint %u", constraintOid);
@@ -2498,6 +2653,24 @@ ri_LoadConstraintInfo(Oid constraintOid)
if ((Pointer) arr != DatumGetPointer(adatum))
pfree(arr); /* free de-toasted
copy, if any */
+ /*
+ * Fix up some stuff for Foreign Key Arrays. We need a has_array flag
+ * indicating whether there's an array foreign key, and we want to set
+ * ff_eq_oprs[i] to array_eq() for array columns, because that's what
+ * makes sense for ri_KeysEqual, and we have no other use for ff_eq_oprs
+ * in this module. (If we did, substituting the array comparator at the
+ * call point in ri_KeysEqual might be more appropriate.)
+ */
+ riinfo->has_array = false;
+ for (i = 0; i < numkeys; i++)
+ {
+ if (riinfo->fk_reftypes[i] != FKCONSTR_REF_PLAIN)
+ {
+ riinfo->has_array = true;
+ riinfo->ff_eq_oprs[i] = ARRAY_EQ_OP;
+ }
+ }
+
ReleaseSysCache(tup);
/*
diff --git a/src/backend/utils/adt/ruleutils.c
b/src/backend/utils/adt/ruleutils.c
index 9cdbb06add..229d5d8c93 100644
--- a/src/backend/utils/adt/ruleutils.c
+++ b/src/backend/utils/adt/ruleutils.c
@@ -314,6 +314,9 @@ static char *pg_get_viewdef_worker(Oid viewoid,
static char *pg_get_triggerdef_worker(Oid trigid, bool pretty);
static void 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,
@@ -1899,7 +1902,8 @@ pg_get_constraintdef_worker(Oid constraintId, bool
fullCommand,
{
case CONSTRAINT_FOREIGN:
{
- Datum val;
+ Datum colindexes;
+ Datum reftypes;
bool isnull;
const char *string;
@@ -1907,13 +1911,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(",
@@ -1921,13 +1933,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, ')');
@@ -2202,6 +2216,66 @@ decompile_column_index_array(Datum column_index_array,
Oid relId,
}
}
+ /*
+ * 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_relid_attribute_name(relId,
DatumGetInt16(keys[j]));
+
+ 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));
+ }
+}
/* ----------
* get_expr - Decompile an expression tree
diff --git a/src/include/catalog/pg_constraint.h
b/src/include/catalog/pg_constraint.h
index 8fca86d71e..96eea9753d 100644
--- a/src/include/catalog/pg_constraint.h
+++ b/src/include/catalog/pg_constraint.h
@@ -104,8 +104,16 @@ CATALOG(pg_constraint,2606)
int16 confkey[1];
/*
+ * If a foreign key, the reference semantics for each column
+ */
+ char confreftype[1];
+
+ /*
* If a foreign key, the OIDs of the PK = FK equality operators for each
* column of the constraint
+ *
+ * Note: for Foreign Key Arrays, all these operators are for the array's
+ * element type.
*/
Oid conpfeqop[1];
@@ -150,7 +158,7 @@ typedef FormData_pg_constraint *Form_pg_constraint;
* compiler constants for pg_constraint
* ----------------
*/
-#define Natts_pg_constraint 24
+#define Natts_pg_constraint 25
#define Anum_pg_constraint_conname 1
#define Anum_pg_constraint_connamespace 2
#define Anum_pg_constraint_contype 3
@@ -169,12 +177,13 @@ typedef FormData_pg_constraint *Form_pg_constraint;
#define Anum_pg_constraint_connoinherit 16
#define Anum_pg_constraint_conkey 17
#define Anum_pg_constraint_confkey 18
-#define Anum_pg_constraint_conpfeqop 19
-#define Anum_pg_constraint_conppeqop 20
-#define Anum_pg_constraint_conffeqop 21
-#define Anum_pg_constraint_conexclop 22
-#define Anum_pg_constraint_conbin 23
-#define Anum_pg_constraint_consrc 24
+#define Anum_pg_constraint_confreftype 19
+#define Anum_pg_constraint_conpfeqop 20
+#define Anum_pg_constraint_conppeqop 21
+#define Anum_pg_constraint_conffeqop 22
+#define Anum_pg_constraint_conexclop 23
+#define Anum_pg_constraint_conbin 24
+#define Anum_pg_constraint_consrc 25
/* ----------------
* initial contents of pg_constraint
@@ -195,7 +204,9 @@ typedef FormData_pg_constraint *Form_pg_constraint;
/*
* Valid values for confupdtype and confdeltype are the FKCONSTR_ACTION_xxx
* constants defined in parsenodes.h. Valid values for confmatchtype are
- * the FKCONSTR_MATCH_xxx constants defined in parsenodes.h.
+ * the FKCONSTR_MATCH_xxx constants defined in parsenodes.h. Valid values
+ * for elements of confreftype[] are the FKCONSTR_REF_xxx constants defined
+ * in parsenodes.h.
*/
#endif /* PG_CONSTRAINT_H */
diff --git a/src/include/catalog/pg_constraint_fn.h
b/src/include/catalog/pg_constraint_fn.h
index 6bb1b09714..e720745546 100644
--- a/src/include/catalog/pg_constraint_fn.h
+++ b/src/include/catalog/pg_constraint_fn.h
@@ -40,6 +40,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,
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index b72178efd1..2b84d8a960 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -2086,6 +2086,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;
@@ -2121,6 +2125,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 26af944e03..af00143580 100644
--- a/src/include/parser/kwlist.h
+++ b/src/include/parser/kwlist.h
@@ -141,6 +141,7 @@ PG_KEYWORD("domain", DOMAIN_P, UNRESERVED_KEYWORD)
PG_KEYWORD("double", DOUBLE_P, UNRESERVED_KEYWORD)
PG_KEYWORD("drop", DROP, UNRESERVED_KEYWORD)
PG_KEYWORD("each", EACH, UNRESERVED_KEYWORD)
+PG_KEYWORD("element", ELEMENT, UNRESERVED_KEYWORD)
PG_KEYWORD("else", ELSE, RESERVED_KEYWORD)
PG_KEYWORD("enable", ENABLE_P, UNRESERVED_KEYWORD)
PG_KEYWORD("encoding", ENCODING, UNRESERVED_KEYWORD)
diff --git a/src/include/parser/parse_coerce.h
b/src/include/parser/parse_coerce.h
index af12c136ef..2955b57ecc 100644
--- a/src/include/parser/parse_coerce.h
+++ b/src/include/parser/parse_coerce.h
@@ -67,6 +67,7 @@ extern int parser_coercion_errposition(ParseState *pstate,
extern Oid select_common_type(ParseState *pstate, List *exprs,
const char *context, Node **which_expr);
+extern Oid select_common_type_2args(Oid ptype, Oid ntype);
extern Node *coerce_to_common_type(ParseState *pstate, Node *node,
Oid targetTypeId,
const char *context);
diff --git a/src/test/regress/expected/element_foreign_key.out
b/src/test/regress/expected/element_foreign_key.out
new file mode 100644
index 0000000000..ed991e2fae
--- /dev/null
+++ b/src/test/regress/expected/element_foreign_key.out
@@ -0,0 +1,639 @@
+-- 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" 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: Foreign Key Arrays 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: Foreign Key Arrays 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: Foreign Key Arrays 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: Foreign Key Arrays 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: Foreign Key Arrays 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: Foreign Key Arrays 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: Foreign Key Arrays 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: Foreign Key Arrays 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: Foreign Key Arrays 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: Foreign Key Arrays 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: Foreign Key Arrays 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: Foreign Key Arrays 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: Foreign Key Arrays 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: Foreign Key Arrays 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: Foreign Key Arrays 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: Foreign Key Arrays 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: Foreign Key Arrays 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: Foreign Key Arrays 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: Foreign Key Arrays 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: Foreign Key Arrays 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: Foreign Key Arrays 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 refrencing 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;
+-- -- Repeat a similar test using INT2 keys coerced from INT4
+CREATE TABLE PKTABLEFORARRAY ( ptest1 int4 PRIMARY KEY, ptest2 text );
+-- Populate the primary table
+INSERT INTO PKTABLEFORARRAY VALUES (1, 'Test 1');
+INSERT INTO PKTABLEFORARRAY VALUES (2, 'Test 2');
+INSERT INTO PKTABLEFORARRAY VALUES (3, 'Test 3');
+-- Create the refrencing table
+CREATE TABLE FKTABLEFORARRAY ( ftest1 int2[], FOREIGN KEY (EACH ELEMENT OF
ftest1) REFERENCES PKTABLEFORARRAY, ftest2 int );
+-- Insert valid rows into FK TABLE
+INSERT INTO FKTABLEFORARRAY VALUES (ARRAY[1], 1);
+INSERT INTO FKTABLEFORARRAY VALUES (ARRAY[2], 2);
+INSERT INTO FKTABLEFORARRAY VALUES (ARRAY[3], 3);
+INSERT INTO FKTABLEFORARRAY VALUES (ARRAY[1,2,3], 4);
+-- Insert invalid rows into FK TABLE
+INSERT INTO FKTABLEFORARRAY VALUES (ARRAY[4], 5);
+ERROR: insert or update on table "fktableforarray" violates foreign key
constraint "fktableforarray_ftest1_fkey"
+DETAIL: Key (ftest1)=({4}) is not present in table "pktableforarray".
+INSERT INTO FKTABLEFORARRAY VALUES (ARRAY[1,2,5], 6);
+ERROR: insert or update on table "fktableforarray" violates foreign key
constraint "fktableforarray_ftest1_fkey"
+DETAIL: Key (ftest1)=({1,2,5}) is not present in table "pktableforarray".
+-- Cleanup
+DROP TABLE FKTABLEFORARRAY;
+DROP TABLE PKTABLEFORARRAY;
+-- Repeat a similar test using INT4 keys coerced from INT2
+CREATE TABLE PKTABLEFORARRAY ( ptest1 int2 PRIMARY KEY, ptest2 text );
+-- Populate the primary table
+INSERT INTO PKTABLEFORARRAY VALUES (1, 'Test 1');
+INSERT INTO PKTABLEFORARRAY VALUES (2, 'Test 2');
+INSERT INTO PKTABLEFORARRAY VALUES (3, 'Test 3');
+-- Create the refrencing table
+CREATE TABLE FKTABLEFORARRAY ( ftest1 int4[], FOREIGN KEY (EACH ELEMENT OF
ftest1) REFERENCES PKTABLEFORARRAY, ftest2 int );
+-- Insert valid rows into FK TABLE
+INSERT INTO FKTABLEFORARRAY VALUES (ARRAY[1], 1);
+INSERT INTO FKTABLEFORARRAY VALUES (ARRAY[2], 2);
+INSERT INTO FKTABLEFORARRAY VALUES (ARRAY[3], 3);
+INSERT INTO FKTABLEFORARRAY VALUES (ARRAY[1,2,3], 4);
+-- Insert invalid rows into FK TABLE
+INSERT INTO FKTABLEFORARRAY VALUES (ARRAY[4], 5);
+ERROR: insert or update on table "fktableforarray" violates foreign key
constraint "fktableforarray_ftest1_fkey"
+DETAIL: Key (ftest1)=({4}) is not present in table "pktableforarray".
+INSERT INTO FKTABLEFORARRAY VALUES (ARRAY[1,2,5], 6);
+ERROR: insert or update on table "fktableforarray" violates foreign key
constraint "fktableforarray_ftest1_fkey"
+DETAIL: Key (ftest1)=({1,2,5}) is not present in table "pktableforarray".
+-- Cleanup
+DROP TABLE FKTABLEFORARRAY;
+DROP TABLE PKTABLEFORARRAY;
+-- Repeat a similar test using FLOAT8 keys coerced from INTEGER
+CREATE TABLE PKTABLEFORARRAY ( ptest1 float8 PRIMARY KEY, ptest2 text );
+-- XXX this really ought to work, but currently we must disallow it
+CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF
ftest1) REFERENCES PKTABLEFORARRAY, ftest2 int );
+ERROR: foreign key constraint "fktableforarray_ftest1_fkey" cannot be
implemented
+DETAIL: Key column "ftest1" has element type integer which does not have a
default btree operator class that's compatible with class "float8_ops".
+-- Cleanup
+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 refrencing 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_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_fkey"
+DETAIL: Key (fid1, fid2)=(B, {A}) is not present in table "pktableforarray".
+-- Cleanup
+DROP TABLE FKTABLEFORARRAY;
+DROP TABLE PKTABLEFORARRAY;
+-- Test Foreign Key Arrays 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 refrencing 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 refrencing 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_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_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_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_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_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');
+-- 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);
+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 using the indexable operator
+SELECT * FROM FKTABLEFORARRAYGIN WHERE ftest1 @> ARRAY[5];
+ ftest1 | ftest2
+-------------+--------
+ {5} | 1
+ {3,5,2,5} | 3
+ {3,5,4,1,3} | 5
+ {5,1} | 7
+ {3,4,5,3} | 10
+(5 rows)
+
+-- Cleanup
+DROP TABLE FKTABLEFORARRAYGIN;
+DROP TABLE PKTABLEFORARRAYGIN;
diff --git a/src/test/regress/parallel_schedule
b/src/test/regress/parallel_schedule
index e224977791..52505e7539 100644
--- a/src/test/regress/parallel_schedule
+++ b/src/test/regress/parallel_schedule
@@ -104,7 +104,12 @@ test: publication subscription
# ----------
# Another group of parallel tests
# ----------
-test: select_views portals_p2 foreign_key cluster dependency guc bitmapops
combocid tsearch tsdicts foreign_data window xmlmap functional_deps
advisory_lock json jsonb json_encoding indirect_toast equivclass
+test: select_views portals_p2 foreign_key cluster dependency guc bitmapops
combocid tsearch tsdicts foreign_data window xmlmap functional_deps
advisory_lock json jsonb json_encoding indirect_toast equivclass
+
+# ----------
+# Another group of parallel tests (no room in the previouse group)
+# ----------
+test: element_foreign_key
# ----------
# Another group of parallel tests
diff --git a/src/test/regress/serial_schedule b/src/test/regress/serial_schedule
index 9fc5f1a268..eefd1df73b 100644
--- a/src/test/regress/serial_schedule
+++ b/src/test/regress/serial_schedule
@@ -143,6 +143,7 @@ test: amutils
test: select_views
test: portals_p2
test: foreign_key
+test: element_foreign_key
test: cluster
test: dependency
test: guc
diff --git a/src/test/regress/sql/element_foreign_key.sql
b/src/test/regress/sql/element_foreign_key.sql
new file mode 100644
index 0000000000..d012589430
--- /dev/null
+++ b/src/test/regress/sql/element_foreign_key.sql
@@ -0,0 +1,503 @@
+-- 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 refrencing 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;
+
+-- -- Repeat a similar test using INT2 keys coerced from INT4
+CREATE TABLE PKTABLEFORARRAY ( ptest1 int4 PRIMARY KEY, ptest2 text );
+
+-- Populate the primary table
+INSERT INTO PKTABLEFORARRAY VALUES (1, 'Test 1');
+INSERT INTO PKTABLEFORARRAY VALUES (2, 'Test 2');
+INSERT INTO PKTABLEFORARRAY VALUES (3, 'Test 3');
+
+-- Create the refrencing table
+CREATE TABLE FKTABLEFORARRAY ( ftest1 int2[], FOREIGN KEY (EACH ELEMENT OF
ftest1) REFERENCES PKTABLEFORARRAY, ftest2 int );
+
+-- Insert valid rows into FK TABLE
+INSERT INTO FKTABLEFORARRAY VALUES (ARRAY[1], 1);
+INSERT INTO FKTABLEFORARRAY VALUES (ARRAY[2], 2);
+INSERT INTO FKTABLEFORARRAY VALUES (ARRAY[3], 3);
+INSERT INTO FKTABLEFORARRAY VALUES (ARRAY[1,2,3], 4);
+
+-- Insert invalid rows into FK TABLE
+INSERT INTO FKTABLEFORARRAY VALUES (ARRAY[4], 5);
+INSERT INTO FKTABLEFORARRAY VALUES (ARRAY[1,2,5], 6);
+
+-- Cleanup
+DROP TABLE FKTABLEFORARRAY;
+DROP TABLE PKTABLEFORARRAY;
+
+-- Repeat a similar test using INT4 keys coerced from INT2
+CREATE TABLE PKTABLEFORARRAY ( ptest1 int2 PRIMARY KEY, ptest2 text );
+
+-- Populate the primary table
+INSERT INTO PKTABLEFORARRAY VALUES (1, 'Test 1');
+INSERT INTO PKTABLEFORARRAY VALUES (2, 'Test 2');
+INSERT INTO PKTABLEFORARRAY VALUES (3, 'Test 3');
+
+-- Create the refrencing table
+CREATE TABLE FKTABLEFORARRAY ( ftest1 int4[], FOREIGN KEY (EACH ELEMENT OF
ftest1) REFERENCES PKTABLEFORARRAY, ftest2 int );
+
+-- Insert valid rows into FK TABLE
+INSERT INTO FKTABLEFORARRAY VALUES (ARRAY[1], 1);
+INSERT INTO FKTABLEFORARRAY VALUES (ARRAY[2], 2);
+INSERT INTO FKTABLEFORARRAY VALUES (ARRAY[3], 3);
+INSERT INTO FKTABLEFORARRAY VALUES (ARRAY[1,2,3], 4);
+
+-- Insert invalid rows into FK TABLE
+INSERT INTO FKTABLEFORARRAY VALUES (ARRAY[4], 5);
+INSERT INTO FKTABLEFORARRAY VALUES (ARRAY[1,2,5], 6);
+
+-- Cleanup
+DROP TABLE FKTABLEFORARRAY;
+DROP TABLE PKTABLEFORARRAY;
+
+-- Repeat a similar test using FLOAT8 keys coerced from INTEGER
+CREATE TABLE PKTABLEFORARRAY ( ptest1 float8 PRIMARY KEY, ptest2 text );
+-- XXX this really ought to work, but currently we must disallow it
+CREATE TABLE FKTABLEFORARRAY ( ftest1 int[], FOREIGN KEY (EACH ELEMENT OF
ftest1) REFERENCES PKTABLEFORARRAY, ftest2 int );
+
+-- Cleanup
+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 refrencing 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 Foreign Key Arrays 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 refrencing 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 refrencing 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');
+
+-- 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);
+
+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 using the indexable operator
+SELECT * FROM FKTABLEFORARRAYGIN WHERE ftest1 @> ARRAY[5];
+
+-- Cleanup
+DROP TABLE FKTABLEFORARRAYGIN;
+DROP TABLE PKTABLEFORARRAYGIN;
\ No newline at end of file
--
2.11.0
