database/postgres
1
0
Fork 0
mirror of https://github.com/postgres/postgres.git synced 2025-06-17 17:02:08 +03:00
Code Activity

Indexes with INCLUDE columns and their support in B-tree

Browse Source 
This patch introduces INCLUDE clause to index definition.  This clause
specifies a list of columns which will be included as a non-key part in
the index.  The INCLUDE columns exist solely to allow more queries to
benefit from index-only scans.  Also, such columns don't need to have
appropriate operator classes.  Expressions are not supported as INCLUDE
columns since they cannot be used in index-only scans.

Index access methods supporting INCLUDE are indicated by amcaninclude flag
in IndexAmRoutine.  For now, only B-tree indexes support INCLUDE clause.

In B-tree indexes INCLUDE columns are truncated from pivot index tuples
(tuples located in non-leaf pages and high keys).  Therefore, B-tree indexes
now might have variable number of attributes.  This patch also provides
generic facility to support that: pivot tuples contain number of their
attributes in t_tid.ip_posid.  Free 13th bit of t_info is used for indicating
that.  This facility will simplify further support of index suffix truncation.
The changes of above are backward-compatible, pg_upgrade doesn't need special
handling of B-tree indexes for that.

Bump catalog version

Author: Anastasia Lubennikova with contribition by Alexander Korotkov and me
Reviewed by: Peter Geoghegan, Tomas Vondra, Antonin Houska, Jeff Janes,
			 David Rowley, Alexander Korotkov
Discussion: https://www.postgresql.org/message-id/flat/56168952.4010101@postgrespro.ru
This commit is contained in:
Teodor Sigaev
2018-04-07 23:00:39 +03:00
parent 01bb85169a
commit 8224de4f42
89 changed files with 2112 additions and 467 deletions
Download Patch File Download Diff File Expand all files Collapse all files

334
src/backend/parser/parse_utilcmd.c
View File

@ -1468,9 +1468,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];
@ -1559,6 +1560,40 @@ 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);
@ -1829,6 +1864,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 &&
@ -1900,6 +1936,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;
@ -2049,24 +2086,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 */
@ -2095,8 +2137,6 @@ transformIndexConstraint(Constraint *constraint, CreateStmtContext *cxt)
index->indexParams = lappend(index->indexParams, elem);
index->excludeOpNames = lappend(index->excludeOpNames, opname);
}
return index;
}
/*
@ -2107,7 +2147,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;
@ -2124,65 +2293,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;
/*
* 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;
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++)
foreach(inher, cxt->inhRelations)
{
Form_pg_attribute inhattr = TupleDescAttr(rel->rd_att,
count);
char *inhname = NameStr(inhattr->attname);
RangeVar *inh = lfirst_node(RangeVar, inher);
Relation rel;
int count;
if (inhattr->attisdropped)
continue;
if (strcmp(key, inhname) == 0)
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;
Form_pg_attribute inhattr = TupleDescAttr(rel->rd_att,
count);
char *inhname = NameStr(inhattr->attname);
/*
* 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;
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;
}
}
@ -2198,27 +2365,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);
@ -2226,9 +2372,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;