pg.ddx.io pgsql-hackers@postgresql.org mailing list archive help / color / mirror / Atom feed From: Anastasia Lubennikova <a.lubennikova@postgrespro.ru>
To: Thomas Munro <thomas.munro@enterprisedb.com>
Cc: Andrey Borodin <x4mmm@yandex-team.ru>
Cc: PostgreSQL Hackers <pgsql-hackers@lists.postgresql.org>
Cc: Anastasia Lubennikova <lubennikovaav@gmail.com>
Subject: Re: WIP: Covering + unique indexes.
Date: Thu, 8 Mar 2018 19:13:50 +0300
Message-ID: <78ca1586-647c-e159-ebf3-30957e270c66@postgrespro.ru> (raw )
In-Reply-To: <CAEepm=07Lry2eH+ORvnGn=zTrSQCR0oVgz_0JNTQKUXdzaAszg@mail.gmail.com >
References: <56168952.4010101@postgrespro.ru >
<7FEB0867-F4F2-4F18-8AA9-7BAA0BB32868@yandex-team.ru >
<151654594496.23710.401435331419974842.pgcf@coridan.postgresql.org >
<1d2b6e12-9e1c-8df5-fdd6-4d20d3791b86@postgrespro.ru >
<CAEepm=2gh3nKi2u18i4r=TLhMyZf6guWSwmrHEcwm7dyJahnow@mail.gmail.com >
<b91d08de-db32-490d-6287-5c47b506767c@postgrespro.ru >
<CAEepm=07Lry2eH+ORvnGn=zTrSQCR0oVgz_0JNTQKUXdzaAszg@mail.gmail.com >
06.03.2018 11:52, Thomas Munro:
> On Wed, Jan 31, 2018 at 3:09 AM, Anastasia Lubennikova
> <a.lubennikova@postgrespro.ru> wrote:
>> Thank you for reviewing. All mentioned issues are fixed.
> == Applying patch 0002-covering-btree_v4.patch...
> 1 out of 1 hunk FAILED -- saving rejects to file
> src/backend/access/nbtree/README.rej
> 1 out of 1 hunk FAILED -- saving rejects to file
> src/backend/access/nbtree/nbtxlog.c.rej
>
> Can we please have a new patch set?
Here it is.
Many thanks to Andrey Borodin.
--
Anastasia Lubennikova
Postgres Professional: http://www.postgrespro.com
The Russian Postgres Company
Attachments:
[text/x-patch] 0001-Covering-core-v7.patch (111.1K, ../78ca1586-647c-e159-ebf3-30957e270c66@postgrespro.ru/2-0001-Covering-core-v7.patch)
download | inline diff:
From a098fd4d4e1a2edc656d9fa10f573700ce87db5f Mon Sep 17 00:00:00 2001
From: Andrey Borodin <amborodin@acm.org>
Date: Thu, 8 Mar 2018 15:50:33 +0500
Subject: [PATCH 1/3] Covering core v7
---
contrib/bloom/blutils.c | 1 +
contrib/dblink/dblink.c | 26 +--
contrib/tcn/tcn.c | 6 +-
doc/src/sgml/catalogs.sgml | 8 +
doc/src/sgml/indexam.sgml | 5 +-
doc/src/sgml/indices.sgml | 7 +-
doc/src/sgml/ref/create_index.sgml | 39 +++-
doc/src/sgml/ref/create_table.sgml | 33 +++-
src/backend/access/brin/brin.c | 1 +
src/backend/access/common/indextuple.c | 30 +++
src/backend/access/gin/ginutil.c | 1 +
src/backend/access/gist/gist.c | 1 +
src/backend/access/hash/hash.c | 1 +
src/backend/access/index/genam.c | 18 +-
src/backend/access/nbtree/nbtree.c | 1 +
src/backend/access/spgist/spgutils.c | 1 +
src/backend/bootstrap/bootparse.y | 2 +
src/backend/bootstrap/bootstrap.c | 2 +-
src/backend/catalog/heap.c | 3 +-
src/backend/catalog/index.c | 69 ++++---
src/backend/catalog/indexing.c | 1 +
src/backend/catalog/pg_constraint.c | 26 ++-
src/backend/catalog/toasting.c | 1 +
src/backend/commands/indexcmds.c | 56 +++++-
src/backend/commands/matview.c | 6 +-
src/backend/commands/tablecmds.c | 9 +-
src/backend/commands/trigger.c | 1 +
src/backend/commands/typecmds.c | 1 +
src/backend/executor/execIndexing.c | 14 +-
src/backend/executor/nodeIndexscan.c | 8 +-
src/backend/nodes/copyfuncs.c | 2 +
src/backend/nodes/equalfuncs.c | 2 +
src/backend/nodes/outfuncs.c | 4 +
src/backend/optimizer/path/indxpath.c | 2 +-
src/backend/optimizer/path/pathkeys.c | 7 +
src/backend/optimizer/util/plancat.c | 32 ++--
src/backend/parser/analyze.c | 6 +-
src/backend/parser/gram.y | 65 ++++---
src/backend/parser/parse_relation.c | 2 +-
src/backend/parser/parse_target.c | 3 +-
src/backend/parser/parse_utilcmd.c | 339 +++++++++++++++++++++++----------
src/backend/utils/adt/ruleutils.c | 31 +++
src/backend/utils/adt/selfuncs.c | 4 +-
src/backend/utils/cache/relcache.c | 89 +++++----
src/backend/utils/sort/tuplesort.c | 5 +-
src/bin/pg_dump/pg_dump.c | 40 +++-
src/bin/pg_dump/pg_dump.h | 6 +-
src/include/access/amapi.h | 2 +
src/include/access/itup.h | 2 +
src/include/catalog/pg_constraint.h | 23 ++-
src/include/catalog/pg_constraint_fn.h | 21 +-
src/include/catalog/pg_index.h | 38 ++--
src/include/nodes/execnodes.h | 9 +-
src/include/nodes/parsenodes.h | 5 +-
src/include/nodes/relation.h | 13 +-
src/include/parser/kwlist.h | 1 +
src/include/utils/rel.h | 16 +-
57 files changed, 824 insertions(+), 323 deletions(-)
diff --git a/contrib/bloom/blutils.c b/contrib/bloom/blutils.c
index bbe7183207..6b2b9e3742 100644
--- a/contrib/bloom/blutils.c
+++ b/contrib/bloom/blutils.c
@@ -120,6 +120,7 @@ blhandler(PG_FUNCTION_ARGS)
amroutine->amclusterable = false;
amroutine->ampredlocks = false;
amroutine->amcanparallel = false;
+ amroutine->amcaninclude = false;
amroutine->amkeytype = InvalidOid;
amroutine->ambuild = blbuild;
diff --git a/contrib/dblink/dblink.c b/contrib/dblink/dblink.c
index ae7e24ad08..48ef3cdde1 100644
--- a/contrib/dblink/dblink.c
+++ b/contrib/dblink/dblink.c
@@ -100,7 +100,7 @@ static remoteConn *getConnectionByName(const char *name);
static HTAB *createConnHash(void);
static void createNewConnection(const char *name, remoteConn *rconn);
static void deleteConnection(const char *name);
-static char **get_pkey_attnames(Relation rel, int16 *numatts);
+static char **get_pkey_attnames(Relation rel, int16 *indnkeyatts);
static char **get_text_array_contents(ArrayType *array, int *numitems);
static char *get_sql_insert(Relation rel, int *pkattnums, int pknumatts, char **src_pkattvals, char **tgt_pkattvals);
static char *get_sql_delete(Relation rel, int *pkattnums, int pknumatts, char **tgt_pkattvals);
@@ -1491,7 +1491,7 @@ PG_FUNCTION_INFO_V1(dblink_get_pkey);
Datum
dblink_get_pkey(PG_FUNCTION_ARGS)
{
- int16 numatts;
+ int16 indnkeyatts;
char **results;
FuncCallContext *funcctx;
int32 call_cntr;
@@ -1517,7 +1517,7 @@ dblink_get_pkey(PG_FUNCTION_ARGS)
rel = get_rel_from_relname(PG_GETARG_TEXT_PP(0), AccessShareLock, ACL_SELECT);
/* get the array of attnums */
- results = get_pkey_attnames(rel, &numatts);
+ results = get_pkey_attnames(rel, &indnkeyatts);
relation_close(rel, AccessShareLock);
@@ -1537,9 +1537,9 @@ dblink_get_pkey(PG_FUNCTION_ARGS)
attinmeta = TupleDescGetAttInMetadata(tupdesc);
funcctx->attinmeta = attinmeta;
- if ((results != NULL) && (numatts > 0))
+ if ((results != NULL) && (indnkeyatts > 0))
{
- funcctx->max_calls = numatts;
+ funcctx->max_calls = indnkeyatts;
/* got results, keep track of them */
funcctx->user_fctx = results;
@@ -2027,10 +2027,10 @@ dblink_fdw_validator(PG_FUNCTION_ARGS)
* get_pkey_attnames
*
* Get the primary key attnames for the given relation.
- * Return NULL, and set numatts = 0, if no primary key exists.
+ * Return NULL, and set indnkeyatts = 0, if no primary key exists.
*/
static char **
-get_pkey_attnames(Relation rel, int16 *numatts)
+get_pkey_attnames(Relation rel, int16 *indnkeyatts)
{
Relation indexRelation;
ScanKeyData skey;
@@ -2040,8 +2040,8 @@ get_pkey_attnames(Relation rel, int16 *numatts)
char **result = NULL;
TupleDesc tupdesc;
- /* initialize numatts to 0 in case no primary key exists */
- *numatts = 0;
+ /* initialize indnkeyatts to 0 in case no primary key exists */
+ *indnkeyatts = 0;
tupdesc = rel->rd_att;
@@ -2062,12 +2062,12 @@ get_pkey_attnames(Relation rel, int16 *numatts)
/* we're only interested if it is the primary key */
if (index->indisprimary)
{
- *numatts = index->indnatts;
- if (*numatts > 0)
+ *indnkeyatts = index->indnkeyatts;
+ if (*indnkeyatts > 0)
{
- result = (char **) palloc(*numatts * sizeof(char *));
+ result = (char **) palloc(*indnkeyatts * sizeof(char *));
- for (i = 0; i < *numatts; i++)
+ for (i = 0; i < *indnkeyatts; i++)
result[i] = SPI_fname(tupdesc, index->indkey.values[i]);
}
break;
diff --git a/contrib/tcn/tcn.c b/contrib/tcn/tcn.c
index 41186fdd8f..43bdd92749 100644
--- a/contrib/tcn/tcn.c
+++ b/contrib/tcn/tcn.c
@@ -138,9 +138,9 @@ triggered_change_notification(PG_FUNCTION_ARGS)
/* we're only interested if it is the primary key and valid */
if (index->indisprimary && IndexIsValid(index))
{
- int numatts = index->indnatts;
+ int indnkeyatts = index->indnkeyatts;
- if (numatts > 0)
+ if (indnkeyatts > 0)
{
int i;
@@ -150,7 +150,7 @@ triggered_change_notification(PG_FUNCTION_ARGS)
appendStringInfoCharMacro(payload, ',');
appendStringInfoCharMacro(payload, operation);
- for (i = 0; i < numatts; i++)
+ for (i = 0; i < indnkeyatts; i++)
{
int colno = index->indkey.values[i];
Form_pg_attribute attr = TupleDescAttr(tupdesc, colno - 1);
diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml
index a0e6d7062b..b7f00eff96 100644
--- a/doc/src/sgml/catalogs.sgml
+++ b/doc/src/sgml/catalogs.sgml
@@ -3709,6 +3709,14 @@ SCRAM-SHA-256$<replaceable><iteration count></replaceable>:<replaceable>&l
<literal>pg_class.relnatts</literal>)</entry>
</row>
+ <row>
+ <entry><structfield>indnkeyatts</structfield></entry>
+ <entry><type>int2</type></entry>
+ <entry></entry>
+ <entry>The number of key columns in the index. "Key columns" are ordinary
+ index columns (as opposed to "included" columns).</entry>
+ </row>
+
<row>
<entry><structfield>indisunique</structfield></entry>
<entry><type>bool</type></entry>
diff --git a/doc/src/sgml/indexam.sgml b/doc/src/sgml/indexam.sgml
index a7f6c8dc6a..7f6cb81e52 100644
--- a/doc/src/sgml/indexam.sgml
+++ b/doc/src/sgml/indexam.sgml
@@ -114,6 +114,8 @@ typedef struct IndexAmRoutine
bool amcanparallel;
/* type of data stored in index, or InvalidOid if variable */
Oid amkeytype;
+ /* does AM support columns included with clause INCLUDE? */
+ bool amcaninclude;
/* interface functions */
ambuild_function ambuild;
@@ -985,7 +987,8 @@ amparallelrescan (IndexScanDesc scan);
using <firstterm>unique indexes</firstterm>, which are indexes that disallow
multiple entries with identical keys. An access method that supports this
feature sets <structfield>amcanunique</structfield> true.
- (At present, only b-tree supports it.)
+ (At present, only b-tree supports it.) Columns which are present in the
+ <literal>INCLUDE</literal> clause are not used to enforce uniqueness.
</para>
<para>
diff --git a/doc/src/sgml/indices.sgml b/doc/src/sgml/indices.sgml
index 0818196e76..4d82b8abba 100644
--- a/doc/src/sgml/indices.sgml
+++ b/doc/src/sgml/indices.sgml
@@ -638,7 +638,8 @@ CREATE INDEX test3_desc_index ON test3 (id DESC NULLS LAST);
Indexes can also be used to enforce uniqueness of a column's value,
or the uniqueness of the combined values of more than one column.
<synopsis>
-CREATE UNIQUE INDEX <replaceable>name</replaceable> ON <replaceable>table</replaceable> (<replaceable>column</replaceable> <optional>, ...</optional>);
+CREATE UNIQUE INDEX <replaceable>name</replaceable> ON <replaceable>table</replaceable> (<replaceable>column</replaceable> <optional>, ...</optional>)
+[ INCLUDE (<replaceable>column</replaceable> <optional>, ...</optional>) ];
</synopsis>
Currently, only B-tree indexes can be declared unique.
</para>
@@ -647,7 +648,9 @@ CREATE UNIQUE INDEX <replaceable>name</replaceable> ON <replaceable>table</repla
When an index is declared unique, multiple table rows with equal
indexed values are not allowed. Null values are not considered
equal. A multicolumn unique index will only reject cases where all
- indexed columns are equal in multiple rows.
+ indexed columns are equal in multiple rows. Columns included with clause
+ <literal>INCLUDE</literal> aren't used to enforce constraints (UNIQUE,
+ PRIMARY KEY, etc).
</para>
<para>
diff --git a/doc/src/sgml/ref/create_index.sgml b/doc/src/sgml/ref/create_index.sgml
index 1fd21e12bd..61f4455a3d 100644
--- a/doc/src/sgml/ref/create_index.sgml
+++ b/doc/src/sgml/ref/create_index.sgml
@@ -23,6 +23,7 @@ PostgreSQL documentation
<synopsis>
CREATE [ UNIQUE ] INDEX [ CONCURRENTLY ] [ [ IF NOT EXISTS ] <replaceable class="parameter">name</replaceable> ] ON [ ONLY ] <replaceable class="parameter">table_name</replaceable> [ USING <replaceable class="parameter">method</replaceable> ]
( { <replaceable class="parameter">column_name</replaceable> | ( <replaceable class="parameter">expression</replaceable> ) } [ COLLATE <replaceable class="parameter">collation</replaceable> ] [ <replaceable class="parameter">opclass</replaceable> ] [ ASC | DESC ] [ NULLS { FIRST | LAST } ] [, ...] )
+ <optional> INCLUDE ( <replaceable class="parameter">column_name</replaceable> [, ...] ) </optional>
[ WITH ( <replaceable class="parameter">storage_parameter</replaceable> = <replaceable class="parameter">value</replaceable> [, ... ] ) ]
[ TABLESPACE <replaceable class="parameter">tablespace_name</replaceable> ]
[ WHERE <replaceable class="parameter">predicate</replaceable> ]
@@ -143,6 +144,33 @@ CREATE [ UNIQUE ] INDEX [ CONCURRENTLY ] [ [ IF NOT EXISTS ] <replaceable class=
</listitem>
</varlistentry>
+ <varlistentry>
+ <term><literal>INCLUDE</literal></term>
+ <listitem>
+ <para>
+ An optional <literal>INCLUDE</literal> clause allows a list of columns to be
+ specified which will be included in the non-key portion of the index.
+ Columns which are part of this clause cannot also exist in the
+ key columns portion of the index, and vice versa. The
+ <literal>INCLUDE</literal> columns exist solely to allow more queries to benefit
+ from <firstterm>index-only scans</firstterm> by including certain columns in the
+ index, the value of which would otherwise have to be obtained by reading
+ the table's heap. Having these columns in the <literal>INCLUDE</literal> clause
+ in some cases allows <productname>PostgreSQL</productname> to skip the heap read
+ completely. This also allows <literal>UNIQUE</literal> indexes to be defined on
+ one set of columns, which can include another set of columns in the
+ <literal>INCLUDE</literal> clause, on which the uniqueness is not enforced.
+ It's the same with other constraints (PRIMARY KEY and EXCLUDE). This can
+ also can be used for non-unique indexes as any columns which are not required
+ for the searching or ordering of records can be used in the
+ <literal>INCLUDE</literal> clause, which can slightly reduce the size of the index.
+ Currently, only the B-tree access method supports this feature.
+ Expressions as included columns are not supported since they cannot be used
+ in index-only scans.
+ </para>
+ </listitem>
+ </varlistentry>
+
<varlistentry>
<term><replaceable class="parameter">name</replaceable></term>
<listitem>
@@ -681,13 +709,22 @@ Indexes:
<title>Examples</title>
<para>
- To create a B-tree index on the column <literal>title</literal> in
+ To create a unique B-tree index on the column <literal>title</literal> in
the table <literal>films</literal>:
<programlisting>
CREATE UNIQUE INDEX title_idx ON films (title);
</programlisting>
</para>
+ <para>
+ To create a unique B-tree index on the column <literal>title</literal>
+ and included columns <literal>director</literal> and <literal>rating</literal>
+ in the table <literal>films</literal>:
+<programlisting>
+CREATE UNIQUE INDEX title_idx ON films (title) INCLUDE (director, rating);
+</programlisting>
+ </para>
+
<para>
To create an index on the expression <literal>lower(title)</literal>,
allowing efficient case-insensitive searches:
diff --git a/doc/src/sgml/ref/create_table.sgml b/doc/src/sgml/ref/create_table.sgml
index 9e8e9d8f1c..1f5bd93696 100644
--- a/doc/src/sgml/ref/create_table.sgml
+++ b/doc/src/sgml/ref/create_table.sgml
@@ -73,8 +73,8 @@ CREATE [ [ GLOBAL | LOCAL ] { TEMPORARY | TEMP } | UNLOGGED ] TABLE [ IF NOT EXI
[ CONSTRAINT <replaceable class="parameter">constraint_name</replaceable> ]
{ CHECK ( <replaceable class="parameter">expression</replaceable> ) [ NO INHERIT ] |
- UNIQUE ( <replaceable class="parameter">column_name</replaceable> [, ... ] ) <replaceable class="parameter">index_parameters</replaceable> |
- PRIMARY KEY ( <replaceable class="parameter">column_name</replaceable> [, ... ] ) <replaceable class="parameter">index_parameters</replaceable> |
+ UNIQUE ( <replaceable class="parameter">column_name</replaceable> [, ... ] ) <replaceable class="parameter">index_parameters</replaceable> <optional> INCLUDE (<replaceable class="parameter">column_name</replaceable> [, ...]) </optional> |
+ PRIMARY KEY ( <replaceable class="parameter">column_name</replaceable> [, ... ] ) <replaceable class="parameter">index_parameters</replaceable> <optional> INCLUDE (<replaceable class="parameter">column_name</replaceable> [, ...]) </optional> |
EXCLUDE [ USING <replaceable class="parameter">index_method</replaceable> ] ( <replaceable class="parameter">exclude_element</replaceable> WITH <replaceable class="parameter">operator</replaceable> [, ... ] ) <replaceable class="parameter">index_parameters</replaceable> [ WHERE ( <replaceable class="parameter">predicate</replaceable> ) ] |
FOREIGN KEY ( <replaceable class="parameter">column_name</replaceable> [, ... ] ) 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> ] }
@@ -767,7 +767,8 @@ WITH ( MODULUS <replaceable class="parameter">numeric_literal</replaceable>, REM
<varlistentry>
<term><literal>UNIQUE</literal> (column constraint)</term>
- <term><literal>UNIQUE ( <replaceable class="parameter">column_name</replaceable> [, ... ] )</literal> (table constraint)</term>
+ <term><literal>UNIQUE ( <replaceable class="parameter">column_name</replaceable> [, ... ] )</literal>
+ <optional> INCLUDE ( <replaceable class="parameter">column_name</replaceable> [, ...]) </optional> (table constraint)</term>
<listitem>
<para>
@@ -797,12 +798,25 @@ WITH ( MODULUS <replaceable class="parameter">numeric_literal</replaceable>, REM
key are considered at each level below the <literal>UNIQUE</literal>
constraint.
</para>
+
+ <para>
+ Adding a unique constraint will automatically create a unique btree
+ index on the column or group of columns used in the constraint.
+ The optional clause <literal>INCLUDE</literal> adds to that index
+ one or more columns on which the uniqueness is not enforced.
+ Note that although the constraint is not enforced on the included columns, it still
+ depends on them. Consequently, some operations on these columns (e.g. <literal>DROP COLUMN</literal>)
+ can cause cascade constraint and index deletion.
+ See paragraph about <literal>INCLUDE</literal> in
+ <xref linkend="sql-createindex"/> for more information.
+ </para>
</listitem>
</varlistentry>
<varlistentry>
<term><literal>PRIMARY KEY</literal> (column constraint)</term>
- <term><literal>PRIMARY KEY ( <replaceable class="parameter">column_name</replaceable> [, ... ] )</literal> (table constraint)</term>
+ <term><literal>PRIMARY KEY ( <replaceable class="parameter">column_name</replaceable> [, ... ] )</literal>
+ <optional> INCLUDE ( <replaceable class="parameter">column_name</replaceable> [, ...]) </optional> (table constraint)</term>
<listitem>
<para>
The <literal>PRIMARY KEY</literal> constraint specifies that a column or
@@ -832,6 +846,17 @@ WITH ( MODULUS <replaceable class="parameter">numeric_literal</replaceable>, REM
tables.
</para>
+ <para>
+ Adding a <literal>PRIMARY KEY</literal> constraint will automatically create a unique btree
+ index on the column or group of columns used in the constraint.
+ An optional <literal>INCLUDE</literal> clause allows a list of columns
+ to be specified which will be included in the non-key portion of the index.
+ Although uniqueness is not enforced on the included columns, the constraint
+ still depends on them. Consequently, some operations on the included columns
+ (e.g. <literal>DROP COLUMN</literal>) can cause cascade constraint and index deletion.
+ See paragraph about <literal>INCLUDE</literal> in
+ <xref linkend="sql-createindex"/> for more information.
+ </para>
</listitem>
</varlistentry>
diff --git a/src/backend/access/brin/brin.c b/src/backend/access/brin/brin.c
index 68b3371665..4487a03d38 100644
--- a/src/backend/access/brin/brin.c
+++ b/src/backend/access/brin/brin.c
@@ -97,6 +97,7 @@ brinhandler(PG_FUNCTION_ARGS)
amroutine->amclusterable = false;
amroutine->ampredlocks = false;
amroutine->amcanparallel = false;
+ amroutine->amcaninclude = false;
amroutine->amkeytype = InvalidOid;
amroutine->ambuild = brinbuild;
diff --git a/src/backend/access/common/indextuple.c b/src/backend/access/common/indextuple.c
index f7103e53bc..662e189335 100644
--- a/src/backend/access/common/indextuple.c
+++ b/src/backend/access/common/indextuple.c
@@ -19,6 +19,7 @@
#include "access/heapam.h"
#include "access/itup.h"
#include "access/tuptoaster.h"
+#include "utils/rel.h"
/* ----------------------------------------------------------------
@@ -445,3 +446,32 @@ CopyIndexTuple(IndexTuple source)
memcpy(result, source, size);
return result;
}
+
+/*
+ * Reform index tuple. Truncate nonkey (INCLUDE) attributes.
+ */
+IndexTuple
+index_truncate_tuple(Relation idxrel, IndexTuple olditup)
+{
+ TupleDesc itupdesc = CreateTupleDescCopyConstr(RelationGetDescr(idxrel));
+ Datum values[INDEX_MAX_KEYS];
+ bool isnull[INDEX_MAX_KEYS];
+ IndexTuple newitup;
+ int indnatts = IndexRelationGetNumberOfAttributes(idxrel);
+ int indnkeyatts = IndexRelationGetNumberOfKeyAttributes(idxrel);
+
+ Assert(indnatts <= INDEX_MAX_KEYS);
+ Assert(indnkeyatts > 0);
+ Assert(indnkeyatts < indnatts);
+
+ index_deform_tuple(olditup, itupdesc, values, isnull);
+
+ /* form new tuple that will contain only key attributes */
+ itupdesc->natts = indnkeyatts;
+ newitup = index_form_tuple(itupdesc, values, isnull);
+ newitup->t_tid = olditup->t_tid;
+
+ FreeTupleDesc(itupdesc);
+ Assert(IndexTupleSize(newitup) <= IndexTupleSize(olditup));
+ return newitup;
+}
diff --git a/src/backend/access/gin/ginutil.c b/src/backend/access/gin/ginutil.c
index 7bac7a1252..347cd55903 100644
--- a/src/backend/access/gin/ginutil.c
+++ b/src/backend/access/gin/ginutil.c
@@ -51,6 +51,7 @@ ginhandler(PG_FUNCTION_ARGS)
amroutine->amclusterable = false;
amroutine->ampredlocks = false;
amroutine->amcanparallel = false;
+ amroutine->amcaninclude = false;
amroutine->amkeytype = InvalidOid;
amroutine->ambuild = ginbuild;
diff --git a/src/backend/access/gist/gist.c b/src/backend/access/gist/gist.c
index 51c32e4afe..21185b3659 100644
--- a/src/backend/access/gist/gist.c
+++ b/src/backend/access/gist/gist.c
@@ -72,6 +72,7 @@ gisthandler(PG_FUNCTION_ARGS)
amroutine->amclusterable = true;
amroutine->ampredlocks = false;
amroutine->amcanparallel = false;
+ amroutine->amcaninclude = false;
amroutine->amkeytype = InvalidOid;
amroutine->ambuild = gistbuild;
diff --git a/src/backend/access/hash/hash.c b/src/backend/access/hash/hash.c
index e337439ada..20dac57248 100644
--- a/src/backend/access/hash/hash.c
+++ b/src/backend/access/hash/hash.c
@@ -70,6 +70,7 @@ hashhandler(PG_FUNCTION_ARGS)
amroutine->amclusterable = false;
amroutine->ampredlocks = false;
amroutine->amcanparallel = false;
+ amroutine->amcaninclude = false;
amroutine->amkeytype = INT4OID;
amroutine->ambuild = hashbuild;
diff --git a/src/backend/access/index/genam.c b/src/backend/access/index/genam.c
index 214825114e..9e9d412973 100644
--- a/src/backend/access/index/genam.c
+++ b/src/backend/access/index/genam.c
@@ -158,7 +158,8 @@ IndexScanEnd(IndexScanDesc scan)
*
* Construct a string describing the contents of an index entry, in the
* form "(key_name, ...)=(key_value, ...)". This is currently used
- * for building unique-constraint and exclusion-constraint error messages.
+ * for building unique-constraint and exclusion-constraint error messages,
+ * so only key columns of the index are checked and printed.
*
* Note that if the user does not have permissions to view all of the
* columns involved then a NULL is returned. Returning a partial key seems
@@ -180,13 +181,14 @@ BuildIndexValueDescription(Relation indexRelation,
StringInfoData buf;
Form_pg_index idxrec;
HeapTuple ht_idx;
- int natts = indexRelation->rd_rel->relnatts;
+ int indnkeyatts;
int i;
int keyno;
Oid indexrelid = RelationGetRelid(indexRelation);
Oid indrelid;
AclResult aclresult;
+ indnkeyatts = IndexRelationGetNumberOfKeyAttributes(indexRelation);
/*
* Check permissions- if the user does not have access to view all of the
* key columns then return NULL to avoid leaking data.
@@ -224,7 +226,7 @@ BuildIndexValueDescription(Relation indexRelation,
* No table-level access, so step through the columns in the index and
* make sure the user has SELECT rights on all of them.
*/
- for (keyno = 0; keyno < idxrec->indnatts; keyno++)
+ for (keyno = 0; keyno < idxrec->indnkeyatts; keyno++)
{
AttrNumber attnum = idxrec->indkey.values[keyno];
@@ -250,7 +252,7 @@ BuildIndexValueDescription(Relation indexRelation,
appendStringInfo(&buf, "(%s)=(",
pg_get_indexdef_columns(indexrelid, true));
- for (i = 0; i < natts; i++)
+ for (i = 0; i < indnkeyatts; i++)
{
char *val;
@@ -368,7 +370,7 @@ systable_beginscan(Relation heapRelation,
{
int j;
- for (j = 0; j < irel->rd_index->indnatts; j++)
+ for (j = 0; j < IndexRelationGetNumberOfAttributes(irel); j++)
{
if (key[i].sk_attno == irel->rd_index->indkey.values[j])
{
@@ -376,7 +378,7 @@ systable_beginscan(Relation heapRelation,
break;
}
}
- if (j == irel->rd_index->indnatts)
+ if (j == IndexRelationGetNumberOfAttributes(irel))
elog(ERROR, "column is not in index");
}
@@ -570,7 +572,7 @@ systable_beginscan_ordered(Relation heapRelation,
{
int j;
- for (j = 0; j < indexRelation->rd_index->indnatts; j++)
+ for (j = 0; j < IndexRelationGetNumberOfAttributes(indexRelation); j++)
{
if (key[i].sk_attno == indexRelation->rd_index->indkey.values[j])
{
@@ -578,7 +580,7 @@ systable_beginscan_ordered(Relation heapRelation,
break;
}
}
- if (j == indexRelation->rd_index->indnatts)
+ if (j == IndexRelationGetNumberOfAttributes(indexRelation))
elog(ERROR, "column is not in index");
}
diff --git a/src/backend/access/nbtree/nbtree.c b/src/backend/access/nbtree/nbtree.c
index 8158508d8c..675e6aa80d 100644
--- a/src/backend/access/nbtree/nbtree.c
+++ b/src/backend/access/nbtree/nbtree.c
@@ -117,6 +117,7 @@ bthandler(PG_FUNCTION_ARGS)
amroutine->amclusterable = true;
amroutine->ampredlocks = true;
amroutine->amcanparallel = true;
+ amroutine->amcaninclude = false;
amroutine->amkeytype = InvalidOid;
amroutine->ambuild = btbuild;
diff --git a/src/backend/access/spgist/spgutils.c b/src/backend/access/spgist/spgutils.c
index c4278b0160..4a9b5da268 100644
--- a/src/backend/access/spgist/spgutils.c
+++ b/src/backend/access/spgist/spgutils.c
@@ -50,6 +50,7 @@ spghandler(PG_FUNCTION_ARGS)
amroutine->amclusterable = false;
amroutine->ampredlocks = false;
amroutine->amcanparallel = false;
+ amroutine->amcaninclude = false;
amroutine->amkeytype = InvalidOid;
amroutine->ambuild = spgbuild;
diff --git a/src/backend/bootstrap/bootparse.y b/src/backend/bootstrap/bootparse.y
index 9e81f9514d..df9d473ae7 100644
--- a/src/backend/bootstrap/bootparse.y
+++ b/src/backend/bootstrap/bootparse.y
@@ -299,6 +299,7 @@ Boot_DeclareIndexStmt:
stmt->accessMethod = $8;
stmt->tableSpace = NULL;
stmt->indexParams = $10;
+ stmt->indexIncludingParams = NIL;
stmt->options = NIL;
stmt->whereClause = NULL;
stmt->excludeOpNames = NIL;
@@ -345,6 +346,7 @@ Boot_DeclareUniqueIndexStmt:
stmt->accessMethod = $9;
stmt->tableSpace = NULL;
stmt->indexParams = $11;
+ stmt->indexIncludingParams = NIL;
stmt->options = NIL;
stmt->whereClause = NULL;
stmt->excludeOpNames = NIL;
diff --git a/src/backend/bootstrap/bootstrap.c b/src/backend/bootstrap/bootstrap.c
index 28ff2f0979..f927fc8cc2 100644
--- a/src/backend/bootstrap/bootstrap.c
+++ b/src/backend/bootstrap/bootstrap.c
@@ -616,7 +616,7 @@ boot_openrel(char *relname)
relname, (int) ATTRIBUTE_FIXED_PART_SIZE);
boot_reldesc = heap_openrv(makeRangeVar(NULL, relname, -1), NoLock);
- numattr = boot_reldesc->rd_rel->relnatts;
+ numattr = RelationGetNumberOfAttributes(boot_reldesc);
for (i = 0; i < numattr; i++)
{
if (attrtypes[i] == NULL)
diff --git a/src/backend/catalog/heap.c b/src/backend/catalog/heap.c
index cf36ce4add..12240f63ed 100644
--- a/src/backend/catalog/heap.c
+++ b/src/backend/catalog/heap.c
@@ -2119,7 +2119,8 @@ StoreRelCheck(Relation rel, const char *ccname, Node *expr,
is_validated,
RelationGetRelid(rel), /* relation */
attNos, /* attrs in the constraint */
- keycount, /* # attrs in the constraint */
+ keycount, /* # key attrs in the constraint */
+ keycount, /* # total attrs in the constraint */
InvalidOid, /* not a domain constraint */
InvalidOid, /* no associated index */
InvalidOid, /* Foreign key fields */
diff --git a/src/backend/catalog/index.c b/src/backend/catalog/index.c
index 431bc31969..39d1cd9497 100644
--- a/src/backend/catalog/index.c
+++ b/src/backend/catalog/index.c
@@ -237,7 +237,7 @@ index_check_primary_key(Relation heapRel,
* null, otherwise attempt to ALTER TABLE .. SET NOT NULL
*/
cmds = NIL;
- for (i = 0; i < indexInfo->ii_NumIndexAttrs; i++)
+ for (i = 0; i < indexInfo->ii_NumIndexKeyAttrs; i++)
{
AttrNumber attnum = indexInfo->ii_KeyAttrNumbers[i];
HeapTuple atttuple;
@@ -445,17 +445,26 @@ ConstructTupleDescriptor(Relation heapRelation,
/*
* Check the opclass and index AM to see if either provides a keytype
- * (overriding the attribute type). Opclass takes precedence.
+ * (overriding the attribute type). Opclass (if exists) takes precedence.
*/
- tuple = SearchSysCache1(CLAOID, ObjectIdGetDatum(classObjectId[i]));
- if (!HeapTupleIsValid(tuple))
- elog(ERROR, "cache lookup failed for opclass %u",
- classObjectId[i]);
- opclassTup = (Form_pg_opclass) GETSTRUCT(tuple);
- if (OidIsValid(opclassTup->opckeytype))
- keyType = opclassTup->opckeytype;
- else
- keyType = amroutine->amkeytype;
+ keyType = amroutine->amkeytype;
+
+ /*
+ * Code below is concerned to the opclasses which are not used
+ * with the included columns.
+ */
+ if (i < indexInfo->ii_NumIndexKeyAttrs)
+ {
+ tuple = SearchSysCache1(CLAOID, ObjectIdGetDatum(classObjectId[i]));
+ if (!HeapTupleIsValid(tuple))
+ elog(ERROR, "cache lookup failed for opclass %u",
+ classObjectId[i]);
+ opclassTup = (Form_pg_opclass) GETSTRUCT(tuple);
+ if (OidIsValid(opclassTup->opckeytype))
+ keyType = opclassTup->opckeytype;
+
+ ReleaseSysCache(tuple);
+ }
/*
* If keytype is specified as ANYELEMENT, and opcintype is ANYARRAY,
@@ -470,8 +479,6 @@ ConstructTupleDescriptor(Relation heapRelation,
to->atttypid);
}
- ReleaseSysCache(tuple);
-
/*
* If a key type different from the heap value is specified, update
* the type-related fields in the index tupdesc.
@@ -600,7 +607,7 @@ UpdateIndexRelation(Oid indexoid,
for (i = 0; i < indexInfo->ii_NumIndexAttrs; i++)
indkey->values[i] = indexInfo->ii_KeyAttrNumbers[i];
indcollation = buildoidvector(collationOids, indexInfo->ii_NumIndexAttrs);
- indclass = buildoidvector(classOids, indexInfo->ii_NumIndexAttrs);
+ indclass = buildoidvector(classOids, indexInfo->ii_NumIndexKeyAttrs);
indoption = buildint2vector(coloptions, indexInfo->ii_NumIndexAttrs);
/*
@@ -645,6 +652,7 @@ UpdateIndexRelation(Oid indexoid,
values[Anum_pg_index_indexrelid - 1] = ObjectIdGetDatum(indexoid);
values[Anum_pg_index_indrelid - 1] = ObjectIdGetDatum(heapoid);
values[Anum_pg_index_indnatts - 1] = Int16GetDatum(indexInfo->ii_NumIndexAttrs);
+ values[Anum_pg_index_indnkeyatts - 1] = Int16GetDatum(indexInfo->ii_NumIndexKeyAttrs);
values[Anum_pg_index_indisunique - 1] = BoolGetDatum(indexInfo->ii_Unique);
values[Anum_pg_index_indisprimary - 1] = BoolGetDatum(primary);
values[Anum_pg_index_indisexclusion - 1] = BoolGetDatum(isexclusion);
@@ -1084,7 +1092,7 @@ index_create(Relation heapRelation,
}
/* Store dependency on operator classes */
- for (i = 0; i < indexInfo->ii_NumIndexAttrs; i++)
+ for (i = 0; i < indexInfo->ii_NumIndexKeyAttrs; i++)
{
referenced.classId = OperatorClassRelationId;
referenced.objectId = classObjectId[i];
@@ -1140,6 +1148,8 @@ index_create(Relation heapRelation,
else
Assert(indexRelation->rd_indexcxt != NULL);
+ indexRelation->rd_index->indnkeyatts = indexInfo->ii_NumIndexKeyAttrs;
+
/*
* If this is bootstrap (initdb) time, then we don't actually fill in the
* index yet. We'll be creating more indexes and classes later, so we
@@ -1285,6 +1295,7 @@ index_constraint_create(Relation heapRelation,
true,
RelationGetRelid(heapRelation),
indexInfo->ii_KeyAttrNumbers,
+ indexInfo->ii_NumIndexKeyAttrs,
indexInfo->ii_NumIndexAttrs,
InvalidOid, /* no domain */
indexRelationId, /* index OID */
@@ -1743,15 +1754,19 @@ BuildIndexInfo(Relation index)
IndexInfo *ii = makeNode(IndexInfo);
Form_pg_index indexStruct = index->rd_index;
int i;
- int numKeys;
+ int numAtts;
/* check the number of keys, and copy attr numbers into the IndexInfo */
- numKeys = indexStruct->indnatts;
- if (numKeys < 1 || numKeys > INDEX_MAX_KEYS)
+ numAtts = indexStruct->indnatts;
+ if (numAtts < 1 || numAtts > INDEX_MAX_KEYS)
elog(ERROR, "invalid indnatts %d for index %u",
- numKeys, RelationGetRelid(index));
- ii->ii_NumIndexAttrs = numKeys;
- for (i = 0; i < numKeys; i++)
+ numAtts, RelationGetRelid(index));
+ ii->ii_NumIndexAttrs = numAtts;
+ ii->ii_NumIndexKeyAttrs = indexStruct->indnkeyatts;
+ Assert(ii->ii_NumIndexKeyAttrs != 0);
+ Assert(ii->ii_NumIndexKeyAttrs <= ii->ii_NumIndexAttrs);
+
+ for (i = 0; i < numAtts; i++)
ii->ii_KeyAttrNumbers[i] = indexStruct->indkey.values[i];
/* fetch any expressions needed for expressional indexes */
@@ -1922,9 +1937,11 @@ CompareIndexInfo(IndexInfo *info1, IndexInfo *info2,
void
BuildSpeculativeIndexInfo(Relation index, IndexInfo *ii)
{
- int ncols = index->rd_rel->relnatts;
+ int indnkeyatts;
int i;
+ indnkeyatts = IndexRelationGetNumberOfKeyAttributes(index);
+
/*
* fetch info for checking unique indexes
*/
@@ -1933,16 +1950,16 @@ BuildSpeculativeIndexInfo(Relation index, IndexInfo *ii)
if (index->rd_rel->relam != BTREE_AM_OID)
elog(ERROR, "unexpected non-btree speculative unique index");
- ii->ii_UniqueOps = (Oid *) palloc(sizeof(Oid) * ncols);
- ii->ii_UniqueProcs = (Oid *) palloc(sizeof(Oid) * ncols);
- ii->ii_UniqueStrats = (uint16 *) palloc(sizeof(uint16) * ncols);
+ ii->ii_UniqueOps = (Oid *) palloc(sizeof(Oid) * indnkeyatts);
+ ii->ii_UniqueProcs = (Oid *) palloc(sizeof(Oid) * indnkeyatts);
+ ii->ii_UniqueStrats = (uint16 *) palloc(sizeof(uint16) * indnkeyatts);
/*
* We have to look up the operator's strategy number. This provides a
* cross-check that the operator does match the index.
*/
/* We need the func OIDs and strategy numbers too */
- for (i = 0; i < ncols; i++)
+ for (i = 0; i < indnkeyatts; i++)
{
ii->ii_UniqueStrats[i] = BTEqualStrategyNumber;
ii->ii_UniqueOps[i] =
diff --git a/src/backend/catalog/indexing.c b/src/backend/catalog/indexing.c
index a84b7da114..5a361683da 100644
--- a/src/backend/catalog/indexing.c
+++ b/src/backend/catalog/indexing.c
@@ -119,6 +119,7 @@ CatalogIndexInsert(CatalogIndexState indstate, HeapTuple heapTuple)
Assert(indexInfo->ii_Predicate == NIL);
Assert(indexInfo->ii_ExclusionOps == NULL);
Assert(relationDescs[i]->rd_index->indimmediate);
+ Assert(indexInfo->ii_NumIndexKeyAttrs != 0);
/*
* FormIndexDatum fills in its values and isnull parameters with the
diff --git a/src/backend/catalog/pg_constraint.c b/src/backend/catalog/pg_constraint.c
index 731c5e4317..fe846a0aa8 100644
--- a/src/backend/catalog/pg_constraint.c
+++ b/src/backend/catalog/pg_constraint.c
@@ -55,6 +55,7 @@ CreateConstraintEntry(const char *constraintName,
Oid relId,
const int16 *constraintKey,
int constraintNKeys,
+ int constraintNTotalKeys,
Oid domainId,
Oid indexRelId,
Oid foreignRelId,
@@ -81,6 +82,7 @@ CreateConstraintEntry(const char *constraintName,
bool nulls[Natts_pg_constraint];
Datum values[Natts_pg_constraint];
ArrayType *conkeyArray;
+ ArrayType *conincludingArray;
ArrayType *confkeyArray;
ArrayType *conpfeqopArray;
ArrayType *conppeqopArray;
@@ -111,6 +113,21 @@ CreateConstraintEntry(const char *constraintName,
else
conkeyArray = NULL;
+ if (constraintNTotalKeys > constraintNKeys)
+ {
+ Datum *conincluding;
+ int j = 0;
+ int constraintNIncludedKeys = constraintNTotalKeys - constraintNKeys;
+
+ conincluding = (Datum *) palloc(constraintNIncludedKeys* sizeof(Datum));
+ for (i = constraintNKeys; i < constraintNTotalKeys; i++)
+ conincluding[j++] = Int16GetDatum(constraintKey[i]);
+ conincludingArray = construct_array(conincluding, constraintNIncludedKeys,
+ INT2OID, 2, true, 's');
+ }
+ else
+ conincludingArray = NULL;
+
if (foreignNKeys > 0)
{
Datum *fkdatums;
@@ -183,6 +200,11 @@ CreateConstraintEntry(const char *constraintName,
else
nulls[Anum_pg_constraint_conkey - 1] = true;
+ if (conincludingArray)
+ values[Anum_pg_constraint_conincluding - 1] = PointerGetDatum(conincludingArray);
+ else
+ nulls[Anum_pg_constraint_conincluding - 1] = true;
+
if (confkeyArray)
values[Anum_pg_constraint_confkey - 1] = PointerGetDatum(confkeyArray);
else
@@ -244,9 +266,9 @@ CreateConstraintEntry(const char *constraintName,
relobject.classId = RelationRelationId;
relobject.objectId = relId;
- if (constraintNKeys > 0)
+ if (constraintNTotalKeys > 0)
{
- for (i = 0; i < constraintNKeys; i++)
+ for (i = 0; i < constraintNTotalKeys; i++)
{
relobject.objectSubId = constraintKey[i];
diff --git a/src/backend/catalog/toasting.c b/src/backend/catalog/toasting.c
index 8bf2698545..61a037523c 100644
--- a/src/backend/catalog/toasting.c
+++ b/src/backend/catalog/toasting.c
@@ -302,6 +302,7 @@ create_toast_table(Relation rel, Oid toastOid, Oid toastIndexOid,
indexInfo = makeNode(IndexInfo);
indexInfo->ii_NumIndexAttrs = 2;
+ indexInfo->ii_NumIndexKeyAttrs = 2;
indexInfo->ii_KeyAttrNumbers[0] = 1;
indexInfo->ii_KeyAttrNumbers[1] = 2;
indexInfo->ii_Expressions = NIL;
diff --git a/src/backend/commands/indexcmds.c b/src/backend/commands/indexcmds.c
index 504806b25b..1b7fe75134 100644
--- a/src/backend/commands/indexcmds.c
+++ b/src/backend/commands/indexcmds.c
@@ -224,7 +224,7 @@ CheckIndexCompatible(Oid oldId,
}
/* Any change in operator class or collation breaks compatibility. */
- old_natts = indexForm->indnatts;
+ old_natts = indexForm->indnkeyatts;
Assert(old_natts == numberOfAttributes);
d = SysCacheGetAttr(INDEXRELID, tuple, Anum_pg_index_indcollation, &isnull);
@@ -351,6 +351,7 @@ DefineIndex(Oid relationId,
bits16 flags;
bits16 constr_flags;
int numberOfAttributes;
+ int numberOfKeyAttributes;
TransactionId limitXmin;
VirtualTransactionId *old_snapshots;
ObjectAddress address;
@@ -361,10 +362,27 @@ DefineIndex(Oid relationId,
Snapshot snapshot;
int i;
+ if(list_intersection(stmt->indexParams, stmt->indexIncludingParams) != NIL)
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
+ errmsg("included columns must not intersect with key columns")));
+ /*
+ * count key attributes in index
+ */
+ numberOfKeyAttributes = list_length(stmt->indexParams);
+
/*
- * count attributes in index
+ * We append any INCLUDE columns onto the indexParams list so that
+ * we have one list with all columns. Later we can determine which of these
+ * are key columns, and which are just part of the INCLUDE list by checking
+ * the list position. A list item in a position less than
+ * ii_NumIndexKeyAttrs is part of the key columns, and anything equal to
+ * and over is part of the INCLUDE columns.
*/
+ stmt->indexParams = list_concat(stmt->indexParams,
+ stmt->indexIncludingParams);
numberOfAttributes = list_length(stmt->indexParams);
+
if (numberOfAttributes <= 0)
ereport(ERROR,
(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
@@ -568,6 +586,11 @@ DefineIndex(Oid relationId,
(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
errmsg("access method \"%s\" does not support unique indexes",
accessMethodName)));
+ if (list_length(stmt->indexIncludingParams) > 0 && !amRoutine->amcaninclude)
+ ereport(ERROR,
+ (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("access method \"%s\" does not support included columns",
+ accessMethodName)));
if (numberOfAttributes > 1 && !amRoutine->amcanmulticol)
ereport(ERROR,
(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
@@ -605,6 +628,7 @@ DefineIndex(Oid relationId,
*/
indexInfo = makeNode(IndexInfo);
indexInfo->ii_NumIndexAttrs = numberOfAttributes;
+ indexInfo->ii_NumIndexKeyAttrs = numberOfKeyAttributes;
indexInfo->ii_Expressions = NIL; /* for now */
indexInfo->ii_ExpressionsState = NIL;
indexInfo->ii_Predicate = make_ands_implicit((Expr *) stmt->whereClause);
@@ -624,7 +648,7 @@ DefineIndex(Oid relationId,
typeObjectId = (Oid *) palloc(numberOfAttributes * sizeof(Oid));
collationObjectId = (Oid *) palloc(numberOfAttributes * sizeof(Oid));
- classObjectId = (Oid *) palloc(numberOfAttributes * sizeof(Oid));
+ classObjectId = (Oid *) palloc(numberOfKeyAttributes * sizeof(Oid));
coloptions = (int16 *) palloc(numberOfAttributes * sizeof(int16));
ComputeIndexAttrs(indexInfo,
typeObjectId, collationObjectId, classObjectId,
@@ -1348,16 +1372,15 @@ ComputeIndexAttrs(IndexInfo *indexInfo,
ListCell *nextExclOp;
ListCell *lc;
int attn;
+ int nkeycols = indexInfo->ii_NumIndexKeyAttrs;
/* Allocate space for exclusion operator info, if needed */
if (exclusionOpNames)
{
- int ncols = list_length(attList);
-
- Assert(list_length(exclusionOpNames) == ncols);
- indexInfo->ii_ExclusionOps = (Oid *) palloc(sizeof(Oid) * ncols);
- indexInfo->ii_ExclusionProcs = (Oid *) palloc(sizeof(Oid) * ncols);
- indexInfo->ii_ExclusionStrats = (uint16 *) palloc(sizeof(uint16) * ncols);
+ Assert(list_length(exclusionOpNames) == nkeycols);
+ indexInfo->ii_ExclusionOps = (Oid *) palloc(sizeof(Oid) * nkeycols);
+ indexInfo->ii_ExclusionProcs = (Oid *) palloc(sizeof(Oid) * nkeycols);
+ indexInfo->ii_ExclusionStrats = (uint16 *) palloc(sizeof(uint16) * nkeycols);
nextExclOp = list_head(exclusionOpNames);
}
else
@@ -1410,6 +1433,11 @@ ComputeIndexAttrs(IndexInfo *indexInfo,
Node *expr = attribute->expr;
Assert(expr != NULL);
+
+ if (attn >= nkeycols)
+ ereport(ERROR,
+ (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("expressions are not supported in included columns")));
atttype = exprType(expr);
attcollation = exprCollation(expr);
@@ -1487,6 +1515,16 @@ ComputeIndexAttrs(IndexInfo *indexInfo,
collationOidP[attn] = attcollation;
+ /*
+ * Skip opclass and ordering options for included columns.
+ */
+ if (attn >= nkeycols)
+ {
+ colOptionP[attn] = 0;
+ attn++;
+ continue;
+ }
+
/*
* Identify the opclass to use.
*/
diff --git a/src/backend/commands/matview.c b/src/backend/commands/matview.c
index ab6a889b12..15df8322ca 100644
--- a/src/backend/commands/matview.c
+++ b/src/backend/commands/matview.c
@@ -634,7 +634,7 @@ refresh_by_match_merge(Oid matviewOid, Oid tempOid, Oid relowner,
RelationGetRelationName(tempRel));
diffname = make_temptable_name_n(tempname, 2);
- relnatts = matviewRel->rd_rel->relnatts;
+ relnatts = RelationGetNumberOfAttributes(matviewRel);
usedForQual = (bool *) palloc0(sizeof(bool) * relnatts);
/* Open SPI context. */
@@ -720,11 +720,11 @@ refresh_by_match_merge(Oid matviewOid, Oid tempOid, Oid relowner,
RelationGetIndexExpressions(indexRel) == NIL &&
RelationGetIndexPredicate(indexRel) == NIL)
{
- int numatts = indexStruct->indnatts;
+ int indnkeyatts = indexStruct->indnkeyatts;
int i;
/* Add quals for all columns from this index. */
- for (i = 0; i < numatts; i++)
+ for (i = 0; i < indnkeyatts; i++)
{
int attnum = indexStruct->indkey.values[i];
Form_pg_attribute attr = TupleDescAttr(tupdesc, attnum - 1);
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index 74e020bffc..1a85f76ad2 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -5825,7 +5825,7 @@ ATExecDropNotNull(Relation rel, const char *colName, LOCKMODE lockmode)
* Loop over each attribute in the primary key and see if it
* matches the to-be-altered attribute
*/
- for (i = 0; i < indexStruct->indnatts; i++)
+ for (i = 0; i < indexStruct->indnkeyatts; i++)
{
if (indexStruct->indkey.values[i] == attnum)
ereport(ERROR,
@@ -7504,6 +7504,7 @@ ATAddForeignKeyConstraint(AlteredTableInfo *tab, Relation rel,
RelationGetRelid(rel),
fkattnum,
numfks,
+ numfks,
InvalidOid, /* not a domain constraint */
indexOid,
RelationGetRelid(pkrel),
@@ -8026,7 +8027,7 @@ transformFkeyGetPrimaryKey(Relation pkrel, Oid *indexOid,
* assume a primary key cannot have expressional elements)
*/
*attnamelist = NIL;
- for (i = 0; i < indexStruct->indnatts; i++)
+ for (i = 0; i < indexStruct->indnkeyatts; i++)
{
int pkattno = indexStruct->indkey.values[i];
@@ -8104,7 +8105,7 @@ transformFkeyCheckAttrs(Relation pkrel,
* partial index; forget it if there are any expressions, too. Invalid
* indexes are out as well.
*/
- if (indexStruct->indnatts == numattrs &&
+ if (indexStruct->indnkeyatts == numattrs &&
indexStruct->indisunique &&
IndexIsValid(indexStruct) &&
heap_attisnull(indexTuple, Anum_pg_index_indpred) &&
@@ -12326,7 +12327,7 @@ ATExecReplicaIdentity(Relation rel, ReplicaIdentityStmt *stmt, LOCKMODE lockmode
RelationGetRelationName(indexRel))));
/* Check index for nullable columns. */
- for (key = 0; key < indexRel->rd_index->indnatts; key++)
+ for (key = 0; key < IndexRelationGetNumberOfKeyAttributes(indexRel); key++)
{
int16 attno = indexRel->rd_index->indkey.values[key];
Form_pg_attribute attr;
diff --git a/src/backend/commands/trigger.c b/src/backend/commands/trigger.c
index fbd176b5d0..af529543ce 100644
--- a/src/backend/commands/trigger.c
+++ b/src/backend/commands/trigger.c
@@ -654,6 +654,7 @@ CreateTrigger(CreateTrigStmt *stmt, const char *queryString,
RelationGetRelid(rel),
NULL, /* no conkey */
0,
+ 0,
InvalidOid, /* no domain */
InvalidOid, /* no index */
InvalidOid, /* no foreign key */
diff --git a/src/backend/commands/typecmds.c b/src/backend/commands/typecmds.c
index bf3cd3a454..ae1996b031 100644
--- a/src/backend/commands/typecmds.c
+++ b/src/backend/commands/typecmds.c
@@ -3156,6 +3156,7 @@ domainAddConstraint(Oid domainOid, Oid domainNamespace, Oid baseTypeOid,
InvalidOid, /* not a relation constraint */
NULL,
0,
+ 0,
domainOid, /* domain constraint */
InvalidOid, /* no associated index */
InvalidOid, /* Foreign key fields */
diff --git a/src/backend/executor/execIndexing.c b/src/backend/executor/execIndexing.c
index 62e51f1ef3..903076ee3c 100644
--- a/src/backend/executor/execIndexing.c
+++ b/src/backend/executor/execIndexing.c
@@ -648,7 +648,7 @@ check_exclusion_or_unique_constraint(Relation heap, Relation index,
Oid *constr_procs;
uint16 *constr_strats;
Oid *index_collations = index->rd_indcollation;
- int index_natts = index->rd_index->indnatts;
+ int indnkeyatts = IndexRelationGetNumberOfKeyAttributes(index);
IndexScanDesc index_scan;
HeapTuple tup;
ScanKeyData scankeys[INDEX_MAX_KEYS];
@@ -675,7 +675,7 @@ check_exclusion_or_unique_constraint(Relation heap, Relation index,
* If any of the input values are NULL, the constraint check is assumed to
* pass (i.e., we assume the operators are strict).
*/
- for (i = 0; i < index_natts; i++)
+ for (i = 0; i < indnkeyatts; i++)
{
if (isnull[i])
return true;
@@ -687,7 +687,7 @@ check_exclusion_or_unique_constraint(Relation heap, Relation index,
*/
InitDirtySnapshot(DirtySnapshot);
- for (i = 0; i < index_natts; i++)
+ for (i = 0; i < indnkeyatts; i++)
{
ScanKeyEntryInitialize(&scankeys[i],
0,
@@ -719,8 +719,8 @@ check_exclusion_or_unique_constraint(Relation heap, Relation index,
retry:
conflict = false;
found_self = false;
- index_scan = index_beginscan(heap, index, &DirtySnapshot, index_natts, 0);
- index_rescan(index_scan, scankeys, index_natts, NULL, 0);
+ index_scan = index_beginscan(heap, index, &DirtySnapshot, indnkeyatts, 0);
+ index_rescan(index_scan, scankeys, indnkeyatts, NULL, 0);
while ((tup = index_getnext(index_scan,
ForwardScanDirection)) != NULL)
@@ -881,10 +881,10 @@ index_recheck_constraint(Relation index, Oid *constr_procs,
Datum *existing_values, bool *existing_isnull,
Datum *new_values)
{
- int index_natts = index->rd_index->indnatts;
+ int indnkeyatts = IndexRelationGetNumberOfKeyAttributes(index);
int i;
- for (i = 0; i < index_natts; i++)
+ for (i = 0; i < indnkeyatts; i++)
{
/* Assume the exclusion operators are strict */
if (existing_isnull[i])
diff --git a/src/backend/executor/nodeIndexscan.c b/src/backend/executor/nodeIndexscan.c
index 01c9de88f4..d6012192a1 100644
--- a/src/backend/executor/nodeIndexscan.c
+++ b/src/backend/executor/nodeIndexscan.c
@@ -1227,7 +1227,9 @@ ExecIndexBuildScanKeys(PlanState *planstate, Relation index,
Expr *leftop; /* expr on lhs of operator */
Expr *rightop; /* expr on rhs ... */
AttrNumber varattno; /* att number used in scan */
+ int indnkeyatts;
+ indnkeyatts = IndexRelationGetNumberOfKeyAttributes(index);
if (IsA(clause, OpExpr))
{
/* indexkey op const or indexkey op expression */
@@ -1252,7 +1254,7 @@ ExecIndexBuildScanKeys(PlanState *planstate, Relation index,
elog(ERROR, "indexqual doesn't have key on left side");
varattno = ((Var *) leftop)->varattno;
- if (varattno < 1 || varattno > index->rd_index->indnatts)
+ if (varattno < 1 || varattno > indnkeyatts)
elog(ERROR, "bogus index qualification");
/*
@@ -1375,7 +1377,7 @@ ExecIndexBuildScanKeys(PlanState *planstate, Relation index,
opnos_cell = lnext(opnos_cell);
if (index->rd_rel->relam != BTREE_AM_OID ||
- varattno < 1 || varattno > index->rd_index->indnatts)
+ varattno < 1 || varattno > indnkeyatts)
elog(ERROR, "bogus RowCompare index qualification");
opfamily = index->rd_opfamily[varattno - 1];
@@ -1499,7 +1501,7 @@ ExecIndexBuildScanKeys(PlanState *planstate, Relation index,
elog(ERROR, "indexqual doesn't have key on left side");
varattno = ((Var *) leftop)->varattno;
- if (varattno < 1 || varattno > index->rd_index->indnatts)
+ if (varattno < 1 || varattno > indnkeyatts)
elog(ERROR, "bogus index qualification");
/*
diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c
index f84da801c6..12fc428944 100644
--- a/src/backend/nodes/copyfuncs.c
+++ b/src/backend/nodes/copyfuncs.c
@@ -2854,6 +2854,7 @@ _copyConstraint(const Constraint *from)
COPY_STRING_FIELD(cooked_expr);
COPY_SCALAR_FIELD(generated_when);
COPY_NODE_FIELD(keys);
+ COPY_NODE_FIELD(including);
COPY_NODE_FIELD(exclusions);
COPY_NODE_FIELD(options);
COPY_STRING_FIELD(indexname);
@@ -3397,6 +3398,7 @@ _copyIndexStmt(const IndexStmt *from)
COPY_STRING_FIELD(accessMethod);
COPY_STRING_FIELD(tableSpace);
COPY_NODE_FIELD(indexParams);
+ COPY_NODE_FIELD(indexIncludingParams);
COPY_NODE_FIELD(options);
COPY_NODE_FIELD(whereClause);
COPY_NODE_FIELD(excludeOpNames);
diff --git a/src/backend/nodes/equalfuncs.c b/src/backend/nodes/equalfuncs.c
index ee8d925db1..9fa58f4cfb 100644
--- a/src/backend/nodes/equalfuncs.c
+++ b/src/backend/nodes/equalfuncs.c
@@ -1338,6 +1338,7 @@ _equalIndexStmt(const IndexStmt *a, const IndexStmt *b)
COMPARE_STRING_FIELD(accessMethod);
COMPARE_STRING_FIELD(tableSpace);
COMPARE_NODE_FIELD(indexParams);
+ COMPARE_NODE_FIELD(indexIncludingParams);
COMPARE_NODE_FIELD(options);
COMPARE_NODE_FIELD(whereClause);
COMPARE_NODE_FIELD(excludeOpNames);
@@ -2589,6 +2590,7 @@ _equalConstraint(const Constraint *a, const Constraint *b)
COMPARE_STRING_FIELD(cooked_expr);
COMPARE_SCALAR_FIELD(generated_when);
COMPARE_NODE_FIELD(keys);
+ COMPARE_NODE_FIELD(including);
COMPARE_NODE_FIELD(exclusions);
COMPARE_NODE_FIELD(options);
COMPARE_STRING_FIELD(indexname);
diff --git a/src/backend/nodes/outfuncs.c b/src/backend/nodes/outfuncs.c
index 1785ea3918..e7ece3072a 100644
--- a/src/backend/nodes/outfuncs.c
+++ b/src/backend/nodes/outfuncs.c
@@ -2662,6 +2662,7 @@ _outIndexStmt(StringInfo str, const IndexStmt *node)
WRITE_STRING_FIELD(accessMethod);
WRITE_STRING_FIELD(tableSpace);
WRITE_NODE_FIELD(indexParams);
+ WRITE_NODE_FIELD(indexIncludingParams);
WRITE_NODE_FIELD(options);
WRITE_NODE_FIELD(whereClause);
WRITE_NODE_FIELD(excludeOpNames);
@@ -3487,6 +3488,7 @@ _outConstraint(StringInfo str, const Constraint *node)
case CONSTR_PRIMARY:
appendStringInfoString(str, "PRIMARY_KEY");
WRITE_NODE_FIELD(keys);
+ WRITE_NODE_FIELD(including);
WRITE_NODE_FIELD(options);
WRITE_STRING_FIELD(indexname);
WRITE_STRING_FIELD(indexspace);
@@ -3496,6 +3498,7 @@ _outConstraint(StringInfo str, const Constraint *node)
case CONSTR_UNIQUE:
appendStringInfoString(str, "UNIQUE");
WRITE_NODE_FIELD(keys);
+ WRITE_NODE_FIELD(including);
WRITE_NODE_FIELD(options);
WRITE_STRING_FIELD(indexname);
WRITE_STRING_FIELD(indexspace);
@@ -3505,6 +3508,7 @@ _outConstraint(StringInfo str, const Constraint *node)
case CONSTR_EXCLUSION:
appendStringInfoString(str, "EXCLUSION");
WRITE_NODE_FIELD(exclusions);
+ WRITE_NODE_FIELD(including);
WRITE_NODE_FIELD(options);
WRITE_STRING_FIELD(indexname);
WRITE_STRING_FIELD(indexspace);
diff --git a/src/backend/optimizer/path/indxpath.c b/src/backend/optimizer/path/indxpath.c
index 594ac8eacb..8e16a79a90 100644
--- a/src/backend/optimizer/path/indxpath.c
+++ b/src/backend/optimizer/path/indxpath.c
@@ -2164,7 +2164,7 @@ match_eclass_clauses_to_index(PlannerInfo *root, IndexOptInfo *index,
if (!index->rel->has_eclass_joins)
return;
- for (indexcol = 0; indexcol < index->ncolumns; indexcol++)
+ for (indexcol = 0; indexcol < index->nkeycolumns; indexcol++)
{
ec_member_matches_arg arg;
List *clauses;
diff --git a/src/backend/optimizer/path/pathkeys.c b/src/backend/optimizer/path/pathkeys.c
index 6d1cc3b8a0..c971dc78d9 100644
--- a/src/backend/optimizer/path/pathkeys.c
+++ b/src/backend/optimizer/path/pathkeys.c
@@ -477,6 +477,13 @@ build_index_pathkeys(PlannerInfo *root,
bool nulls_first;
PathKey *cpathkey;
+ /*
+ * INCLUDE columns are stored in index unordered,
+ * so they don't support ordered index scan.
+ */
+ if(i >= index->nkeycolumns)
+ break;
+
/* We assume we don't need to make a copy of the tlist item */
indexkey = indextle->expr;
diff --git a/src/backend/optimizer/util/plancat.c b/src/backend/optimizer/util/plancat.c
index b799e249db..cf13dc8e4b 100644
--- a/src/backend/optimizer/util/plancat.c
+++ b/src/backend/optimizer/util/plancat.c
@@ -185,7 +185,7 @@ get_relation_info(PlannerInfo *root, Oid relationObjectId, bool inhparent,
Form_pg_index index;
IndexAmRoutine *amroutine;
IndexOptInfo *info;
- int ncolumns;
+ int ncolumns, nkeycolumns;
int i;
/*
@@ -238,19 +238,25 @@ get_relation_info(PlannerInfo *root, Oid relationObjectId, bool inhparent,
RelationGetForm(indexRelation)->reltablespace;
info->rel = rel;
info->ncolumns = ncolumns = index->indnatts;
+ info->nkeycolumns = nkeycolumns = index->indnkeyatts;
+
info->indexkeys = (int *) palloc(sizeof(int) * ncolumns);
info->indexcollations = (Oid *) palloc(sizeof(Oid) * ncolumns);
- info->opfamily = (Oid *) palloc(sizeof(Oid) * ncolumns);
- info->opcintype = (Oid *) palloc(sizeof(Oid) * ncolumns);
+ info->opfamily = (Oid *) palloc(sizeof(Oid) * nkeycolumns);
+ info->opcintype = (Oid *) palloc(sizeof(Oid) * nkeycolumns);
info->canreturn = (bool *) palloc(sizeof(bool) * ncolumns);
for (i = 0; i < ncolumns; i++)
{
info->indexkeys[i] = index->indkey.values[i];
info->indexcollations[i] = indexRelation->rd_indcollation[i];
+ info->canreturn[i] = index_can_return(indexRelation, i + 1);
+ }
+
+ for (i = 0; i < nkeycolumns; i++)
+ {
info->opfamily[i] = indexRelation->rd_opfamily[i];
info->opcintype[i] = indexRelation->rd_opcintype[i];
- info->canreturn[i] = index_can_return(indexRelation, i + 1);
}
info->relam = indexRelation->rd_rel->relam;
@@ -279,10 +285,10 @@ get_relation_info(PlannerInfo *root, Oid relationObjectId, bool inhparent,
Assert(amroutine->amcanorder);
info->sortopfamily = info->opfamily;
- info->reverse_sort = (bool *) palloc(sizeof(bool) * ncolumns);
- info->nulls_first = (bool *) palloc(sizeof(bool) * ncolumns);
+ info->reverse_sort = (bool *) palloc(sizeof(bool) * nkeycolumns);
+ info->nulls_first = (bool *) palloc(sizeof(bool) * nkeycolumns);
- for (i = 0; i < ncolumns; i++)
+ for (i = 0; i < nkeycolumns; i++)
{
int16 opt = indexRelation->rd_indoption[i];
@@ -306,11 +312,11 @@ get_relation_info(PlannerInfo *root, Oid relationObjectId, bool inhparent,
* of current or foreseeable amcanorder index types, it's not
* worth expending more effort on now.
*/
- info->sortopfamily = (Oid *) palloc(sizeof(Oid) * ncolumns);
- info->reverse_sort = (bool *) palloc(sizeof(bool) * ncolumns);
- info->nulls_first = (bool *) palloc(sizeof(bool) * ncolumns);
+ info->sortopfamily = (Oid *) palloc(sizeof(Oid) * nkeycolumns);
+ info->reverse_sort = (bool *) palloc(sizeof(bool) * nkeycolumns);
+ info->nulls_first = (bool *) palloc(sizeof(bool) * nkeycolumns);
- for (i = 0; i < ncolumns; i++)
+ for (i = 0; i < nkeycolumns; i++)
{
int16 opt = indexRelation->rd_indoption[i];
Oid ltopr;
@@ -731,7 +737,7 @@ infer_arbiter_indexes(PlannerInfo *root)
/* Build BMS representation of plain (non expression) index attrs */
indexedAttrs = NULL;
- for (natt = 0; natt < idxForm->indnatts; natt++)
+ for (natt = 0; natt < idxForm->indnkeyatts; natt++)
{
int attno = idxRel->rd_index->indkey.values[natt];
@@ -1780,7 +1786,7 @@ has_unique_index(RelOptInfo *rel, AttrNumber attno)
* just the specified attr is unique.
*/
if (index->unique &&
- index->ncolumns == 1 &&
+ index->nkeycolumns == 1 &&
index->indexkeys[0] == attno &&
(index->indpred == NIL || index->predOK))
return true;
diff --git a/src/backend/parser/analyze.c b/src/backend/parser/analyze.c
index c3a9617f67..b5bd835827 100644
--- a/src/backend/parser/analyze.c
+++ b/src/backend/parser/analyze.c
@@ -1055,7 +1055,7 @@ transformOnConflictClause(ParseState *pstate,
* relation. Have to be careful to use resnos that correspond to
* attnos of the underlying relation.
*/
- for (attno = 0; attno < targetrel->rd_rel->relnatts; attno++)
+ for (attno = 0; attno < RelationGetNumberOfAttributes(targetrel); attno++)
{
Form_pg_attribute attr = TupleDescAttr(targetrel->rd_att, attno);
char *name;
@@ -2280,8 +2280,8 @@ transformUpdateTargetList(ParseState *pstate, List *origTlist)
EXPR_KIND_UPDATE_SOURCE);
/* Prepare to assign non-conflicting resnos to resjunk attributes */
- if (pstate->p_next_resno <= pstate->p_target_relation->rd_rel->relnatts)
- pstate->p_next_resno = pstate->p_target_relation->rd_rel->relnatts + 1;
+ if (pstate->p_next_resno <= RelationGetNumberOfAttributes(pstate->p_target_relation))
+ pstate->p_next_resno = RelationGetNumberOfAttributes(pstate->p_target_relation) + 1;
/* Prepare non-junk columns for assignment to target table */
target_rte = pstate->p_target_rangetblentry;
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index 06c03dff3c..68d3eaf5ba 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -380,6 +380,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);
oper_argtypes RuleActionList RuleActionMulti
opt_column_list columnList opt_name_list
sort_clause opt_sort_clause sortby_list index_params
+ opt_include opt_c_include index_including_params
name_list role_list from_clause from_list opt_array_bounds
qualified_name_list any_name any_name_list type_name_list
any_operator expr_list attrs
@@ -638,7 +639,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);
HANDLER HAVING HEADER_P HOLD HOUR_P
- IDENTITY_P IF_P ILIKE IMMEDIATE IMMUTABLE IMPLICIT_P IMPORT_P IN_P
+ IDENTITY_P IF_P ILIKE IMMEDIATE IMMUTABLE IMPLICIT_P IMPORT_P IN_P INCLUDE
INCLUDING INCREMENT INDEX INDEXES INHERIT INHERITS INITIALLY INLINE_P
INNER_P INOUT INPUT_P INSENSITIVE INSERT INSTEAD INT_P INTEGER
INTERSECT INTERVAL INTO INVOKER IS ISNULL ISOLATION
@@ -3677,17 +3678,18 @@ ConstraintElem:
n->initially_valid = !n->skip_validation;
$$ = (Node *)n;
}
- | UNIQUE '(' columnList ')' opt_definition OptConsTableSpace
+ | UNIQUE '(' columnList ')' opt_c_include opt_definition OptConsTableSpace
ConstraintAttributeSpec
{
Constraint *n = makeNode(Constraint);
n->contype = CONSTR_UNIQUE;
n->location = @1;
n->keys = $3;
- n->options = $5;
+ n->including = $5;
+ n->options = $6;
n->indexname = NULL;
- n->indexspace = $6;
- processCASbits($7, @7, "UNIQUE",
+ n->indexspace = $7;
+ processCASbits($8, @8, "UNIQUE",
&n->deferrable, &n->initdeferred, NULL,
NULL, yyscanner);
$$ = (Node *)n;
@@ -3698,6 +3700,7 @@ ConstraintElem:
n->contype = CONSTR_UNIQUE;
n->location = @1;
n->keys = NIL;
+ n->including = NIL;
n->options = NIL;
n->indexname = $2;
n->indexspace = NULL;
@@ -3706,17 +3709,18 @@ ConstraintElem:
NULL, yyscanner);
$$ = (Node *)n;
}
- | PRIMARY KEY '(' columnList ')' opt_definition OptConsTableSpace
+ | PRIMARY KEY '(' columnList ')' opt_c_include opt_definition OptConsTableSpace
ConstraintAttributeSpec
{
Constraint *n = makeNode(Constraint);
n->contype = CONSTR_PRIMARY;
n->location = @1;
n->keys = $4;
- n->options = $6;
+ n->including = $6;
+ n->options = $7;
n->indexname = NULL;
- n->indexspace = $7;
- processCASbits($8, @8, "PRIMARY KEY",
+ n->indexspace = $8;
+ processCASbits($9, @9, "PRIMARY KEY",
&n->deferrable, &n->initdeferred, NULL,
NULL, yyscanner);
$$ = (Node *)n;
@@ -3727,6 +3731,7 @@ ConstraintElem:
n->contype = CONSTR_PRIMARY;
n->location = @1;
n->keys = NIL;
+ n->including = NIL;
n->options = NIL;
n->indexname = $3;
n->indexspace = NULL;
@@ -3736,7 +3741,7 @@ ConstraintElem:
$$ = (Node *)n;
}
| EXCLUDE access_method_clause '(' ExclusionConstraintList ')'
- opt_definition OptConsTableSpace ExclusionWhereClause
+ opt_c_include opt_definition OptConsTableSpace ExclusionWhereClause
ConstraintAttributeSpec
{
Constraint *n = makeNode(Constraint);
@@ -3744,11 +3749,12 @@ ConstraintElem:
n->location = @1;
n->access_method = $2;
n->exclusions = $4;
- n->options = $6;
+ n->including = $6;
+ n->options = $7;
n->indexname = NULL;
- n->indexspace = $7;
- n->where_clause = $8;
- processCASbits($9, @9, "EXCLUDE",
+ n->indexspace = $8;
+ n->where_clause = $9;
+ processCASbits($10, @10, "EXCLUDE",
&n->deferrable, &n->initdeferred, NULL,
NULL, yyscanner);
$$ = (Node *)n;
@@ -3794,6 +3800,10 @@ columnElem: ColId
}
;
+opt_c_include: INCLUDE '(' columnList ')' { $$ = $3; }
+ | /* EMPTY */ { $$ = NIL; }
+ ;
+
key_match: MATCH FULL
{
$$ = FKCONSTR_MATCH_FULL;
@@ -7364,7 +7374,7 @@ defacl_privilege_target:
IndexStmt: CREATE opt_unique INDEX opt_concurrently opt_index_name
ON relation_expr access_method_clause '(' index_params ')'
- opt_reloptions OptTableSpace where_clause
+ opt_include opt_reloptions OptTableSpace where_clause
{
IndexStmt *n = makeNode(IndexStmt);
n->unique = $2;
@@ -7374,9 +7384,10 @@ IndexStmt: CREATE opt_unique INDEX opt_concurrently opt_index_name
n->relationId = InvalidOid;
n->accessMethod = $8;
n->indexParams = $10;
- n->options = $12;
- n->tableSpace = $13;
- n->whereClause = $14;
+ n->indexIncludingParams = $12;
+ n->options = $13;
+ n->tableSpace = $14;
+ n->whereClause = $15;
n->excludeOpNames = NIL;
n->idxcomment = NULL;
n->indexOid = InvalidOid;
@@ -7391,7 +7402,7 @@ IndexStmt: CREATE opt_unique INDEX opt_concurrently opt_index_name
}
| CREATE opt_unique INDEX opt_concurrently IF_P NOT EXISTS index_name
ON relation_expr access_method_clause '(' index_params ')'
- opt_reloptions OptTableSpace where_clause
+ opt_include opt_reloptions OptTableSpace where_clause
{
IndexStmt *n = makeNode(IndexStmt);
n->unique = $2;
@@ -7401,9 +7412,10 @@ IndexStmt: CREATE opt_unique INDEX opt_concurrently opt_index_name
n->relationId = InvalidOid;
n->accessMethod = $11;
n->indexParams = $13;
- n->options = $15;
- n->tableSpace = $16;
- n->whereClause = $17;
+ n->indexIncludingParams = $15;
+ n->options = $16;
+ n->tableSpace = $17;
+ n->whereClause = $18;
n->excludeOpNames = NIL;
n->idxcomment = NULL;
n->indexOid = InvalidOid;
@@ -7482,6 +7494,14 @@ index_elem: ColId opt_collate opt_class opt_asc_desc opt_nulls_order
}
;
+opt_include: INCLUDE '(' index_including_params ')' { $$ = $3; }
+ | /* EMPTY */ { $$ = NIL; }
+ ;
+
+index_including_params: index_elem { $$ = list_make1($1); }
+ | index_including_params ',' index_elem { $$ = lappend($1, $3); }
+ ;
+
opt_collate: COLLATE any_name { $$ = $2; }
| /*EMPTY*/ { $$ = NIL; }
;
@@ -15064,6 +15084,7 @@ unreserved_keyword:
| IMMUTABLE
| IMPLICIT_P
| IMPORT_P
+ | INCLUDE
| INCLUDING
| INCREMENT
| INDEX
diff --git a/src/backend/parser/parse_relation.c b/src/backend/parser/parse_relation.c
index 053ae02c9f..bf5df26009 100644
--- a/src/backend/parser/parse_relation.c
+++ b/src/backend/parser/parse_relation.c
@@ -3079,7 +3079,7 @@ attnameAttNum(Relation rd, const char *attname, bool sysColOK)
{
int i;
- for (i = 0; i < rd->rd_rel->relnatts; i++)
+ for (i = 0; i < RelationGetNumberOfAttributes(rd); i++)
{
Form_pg_attribute att = TupleDescAttr(rd->rd_att, i);
diff --git a/src/backend/parser/parse_target.c b/src/backend/parser/parse_target.c
index ea209cdab6..4932e58022 100644
--- a/src/backend/parser/parse_target.c
+++ b/src/backend/parser/parse_target.c
@@ -978,7 +978,8 @@ checkInsertTargets(ParseState *pstate, List *cols, List **attrnos)
/*
* Generate default column list for INSERT.
*/
- int numcol = pstate->p_target_relation->rd_rel->relnatts;
+ int numcol = RelationGetNumberOfAttributes(pstate->p_target_relation);
+
int i;
for (i = 0; i < numcol; i++)
diff --git a/src/backend/parser/parse_utilcmd.c b/src/backend/parser/parse_utilcmd.c
index 0fd14f43c6..950b1530cf 100644
--- a/src/backend/parser/parse_utilcmd.c
+++ b/src/backend/parser/parse_utilcmd.c
@@ -1480,9 +1480,10 @@ generateClonedIndexStmt(RangeVar *heapRel, Oid heapRelid, Relation source_idx,
/* Build the list of IndexElem */
index->indexParams = NIL;
+ index->indexIncludingParams = NIL;
indexpr_item = list_head(indexprs);
- for (keyno = 0; keyno < idxrec->indnatts; keyno++)
+ for (keyno = 0; keyno < idxrec->indnkeyatts; keyno++)
{
IndexElem *iparam;
AttrNumber attnum = idxrec->indkey.values[keyno];
@@ -1571,6 +1572,39 @@ generateClonedIndexStmt(RangeVar *heapRel, Oid heapRelid, Relation source_idx,
index->indexParams = lappend(index->indexParams, iparam);
}
+ /* Handle included columns separately */
+ for (keyno = idxrec->indnkeyatts; keyno < idxrec->indnatts; keyno++)
+ {
+ IndexElem *iparam;
+ AttrNumber attnum = idxrec->indkey.values[keyno];
+ Form_pg_attribute attr = TupleDescAttr(RelationGetDescr(source_idx),
+ keyno);
+ iparam = makeNode(IndexElem);
+
+ if (AttributeNumberIsValid(attnum))
+ {
+ /* Simple index column */
+ char *attname;
+
+ attname = get_attname(indrelid, attnum, false);
+ keycoltype = get_atttype(indrelid, attnum);
+
+ iparam->name = attname;
+ iparam->expr = NULL;
+ }
+ else
+ ereport(ERROR,
+ (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("expressions are not supported in included columns")));
+
+ /* Copy the original index column name */
+ iparam->indexcolname = pstrdup(NameStr(attr->attname));
+
+ /* Add the collation name, if non-default */
+ iparam->collation = get_collation(indcollation->values[keyno], keycoltype);
+
+ index->indexIncludingParams = lappend(index->indexIncludingParams, iparam);
+ }
/* Copy reloptions if any */
datum = SysCacheGetAttr(RELOID, ht_idxrel,
Anum_pg_class_reloptions, &isnull);
@@ -1841,6 +1875,7 @@ transformIndexConstraints(CreateStmtContext *cxt)
IndexStmt *priorindex = lfirst(k);
if (equal(index->indexParams, priorindex->indexParams) &&
+ equal(index->indexIncludingParams, priorindex->indexIncludingParams) &&
equal(index->whereClause, priorindex->whereClause) &&
equal(index->excludeOpNames, priorindex->excludeOpNames) &&
strcmp(index->accessMethod, priorindex->accessMethod) == 0 &&
@@ -1912,6 +1947,7 @@ transformIndexConstraint(Constraint *constraint, CreateStmtContext *cxt)
index->tableSpace = constraint->indexspace;
index->whereClause = constraint->where_clause;
index->indexParams = NIL;
+ index->indexIncludingParams = NIL;
index->excludeOpNames = NIL;
index->idxcomment = NULL;
index->indexOid = InvalidOid;
@@ -2061,24 +2097,29 @@ transformIndexConstraint(Constraint *constraint, CreateStmtContext *cxt)
heap_rel->rd_rel->relhasoids);
attname = pstrdup(NameStr(attform->attname));
- /*
- * Insist on default opclass and sort options. While the index
- * would still work as a constraint with non-default settings, it
- * might not provide exactly the same uniqueness semantics as
- * you'd get from a normally-created constraint; and there's also
- * the dump/reload problem mentioned above.
- */
- defopclass = GetDefaultOpClass(attform->atttypid,
- index_rel->rd_rel->relam);
- if (indclass->values[i] != defopclass ||
- index_rel->rd_indoption[i] != 0)
- ereport(ERROR,
- (errcode(ERRCODE_WRONG_OBJECT_TYPE),
- errmsg("index \"%s\" does not have default sorting behavior", index_name),
- errdetail("Cannot create a primary key or unique constraint using such an index."),
- parser_errposition(cxt->pstate, constraint->location)));
+ if (i < index_form->indnkeyatts)
+ {
+ /*
+ * Insist on default opclass and sort options. While the index
+ * would still work as a constraint with non-default settings, it
+ * might not provide exactly the same uniqueness semantics as
+ * you'd get from a normally-created constraint; and there's also
+ * the dump/reload problem mentioned above.
+ */
+ defopclass = GetDefaultOpClass(attform->atttypid,
+ index_rel->rd_rel->relam);
+ if (indclass->values[i] != defopclass ||
+ index_rel->rd_indoption[i] != 0)
+ ereport(ERROR,
+ (errcode(ERRCODE_WRONG_OBJECT_TYPE),
+ errmsg("index \"%s\" does not have default sorting behavior", index_name),
+ errdetail("Cannot create a primary key or unique constraint using such an index."),
+ parser_errposition(cxt->pstate, constraint->location)));
- constraint->keys = lappend(constraint->keys, makeString(attname));
+ constraint->keys = lappend(constraint->keys, makeString(attname));
+ }
+ else
+ constraint->including = lappend(constraint->including, makeString(attname));
}
/* Close the index relation but keep the lock */
@@ -2107,8 +2148,6 @@ transformIndexConstraint(Constraint *constraint, CreateStmtContext *cxt)
index->indexParams = lappend(index->indexParams, elem);
index->excludeOpNames = lappend(index->excludeOpNames, opname);
}
-
- return index;
}
/*
@@ -2119,7 +2158,136 @@ transformIndexConstraint(Constraint *constraint, CreateStmtContext *cxt)
* it to DefineIndex to mark the columns NOT NULL, it's more efficient to
* get it right the first time.)
*/
- foreach(lc, constraint->keys)
+ else
+ {
+ foreach(lc, constraint->keys)
+ {
+ char *key = strVal(lfirst(lc));
+ bool found = false;
+ ColumnDef *column = NULL;
+ ListCell *columns;
+ IndexElem *iparam;
+
+ /* Make sure referenced column exist. */
+ foreach(columns, cxt->columns)
+ {
+ column = castNode(ColumnDef, lfirst(columns));
+ if (strcmp(column->colname, key) == 0)
+ {
+ found = true;
+ break;
+ }
+ }
+ if (found)
+ {
+ /* found column in the new table; force it to be NOT NULL */
+ if (constraint->contype == CONSTR_PRIMARY)
+ column->is_not_null = true;
+ }
+ else if (SystemAttributeByName(key, cxt->hasoids) != NULL)
+ {
+ /*
+ * column will be a system column in the new table, so accept it.
+ * System columns can't ever be null, so no need to worry about
+ * PRIMARY/NOT NULL constraint.
+ */
+ found = true;
+ }
+ else if (cxt->inhRelations)
+ {
+ /* try inherited tables */
+ ListCell *inher;
+
+ foreach(inher, cxt->inhRelations)
+ {
+ RangeVar *inh = castNode(RangeVar, lfirst(inher));
+ Relation rel;
+ int count;
+
+ rel = heap_openrv(inh, AccessShareLock);
+ /* check user requested inheritance from valid relkind */
+ if (rel->rd_rel->relkind != RELKIND_RELATION &&
+ rel->rd_rel->relkind != RELKIND_FOREIGN_TABLE &&
+ rel->rd_rel->relkind != RELKIND_PARTITIONED_TABLE)
+ ereport(ERROR,
+ (errcode(ERRCODE_WRONG_OBJECT_TYPE),
+ errmsg("inherited relation \"%s\" is not a table or foreign table",
+ inh->relname)));
+ for (count = 0; count < rel->rd_att->natts; count++)
+ {
+ Form_pg_attribute inhattr = TupleDescAttr(rel->rd_att,
+ count);
+ char *inhname = NameStr(inhattr->attname);
+
+ if (inhattr->attisdropped)
+ continue;
+ if (strcmp(key, inhname) == 0)
+ {
+ found = true;
+
+ /*
+ * We currently have no easy way to force an inherited
+ * column to be NOT NULL at creation, if its parent
+ * wasn't so already. We leave it to DefineIndex to
+ * fix things up in this case.
+ */
+ break;
+ }
+ }
+ heap_close(rel, NoLock);
+ if (found)
+ break;
+ }
+ }
+
+ /*
+ * In the ALTER TABLE case, don't complain about index keys not
+ * created in the command; they may well exist already. DefineIndex
+ * will complain about them if not, and will also take care of marking
+ * them NOT NULL.
+ */
+ if (!found && !cxt->isalter)
+ ereport(ERROR,
+ (errcode(ERRCODE_UNDEFINED_COLUMN),
+ errmsg("column \"%s\" named in key does not exist", key),
+ parser_errposition(cxt->pstate, constraint->location)));
+
+ /* Check for PRIMARY KEY(foo, foo) */
+ foreach(columns, index->indexParams)
+ {
+ iparam = (IndexElem *) lfirst(columns);
+ if (iparam->name && strcmp(key, iparam->name) == 0)
+ {
+ if (index->primary)
+ ereport(ERROR,
+ (errcode(ERRCODE_DUPLICATE_COLUMN),
+ errmsg("column \"%s\" appears twice in primary key constraint",
+ key),
+ parser_errposition(cxt->pstate, constraint->location)));
+ else
+ ereport(ERROR,
+ (errcode(ERRCODE_DUPLICATE_COLUMN),
+ errmsg("column \"%s\" appears twice in unique constraint",
+ key),
+ parser_errposition(cxt->pstate, constraint->location)));
+ }
+ }
+
+ /* OK, add it to the index definition */
+ iparam = makeNode(IndexElem);
+ iparam->name = pstrdup(key);
+ iparam->expr = NULL;
+ iparam->indexcolname = NULL;
+ iparam->collation = NIL;
+ iparam->opclass = NIL;
+ iparam->ordering = SORTBY_DEFAULT;
+ iparam->nulls_ordering = SORTBY_NULLS_DEFAULT;
+ index->indexParams = lappend(index->indexParams, iparam);
+ }
+ }
+
+ /* Add included columns to index definition */
+ foreach(lc, constraint->including)
{
char *key = strVal(lfirst(lc));
bool found = false;
@@ -2136,65 +2304,63 @@ transformIndexConstraint(Constraint *constraint, CreateStmtContext *cxt)
break;
}
}
- if (found)
- {
- /* found column in the new table; force it to be NOT NULL */
- if (constraint->contype == CONSTR_PRIMARY)
- column->is_not_null = true;
- }
- else if (SystemAttributeByName(key, cxt->hasoids) != NULL)
- {
- /*
- * column will be a system column in the new table, so accept it.
- * System columns can't ever be null, so no need to worry about
- * PRIMARY/NOT NULL constraint.
- */
- found = true;
- }
- else if (cxt->inhRelations)
- {
- /* try inherited tables */
- ListCell *inher;
- foreach(inher, cxt->inhRelations)
+ if (!found)
+ {
+ if (SystemAttributeByName(key, cxt->hasoids) != NULL)
{
- RangeVar *inh = lfirst_node(RangeVar, inher);
- Relation rel;
- int count;
-
- rel = heap_openrv(inh, AccessShareLock);
- /* check user requested inheritance from valid relkind */
- if (rel->rd_rel->relkind != RELKIND_RELATION &&
- rel->rd_rel->relkind != RELKIND_FOREIGN_TABLE &&
- rel->rd_rel->relkind != RELKIND_PARTITIONED_TABLE)
- ereport(ERROR,
- (errcode(ERRCODE_WRONG_OBJECT_TYPE),
- errmsg("inherited relation \"%s\" is not a table or foreign table",
- inh->relname)));
- for (count = 0; count < rel->rd_att->natts; count++)
- {
- Form_pg_attribute inhattr = TupleDescAttr(rel->rd_att,
- count);
- char *inhname = NameStr(inhattr->attname);
+ /*
+ * column will be a system column in the new table, so accept it.
+ * System columns can't ever be null, so no need to worry about
+ * PRIMARY/NOT NULL constraint.
+ */
+ found = true;
+ }
+ else if (cxt->inhRelations)
+ {
+ /* try inherited tables */
+ ListCell *inher;
- if (inhattr->attisdropped)
- continue;
- if (strcmp(key, inhname) == 0)
+ foreach(inher, cxt->inhRelations)
+ {
+ RangeVar *inh = lfirst_node(RangeVar, inher);
+ Relation rel;
+ int count;
+
+ rel = heap_openrv(inh, AccessShareLock);
+ /* check user requested inheritance from valid relkind */
+ if (rel->rd_rel->relkind != RELKIND_RELATION &&
+ rel->rd_rel->relkind != RELKIND_FOREIGN_TABLE &&
+ rel->rd_rel->relkind != RELKIND_PARTITIONED_TABLE)
+ ereport(ERROR,
+ (errcode(ERRCODE_WRONG_OBJECT_TYPE),
+ errmsg("inherited relation \"%s\" is not a table or foreign table",
+ inh->relname)));
+ for (count = 0; count < rel->rd_att->natts; count++)
{
- found = true;
-
- /*
- * We currently have no easy way to force an inherited
- * column to be NOT NULL at creation, if its parent
- * wasn't so already. We leave it to DefineIndex to
- * fix things up in this case.
- */
- break;
+ Form_pg_attribute inhattr = TupleDescAttr(rel->rd_att,
+ count);
+ char *inhname = NameStr(inhattr->attname);
+
+ if (inhattr->attisdropped)
+ continue;
+ if (strcmp(key, inhname) == 0)
+ {
+ found = true;
+
+ /*
+ * We currently have no easy way to force an inherited
+ * column to be NOT NULL at creation, if its parent
+ * wasn't so already. We leave it to DefineIndex to
+ * fix things up in this case.
+ */
+ break;
+ }
}
+ heap_close(rel, NoLock);
+ if (found)
+ break;
}
- heap_close(rel, NoLock);
- if (found)
- break;
}
}
@@ -2210,27 +2376,6 @@ transformIndexConstraint(Constraint *constraint, CreateStmtContext *cxt)
errmsg("column \"%s\" named in key does not exist", key),
parser_errposition(cxt->pstate, constraint->location)));
- /* Check for PRIMARY KEY(foo, foo) */
- foreach(columns, index->indexParams)
- {
- iparam = (IndexElem *) lfirst(columns);
- if (iparam->name && strcmp(key, iparam->name) == 0)
- {
- if (index->primary)
- ereport(ERROR,
- (errcode(ERRCODE_DUPLICATE_COLUMN),
- errmsg("column \"%s\" appears twice in primary key constraint",
- key),
- parser_errposition(cxt->pstate, constraint->location)));
- else
- ereport(ERROR,
- (errcode(ERRCODE_DUPLICATE_COLUMN),
- errmsg("column \"%s\" appears twice in unique constraint",
- key),
- parser_errposition(cxt->pstate, constraint->location)));
- }
- }
-
/* OK, add it to the index definition */
iparam = makeNode(IndexElem);
iparam->name = pstrdup(key);
@@ -2238,9 +2383,7 @@ transformIndexConstraint(Constraint *constraint, CreateStmtContext *cxt)
iparam->indexcolname = NULL;
iparam->collation = NIL;
iparam->opclass = NIL;
- iparam->ordering = SORTBY_DEFAULT;
- iparam->nulls_ordering = SORTBY_NULLS_DEFAULT;
- index->indexParams = lappend(index->indexParams, iparam);
+ index->indexIncludingParams = lappend(index->indexIncludingParams, iparam);
}
return index;
diff --git a/src/backend/utils/adt/ruleutils.c b/src/backend/utils/adt/ruleutils.c
index b58ee3c387..0f7a2b7296 100644
--- a/src/backend/utils/adt/ruleutils.c
+++ b/src/backend/utils/adt/ruleutils.c
@@ -1294,6 +1294,21 @@ pg_get_indexdef_worker(Oid indexrelid, int colno,
Oid keycoltype;
Oid keycolcollation;
+ /*
+ * attrsOnly flag is used for building unique-constraint and
+ * exclusion-constraint error messages. Included attrs are
+ * meaningless there, so do not include them in the message.
+ */
+ if (attrsOnly && keyno >= idxrec->indnkeyatts)
+ break;
+
+ /* Report the INCLUDED attributes, if any. */
+ if ((!attrsOnly) && keyno == idxrec->indnkeyatts)
+ {
+ appendStringInfoString(&buf, ") INCLUDE (");
+ sep = "";
+ }
+
if (!colno)
appendStringInfoString(&buf, sep);
sep = ", ";
@@ -1345,6 +1360,9 @@ pg_get_indexdef_worker(Oid indexrelid, int colno,
appendStringInfo(&buf, " COLLATE %s",
generate_collation_name((indcoll)));
+ if(keyno >= idxrec->indnkeyatts)
+ continue;
+
/* Add the operator class name, if not default */
get_opclass_name(indclass->values[keyno], keycoltype, &buf);
@@ -2045,6 +2063,19 @@ pg_get_constraintdef_worker(Oid constraintId, bool fullCommand,
appendStringInfoChar(&buf, ')');
+ /* Fetch and build including column list */
+ isnull = true;
+ val = SysCacheGetAttr(CONSTROID, tup,
+ Anum_pg_constraint_conincluding, &isnull);
+ if (!isnull)
+ {
+ appendStringInfoString(&buf, " INCLUDE (");
+
+ decompile_column_index_array(val, conForm->conrelid, &buf);
+
+ appendStringInfoChar(&buf, ')');
+ }
+
indexId = get_constraint_index(constraintId);
/* XXX why do we only print these bits if fullCommand? */
diff --git a/src/backend/utils/adt/selfuncs.c b/src/backend/utils/adt/selfuncs.c
index bf240aa9c5..fa33c8331e 100644
--- a/src/backend/utils/adt/selfuncs.c
+++ b/src/backend/utils/adt/selfuncs.c
@@ -4883,7 +4883,7 @@ examine_variable(PlannerInfo *root, Node *node, int varRelid,
* should match has_unique_index().
*/
if (index->unique &&
- index->ncolumns == 1 &&
+ index->nkeycolumns == 1 &&
(index->indpred == NIL || index->predOK))
vardata->isunique = true;
@@ -7020,7 +7020,7 @@ btcostestimate(PlannerInfo *root, IndexPath *path, double loop_count,
* NullTest invalidates that theory, even though it sets eqQualHere.
*/
if (index->unique &&
- indexcol == index->ncolumns - 1 &&
+ indexcol == index->nkeycolumns - 1 &&
eqQualHere &&
!found_saop &&
!found_is_null_op)
diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c
index 9ee78f885f..5a893ebc4a 100644
--- a/src/backend/utils/cache/relcache.c
+++ b/src/backend/utils/cache/relcache.c
@@ -533,7 +533,7 @@ RelationBuildTupleDesc(Relation relation)
/*
* add attribute data to relation->rd_att
*/
- need = relation->rd_rel->relnatts;
+ need = RelationGetNumberOfAttributes(relation);
while (HeapTupleIsValid(pg_attribute_tuple = systable_getnext(pg_attribute_scan)))
{
@@ -542,7 +542,7 @@ RelationBuildTupleDesc(Relation relation)
attp = (Form_pg_attribute) GETSTRUCT(pg_attribute_tuple);
if (attp->attnum <= 0 ||
- attp->attnum > relation->rd_rel->relnatts)
+ attp->attnum > RelationGetNumberOfAttributes(relation))
elog(ERROR, "invalid attribute number %d for %s",
attp->attnum, RelationGetRelationName(relation));
@@ -559,7 +559,7 @@ RelationBuildTupleDesc(Relation relation)
if (attrdef == NULL)
attrdef = (AttrDefault *)
MemoryContextAllocZero(CacheMemoryContext,
- relation->rd_rel->relnatts *
+ RelationGetNumberOfAttributes(relation) *
sizeof(AttrDefault));
attrdef[ndef].adnum = attp->attnum;
attrdef[ndef].adbin = NULL;
@@ -589,7 +589,7 @@ RelationBuildTupleDesc(Relation relation)
{
int i;
- for (i = 0; i < relation->rd_rel->relnatts; i++)
+ for (i = 0; i < RelationGetNumberOfAttributes(relation); i++)
Assert(TupleDescAttr(relation->rd_att, i)->attcacheoff == -1);
}
#endif
@@ -599,7 +599,7 @@ RelationBuildTupleDesc(Relation relation)
* attribute: it must be zero. This eliminates the need for special cases
* for attnum=1 that used to exist in fastgetattr() and index_getattr().
*/
- if (relation->rd_rel->relnatts > 0)
+ if (RelationGetNumberOfAttributes(relation) > 0)
TupleDescAttr(relation->rd_att, 0)->attcacheoff = 0;
/*
@@ -611,7 +611,7 @@ RelationBuildTupleDesc(Relation relation)
if (ndef > 0) /* DEFAULTs */
{
- if (ndef < relation->rd_rel->relnatts)
+ if (ndef < RelationGetNumberOfAttributes(relation))
constr->defval = (AttrDefault *)
repalloc(attrdef, ndef * sizeof(AttrDefault));
else
@@ -1490,7 +1490,8 @@ RelationInitIndexAccessInfo(Relation relation)
int2vector *indoption;
MemoryContext indexcxt;
MemoryContext oldcontext;
- int natts;
+ int indnatts;
+ int indnkeyatts;
uint16 amsupport;
/*
@@ -1520,10 +1521,11 @@ RelationInitIndexAccessInfo(Relation relation)
relation->rd_amhandler = aform->amhandler;
ReleaseSysCache(tuple);
- natts = relation->rd_rel->relnatts;
- if (natts != relation->rd_index->indnatts)
+ indnatts = RelationGetNumberOfAttributes(relation);
+ if (indnatts != IndexRelationGetNumberOfAttributes(relation))
elog(ERROR, "relnatts disagrees with indnatts for index %u",
RelationGetRelid(relation));
+ indnkeyatts = IndexRelationGetNumberOfKeyAttributes(relation);
/*
* Make the private context to hold index access info. The reason we need
@@ -1542,17 +1544,19 @@ RelationInitIndexAccessInfo(Relation relation)
InitIndexAmRoutine(relation);
/*
- * Allocate arrays to hold data
+ * Allocate arrays to hold data.
+ * Opclasses are not used for included columns, so
+ * allocate them for indnkeyatts only.
*/
relation->rd_opfamily = (Oid *)
- MemoryContextAllocZero(indexcxt, natts * sizeof(Oid));
+ MemoryContextAllocZero(indexcxt, indnkeyatts * sizeof(Oid));
relation->rd_opcintype = (Oid *)
- MemoryContextAllocZero(indexcxt, natts * sizeof(Oid));
+ MemoryContextAllocZero(indexcxt, indnkeyatts * sizeof(Oid));
amsupport = relation->rd_amroutine->amsupport;
if (amsupport > 0)
{
- int nsupport = natts * amsupport;
+ int nsupport = indnatts * amsupport;
relation->rd_support = (RegProcedure *)
MemoryContextAllocZero(indexcxt, nsupport * sizeof(RegProcedure));
@@ -1566,10 +1570,10 @@ RelationInitIndexAccessInfo(Relation relation)
}
relation->rd_indcollation = (Oid *)
- MemoryContextAllocZero(indexcxt, natts * sizeof(Oid));
+ MemoryContextAllocZero(indexcxt, indnatts * sizeof(Oid));
relation->rd_indoption = (int16 *)
- MemoryContextAllocZero(indexcxt, natts * sizeof(int16));
+ MemoryContextAllocZero(indexcxt, indnatts * sizeof(int16));
/*
* indcollation cannot be referenced directly through the C struct,
@@ -1582,7 +1586,7 @@ RelationInitIndexAccessInfo(Relation relation)
&isnull);
Assert(!isnull);
indcoll = (oidvector *) DatumGetPointer(indcollDatum);
- memcpy(relation->rd_indcollation, indcoll->values, natts * sizeof(Oid));
+ memcpy(relation->rd_indcollation, indcoll->values, indnatts * sizeof(Oid));
/*
* indclass cannot be referenced directly through the C struct, because it
@@ -1603,7 +1607,7 @@ RelationInitIndexAccessInfo(Relation relation)
*/
IndexSupportInitialize(indclass, relation->rd_support,
relation->rd_opfamily, relation->rd_opcintype,
- amsupport, natts);
+ amsupport, indnkeyatts);
/*
* Similarly extract indoption and copy it to the cache entry
@@ -1614,7 +1618,7 @@ RelationInitIndexAccessInfo(Relation relation)
&isnull);
Assert(!isnull);
indoption = (int2vector *) DatumGetPointer(indoptionDatum);
- memcpy(relation->rd_indoption, indoption->values, natts * sizeof(int16));
+ memcpy(relation->rd_indoption, indoption->values, indnatts * sizeof(int16));
/*
* expressions, predicate, exclusion caches will be filled later
@@ -4924,20 +4928,29 @@ restart:
{
int attrnum = indexInfo->ii_KeyAttrNumbers[i];
+ /*
+ * Since we have covering indexes with non-key columns,
+ * we must handle them accurately here. non-key columns
+ * must be added into indexattrs, since they are in index,
+ * and HOT-update shouldn't miss them.
+ * Obviously, non-key columns couldn't be referenced by
+ * foreign key or identity key. Hence we do not include
+ * them into uindexattrs, pkindexattrs and idindexattrs bitmaps.
+ */
if (attrnum != 0)
{
indexattrs = bms_add_member(indexattrs,
attrnum - FirstLowInvalidHeapAttributeNumber);
- if (isKey)
+ if (isKey && i < indexInfo->ii_NumIndexKeyAttrs)
uindexattrs = bms_add_member(uindexattrs,
attrnum - FirstLowInvalidHeapAttributeNumber);
- if (isPK)
+ if (isPK && i < indexInfo->ii_NumIndexKeyAttrs)
pkindexattrs = bms_add_member(pkindexattrs,
attrnum - FirstLowInvalidHeapAttributeNumber);
- if (isIDKey)
+ if (isIDKey && i < indexInfo->ii_NumIndexKeyAttrs)
idindexattrs = bms_add_member(idindexattrs,
attrnum - FirstLowInvalidHeapAttributeNumber);
}
@@ -5036,7 +5049,7 @@ RelationGetExclusionInfo(Relation indexRelation,
Oid **procs,
uint16 **strategies)
{
- int ncols = indexRelation->rd_rel->relnatts;
+ int indnkeyatts;
Oid *ops;
Oid *funcs;
uint16 *strats;
@@ -5048,17 +5061,19 @@ RelationGetExclusionInfo(Relation indexRelation,
MemoryContext oldcxt;
int i;
+ indnkeyatts = IndexRelationGetNumberOfKeyAttributes(indexRelation);
+
/* Allocate result space in caller context */
- *operators = ops = (Oid *) palloc(sizeof(Oid) * ncols);
- *procs = funcs = (Oid *) palloc(sizeof(Oid) * ncols);
- *strategies = strats = (uint16 *) palloc(sizeof(uint16) * ncols);
+ *operators = ops = (Oid *) palloc(sizeof(Oid) * indnkeyatts);
+ *procs = funcs = (Oid *) palloc(sizeof(Oid) * indnkeyatts);
+ *strategies = strats = (uint16 *) palloc(sizeof(uint16) * indnkeyatts);
/* Quick exit if we have the data cached already */
if (indexRelation->rd_exclstrats != NULL)
{
- memcpy(ops, indexRelation->rd_exclops, sizeof(Oid) * ncols);
- memcpy(funcs, indexRelation->rd_exclprocs, sizeof(Oid) * ncols);
- memcpy(strats, indexRelation->rd_exclstrats, sizeof(uint16) * ncols);
+ memcpy(ops, indexRelation->rd_exclops, sizeof(Oid) * indnkeyatts);
+ memcpy(funcs, indexRelation->rd_exclprocs, sizeof(Oid) * indnkeyatts);
+ memcpy(strats, indexRelation->rd_exclstrats, sizeof(uint16) * indnkeyatts);
return;
}
@@ -5107,12 +5122,12 @@ RelationGetExclusionInfo(Relation indexRelation,
arr = DatumGetArrayTypeP(val); /* ensure not toasted */
nelem = ARR_DIMS(arr)[0];
if (ARR_NDIM(arr) != 1 ||
- nelem != ncols ||
+ nelem != indnkeyatts ||
ARR_HASNULL(arr) ||
ARR_ELEMTYPE(arr) != OIDOID)
elog(ERROR, "conexclop is not a 1-D Oid array");
- memcpy(ops, ARR_DATA_PTR(arr), sizeof(Oid) * ncols);
+ memcpy(ops, ARR_DATA_PTR(arr), sizeof(Oid) * indnkeyatts);
}
systable_endscan(conscan);
@@ -5123,7 +5138,7 @@ RelationGetExclusionInfo(Relation indexRelation,
RelationGetRelationName(indexRelation));
/* We need the func OIDs and strategy numbers too */
- for (i = 0; i < ncols; i++)
+ for (i = 0; i < indnkeyatts; i++)
{
funcs[i] = get_opcode(ops[i]);
strats[i] = get_op_opfamily_strategy(ops[i],
@@ -5136,12 +5151,12 @@ RelationGetExclusionInfo(Relation indexRelation,
/* Save a copy of the results in the relcache entry. */
oldcxt = MemoryContextSwitchTo(indexRelation->rd_indexcxt);
- indexRelation->rd_exclops = (Oid *) palloc(sizeof(Oid) * ncols);
- indexRelation->rd_exclprocs = (Oid *) palloc(sizeof(Oid) * ncols);
- indexRelation->rd_exclstrats = (uint16 *) palloc(sizeof(uint16) * ncols);
- memcpy(indexRelation->rd_exclops, ops, sizeof(Oid) * ncols);
- memcpy(indexRelation->rd_exclprocs, funcs, sizeof(Oid) * ncols);
- memcpy(indexRelation->rd_exclstrats, strats, sizeof(uint16) * ncols);
+ indexRelation->rd_exclops = (Oid *) palloc(sizeof(Oid) * indnkeyatts);
+ indexRelation->rd_exclprocs = (Oid *) palloc(sizeof(Oid) * indnkeyatts);
+ indexRelation->rd_exclstrats = (uint16 *) palloc(sizeof(uint16) * indnkeyatts);
+ memcpy(indexRelation->rd_exclops, ops, sizeof(Oid) * indnkeyatts);
+ memcpy(indexRelation->rd_exclprocs, funcs, sizeof(Oid) * indnkeyatts);
+ memcpy(indexRelation->rd_exclstrats, strats, sizeof(uint16) * indnkeyatts);
MemoryContextSwitchTo(oldcxt);
}
diff --git a/src/backend/utils/sort/tuplesort.c b/src/backend/utils/sort/tuplesort.c
index 041bdc2fa7..dd1e7882e5 100644
--- a/src/backend/utils/sort/tuplesort.c
+++ b/src/backend/utils/sort/tuplesort.c
@@ -900,7 +900,7 @@ tuplesort_begin_cluster(TupleDesc tupDesc,
workMem, randomAccess ? 't' : 'f');
#endif
- state->nKeys = RelationGetNumberOfAttributes(indexRel);
+ state->nKeys = IndexRelationGetNumberOfKeyAttributes(indexRel);
TRACE_POSTGRESQL_SORT_START(CLUSTER_SORT,
false, /* no unique check */
@@ -995,7 +995,7 @@ tuplesort_begin_index_btree(Relation heapRel,
workMem, randomAccess ? 't' : 'f');
#endif
- state->nKeys = RelationGetNumberOfAttributes(indexRel);
+ state->nKeys = IndexRelationGetNumberOfKeyAttributes(indexRel);
TRACE_POSTGRESQL_SORT_START(INDEX_SORT,
enforceUnique,
@@ -1015,7 +1015,6 @@ tuplesort_begin_index_btree(Relation heapRel,
state->enforceUnique = enforceUnique;
indexScanKey = _bt_mkscankey_nodata(indexRel);
- state->nKeys = RelationGetNumberOfAttributes(indexRel);
/* Prepare SortSupport data for each column */
state->sortKeys = (SortSupport) palloc0(state->nKeys *
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index 566cbf2cda..dd2c0c74c5 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -6705,7 +6705,8 @@ getIndexes(Archive *fout, TableInfo tblinfo[], int numTables)
i_indexname,
i_parentidx,
i_indexdef,
- i_indnkeys,
+ i_indnnkeyatts,
+ i_indnatts,
i_indkey,
i_indisclustered,
i_indisreplident,
@@ -6762,6 +6763,8 @@ getIndexes(Archive *fout, TableInfo tblinfo[], int numTables)
"t.relname AS indexname, "
"inh.inhparent AS parentidx, "
"pg_catalog.pg_get_indexdef(i.indexrelid) AS indexdef, "
+ "i.indnkeyatts AS indnkeyatts, "
+ "i.indnatts AS indnatts, "
"t.relnatts AS indnkeys, "
"i.indkey, i.indisclustered, "
"i.indisreplident, t.relpages, "
@@ -6798,6 +6801,8 @@ getIndexes(Archive *fout, TableInfo tblinfo[], int numTables)
"t.relname AS indexname, "
"0 AS parentidx, "
"pg_catalog.pg_get_indexdef(i.indexrelid) AS indexdef, "
+ "i.indnkeyatts AS indnkeyatts, "
+ "i.indnatts AS indnatts, "
"t.relnatts AS indnkeys, "
"i.indkey, i.indisclustered, "
"i.indisreplident, t.relpages, "
@@ -6830,6 +6835,8 @@ getIndexes(Archive *fout, TableInfo tblinfo[], int numTables)
"t.relname AS indexname, "
"0 AS parentidx, "
"pg_catalog.pg_get_indexdef(i.indexrelid) AS indexdef, "
+ "i.indnkeyatts AS indnkeyatts, "
+ "i.indnatts AS indnatts, "
"t.relnatts AS indnkeys, "
"i.indkey, i.indisclustered, "
"false AS indisreplident, t.relpages, "
@@ -6858,6 +6865,8 @@ getIndexes(Archive *fout, TableInfo tblinfo[], int numTables)
"t.relname AS indexname, "
"0 AS parentidx, "
"pg_catalog.pg_get_indexdef(i.indexrelid) AS indexdef, "
+ "i.indnkeyatts AS indnkeyatts, "
+ "i.indnatts AS indnatts, "
"t.relnatts AS indnkeys, "
"i.indkey, i.indisclustered, "
"false AS indisreplident, t.relpages, "
@@ -6922,7 +6931,8 @@ getIndexes(Archive *fout, TableInfo tblinfo[], int numTables)
i_indexname = PQfnumber(res, "indexname");
i_parentidx = PQfnumber(res, "parentidx");
i_indexdef = PQfnumber(res, "indexdef");
- i_indnkeys = PQfnumber(res, "indnkeys");
+ i_indnnkeyatts = PQfnumber(res, "indnkeyatts");
+ i_indnatts = PQfnumber(res, "indnatts");
i_indkey = PQfnumber(res, "indkey");
i_indisclustered = PQfnumber(res, "indisclustered");
i_indisreplident = PQfnumber(res, "indisreplident");
@@ -6955,12 +6965,13 @@ getIndexes(Archive *fout, TableInfo tblinfo[], int numTables)
indxinfo[j].dobj.namespace = tbinfo->dobj.namespace;
indxinfo[j].indextable = tbinfo;
indxinfo[j].indexdef = pg_strdup(PQgetvalue(res, j, i_indexdef));
- indxinfo[j].indnkeys = atoi(PQgetvalue(res, j, i_indnkeys));
+ indxinfo[j].indnkeyattrs = atoi(PQgetvalue(res, j, i_indnnkeyatts));
+ indxinfo[j].indnattrs = atoi(PQgetvalue(res, j, i_indnatts));
indxinfo[j].tablespace = pg_strdup(PQgetvalue(res, j, i_tablespace));
indxinfo[j].indreloptions = pg_strdup(PQgetvalue(res, j, i_indreloptions));
- indxinfo[j].indkeys = (Oid *) pg_malloc(indxinfo[j].indnkeys * sizeof(Oid));
+ indxinfo[j].indkeys = (Oid *) pg_malloc(indxinfo[j].indnattrs * sizeof(Oid));
parseOidArray(PQgetvalue(res, j, i_indkey),
- indxinfo[j].indkeys, indxinfo[j].indnkeys);
+ indxinfo[j].indkeys, indxinfo[j].indnattrs);
indxinfo[j].indisclustered = (PQgetvalue(res, j, i_indisclustered)[0] == 't');
indxinfo[j].indisreplident = (PQgetvalue(res, j, i_indisreplident)[0] == 't');
indxinfo[j].parentidx = atooid(PQgetvalue(res, j, i_parentidx));
@@ -16307,7 +16318,7 @@ dumpConstraint(Archive *fout, ConstraintInfo *coninfo)
{
appendPQExpBuffer(q, "%s (",
coninfo->contype == 'p' ? "PRIMARY KEY" : "UNIQUE");
- for (k = 0; k < indxinfo->indnkeys; k++)
+ for (k = 0; k < indxinfo->indnkeyattrs; k++)
{
int indkey = (int) indxinfo->indkeys[k];
const char *attname;
@@ -16321,6 +16332,23 @@ dumpConstraint(Archive *fout, ConstraintInfo *coninfo)
fmtId(attname));
}
+ if (indxinfo->indnkeyattrs < indxinfo->indnattrs)
+ appendPQExpBuffer(q, ") INCLUDE (");
+
+ for (k = indxinfo->indnkeyattrs; k < indxinfo->indnattrs; k++)
+ {
+ int indkey = (int) indxinfo->indkeys[k];
+ const char *attname;
+
+ if (indkey == InvalidAttrNumber)
+ break;
+ attname = getAttrName(indkey, tbinfo);
+
+ appendPQExpBuffer(q, "%s%s",
+ (k == indxinfo->indnkeyattrs) ? "" : ", ",
+ fmtId(attname));
+ }
+
appendPQExpBufferChar(q, ')');
if (nonemptyReloptions(indxinfo->indreloptions))
diff --git a/src/bin/pg_dump/pg_dump.h b/src/bin/pg_dump/pg_dump.h
index a4d6d926a8..d59591f389 100644
--- a/src/bin/pg_dump/pg_dump.h
+++ b/src/bin/pg_dump/pg_dump.h
@@ -360,8 +360,10 @@ typedef struct _indxInfo
char *indexdef;
char *tablespace; /* tablespace in which index is stored */
char *indreloptions; /* options specified by WITH (...) */
- int indnkeys;
- Oid *indkeys;
+ int indnkeyattrs; /* number of index key attributes */
+ int indnattrs; /* total number of index attributes */
+ Oid *indkeys; /* In spite of the name 'indkeys' this field
+ * contains both key and nonkey attributes */
bool indisclustered;
bool indisreplident;
Oid parentidx; /* if partitioned, parent index OID */
diff --git a/src/include/access/amapi.h b/src/include/access/amapi.h
index 8d7bc246e6..d16fa6823b 100644
--- a/src/include/access/amapi.h
+++ b/src/include/access/amapi.h
@@ -191,6 +191,8 @@ typedef struct IndexAmRoutine
bool ampredlocks;
/* does AM support parallel scan? */
bool amcanparallel;
+ /* does AM support columns included with clause INCLUDE? */
+ bool amcaninclude;
/* type of data stored in index, or InvalidOid if variable */
Oid amkeytype;
diff --git a/src/include/access/itup.h b/src/include/access/itup.h
index 9be3442c66..82cf980fc1 100644
--- a/src/include/access/itup.h
+++ b/src/include/access/itup.h
@@ -18,6 +18,7 @@
#include "access/tupmacs.h"
#include "storage/bufpage.h"
#include "storage/itemptr.h"
+#include "utils/rel.h"
/*
* Index tuple header structure
@@ -146,5 +147,6 @@ extern Datum nocache_index_getattr(IndexTuple tup, int attnum,
extern void index_deform_tuple(IndexTuple tup, TupleDesc tupleDescriptor,
Datum *values, bool *isnull);
extern IndexTuple CopyIndexTuple(IndexTuple source);
+extern IndexTuple index_truncate_tuple(Relation idxrel, IndexTuple olditup);
#endif /* ITUP_H */
diff --git a/src/include/catalog/pg_constraint.h b/src/include/catalog/pg_constraint.h
index 8fca86d71e..3c0331c5c4 100644
--- a/src/include/catalog/pg_constraint.h
+++ b/src/include/catalog/pg_constraint.h
@@ -98,6 +98,12 @@ CATALOG(pg_constraint,2606)
*/
int16 conkey[1];
+ /*
+ * Columns of conrelid that the constraint does not apply to,
+ * but included into the same index with key columns.
+ */
+ int16 conincluding[1];
+
/*
* If a foreign key, the referenced columns of confrelid
*/
@@ -150,7 +156,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
@@ -168,13 +174,14 @@ typedef FormData_pg_constraint *Form_pg_constraint;
#define Anum_pg_constraint_coninhcount 15
#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_conincluding 18
+#define Anum_pg_constraint_confkey 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
diff --git a/src/include/catalog/pg_constraint_fn.h b/src/include/catalog/pg_constraint_fn.h
index d3351f4a83..b937f5042c 100644
--- a/src/include/catalog/pg_constraint_fn.h
+++ b/src/include/catalog/pg_constraint_fn.h
@@ -27,30 +27,31 @@ typedef enum ConstraintCategory
CONSTRAINT_ASSERTION /* for future expansion */
} ConstraintCategory;
-extern Oid CreateConstraintEntry(const char *constraintName,
+extern Oid CreateConstraintEntry(const char* constraintName,
Oid constraintNamespace,
char constraintType,
bool isDeferrable,
bool isDeferred,
bool isValidated,
Oid relId,
- const int16 *constraintKey,
+ const int16* constraintKey,
int constraintNKeys,
+ int constraintNTotalKeys,
Oid domainId,
Oid indexRelId,
Oid foreignRelId,
- const int16 *foreignKey,
- const Oid *pfEqOp,
- const Oid *ppEqOp,
- const Oid *ffEqOp,
+ const int16* foreignKey,
+ const Oid* pfEqOp,
+ const Oid* ppEqOp,
+ const Oid* ffEqOp,
int foreignNKeys,
char foreignUpdateType,
char foreignDeleteType,
char foreignMatchType,
- const Oid *exclOp,
- Node *conExpr,
- const char *conBin,
- const char *conSrc,
+ const Oid* exclOp,
+ Node* conExpr,
+ const char* conBin,
+ const char* conSrc,
bool conIsLocal,
int conInhCount,
bool conNoInherit,
diff --git a/src/include/catalog/pg_index.h b/src/include/catalog/pg_index.h
index 057a9f7fe4..6ae03dbcbb 100644
--- a/src/include/catalog/pg_index.h
+++ b/src/include/catalog/pg_index.h
@@ -32,7 +32,8 @@ CATALOG(pg_index,2610) BKI_WITHOUT_OIDS BKI_SCHEMA_MACRO
{
Oid indexrelid; /* OID of the index */
Oid indrelid; /* OID of the relation it indexes */
- int16 indnatts; /* number of columns in index */
+ int16 indnatts; /* total number of columns in index */
+ int16 indnkeyatts; /* number of key columns in index */
bool indisunique; /* is this a unique index? */
bool indisprimary; /* is this index for primary key? */
bool indisexclusion; /* is this index for exclusion constraint? */
@@ -70,26 +71,27 @@ typedef FormData_pg_index *Form_pg_index;
* compiler constants for pg_index
* ----------------
*/
-#define Natts_pg_index 19
+#define Natts_pg_index 20
#define Anum_pg_index_indexrelid 1
#define Anum_pg_index_indrelid 2
#define Anum_pg_index_indnatts 3
-#define Anum_pg_index_indisunique 4
-#define Anum_pg_index_indisprimary 5
-#define Anum_pg_index_indisexclusion 6
-#define Anum_pg_index_indimmediate 7
-#define Anum_pg_index_indisclustered 8
-#define Anum_pg_index_indisvalid 9
-#define Anum_pg_index_indcheckxmin 10
-#define Anum_pg_index_indisready 11
-#define Anum_pg_index_indislive 12
-#define Anum_pg_index_indisreplident 13
-#define Anum_pg_index_indkey 14
-#define Anum_pg_index_indcollation 15
-#define Anum_pg_index_indclass 16
-#define Anum_pg_index_indoption 17
-#define Anum_pg_index_indexprs 18
-#define Anum_pg_index_indpred 19
+#define Anum_pg_index_indnkeyatts 4
+#define Anum_pg_index_indisunique 5
+#define Anum_pg_index_indisprimary 6
+#define Anum_pg_index_indisexclusion 7
+#define Anum_pg_index_indimmediate 8
+#define Anum_pg_index_indisclustered 9
+#define Anum_pg_index_indisvalid 10
+#define Anum_pg_index_indcheckxmin 11
+#define Anum_pg_index_indisready 12
+#define Anum_pg_index_indislive 13
+#define Anum_pg_index_indisreplident 14
+#define Anum_pg_index_indkey 15
+#define Anum_pg_index_indcollation 16
+#define Anum_pg_index_indclass 17
+#define Anum_pg_index_indoption 18
+#define Anum_pg_index_indexprs 19
+#define Anum_pg_index_indpred 20
/*
* Index AMs that support ordered scans must support these two indoption
diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h
index a953820f43..1b4d25ef0f 100644
--- a/src/include/nodes/execnodes.h
+++ b/src/include/nodes/execnodes.h
@@ -115,9 +115,11 @@ typedef struct ExprState
* entries for a particular index. Used for both index_build and
* retail creation of index entries.
*
- * NumIndexAttrs number of columns in this index
+ * NumIndexAttrs total number of columns in this index
+ * NumIndexKeyAttrs number of key columns in index
* KeyAttrNumbers underlying-rel attribute numbers used as keys
- * (zeroes indicate expressions)
+ * (zeroes indicate expressions). It also contains
+ * info about included columns.
* Expressions expr trees for expression entries, or NIL if none
* ExpressionsState exec state for expressions, or NIL if none
* Predicate partial-index predicate, or NIL if none
@@ -143,7 +145,8 @@ typedef struct ExprState
typedef struct IndexInfo
{
NodeTag type;
- int ii_NumIndexAttrs;
+ int ii_NumIndexAttrs; /* total number of columns in index */
+ int ii_NumIndexKeyAttrs; /* number of key columns in index */
AttrNumber ii_KeyAttrNumbers[INDEX_MAX_KEYS];
List *ii_Expressions; /* list of Expr */
List *ii_ExpressionsState; /* list of ExprState */
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index f668cbad34..c179ea6f88 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -2102,7 +2102,8 @@ typedef struct Constraint
char generated_when;
/* Fields used for unique constraints (UNIQUE and PRIMARY KEY): */
- List *keys; /* String nodes naming referenced column(s) */
+ List *keys; /* String nodes naming referenced key column(s) */
+ List *including; /* String nodes naming referenced nonkey column(s) */
/* Fields used for EXCLUSION constraints: */
List *exclusions; /* list of (IndexElem, operator name) pairs */
@@ -2715,6 +2716,8 @@ typedef struct IndexStmt
char *accessMethod; /* name of access method (eg. btree) */
char *tableSpace; /* tablespace, or NULL for default */
List *indexParams; /* columns to index: a list of IndexElem */
+ List *indexIncludingParams; /* additional columns to index:
+ * a list of IndexElem */
List *options; /* WITH clause options: a list of DefElem */
Node *whereClause; /* qualification (partial-index predicate) */
List *excludeOpNames; /* exclusion operator names, or NIL if none */
diff --git a/src/include/nodes/relation.h b/src/include/nodes/relation.h
index d576aa7350..7186e35e68 100644
--- a/src/include/nodes/relation.h
+++ b/src/include/nodes/relation.h
@@ -692,11 +692,12 @@ typedef struct RelOptInfo
* IndexOptInfo
* Per-index information for planning/optimization
*
- * indexkeys[], indexcollations[], opfamily[], and opcintype[]
- * each have ncolumns entries.
+ * indexkeys[], indexcollations[] each have ncolumns entries.
+ * opfamily[], and opcintype[] each have nkeycolumns entries. They do
+ * not contain any information about included attributes.
*
- * sortopfamily[], reverse_sort[], and nulls_first[] likewise have
- * ncolumns entries, if the index is ordered; but if it is unordered,
+ * sortopfamily[], reverse_sort[], and nulls_first[] have
+ * nkeycolumns entries, if the index is ordered; but if it is unordered,
* those pointers are NULL.
*
* Zeroes in the indexkeys[] array indicate index columns that are
@@ -733,7 +734,9 @@ typedef struct IndexOptInfo
/* index descriptor information */
int ncolumns; /* number of columns in index */
- int *indexkeys; /* column numbers of index's keys, or 0 */
+ int nkeycolumns; /* number of key columns in index */
+ int *indexkeys; /* column numbers of index's attributes
+ * both key and included columns, or 0 */
Oid *indexcollations; /* OIDs of collations of index columns */
Oid *opfamily; /* OIDs of operator families for columns */
Oid *opcintype; /* OIDs of opclass declared input data types */
diff --git a/src/include/parser/kwlist.h b/src/include/parser/kwlist.h
index cf32197bc3..23db40147b 100644
--- a/src/include/parser/kwlist.h
+++ b/src/include/parser/kwlist.h
@@ -196,6 +196,7 @@ PG_KEYWORD("immutable", IMMUTABLE, UNRESERVED_KEYWORD)
PG_KEYWORD("implicit", IMPLICIT_P, UNRESERVED_KEYWORD)
PG_KEYWORD("import", IMPORT_P, UNRESERVED_KEYWORD)
PG_KEYWORD("in", IN_P, RESERVED_KEYWORD)
+PG_KEYWORD("include", INCLUDE, UNRESERVED_KEYWORD)
PG_KEYWORD("including", INCLUDING, UNRESERVED_KEYWORD)
PG_KEYWORD("increment", INCREMENT, UNRESERVED_KEYWORD)
PG_KEYWORD("index", INDEX, UNRESERVED_KEYWORD)
diff --git a/src/include/utils/rel.h b/src/include/utils/rel.h
index aa8add544a..da9f77e91a 100644
--- a/src/include/utils/rel.h
+++ b/src/include/utils/rel.h
@@ -426,10 +426,24 @@ typedef struct ViewOptions
/*
* RelationGetNumberOfAttributes
- * Returns the number of attributes in a relation.
+ * Returns the total number of attributes in a relation.
*/
#define RelationGetNumberOfAttributes(relation) ((relation)->rd_rel->relnatts)
+/*
+ * IndexRelationGetNumberOfAttributes
+ * Returns the number of attributes in an index.
+ */
+#define IndexRelationGetNumberOfAttributes(relation) \
+ ((relation)->rd_index->indnatts)
+
+/*
+ * IndexRelationGetNumberOfKeyAttributes
+ * Returns the number of key attributes in an index.
+ */
+#define IndexRelationGetNumberOfKeyAttributes(relation) \
+ ((relation)->rd_index->indnkeyatts)
+
/*
* RelationGetDescr
* Returns tuple descriptor for a relation.
--
2.14.3 (Apple Git-98)
[text/x-patch] 0002-Covering-btree-v5.patch (46.5K, ../78ca1586-647c-e159-ebf3-30957e270c66@postgrespro.ru/3-0002-Covering-btree-v5.patch)
download | inline diff:
From 0b9d210f3291a5b20fa9d61b4f58d3fa5a21639b Mon Sep 17 00:00:00 2001
From: Andrey Borodin <amborodin@acm.org>
Date: Thu, 8 Mar 2018 15:53:51 +0500
Subject: [PATCH 2/3] Covering btree v5
---
src/backend/access/nbtree/README | 65 ++++++
src/backend/access/nbtree/nbtinsert.c | 68 +++---
src/backend/access/nbtree/nbtpage.c | 5 +-
src/backend/access/nbtree/nbtree.c | 2 +-
src/backend/access/nbtree/nbtsort.c | 42 +++-
src/backend/access/nbtree/nbtutils.c | 25 +-
src/backend/access/nbtree/nbtxlog.c | 12 +-
src/include/access/nbtree.h | 2 +
src/test/regress/expected/create_index.out | 19 ++
src/test/regress/expected/index_including.out | 320 ++++++++++++++++++++++++++
src/test/regress/parallel_schedule | 2 +-
src/test/regress/serial_schedule | 1 +
src/test/regress/sql/create_index.sql | 20 ++
src/test/regress/sql/index_including.sql | 189 +++++++++++++++
14 files changed, 725 insertions(+), 47 deletions(-)
create mode 100644 src/test/regress/expected/index_including.out
create mode 100644 src/test/regress/sql/index_including.sql
diff --git a/src/backend/access/nbtree/README b/src/backend/access/nbtree/README
index 34f78b2f50..f0657e4244 100644
--- a/src/backend/access/nbtree/README
+++ b/src/backend/access/nbtree/README
@@ -623,3 +623,68 @@ routines must treat it accordingly. The actual key stored in the
item is irrelevant, and need not be stored at all. This arrangement
corresponds to the fact that an L&Y non-leaf page has one more pointer
than key.
+
+Notes to Operator Class Implementors
+------------------------------------
+
+With this implementation, we require each supported combination of
+datatypes to supply us with a comparison procedure via pg_amproc.
+This procedure must take two nonnull values A and B and return an int32 < 0,
+0, or > 0 if A < B, A = B, or A > B, respectively. The procedure must
+not return INT_MIN for "A < B", since the value may be negated before
+being tested for sign. A null result is disallowed, too. See nbtcompare.c
+for examples.
+
+There are some basic assumptions that a btree operator family must satisfy:
+
+An = operator must be an equivalence relation; that is, for all non-null
+values A,B,C of the datatype:
+
+ A = A is true reflexive law
+ if A = B, then B = A symmetric law
+ if A = B and B = C, then A = C transitive law
+
+A < operator must be a strong ordering relation; that is, for all non-null
+values A,B,C:
+
+ A < A is false irreflexive law
+ if A < B and B < C, then A < C transitive law
+
+Furthermore, the ordering is total; that is, for all non-null values A,B:
+
+ exactly one of A < B, A = B, and B < A is true trichotomy law
+
+(The trichotomy law justifies the definition of the comparison support
+procedure, of course.)
+
+The other three operators are defined in terms of these two in the obvious way,
+and must act consistently with them.
+
+For an operator family supporting multiple datatypes, the above laws must hold
+when A,B,C are taken from any datatypes in the family. The transitive laws
+are the trickiest to ensure, as in cross-type situations they represent
+statements that the behaviors of two or three different operators are
+consistent. As an example, it would not work to put float8 and numeric into
+an opfamily, at least not with the current semantics that numerics are
+converted to float8 for comparison to a float8. Because of the limited
+accuracy of float8, this means there are distinct numeric values that will
+compare equal to the same float8 value, and thus the transitive law fails.
+
+It should be fairly clear why a btree index requires these laws to hold within
+a single datatype: without them there is no ordering to arrange the keys with.
+Also, index searches using a key of a different datatype require comparisons
+to behave sanely across two datatypes. The extensions to three or more
+datatypes within a family are not strictly required by the btree index
+mechanism itself, but the planner relies on them for optimization purposes.
+
+Included attributes in B-tree indexes
+-------------------------------------
+
+Since 11.0 there is an optional INCLUDE clause, that allows to add
+a portion of non-key attributes to index. They exist to allow more queries
+to benefit from index-only scans. We never use included attributes in
+ScanKeys, neither for search nor for inserts. That allows us to include
+into B-tree any datatypes, even those which don't have suitable opclass.
+Included columns only stored in regular items on leaf pages. All inner
+keys and high keys are truncated and contain only key attributes.
+That helps to reduce the size of index.
diff --git a/src/backend/access/nbtree/nbtinsert.c b/src/backend/access/nbtree/nbtinsert.c
index 2fe9867353..7f5f198279 100644
--- a/src/backend/access/nbtree/nbtinsert.c
+++ b/src/backend/access/nbtree/nbtinsert.c
@@ -79,8 +79,6 @@ static OffsetNumber _bt_findsplitloc(Relation rel, Page page,
static void _bt_checksplitloc(FindSplitData *state,
OffsetNumber firstoldonright, bool newitemonleft,
int dataitemstoleft, Size firstoldonrightsz);
-static bool _bt_pgaddtup(Page page, Size itemsize, IndexTuple itup,
- OffsetNumber itup_off);
static bool _bt_isequal(TupleDesc itupdesc, Page page, OffsetNumber offnum,
int keysz, ScanKey scankey);
static void _bt_vacuum_one_page(Relation rel, Buffer buffer, Relation heapRel);
@@ -109,18 +107,22 @@ _bt_doinsert(Relation rel, IndexTuple itup,
IndexUniqueCheck checkUnique, Relation heapRel)
{
bool is_unique = false;
- int natts = rel->rd_rel->relnatts;
+ int indnkeyatts;
ScanKey itup_scankey;
BTStack stack;
Buffer buf;
OffsetNumber offset;
+ Assert(IndexRelationGetNumberOfAttributes(rel) != 0);
+ indnkeyatts = IndexRelationGetNumberOfKeyAttributes(rel);
+ Assert(indnkeyatts != 0);
+
/* we need an insertion scan key to do our search, so build one */
itup_scankey = _bt_mkscankey(rel, itup);
top:
/* find the first page containing this key */
- stack = _bt_search(rel, natts, itup_scankey, false, &buf, BT_WRITE, NULL);
+ stack = _bt_search(rel, indnkeyatts, itup_scankey, false, &buf, BT_WRITE, NULL);
offset = InvalidOffsetNumber;
@@ -135,7 +137,7 @@ top:
* move right in the tree. See Lehman and Yao for an excruciatingly
* precise description.
*/
- buf = _bt_moveright(rel, buf, natts, itup_scankey, false,
+ buf = _bt_moveright(rel, buf, indnkeyatts, itup_scankey, false,
true, stack, BT_WRITE, NULL);
/*
@@ -164,7 +166,7 @@ top:
TransactionId xwait;
uint32 speculativeToken;
- offset = _bt_binsrch(rel, buf, natts, itup_scankey, false);
+ offset = _bt_binsrch(rel, buf, indnkeyatts, itup_scankey, false);
xwait = _bt_check_unique(rel, itup, heapRel, buf, offset, itup_scankey,
checkUnique, &is_unique, &speculativeToken);
@@ -200,7 +202,7 @@ top:
*/
CheckForSerializableConflictIn(rel, NULL, buf);
/* do the insertion */
- _bt_findinsertloc(rel, &buf, &offset, natts, itup_scankey, itup,
+ _bt_findinsertloc(rel, &buf, &offset, indnkeyatts, itup_scankey, itup,
stack, heapRel);
_bt_insertonpg(rel, buf, InvalidBuffer, stack, itup, offset, false);
}
@@ -243,7 +245,7 @@ _bt_check_unique(Relation rel, IndexTuple itup, Relation heapRel,
uint32 *speculativeToken)
{
TupleDesc itupdesc = RelationGetDescr(rel);
- int natts = rel->rd_rel->relnatts;
+ int indnkeyatts = IndexRelationGetNumberOfKeyAttributes(rel);
SnapshotData SnapshotDirty;
OffsetNumber maxoff;
Page page;
@@ -302,7 +304,7 @@ _bt_check_unique(Relation rel, IndexTuple itup, Relation heapRel,
* in real comparison, but only for ordering/finding items on
* pages. - vadim 03/24/97
*/
- if (!_bt_isequal(itupdesc, page, offset, natts, itup_scankey))
+ if (!_bt_isequal(itupdesc, page, offset, indnkeyatts, itup_scankey))
break; /* we're past all the equal tuples */
/* okay, we gotta fetch the heap tuple ... */
@@ -467,7 +469,7 @@ _bt_check_unique(Relation rel, IndexTuple itup, Relation heapRel,
if (P_RIGHTMOST(opaque))
break;
if (!_bt_isequal(itupdesc, page, P_HIKEY,
- natts, itup_scankey))
+ indnkeyatts, itup_scankey))
break;
/* Advance to next non-dead page --- there must be one */
for (;;)
@@ -981,6 +983,9 @@ _bt_split(Relation rel, Buffer buf, Buffer cbuf, OffsetNumber firstright,
OffsetNumber maxoff;
OffsetNumber i;
bool isleaf;
+ IndexTuple lefthikey;
+ int indnatts = IndexRelationGetNumberOfAttributes(rel);
+ int indnkeyatts = IndexRelationGetNumberOfKeyAttributes(rel);
/* Acquire a new page to split into */
rbuf = _bt_getbuf(rel, P_NEW, BT_WRITE);
@@ -1080,7 +1085,22 @@ _bt_split(Relation rel, Buffer buf, Buffer cbuf, OffsetNumber firstright,
itemsz = ItemIdGetLength(itemid);
item = (IndexTuple) PageGetItem(origpage, itemid);
}
- if (PageAddItem(leftpage, (Item) item, itemsz, leftoff,
+
+ /*
+ * We must truncate the "high key" item, before insert it onto the leaf page.
+ * It's the only point in insertion process, where we perform truncation.
+ * All other functions work with this high key and do not change it.
+ */
+ if (indnatts != indnkeyatts && P_ISLEAF(lopaque))
+ {
+ lefthikey = index_truncate_tuple(rel, item);
+ itemsz = IndexTupleSize(lefthikey);
+ itemsz = MAXALIGN(itemsz);
+ }
+ else
+ lefthikey = item;
+
+ if (PageAddItem(leftpage, (Item) lefthikey, itemsz, leftoff,
false, false) == InvalidOffsetNumber)
{
memset(rightpage, 0, BufferGetPageSize(rbuf));
@@ -1297,20 +1317,16 @@ _bt_split(Relation rel, Buffer buf, Buffer cbuf, OffsetNumber firstright,
if (newitemonleft)
XLogRegisterBufData(0, (char *) newitem, MAXALIGN(newitemsz));
- /* Log left page */
- if (!isleaf)
- {
- /*
- * We must also log the left page's high key, because the right
- * page's leftmost key is suppressed on non-leaf levels. Show it
- * as belonging to the left page buffer, so that it is not stored
- * if XLogInsert decides it needs a full-page image of the left
- * page.
- */
- itemid = PageGetItemId(origpage, P_HIKEY);
- item = (IndexTuple) PageGetItem(origpage, itemid);
- XLogRegisterBufData(0, (char *) item, MAXALIGN(IndexTupleSize(item)));
- }
+ /*
+ * We must also log the left page's high key, because the right
+ * page's leftmost key is suppressed on non-leaf levels. Show it
+ * as belonging to the left page buffer, so that it is not stored
+ * if XLogInsert decides it needs a full-page image of the left
+ * page.
+ */
+ itemid = PageGetItemId(origpage, P_HIKEY);
+ item = (IndexTuple) PageGetItem(origpage, itemid);
+ XLogRegisterBufData(0, (char *) item, MAXALIGN(IndexTupleSize(item)));
/*
* Log the contents of the right page in the format understood by
@@ -2083,7 +2099,7 @@ _bt_newroot(Relation rel, Buffer lbuf, Buffer rbuf)
* we insert the tuples in order, so that the given itup_off does
* represent the final position of the tuple!
*/
-static bool
+bool
_bt_pgaddtup(Page page,
Size itemsize,
IndexTuple itup,
diff --git a/src/backend/access/nbtree/nbtpage.c b/src/backend/access/nbtree/nbtpage.c
index 92afe2de38..e6bfb18e7b 100644
--- a/src/backend/access/nbtree/nbtpage.c
+++ b/src/backend/access/nbtree/nbtpage.c
@@ -1256,8 +1256,9 @@ _bt_pagedel(Relation rel, Buffer buf)
/* we need an insertion scan key for the search, so build one */
itup_scankey = _bt_mkscankey(rel, targetkey);
/* find the leftmost leaf page containing this key */
- stack = _bt_search(rel, rel->rd_rel->relnatts, itup_scankey,
- false, &lbuf, BT_READ, NULL);
+ stack = _bt_search(rel,
+ IndexRelationGetNumberOfKeyAttributes(rel),
+ itup_scankey, false, &lbuf, BT_READ, NULL);
/* don't need a pin on the page */
_bt_relbuf(rel, lbuf);
diff --git a/src/backend/access/nbtree/nbtree.c b/src/backend/access/nbtree/nbtree.c
index 675e6aa80d..34c46509aa 100644
--- a/src/backend/access/nbtree/nbtree.c
+++ b/src/backend/access/nbtree/nbtree.c
@@ -117,7 +117,7 @@ bthandler(PG_FUNCTION_ARGS)
amroutine->amclusterable = true;
amroutine->ampredlocks = true;
amroutine->amcanparallel = true;
- amroutine->amcaninclude = false;
+ amroutine->amcaninclude = true;
amroutine->amkeytype = InvalidOid;
amroutine->ambuild = btbuild;
diff --git a/src/backend/access/nbtree/nbtsort.c b/src/backend/access/nbtree/nbtsort.c
index f0c276b52a..b36e8e5f6e 100644
--- a/src/backend/access/nbtree/nbtsort.c
+++ b/src/backend/access/nbtree/nbtsort.c
@@ -801,6 +801,9 @@ _bt_buildadd(BTWriteState *wstate, BTPageState *state, IndexTuple itup)
OffsetNumber last_off;
Size pgspc;
Size itupsz;
+ BTPageOpaque pageop;
+ int indnatts = IndexRelationGetNumberOfAttributes(wstate->index);
+ int indnkeyatts = IndexRelationGetNumberOfKeyAttributes(wstate->index);
/*
* This is a handy place to check for cancel interrupts during the btree
@@ -855,6 +858,8 @@ _bt_buildadd(BTWriteState *wstate, BTPageState *state, IndexTuple itup)
ItemId ii;
ItemId hii;
IndexTuple oitup;
+ IndexTuple keytup;
+ BTPageOpaque opageop = (BTPageOpaque) PageGetSpecialPointer(opage);
/* Create new page of same level */
npage = _bt_blnewpage(state->btps_level);
@@ -882,6 +887,27 @@ _bt_buildadd(BTWriteState *wstate, BTPageState *state, IndexTuple itup)
ItemIdSetUnused(ii); /* redundant */
((PageHeader) opage)->pd_lower -= sizeof(ItemIdData);
+ if (indnkeyatts != indnatts && P_ISLEAF(opageop))
+ {
+ /*
+ * We truncate included attributes of High key here.
+ * Subsequent insertions assume that hikey is already truncated,
+ * and so they need not worry about it, when copying the high key
+ * into the parent page as a downlink.
+ * NOTE: It is not crucial for reliability in present, but maybe
+ * it will be that in the future. Now the purpose is just to save
+ * more space on inner pages of btree.
+ */
+ keytup = index_truncate_tuple(wstate->index, oitup);
+
+ /* delete "wrong" high key, insert keytup as P_HIKEY. */
+ PageIndexTupleDelete(opage, P_HIKEY);
+
+ if (!_bt_pgaddtup(opage, IndexTupleSize(keytup), keytup, P_HIKEY))
+ elog(ERROR, "failed to rewrite compressed item in index \"%s\"",
+ RelationGetRelationName(wstate->index));
+ }
+
/*
* Link the old page into its parent, using its minimum key. If we
* don't have a parent, we have to create one; this adds a new btree
@@ -899,7 +925,11 @@ _bt_buildadd(BTWriteState *wstate, BTPageState *state, IndexTuple itup)
* Save a copy of the minimum key for the new page. We have to copy
* it off the old page, not the new one, in case we are not at leaf
* level.
+ * Despite oitup is already initialized, it's important to get high
+ * key from the page, since we could have replaced it with truncated
+ * copy. See comment above.
*/
+ oitup = (IndexTuple) PageGetItem(opage,PageGetItemId(opage, P_HIKEY));
state->btps_minkey = CopyIndexTuple(oitup);
/*
@@ -926,6 +956,7 @@ _bt_buildadd(BTWriteState *wstate, BTPageState *state, IndexTuple itup)
last_off = P_FIRSTKEY;
}
+ pageop = (BTPageOpaque) PageGetSpecialPointer(npage);
/*
* If the new item is the first for its page, stash a copy for later. Note
* this will only happen for the first item on a level; on later pages,
@@ -935,7 +966,14 @@ _bt_buildadd(BTWriteState *wstate, BTPageState *state, IndexTuple itup)
if (last_off == P_HIKEY)
{
Assert(state->btps_minkey == NULL);
- state->btps_minkey = CopyIndexTuple(itup);
+ /*
+ * Truncate the tuple that we're going to insert
+ * into the parent page as a downlink
+ */
+ if (indnkeyatts != indnatts && P_ISLEAF(pageop))
+ state->btps_minkey = index_truncate_tuple(wstate->index, itup);
+ else
+ state->btps_minkey = CopyIndexTuple(itup);
}
/*
@@ -1028,7 +1066,7 @@ _bt_load(BTWriteState *wstate, BTSpool *btspool, BTSpool *btspool2)
bool load1;
TupleDesc tupdes = RelationGetDescr(wstate->index);
int i,
- keysz = RelationGetNumberOfAttributes(wstate->index);
+ keysz = IndexRelationGetNumberOfKeyAttributes(wstate->index);
ScanKey indexScanKey = NULL;
SortSupport sortKeys;
diff --git a/src/backend/access/nbtree/nbtutils.c b/src/backend/access/nbtree/nbtutils.c
index 752667c885..2fc5924bf0 100644
--- a/src/backend/access/nbtree/nbtutils.c
+++ b/src/backend/access/nbtree/nbtutils.c
@@ -63,17 +63,26 @@ _bt_mkscankey(Relation rel, IndexTuple itup)
{
ScanKey skey;
TupleDesc itupdesc;
- int natts;
+ int indnatts PG_USED_FOR_ASSERTS_ONLY;
+ int indnkeyatts;
int16 *indoption;
int i;
itupdesc = RelationGetDescr(rel);
- natts = RelationGetNumberOfAttributes(rel);
+ indnatts = IndexRelationGetNumberOfAttributes(rel);
+ indnkeyatts = IndexRelationGetNumberOfKeyAttributes(rel);
indoption = rel->rd_indoption;
- skey = (ScanKey) palloc(natts * sizeof(ScanKeyData));
+ Assert(indnkeyatts != 0);
+ Assert(indnkeyatts <= indnatts);
- for (i = 0; i < natts; i++)
+ /*
+ * We'll execute search using ScanKey constructed on key columns.
+ * Non key (included) columns must be omitted.
+ */
+ skey = (ScanKey) palloc(indnkeyatts * sizeof(ScanKeyData));
+
+ for (i = 0; i < indnkeyatts; i++)
{
FmgrInfo *procinfo;
Datum arg;
@@ -115,16 +124,16 @@ ScanKey
_bt_mkscankey_nodata(Relation rel)
{
ScanKey skey;
- int natts;
+ int indnkeyatts;
int16 *indoption;
int i;
- natts = RelationGetNumberOfAttributes(rel);
+ indnkeyatts = IndexRelationGetNumberOfKeyAttributes(rel);
indoption = rel->rd_indoption;
- skey = (ScanKey) palloc(natts * sizeof(ScanKeyData));
+ skey = (ScanKey) palloc(indnkeyatts * sizeof(ScanKeyData));
- for (i = 0; i < natts; i++)
+ for (i = 0; i < indnkeyatts; i++)
{
FmgrInfo *procinfo;
int flags;
diff --git a/src/backend/access/nbtree/nbtxlog.c b/src/backend/access/nbtree/nbtxlog.c
index 233c3965d9..bbfe860e36 100644
--- a/src/backend/access/nbtree/nbtxlog.c
+++ b/src/backend/access/nbtree/nbtxlog.c
@@ -294,13 +294,11 @@ btree_xlog_split(bool onleft, XLogReaderState *record)
}
/* Extract left hikey and its size (assuming 16-bit alignment) */
- if (!isleaf)
- {
- left_hikey = (IndexTuple) datapos;
- left_hikeysz = MAXALIGN(IndexTupleSize(left_hikey));
- datapos += left_hikeysz;
- datalen -= left_hikeysz;
- }
+ left_hikey = (IndexTuple) datapos;
+ left_hikeysz = MAXALIGN(IndexTupleSize(left_hikey));
+ datapos += left_hikeysz;
+ datalen -= left_hikeysz;
+
Assert(datalen == 0);
newlpage = PageGetTempPageCopySpecial(lpage);
diff --git a/src/include/access/nbtree.h b/src/include/access/nbtree.h
index 2b0b1da763..053f8aa345 100644
--- a/src/include/access/nbtree.h
+++ b/src/include/access/nbtree.h
@@ -476,6 +476,8 @@ extern bool _bt_doinsert(Relation rel, IndexTuple itup,
IndexUniqueCheck checkUnique, Relation heapRel);
extern Buffer _bt_getstackbuf(Relation rel, BTStack stack, int access);
extern void _bt_finish_split(Relation rel, Buffer bbuf, BTStack stack);
+extern bool _bt_pgaddtup(Page page, Size itemsize, IndexTuple itup,
+ OffsetNumber itup_off);
/*
* prototypes for functions in nbtpage.c
diff --git a/src/test/regress/expected/create_index.out b/src/test/regress/expected/create_index.out
index 057faff2e5..024836b335 100644
--- a/src/test/regress/expected/create_index.out
+++ b/src/test/regress/expected/create_index.out
@@ -2395,6 +2395,25 @@ DETAIL: Key ((f1 || f2))=(ABCDEF) already exists.
-- but this shouldn't:
INSERT INTO func_index_heap VALUES('QWERTY');
--
+-- Test unique index with included columns
+--
+CREATE TABLE covering_index_heap (f1 int, f2 int, f3 text);
+CREATE UNIQUE INDEX covering_index_index on covering_index_heap (f1,f2) INCLUDE(f3);
+INSERT INTO covering_index_heap VALUES(1,1,'AAA');
+INSERT INTO covering_index_heap VALUES(1,2,'AAA');
+-- this should fail because of unique index on f1,f2:
+INSERT INTO covering_index_heap VALUES(1,2,'BBB');
+ERROR: duplicate key value violates unique constraint "covering_index_index"
+DETAIL: Key (f1, f2)=(1, 2) already exists.
+-- and this shouldn't:
+INSERT INTO covering_index_heap VALUES(1,4,'AAA');
+-- Try to build index on table that already contains data
+CREATE UNIQUE INDEX covering_pkey on covering_index_heap (f1,f2) INCLUDE(f3);
+-- Try to use existing covering index as primary key
+ALTER TABLE covering_index_heap ADD CONSTRAINT covering_pkey PRIMARY KEY USING INDEX
+covering_pkey;
+DROP TABLE covering_index_heap;
+--
-- Also try building functional, expressional, and partial indexes on
-- tables that already contain data.
--
diff --git a/src/test/regress/expected/index_including.out b/src/test/regress/expected/index_including.out
new file mode 100644
index 0000000000..0a883a492b
--- /dev/null
+++ b/src/test/regress/expected/index_including.out
@@ -0,0 +1,320 @@
+/*
+ * 1.test CREATE INDEX
+ */
+ -- Regular index with included columns
+CREATE TABLE tbl (c1 int, c2 int, c3 int, c4 box);
+INSERT INTO tbl select x, 2*x, 3*x, box('4,4,4,4') from generate_series(1,10) as x;
+CREATE INDEX tbl_idx ON tbl using btree(c1, c2) INCLUDE (c3,c4);
+-- must fail because of intersection of key and included columns
+CREATE INDEX tbl_idx ON tbl using btree(c1, c2) INCLUDE (c1,c3);
+ERROR: included columns must not intersect with key columns
+DROP TABLE tbl;
+-- Unique index and unique constraint
+CREATE TABLE tbl (c1 int, c2 int, c3 int, c4 box);
+INSERT INTO tbl select x, 2*x, 3*x, box('4,4,4,4') from generate_series(1,10) as x;
+CREATE UNIQUE INDEX tbl_idx_unique ON tbl using btree(c1, c2) INCLUDE (c3,c4);
+ALTER TABLE tbl add UNIQUE USING INDEX tbl_idx_unique;
+ALTER TABLE tbl add UNIQUE(c1, c2) INCLUDE (c3, c4);
+DROP TABLE tbl;
+-- Unique index and unique constraint. Both must fail.
+CREATE TABLE tbl (c1 int, c2 int, c3 int, c4 box);
+INSERT INTO tbl select 1, 2, 3*x, box('4,4,4,4') from generate_series(1,10) as x;
+CREATE UNIQUE INDEX tbl_idx_unique ON tbl using btree(c1, c2) INCLUDE (c3,c4);
+ERROR: could not create unique index "tbl_idx_unique"
+DETAIL: Key (c1, c2)=(1, 2) is duplicated.
+ALTER TABLE tbl add UNIQUE(c1, c2) INCLUDE (c3, c4);
+ERROR: could not create unique index "tbl_c1_c2_c3_c4_key"
+DETAIL: Key (c1, c2)=(1, 2) is duplicated.
+DROP TABLE tbl;
+-- PK constraint
+CREATE TABLE tbl (c1 int, c2 int, c3 int, c4 box);
+INSERT INTO tbl select 1, 2, 3*x, box('4,4,4,4') from generate_series(1,10) as x;
+ALTER TABLE tbl add PRIMARY KEY(c1, c2) INCLUDE (c3, c4);
+ERROR: could not create unique index "tbl_pkey"
+DETAIL: Key (c1, c2)=(1, 2) is duplicated.
+DROP TABLE tbl;
+CREATE TABLE tbl (c1 int, c2 int, c3 int, c4 box);
+INSERT INTO tbl select 1, 2, 3*x, box('4,4,4,4') from generate_series(1,10) as x;
+CREATE UNIQUE INDEX tbl_idx_unique ON tbl using btree(c1, c2) INCLUDE (c3,c4);
+ERROR: could not create unique index "tbl_idx_unique"
+DETAIL: Key (c1, c2)=(1, 2) is duplicated.
+ALTER TABLE tbl add PRIMARY KEY USING INDEX tbl_idx_unique;
+ERROR: index "tbl_idx_unique" does not exist
+LINE 1: ALTER TABLE tbl add PRIMARY KEY USING INDEX tbl_idx_unique;
+ ^
+DROP TABLE tbl;
+-- PK constraint. Must fail.
+CREATE TABLE tbl (c1 int, c2 int, c3 int, c4 box);
+INSERT INTO tbl select 1, 2, 3*x, box('4,4,4,4') from generate_series(1,10) as x;
+ALTER TABLE tbl add PRIMARY KEY(c1, c2) INCLUDE (c3, c4);
+ERROR: could not create unique index "tbl_pkey"
+DETAIL: Key (c1, c2)=(1, 2) is duplicated.
+DROP TABLE tbl;
+/*
+ * 2. Test CREATE TABLE with constraint
+ */
+CREATE TABLE tbl (c1 int,c2 int, c3 int, c4 box,
+ CONSTRAINT covering UNIQUE(c1,c2) INCLUDE(c3,c4));
+select indexrelid::regclass, indnatts, indnkeyatts, indisunique, indisprimary, indkey, indclass from pg_index where indrelid = 'tbl'::regclass::oid;
+ indexrelid | indnatts | indnkeyatts | indisunique | indisprimary | indkey | indclass
+------------+----------+-------------+-------------+--------------+---------+-----------
+ covering | 4 | 2 | t | f | 1 2 3 4 | 1978 1978
+(1 row)
+
+select pg_get_constraintdef(oid), conname, conkey, conincluding from pg_constraint where conrelid = 'tbl'::regclass::oid;
+ pg_get_constraintdef | conname | conkey | conincluding
+----------------------------------+----------+--------+--------------
+ UNIQUE (c1, c2) INCLUDE (c3, c4) | covering | {1,2} | {3,4}
+(1 row)
+
+-- ensure that constraint works
+INSERT INTO tbl select 1, 2, 3*x, box('4,4,4,4') from generate_series(1,10) as x;
+ERROR: duplicate key value violates unique constraint "covering"
+DETAIL: Key (c1, c2)=(1, 2) already exists.
+DROP TABLE tbl;
+CREATE TABLE tbl (c1 int,c2 int, c3 int, c4 box,
+ CONSTRAINT covering PRIMARY KEY(c1,c2) INCLUDE(c3,c4));
+select indexrelid::regclass, indnatts, indnkeyatts, indisunique, indisprimary, indkey, indclass from pg_index where indrelid = 'tbl'::regclass::oid;
+ indexrelid | indnatts | indnkeyatts | indisunique | indisprimary | indkey | indclass
+------------+----------+-------------+-------------+--------------+---------+-----------
+ covering | 4 | 2 | t | t | 1 2 3 4 | 1978 1978
+(1 row)
+
+select pg_get_constraintdef(oid), conname, conkey, conincluding from pg_constraint where conrelid = 'tbl'::regclass::oid;
+ pg_get_constraintdef | conname | conkey | conincluding
+---------------------------------------+----------+--------+--------------
+ PRIMARY KEY (c1, c2) INCLUDE (c3, c4) | covering | {1,2} | {3,4}
+(1 row)
+
+-- ensure that constraint works
+INSERT INTO tbl select 1, 2, 3*x, box('4,4,4,4') from generate_series(1,10) as x;
+ERROR: duplicate key value violates unique constraint "covering"
+DETAIL: Key (c1, c2)=(1, 2) already exists.
+INSERT INTO tbl select 1, NULL, 3*x, box('4,4,4,4') from generate_series(1,10) as x;
+ERROR: null value in column "c2" violates not-null constraint
+DETAIL: Failing row contains (1, null, 3, (4,4),(4,4)).
+INSERT INTO tbl select x, 2*x, NULL, NULL from generate_series(1,10) as x;
+DROP TABLE tbl;
+CREATE TABLE tbl (c1 int,c2 int, c3 int, c4 box,
+ UNIQUE(c1,c2) INCLUDE(c3,c4));
+select indexrelid::regclass, indnatts, indnkeyatts, indisunique, indisprimary, indkey, indclass from pg_index where indrelid = 'tbl'::regclass::oid;
+ indexrelid | indnatts | indnkeyatts | indisunique | indisprimary | indkey | indclass
+---------------------+----------+-------------+-------------+--------------+---------+-----------
+ tbl_c1_c2_c3_c4_key | 4 | 2 | t | f | 1 2 3 4 | 1978 1978
+(1 row)
+
+select pg_get_constraintdef(oid), conname, conkey, conincluding from pg_constraint where conrelid = 'tbl'::regclass::oid;
+ pg_get_constraintdef | conname | conkey | conincluding
+----------------------------------+---------------------+--------+--------------
+ UNIQUE (c1, c2) INCLUDE (c3, c4) | tbl_c1_c2_c3_c4_key | {1,2} | {3,4}
+(1 row)
+
+-- ensure that constraint works
+INSERT INTO tbl select 1, 2, 3*x, box('4,4,4,4') from generate_series(1,10) as x;
+ERROR: duplicate key value violates unique constraint "tbl_c1_c2_c3_c4_key"
+DETAIL: Key (c1, c2)=(1, 2) already exists.
+DROP TABLE tbl;
+CREATE TABLE tbl (c1 int,c2 int, c3 int, c4 box,
+ PRIMARY KEY(c1,c2) INCLUDE(c3,c4));
+select indexrelid::regclass, indnatts, indnkeyatts, indisunique, indisprimary, indkey, indclass from pg_index where indrelid = 'tbl'::regclass::oid;
+ indexrelid | indnatts | indnkeyatts | indisunique | indisprimary | indkey | indclass
+------------+----------+-------------+-------------+--------------+---------+-----------
+ tbl_pkey | 4 | 2 | t | t | 1 2 3 4 | 1978 1978
+(1 row)
+
+select pg_get_constraintdef(oid), conname, conkey, conincluding from pg_constraint where conrelid = 'tbl'::regclass::oid;
+ pg_get_constraintdef | conname | conkey | conincluding
+---------------------------------------+----------+--------+--------------
+ PRIMARY KEY (c1, c2) INCLUDE (c3, c4) | tbl_pkey | {1,2} | {3,4}
+(1 row)
+
+-- ensure that constraint works
+INSERT INTO tbl select 1, 2, 3*x, box('4,4,4,4') from generate_series(1,10) as x;
+ERROR: duplicate key value violates unique constraint "tbl_pkey"
+DETAIL: Key (c1, c2)=(1, 2) already exists.
+INSERT INTO tbl select 1, NULL, 3*x, box('4,4,4,4') from generate_series(1,10) as x;
+ERROR: null value in column "c2" violates not-null constraint
+DETAIL: Failing row contains (1, null, 3, (4,4),(4,4)).
+INSERT INTO tbl select x, 2*x, NULL, NULL from generate_series(1,10) as x;
+DROP TABLE tbl;
+CREATE TABLE tbl (c1 int,c2 int, c3 int, c4 box,
+ EXCLUDE USING btree (c1 WITH =) INCLUDE(c3,c4));
+select indexrelid::regclass, indnatts, indnkeyatts, indisunique, indisprimary, indkey, indclass from pg_index where indrelid = 'tbl'::regclass::oid;
+ indexrelid | indnatts | indnkeyatts | indisunique | indisprimary | indkey | indclass
+-------------------+----------+-------------+-------------+--------------+--------+----------
+ tbl_c1_c3_c4_excl | 3 | 1 | f | f | 1 3 4 | 1978
+(1 row)
+
+select pg_get_constraintdef(oid), conname, conkey, conincluding from pg_constraint where conrelid = 'tbl'::regclass::oid;
+ pg_get_constraintdef | conname | conkey | conincluding
+--------------------------------------------------+-------------------+--------+--------------
+ EXCLUDE USING btree (c1 WITH =) INCLUDE (c3, c4) | tbl_c1_c3_c4_excl | {1} | {3,4}
+(1 row)
+
+-- ensure that constraint works
+INSERT INTO tbl select 1, 2, 3*x, box('4,4,4,4') from generate_series(1,10) as x;
+ERROR: conflicting key value violates exclusion constraint "tbl_c1_c3_c4_excl"
+DETAIL: Key (c1)=(1) conflicts with existing key (c1)=(1).
+INSERT INTO tbl select x, 2*x, NULL, NULL from generate_series(1,10) as x;
+DROP TABLE tbl;
+/*
+ * 3.0 Test ALTER TABLE DROP COLUMN.
+ * Any column deletion leads to index deletion.
+ */
+CREATE TABLE tbl (c1 int,c2 int, c3 int, c4 int);
+CREATE UNIQUE INDEX tbl_idx ON tbl using btree(c1, c2, c3, c4);
+select indexdef from pg_indexes where tablename = 'tbl' order by indexname;
+ indexdef
+------------------------------------------------------------------------
+ CREATE UNIQUE INDEX tbl_idx ON public.tbl USING btree (c1, c2, c3, c4)
+(1 row)
+
+ALTER TABLE tbl DROP COLUMN c3;
+select indexdef from pg_indexes where tablename = 'tbl' order by indexname;
+ indexdef
+----------
+(0 rows)
+
+DROP TABLE tbl;
+/*
+ * 3.1 Test ALTER TABLE DROP COLUMN.
+ * Included column deletion leads to the index deletion,
+ * as well as key columns deletion. It's explained in documentation.
+ */
+CREATE TABLE tbl (c1 int,c2 int, c3 int, c4 box);
+CREATE UNIQUE INDEX tbl_idx ON tbl using btree(c1, c2) INCLUDE(c3,c4);
+select indexdef from pg_indexes where tablename = 'tbl' order by indexname;
+ indexdef
+---------------------------------------------------------------------------------
+ CREATE UNIQUE INDEX tbl_idx ON public.tbl USING btree (c1, c2) INCLUDE (c3, c4)
+(1 row)
+
+ALTER TABLE tbl DROP COLUMN c3;
+select indexdef from pg_indexes where tablename = 'tbl' order by indexname;
+ indexdef
+----------
+(0 rows)
+
+DROP TABLE tbl;
+/*
+ * 3.2 Test ALTER TABLE DROP COLUMN.
+ * Included column deletion leads to the index deletion.
+ * as well as key columns deletion. It's explained in documentation.
+ */
+CREATE TABLE tbl (c1 int,c2 int, c3 int, c4 box, UNIQUE(c1, c2) INCLUDE(c3,c4));
+select indexdef from pg_indexes where tablename = 'tbl' order by indexname;
+ indexdef
+---------------------------------------------------------------------------------------------
+ CREATE UNIQUE INDEX tbl_c1_c2_c3_c4_key ON public.tbl USING btree (c1, c2) INCLUDE (c3, c4)
+(1 row)
+
+ALTER TABLE tbl DROP COLUMN c3;
+select indexdef from pg_indexes where tablename = 'tbl' order by indexname;
+ indexdef
+----------
+(0 rows)
+
+ALTER TABLE tbl DROP COLUMN c1;
+select indexdef from pg_indexes where tablename = 'tbl' order by indexname;
+ indexdef
+----------
+(0 rows)
+
+DROP TABLE tbl;
+/*
+ * 4. CREATE INDEX CONCURRENTLY
+ */
+CREATE TABLE tbl (c1 int,c2 int, c3 int, c4 box, UNIQUE(c1, c2) INCLUDE(c3,c4));
+INSERT INTO tbl select x, 2*x, 3*x, box('4,4,4,4') from generate_series(1,1000) as x;
+CREATE UNIQUE INDEX CONCURRENTLY on tbl (c1, c2) INCLUDE (c3, c4);
+select indexdef from pg_indexes where tablename = 'tbl' order by indexname;
+ indexdef
+---------------------------------------------------------------------------------------------
+ CREATE UNIQUE INDEX tbl_c1_c2_c3_c4_idx ON public.tbl USING btree (c1, c2) INCLUDE (c3, c4)
+ CREATE UNIQUE INDEX tbl_c1_c2_c3_c4_key ON public.tbl USING btree (c1, c2) INCLUDE (c3, c4)
+(2 rows)
+
+DROP TABLE tbl;
+/*
+ * 5. REINDEX
+ */
+CREATE TABLE tbl (c1 int,c2 int, c3 int, c4 box, UNIQUE(c1, c2) INCLUDE(c3,c4));
+select indexdef from pg_indexes where tablename = 'tbl' order by indexname;
+ indexdef
+---------------------------------------------------------------------------------------------
+ CREATE UNIQUE INDEX tbl_c1_c2_c3_c4_key ON public.tbl USING btree (c1, c2) INCLUDE (c3, c4)
+(1 row)
+
+ALTER TABLE tbl DROP COLUMN c3;
+select indexdef from pg_indexes where tablename = 'tbl' order by indexname;
+ indexdef
+----------
+(0 rows)
+
+REINDEX INDEX tbl_c1_c2_c3_c4_key;
+ERROR: relation "tbl_c1_c2_c3_c4_key" does not exist
+select indexdef from pg_indexes where tablename = 'tbl' order by indexname;
+ indexdef
+----------
+(0 rows)
+
+ALTER TABLE tbl DROP COLUMN c1;
+select indexdef from pg_indexes where tablename = 'tbl' order by indexname;
+ indexdef
+----------
+(0 rows)
+
+DROP TABLE tbl;
+/*
+ * 7. Check various AMs. All but brtee must fail.
+ */
+CREATE TABLE tbl (c1 int,c2 int, c3 box, c4 box);
+CREATE INDEX on tbl USING brin(c1, c2) INCLUDE (c3, c4);
+ERROR: access method "brin" does not support included columns
+CREATE INDEX on tbl USING gist(c3) INCLUDE (c4);
+ERROR: access method "gist" does not support included columns
+CREATE INDEX on tbl USING spgist(c3) INCLUDE (c4);
+ERROR: access method "spgist" does not support included columns
+CREATE INDEX on tbl USING gin(c1, c2) INCLUDE (c3, c4);
+ERROR: access method "gin" does not support included columns
+CREATE INDEX on tbl USING hash(c1, c2) INCLUDE (c3, c4);
+ERROR: access method "hash" does not support included columns
+CREATE INDEX on tbl USING rtree(c1, c2) INCLUDE (c3, c4);
+NOTICE: substituting access method "gist" for obsolete method "rtree"
+ERROR: access method "gist" does not support included columns
+CREATE INDEX on tbl USING btree(c1, c2) INCLUDE (c3, c4);
+DROP TABLE tbl;
+/*
+ * 8. Update, delete values in indexed table.
+ */
+CREATE TABLE tbl (c1 int, c2 int, c3 int, c4 box);
+INSERT INTO tbl select x, 2*x, 3*x, box('4,4,4,4') from generate_series(1,10) as x;
+CREATE UNIQUE INDEX tbl_idx_unique ON tbl using btree(c1, c2) INCLUDE (c3,c4);
+UPDATE tbl SET c1 = 100 WHERE c1 = 2;
+UPDATE tbl SET c1 = 1 WHERE c1 = 3;
+-- should fail
+UPDATE tbl SET c2 = 2 WHERE c1 = 1;
+ERROR: duplicate key value violates unique constraint "tbl_idx_unique"
+DETAIL: Key (c1, c2)=(1, 2) already exists.
+UPDATE tbl SET c3 = 1;
+DELETE FROM tbl WHERE c1 = 5 OR c3 = 12;
+DROP TABLE tbl;
+/*
+ * 9. Alter column type.
+ */
+CREATE TABLE tbl (c1 int,c2 int, c3 int, c4 box, UNIQUE(c1, c2) INCLUDE(c3,c4));
+INSERT INTO tbl select x, 2*x, 3*x, box('4,4,4,4') from generate_series(1,10) as x;
+ALTER TABLE tbl ALTER c1 TYPE bigint;
+ALTER TABLE tbl ALTER c3 TYPE bigint;
+\d tbl
+ Table "public.tbl"
+ Column | Type | Collation | Nullable | Default
+--------+---------+-----------+----------+---------
+ c1 | bigint | | |
+ c2 | integer | | |
+ c3 | bigint | | |
+ c4 | box | | |
+Indexes:
+ "tbl_c1_c2_c3_c4_key" UNIQUE CONSTRAINT, btree (c1, c2) INCLUDE (c3, c4)
+
+DROP TABLE tbl;
diff --git a/src/test/regress/parallel_schedule b/src/test/regress/parallel_schedule
index ad9434fb87..a9de9fd1fb 100644
--- a/src/test/regress/parallel_schedule
+++ b/src/test/regress/parallel_schedule
@@ -55,7 +55,7 @@ test: copy copyselect copydml
# ----------
test: create_misc create_operator create_procedure
# These depend on the above two
-test: create_index create_view
+test: create_index create_view index_including
# ----------
# Another group of parallel tests
diff --git a/src/test/regress/serial_schedule b/src/test/regress/serial_schedule
index 27cd49845e..0ced2e2e8c 100644
--- a/src/test/regress/serial_schedule
+++ b/src/test/regress/serial_schedule
@@ -65,6 +65,7 @@ test: create_misc
test: create_operator
test: create_procedure
test: create_index
+test: index_including
test: create_view
test: create_aggregate
test: create_function_3
diff --git a/src/test/regress/sql/create_index.sql b/src/test/regress/sql/create_index.sql
index 7f17588b0d..9d4b8883c9 100644
--- a/src/test/regress/sql/create_index.sql
+++ b/src/test/regress/sql/create_index.sql
@@ -731,6 +731,26 @@ INSERT INTO func_index_heap VALUES('ABCD', 'EF');
-- but this shouldn't:
INSERT INTO func_index_heap VALUES('QWERTY');
+--
+-- Test unique index with included columns
+--
+CREATE TABLE covering_index_heap (f1 int, f2 int, f3 text);
+CREATE UNIQUE INDEX covering_index_index on covering_index_heap (f1,f2) INCLUDE(f3);
+
+INSERT INTO covering_index_heap VALUES(1,1,'AAA');
+INSERT INTO covering_index_heap VALUES(1,2,'AAA');
+-- this should fail because of unique index on f1,f2:
+INSERT INTO covering_index_heap VALUES(1,2,'BBB');
+-- and this shouldn't:
+INSERT INTO covering_index_heap VALUES(1,4,'AAA');
+-- Try to build index on table that already contains data
+CREATE UNIQUE INDEX covering_pkey on covering_index_heap (f1,f2) INCLUDE(f3);
+-- Try to use existing covering index as primary key
+ALTER TABLE covering_index_heap ADD CONSTRAINT covering_pkey PRIMARY KEY USING INDEX
+covering_pkey;
+DROP TABLE covering_index_heap;
+
+
--
-- Also try building functional, expressional, and partial indexes on
-- tables that already contain data.
diff --git a/src/test/regress/sql/index_including.sql b/src/test/regress/sql/index_including.sql
new file mode 100644
index 0000000000..e0a700b4ba
--- /dev/null
+++ b/src/test/regress/sql/index_including.sql
@@ -0,0 +1,189 @@
+/*
+ * 1.test CREATE INDEX
+ */
+ -- Regular index with included columns
+CREATE TABLE tbl (c1 int, c2 int, c3 int, c4 box);
+INSERT INTO tbl select x, 2*x, 3*x, box('4,4,4,4') from generate_series(1,10) as x;
+CREATE INDEX tbl_idx ON tbl using btree(c1, c2) INCLUDE (c3,c4);
+-- must fail because of intersection of key and included columns
+CREATE INDEX tbl_idx ON tbl using btree(c1, c2) INCLUDE (c1,c3);
+DROP TABLE tbl;
+
+-- Unique index and unique constraint
+CREATE TABLE tbl (c1 int, c2 int, c3 int, c4 box);
+INSERT INTO tbl select x, 2*x, 3*x, box('4,4,4,4') from generate_series(1,10) as x;
+CREATE UNIQUE INDEX tbl_idx_unique ON tbl using btree(c1, c2) INCLUDE (c3,c4);
+ALTER TABLE tbl add UNIQUE USING INDEX tbl_idx_unique;
+ALTER TABLE tbl add UNIQUE(c1, c2) INCLUDE (c3, c4);
+DROP TABLE tbl;
+
+-- Unique index and unique constraint. Both must fail.
+CREATE TABLE tbl (c1 int, c2 int, c3 int, c4 box);
+INSERT INTO tbl select 1, 2, 3*x, box('4,4,4,4') from generate_series(1,10) as x;
+CREATE UNIQUE INDEX tbl_idx_unique ON tbl using btree(c1, c2) INCLUDE (c3,c4);
+ALTER TABLE tbl add UNIQUE(c1, c2) INCLUDE (c3, c4);
+DROP TABLE tbl;
+
+-- PK constraint
+CREATE TABLE tbl (c1 int, c2 int, c3 int, c4 box);
+INSERT INTO tbl select 1, 2, 3*x, box('4,4,4,4') from generate_series(1,10) as x;
+ALTER TABLE tbl add PRIMARY KEY(c1, c2) INCLUDE (c3, c4);
+DROP TABLE tbl;
+
+CREATE TABLE tbl (c1 int, c2 int, c3 int, c4 box);
+INSERT INTO tbl select 1, 2, 3*x, box('4,4,4,4') from generate_series(1,10) as x;
+CREATE UNIQUE INDEX tbl_idx_unique ON tbl using btree(c1, c2) INCLUDE (c3,c4);
+ALTER TABLE tbl add PRIMARY KEY USING INDEX tbl_idx_unique;
+DROP TABLE tbl;
+-- PK constraint. Must fail.
+CREATE TABLE tbl (c1 int, c2 int, c3 int, c4 box);
+INSERT INTO tbl select 1, 2, 3*x, box('4,4,4,4') from generate_series(1,10) as x;
+ALTER TABLE tbl add PRIMARY KEY(c1, c2) INCLUDE (c3, c4);
+DROP TABLE tbl;
+
+
+/*
+ * 2. Test CREATE TABLE with constraint
+ */
+CREATE TABLE tbl (c1 int,c2 int, c3 int, c4 box,
+ CONSTRAINT covering UNIQUE(c1,c2) INCLUDE(c3,c4));
+select indexrelid::regclass, indnatts, indnkeyatts, indisunique, indisprimary, indkey, indclass from pg_index where indrelid = 'tbl'::regclass::oid;
+select pg_get_constraintdef(oid), conname, conkey, conincluding from pg_constraint where conrelid = 'tbl'::regclass::oid;
+-- ensure that constraint works
+INSERT INTO tbl select 1, 2, 3*x, box('4,4,4,4') from generate_series(1,10) as x;
+DROP TABLE tbl;
+
+CREATE TABLE tbl (c1 int,c2 int, c3 int, c4 box,
+ CONSTRAINT covering PRIMARY KEY(c1,c2) INCLUDE(c3,c4));
+select indexrelid::regclass, indnatts, indnkeyatts, indisunique, indisprimary, indkey, indclass from pg_index where indrelid = 'tbl'::regclass::oid;
+select pg_get_constraintdef(oid), conname, conkey, conincluding from pg_constraint where conrelid = 'tbl'::regclass::oid;
+-- ensure that constraint works
+INSERT INTO tbl select 1, 2, 3*x, box('4,4,4,4') from generate_series(1,10) as x;
+INSERT INTO tbl select 1, NULL, 3*x, box('4,4,4,4') from generate_series(1,10) as x;
+INSERT INTO tbl select x, 2*x, NULL, NULL from generate_series(1,10) as x;
+DROP TABLE tbl;
+
+CREATE TABLE tbl (c1 int,c2 int, c3 int, c4 box,
+ UNIQUE(c1,c2) INCLUDE(c3,c4));
+select indexrelid::regclass, indnatts, indnkeyatts, indisunique, indisprimary, indkey, indclass from pg_index where indrelid = 'tbl'::regclass::oid;
+select pg_get_constraintdef(oid), conname, conkey, conincluding from pg_constraint where conrelid = 'tbl'::regclass::oid;
+-- ensure that constraint works
+INSERT INTO tbl select 1, 2, 3*x, box('4,4,4,4') from generate_series(1,10) as x;
+DROP TABLE tbl;
+
+CREATE TABLE tbl (c1 int,c2 int, c3 int, c4 box,
+ PRIMARY KEY(c1,c2) INCLUDE(c3,c4));
+select indexrelid::regclass, indnatts, indnkeyatts, indisunique, indisprimary, indkey, indclass from pg_index where indrelid = 'tbl'::regclass::oid;
+select pg_get_constraintdef(oid), conname, conkey, conincluding from pg_constraint where conrelid = 'tbl'::regclass::oid;
+-- ensure that constraint works
+INSERT INTO tbl select 1, 2, 3*x, box('4,4,4,4') from generate_series(1,10) as x;
+INSERT INTO tbl select 1, NULL, 3*x, box('4,4,4,4') from generate_series(1,10) as x;
+INSERT INTO tbl select x, 2*x, NULL, NULL from generate_series(1,10) as x;
+DROP TABLE tbl;
+
+CREATE TABLE tbl (c1 int,c2 int, c3 int, c4 box,
+ EXCLUDE USING btree (c1 WITH =) INCLUDE(c3,c4));
+select indexrelid::regclass, indnatts, indnkeyatts, indisunique, indisprimary, indkey, indclass from pg_index where indrelid = 'tbl'::regclass::oid;
+select pg_get_constraintdef(oid), conname, conkey, conincluding from pg_constraint where conrelid = 'tbl'::regclass::oid;
+-- ensure that constraint works
+INSERT INTO tbl select 1, 2, 3*x, box('4,4,4,4') from generate_series(1,10) as x;
+INSERT INTO tbl select x, 2*x, NULL, NULL from generate_series(1,10) as x;
+DROP TABLE tbl;
+
+/*
+ * 3.0 Test ALTER TABLE DROP COLUMN.
+ * Any column deletion leads to index deletion.
+ */
+CREATE TABLE tbl (c1 int,c2 int, c3 int, c4 int);
+CREATE UNIQUE INDEX tbl_idx ON tbl using btree(c1, c2, c3, c4);
+select indexdef from pg_indexes where tablename = 'tbl' order by indexname;
+ALTER TABLE tbl DROP COLUMN c3;
+select indexdef from pg_indexes where tablename = 'tbl' order by indexname;
+DROP TABLE tbl;
+
+/*
+ * 3.1 Test ALTER TABLE DROP COLUMN.
+ * Included column deletion leads to the index deletion,
+ * as well as key columns deletion. It's explained in documentation.
+ */
+CREATE TABLE tbl (c1 int,c2 int, c3 int, c4 box);
+CREATE UNIQUE INDEX tbl_idx ON tbl using btree(c1, c2) INCLUDE(c3,c4);
+select indexdef from pg_indexes where tablename = 'tbl' order by indexname;
+ALTER TABLE tbl DROP COLUMN c3;
+select indexdef from pg_indexes where tablename = 'tbl' order by indexname;
+DROP TABLE tbl;
+
+/*
+ * 3.2 Test ALTER TABLE DROP COLUMN.
+ * Included column deletion leads to the index deletion.
+ * as well as key columns deletion. It's explained in documentation.
+ */
+CREATE TABLE tbl (c1 int,c2 int, c3 int, c4 box, UNIQUE(c1, c2) INCLUDE(c3,c4));
+select indexdef from pg_indexes where tablename = 'tbl' order by indexname;
+ALTER TABLE tbl DROP COLUMN c3;
+select indexdef from pg_indexes where tablename = 'tbl' order by indexname;
+ALTER TABLE tbl DROP COLUMN c1;
+select indexdef from pg_indexes where tablename = 'tbl' order by indexname;
+DROP TABLE tbl;
+
+
+/*
+ * 4. CREATE INDEX CONCURRENTLY
+ */
+CREATE TABLE tbl (c1 int,c2 int, c3 int, c4 box, UNIQUE(c1, c2) INCLUDE(c3,c4));
+INSERT INTO tbl select x, 2*x, 3*x, box('4,4,4,4') from generate_series(1,1000) as x;
+CREATE UNIQUE INDEX CONCURRENTLY on tbl (c1, c2) INCLUDE (c3, c4);
+select indexdef from pg_indexes where tablename = 'tbl' order by indexname;
+DROP TABLE tbl;
+
+
+/*
+ * 5. REINDEX
+ */
+CREATE TABLE tbl (c1 int,c2 int, c3 int, c4 box, UNIQUE(c1, c2) INCLUDE(c3,c4));
+select indexdef from pg_indexes where tablename = 'tbl' order by indexname;
+ALTER TABLE tbl DROP COLUMN c3;
+select indexdef from pg_indexes where tablename = 'tbl' order by indexname;
+REINDEX INDEX tbl_c1_c2_c3_c4_key;
+select indexdef from pg_indexes where tablename = 'tbl' order by indexname;
+ALTER TABLE tbl DROP COLUMN c1;
+select indexdef from pg_indexes where tablename = 'tbl' order by indexname;
+DROP TABLE tbl;
+
+/*
+ * 7. Check various AMs. All but brtee must fail.
+ */
+CREATE TABLE tbl (c1 int,c2 int, c3 box, c4 box);
+CREATE INDEX on tbl USING brin(c1, c2) INCLUDE (c3, c4);
+CREATE INDEX on tbl USING gist(c3) INCLUDE (c4);
+CREATE INDEX on tbl USING spgist(c3) INCLUDE (c4);
+CREATE INDEX on tbl USING gin(c1, c2) INCLUDE (c3, c4);
+CREATE INDEX on tbl USING hash(c1, c2) INCLUDE (c3, c4);
+CREATE INDEX on tbl USING rtree(c1, c2) INCLUDE (c3, c4);
+CREATE INDEX on tbl USING btree(c1, c2) INCLUDE (c3, c4);
+DROP TABLE tbl;
+
+/*
+ * 8. Update, delete values in indexed table.
+ */
+CREATE TABLE tbl (c1 int, c2 int, c3 int, c4 box);
+INSERT INTO tbl select x, 2*x, 3*x, box('4,4,4,4') from generate_series(1,10) as x;
+CREATE UNIQUE INDEX tbl_idx_unique ON tbl using btree(c1, c2) INCLUDE (c3,c4);
+UPDATE tbl SET c1 = 100 WHERE c1 = 2;
+UPDATE tbl SET c1 = 1 WHERE c1 = 3;
+-- should fail
+UPDATE tbl SET c2 = 2 WHERE c1 = 1;
+UPDATE tbl SET c3 = 1;
+DELETE FROM tbl WHERE c1 = 5 OR c3 = 12;
+DROP TABLE tbl;
+
+/*
+ * 9. Alter column type.
+ */
+CREATE TABLE tbl (c1 int,c2 int, c3 int, c4 box, UNIQUE(c1, c2) INCLUDE(c3,c4));
+INSERT INTO tbl select x, 2*x, 3*x, box('4,4,4,4') from generate_series(1,10) as x;
+ALTER TABLE tbl ALTER c1 TYPE bigint;
+ALTER TABLE tbl ALTER c3 TYPE bigint;
+\d tbl
+DROP TABLE tbl;
+
--
2.14.3 (Apple Git-98)
[text/x-patch] 0003-Covering-amcheck-v5.patch (1.8K, ../78ca1586-647c-e159-ebf3-30957e270c66@postgrespro.ru/4-0003-Covering-amcheck-v5.patch)
download | inline diff:
From 7de42661a46668ae7c50a1ed72cc8a822b9ac4dc Mon Sep 17 00:00:00 2001
From: Andrey Borodin <amborodin@acm.org>
Date: Thu, 8 Mar 2018 15:54:57 +0500
Subject: [PATCH 3/3] Covering amcheck v5
---
contrib/amcheck/verify_nbtree.c | 12 ++++++------
1 file changed, 6 insertions(+), 6 deletions(-)
diff --git a/contrib/amcheck/verify_nbtree.c b/contrib/amcheck/verify_nbtree.c
index da518daea3..fb472b38f1 100644
--- a/contrib/amcheck/verify_nbtree.c
+++ b/contrib/amcheck/verify_nbtree.c
@@ -1102,10 +1102,10 @@ static inline bool
invariant_leq_offset(BtreeCheckState *state, ScanKey key,
OffsetNumber upperbound)
{
- int16 natts = state->rel->rd_rel->relnatts;
+ int16 nkeyatts = IndexRelationGetNumberOfKeyAttributes(state->rel);
int32 cmp;
- cmp = _bt_compare(state->rel, natts, key, state->target, upperbound);
+ cmp = _bt_compare(state->rel, nkeyatts, key, state->target, upperbound);
return cmp <= 0;
}
@@ -1121,10 +1121,10 @@ static inline bool
invariant_geq_offset(BtreeCheckState *state, ScanKey key,
OffsetNumber lowerbound)
{
- int16 natts = state->rel->rd_rel->relnatts;
+ int16 nkeyatts = IndexRelationGetNumberOfKeyAttributes(state->rel);
int32 cmp;
- cmp = _bt_compare(state->rel, natts, key, state->target, lowerbound);
+ cmp = _bt_compare(state->rel, nkeyatts, key, state->target, lowerbound);
return cmp >= 0;
}
@@ -1144,10 +1144,10 @@ invariant_leq_nontarget_offset(BtreeCheckState *state,
Page nontarget, ScanKey key,
OffsetNumber upperbound)
{
- int16 natts = state->rel->rd_rel->relnatts;
+ int16 nkeyatts = IndexRelationGetNumberOfKeyAttributes(state->rel);
int32 cmp;
- cmp = _bt_compare(state->rel, natts, key, nontarget, upperbound);
+ cmp = _bt_compare(state->rel, nkeyatts, key, nontarget, upperbound);
return cmp <= 0;
}
--
2.14.3 (Apple Git-98)
view thread (217+ messages) latest in thread
Message-ID: <78ca1586-647c-e159-ebf3-30957e270c66@postgrespro.ru>
Permalink: ../78ca1586-647c-e159-ebf3-30957e270c66@postgrespro.ru/
Also on: postgresql.org/message-id/78ca1586-647c-e159-ebf3-30957e270c66@postgrespro.ru
copy link · copy postgr.es · copy git am
reply Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: pgsql-hackers@postgresql.org
Cc: a.lubennikova@postgrespro.ru, thomas.munro@enterprisedb.com, x4mmm@yandex-team.ru, pgsql-hackers@lists.postgresql.org, lubennikovaav@gmail.com
Subject: Re: WIP: Covering + unique indexes.
In-Reply-To: <78ca1586-647c-e159-ebf3-30957e270c66@postgrespro.ru>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by DDX for PostgreSQL; see mirroring instructions
for how to clone and mirror all data and code used for this inbox