diff --git a/src/backend/bootstrap/bootparse.y b/src/backend/bootstrap/bootparse.y index 387d43e..fed79ff 100644 --- a/src/backend/bootstrap/bootparse.y +++ b/src/backend/bootstrap/bootparse.y @@ -287,7 +287,7 @@ Boot_DeclareIndexStmt: $10, NULL, NIL, NIL, false, false, false, false, false, - false, false, true, false, false); + false, false, true, false, false, false); do_end(); } ; @@ -305,7 +305,7 @@ Boot_DeclareUniqueIndexStmt: $11, NULL, NIL, NIL, true, false, false, false, false, - false, false, true, false, false); + false, false, true, false, false, false); do_end(); } ; diff --git a/src/backend/catalog/index.c b/src/backend/catalog/index.c index 69946fe..0ab523e 100644 --- a/src/backend/catalog/index.c +++ b/src/backend/catalog/index.c @@ -512,6 +512,7 @@ UpdateIndexRelation(Oid indexoid, * allow_system_table_mods: allow table to be a system catalog * skip_build: true to skip the index_build() step for the moment; caller * must do it later (typically via reindex_index()) + * index_exists: the index already exists, we are here just for formalities. * concurrent: if true, do not lock the table against writers. The index * will be marked "invalid" and the caller must take additional steps * to fix it up. @@ -535,6 +536,7 @@ index_create(Oid heapRelationId, bool initdeferred, bool allow_system_table_mods, bool skip_build, + bool index_exists, bool concurrent) { Relation pg_class; @@ -615,6 +617,7 @@ index_create(Oid heapRelationId, if (shared_relation && tableSpaceId != GLOBALTABLESPACE_OID) elog(ERROR, "shared relations must be placed in pg_global tablespace"); + if (!index_exists) if (get_relname_relid(indexRelationName, namespaceId)) ereport(ERROR, (errcode(ERRCODE_DUPLICATE_TABLE), @@ -648,6 +651,17 @@ index_create(Oid heapRelationId, indexRelationId = GetNewRelFileNode(tableSpaceId, pg_class); } + if (index_exists) + { + Assert( skip_build && !IsBootstrapProcessingMode() ); + + indexRelation = relation_open(indexRelationId, AccessExclusiveLock); + + /* done with pg_class */ + heap_close(pg_class, RowExclusiveLock); + } + else + { /* * create the index relation's relcache entry and physical disk file. (If * we fail further down, it's the smgr's responsibility to remove the disk @@ -720,6 +734,7 @@ index_create(Oid heapRelationId, classObjectId, coloptions, isprimary, !deferrable, !concurrent); + } /* * Register constraint and dependencies for the index. @@ -833,7 +848,7 @@ index_create(Oid heapRelationId, true); } } - else + else if(!index_exists) { bool have_simple_col = false; @@ -876,6 +891,8 @@ index_create(Oid heapRelationId, Assert(!initdeferred); } + if (!index_exists) + { /* Store dependency on operator classes */ for (i = 0; i < indexInfo->ii_NumIndexAttrs; i++) { @@ -905,6 +922,7 @@ index_create(Oid heapRelationId, DEPENDENCY_NORMAL, DEPENDENCY_AUTO); } + } } else { @@ -945,6 +963,8 @@ index_create(Oid heapRelationId, } else if (skip_build) { + if (!index_exists) + { /* * Caller is responsible for filling the index later on. However, * we'd better make sure that the heap relation is correctly marked as @@ -958,6 +978,7 @@ index_create(Oid heapRelationId, heapRelation->rd_rel->reltuples); /* Make the above update visible */ CommandCounterIncrement(); + } } else { diff --git a/src/backend/catalog/toasting.c b/src/backend/catalog/toasting.c index 435dfdd..96cbce2 100644 --- a/src/backend/catalog/toasting.c +++ b/src/backend/catalog/toasting.c @@ -269,7 +269,7 @@ create_toast_table(Relation rel, Oid toastOid, Oid toastIndexOid, Datum reloptio rel->rd_rel->reltablespace, classObjectId, coloptions, (Datum) 0, true, false, false, false, - true, false, false); + true, false, false, false); /* * Store the toast table's OID in the parent relation's pg_class row diff --git a/src/backend/commands/indexcmds.c b/src/backend/commands/indexcmds.c index 2da1c86..11ba888 100644 --- a/src/backend/commands/indexcmds.c +++ b/src/backend/commands/indexcmds.c @@ -69,7 +69,9 @@ static void ComputeIndexAttrs(IndexInfo *indexInfo, static Oid GetIndexOpClass(List *opclass, Oid attrType, char *accessMethodName, Oid accessMethodId); static char *ChooseIndexNameAddition(List *colnames); -static bool relationHasPrimaryKey(Relation rel); + +/* TODO: Unable to figure out which header to declare this in */ +Oid relationHasPrimaryKey(Relation rel); /* @@ -122,6 +124,7 @@ DefineIndex(RangeVar *heapRelation, bool is_alter_table, bool check_rights, bool skip_build, + bool index_exists, bool quiet, bool concurrent) { @@ -340,7 +343,7 @@ DefineIndex(RangeVar *heapRelation, * it's no problem either. */ if (is_alter_table && - relationHasPrimaryKey(rel)) + relationHasPrimaryKey(rel) != InvalidOid) { ereport(ERROR, (errcode(ERRCODE_INVALID_TABLE_DEFINITION), @@ -488,6 +491,7 @@ DefineIndex(RangeVar *heapRelation, isconstraint, deferrable, initdeferred, allowSystemTableMods, skip_build || concurrent, + index_exists, concurrent); if (!concurrent) @@ -1541,10 +1545,11 @@ ChooseIndexColumnNames(List *indexElems) * * See whether an existing relation has a primary key. */ -static bool +Oid relationHasPrimaryKey(Relation rel) { - bool result = false; + Oid indexoid = InvalidOid; + bool isprimary = false; List *indexoidlist; ListCell *indexoidscan; @@ -1557,21 +1562,22 @@ relationHasPrimaryKey(Relation rel) foreach(indexoidscan, indexoidlist) { - Oid indexoid = lfirst_oid(indexoidscan); HeapTuple indexTuple; + indexoid = lfirst_oid(indexoidscan); + indexTuple = SearchSysCache1(INDEXRELID, ObjectIdGetDatum(indexoid)); if (!HeapTupleIsValid(indexTuple)) /* should not happen */ elog(ERROR, "cache lookup failed for index %u", indexoid); - result = ((Form_pg_index) GETSTRUCT(indexTuple))->indisprimary; + isprimary = ((Form_pg_index) GETSTRUCT(indexTuple))->indisprimary; ReleaseSysCache(indexTuple); - if (result) + if (isprimary) break; } list_free(indexoidlist); - return result; + return isprimary ? indexoid : InvalidOid; } /* diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c index 2383112..68e916f 100644 --- a/src/backend/commands/tablecmds.c +++ b/src/backend/commands/tablecmds.c @@ -4559,7 +4559,10 @@ ATExecAddIndex(AlteredTableInfo *tab, Relation rel, { bool check_rights; bool skip_build; + bool index_exists = false; bool quiet; + Oid index_oid = InvalidOid; + ListCell *l, *prev = NULL; Assert(IsA(stmt, IndexStmt)); @@ -4570,11 +4573,55 @@ ATExecAddIndex(AlteredTableInfo *tab, Relation rel, /* suppress notices when rebuilding existing index */ quiet = is_rebuild; + /* TODO: + Do stmt->options have 'WITH INDEX' option set + Get OID from the Value* (represented as string) + strspn(opr_name_or_oid, "0123456789") == strlen(opr_name_or_oid)) + result = DatumGetObjectId(DirectFunctionCall1(oidin, + CStringGetDatum(opr_name_or_oid))); + replace InvalidOid, below, with this OID + set 'primary' to on; + set skip_build to on + TODO Check with -hackers that this is the right thing to do even when tab->newvals is non-null? + */ + + foreach(l, stmt->options) + { + DefElem *def = (DefElem*)lfirst(l); + + if (def->defnamespace == NULL && 0 == strcmp(def->defname, "index")) + { + if (!(def->defaction == DEFELEM_UNSPEC || def->defaction == DEFELEM_SET)) + elog(ERROR, "index option for a primary key has a syntax error." ); + } + else + { + prev = l; + continue; + } + + Assert(strspn(strVal(def->arg), "0123456789") == strlen(strVal(def->arg))); + Assert(stmt->primary); + + index_oid = DatumGetObjectId(DirectFunctionCall1(oidin, + CStringGetDatum(strVal(def->arg)))); + + /* We override the skip_build set above */ + skip_build = true; + + index_exists = true; + + break; + } + + if (l) /* unecessary check, but good for readability */ + stmt->options = list_delete_cell(stmt->options, l, prev); + /* The IndexStmt has already been through transformIndexStmt */ DefineIndex(stmt->relation, /* relation */ stmt->idxname, /* index name */ - InvalidOid, /* no predefined OID */ + index_oid, /* no predefined OID */ stmt->accessMethod, /* am name */ stmt->tableSpace, stmt->indexParams, /* parameters */ @@ -4589,6 +4636,7 @@ ATExecAddIndex(AlteredTableInfo *tab, Relation rel, true, /* is_alter_table */ check_rights, skip_build, + index_exists, quiet, false); } diff --git a/src/backend/parser/parse_utilcmd.c b/src/backend/parser/parse_utilcmd.c index 322e689..ad98928 100644 --- a/src/backend/parser/parse_utilcmd.c +++ b/src/backend/parser/parse_utilcmd.c @@ -1931,6 +1931,7 @@ transformAlterTableStmt(AlterTableStmt *stmt, const char *queryString) List *newcmds = NIL; bool skipValidation = true; AlterTableCmd *newcmd; + Oid index_oid = InvalidOid; /* * We must not scribble on the passed-in AlterTableStmt, so copy it. (This @@ -2047,6 +2048,192 @@ transformAlterTableStmt(AlterTableStmt *stmt, const char *queryString) transformFKConstraints(pstate, &cxt, skipValidation, true); + /* TODO: look at cxt->pkey->options and see if it has 'WITH INDEX'; if yes + then cook-up an ALTER TABLE DROP CONSTRAINT for table's primary key, + and push into newcmds + + Check if cxt->pkey->options has a 'WITH INDEX' element + take an exclusive lock on that index + + Does this table have a primary key + check if index mentioned in cxt->pkey matches that PKEY definition, + Does column list match + Do the opclasses match + Does index type match (BTree for now) + Do they have the same owner + + Append a new command to newcmds to drop the PKey constraint + use 'rel' variable to get primary key's OID ( modify and reuse relationHasPrimaryKey() ) + use relation_open() to get pkey's relation + use the returned Relation->rd_rel->relname to build DROP CONSTRAINT command + set missingok member of the command so that this would work even if there was already a DROP CONSTRAINT for the PKey. + push this command to newcmds + + Chenge the 'WITH INDEX' element, and replace index name in Value* to have decimal representation of index's OID. + This will be used by ATExecAddIndex(). + + BIG TODO: convert all elog() calls into ereport() and proper errcode() + calls and write better messages. + */ + + if (cxt.pkey) + foreach(l, cxt.pkey->options) + { + DefElem *def = (DefElem*)lfirst(l); + Oid pkey_oid; + char *index_name; + char *oid_string; + List *index_name_list; + RangeVar *index_rv; + Relation index_rel; + + if (def->defnamespace == NULL && 0 == strcmp(def->defname, "index")) + { + if (!(def->defaction == DEFELEM_UNSPEC || def->defaction == DEFELEM_SET)) + elog(ERROR, "index option for a primary key has a syntax error." ); + } + else + continue; + + /* + * If we don't do this, then this will get to DefineIndex(), and it will + * throw a fit. + */ + if (index_oid != InvalidOid) + elog(ERROR, "only one WITH INDEX option can be specified for a primary key."); + + if (!IsA(def->arg, String)) + elog(ERROR, "WITH INDEX option for primary key should be a string value"); + + index_name = strVal(def->arg); + + index_name_list = stringToQualifiedNameList(index_name); + + /* TODO: Should we assert that this is a non-qualified name? */ + index_rv = makeRangeVarFromNameList(index_name_list); + + index_rel = relation_openrv(index_rv, AccessExclusiveLock); + + index_oid = RelationGetRelid(index_rel); + + oid_string = DatumGetCString(DirectFunctionCall1(oidout, + ObjectIdGetDatum(index_oid))); + + /* replace index name with its Oid::cstring */ + def->arg = (Node*)makeString(oid_string); + + /* TODO: set index name in the statement, to affect the constraint name */ + cxt.pkey->idxname = pstrdup(strVal(llast(index_name_list))); + + /* TODO: make sure the index is in the same schema/namespace as the table */ + + pkey_oid = relationHasPrimaryKey(rel); + + if (pkey_oid != InvalidOid) + { + HeapTuple pkey_tuple; + HeapTuple index_tuple; + Form_pg_index pkey_form; + Form_pg_index index_form; + AlterTableCmd *at_cmd; + int2 i; + Relation pkey_rel; + + /* + * Check pg_class->relam to see if the index type match. As of now, only + * BTree indexes are used to implement primary keys, but it doesn't hurt + * to be future proof. + */ + pkey_rel = relation_open(pkey_oid, AccessExclusiveLock); + + if (pkey_rel->rd_rel->relam != index_rel->rd_rel->relam) + elog(ERROR, "index type of WITH INDEX argument and that of the primary key are not the same."); + + if (pkey_rel->rd_rel->relowner != index_rel->rd_rel->relowner) + elog(ERROR, "owner of WITH INDEX argument and that of the primary key are not same."); + + pkey_tuple = SearchSysCache1(INDEXRELID, ObjectIdGetDatum(pkey_oid)); + if (!HeapTupleIsValid(pkey_tuple)) + elog(ERROR, "cache lookup failed for index %u", pkey_oid); + pkey_form = (Form_pg_index) GETSTRUCT(pkey_tuple); + + index_tuple = SearchSysCache1(INDEXRELID, ObjectIdGetDatum(index_oid)); + if (!HeapTupleIsValid(index_tuple)) + elog(ERROR, "cache lookup failed for index %u", index_oid); + index_form = (Form_pg_index) GETSTRUCT(index_tuple); + + if (!index_form->indisvalid) + elog(ERROR, "index mentioned in the WITH INDEX clause of primary key is not valid."); + + if (!index_form->indisunique) + elog(ERROR, "index mentioned in the WITH INDEX clause of primary key is not a unique index."); + + /* TODO: Should we check for indisready, indcheckxmin ?*/ + + if (index_form->indrelid != pkey_form->indrelid) + elog(ERROR, "index mentioned in the WITH INDEX clause of primary key is not on the same table."); + + if (index_form->indnatts != pkey_form->indnatts) + elog(ERROR, "idefinition of index mentioned in the WITH INDEX clause does not match primary key."); + + /* + * Loop over each attribute in the primary key and see if it + * matches the attributes of the index we are replacing it with. + */ + for (i = 0; i < pkey_form->indnatts; i++) + { + /* We do not expect primary keys on expressions */ + Assert(pkey_form->indkey.values[i] != 0); + + if (pkey_form->indkey.values[i] != index_form->indkey.values[i]) + elog(ERROR, "idefinition of index mentioned in the WITH INDEX clause does not match primary key."); + + if (pkey_form->indclass.values[i] != index_form->indclass.values[i]) + elog(ERROR, "operator classes of index mentioned in the WITH INDEX clause does not match those of primary key."); + } + + ReleaseSysCache(pkey_tuple); + ReleaseSysCache(index_tuple); + + at_cmd = makeNode(AlterTableCmd); + at_cmd->subtype = AT_DropConstraint; + at_cmd->name = pstrdup(NameStr(pkey_rel->rd_rel->relname)); + at_cmd->behavior = DROP_CASCADE;; + at_cmd->missing_ok = true; + + newcmds = lappend(newcmds, at_cmd); + + /* Release the lock so that it can be dropped from cache. */ + relation_close(pkey_rel, AccessExclusiveLock); + } + + /* NOw update pg_index tuple to mark this index as indisprimary */ + { + Relation pg_index; + HeapTuple indexTuple; + Form_pg_index indexForm; + + pg_index = heap_open(IndexRelationId, RowExclusiveLock); + + indexTuple = SearchSysCacheCopy1(INDEXRELID, + ObjectIdGetDatum(index_oid)); + if (!HeapTupleIsValid(indexTuple)) + elog(ERROR, "cache lookup failed for index %u", index_oid); + indexForm = (Form_pg_index) GETSTRUCT(indexTuple); + + indexForm->indisprimary = true; + simple_heap_update(pg_index, &indexTuple->t_self, indexTuple); + CatalogUpdateIndexes(pg_index, indexTuple); + + heap_freetuple(indexTuple); + heap_close(pg_index, RowExclusiveLock); + } + + relation_close(index_rel, NoLock); + + /* do not break; at the end of the loop. Use this opprtunity to catch multiple 'WITH INDEX' clauses*/ + } + /* * Push any index-creation commands into the ALTER, so that they can be * scheduled nicely by tablecmds.c. Note that tablecmds.c assumes that diff --git a/src/backend/tcop/utility.c b/src/backend/tcop/utility.c index b9be06f..adae7c6 100644 --- a/src/backend/tcop/utility.c +++ b/src/backend/tcop/utility.c @@ -890,6 +890,7 @@ standard_ProcessUtility(Node *parsetree, false, /* is_alter_table */ true, /* check_rights */ false, /* skip_build */ + false, /* index_exists */ false, /* quiet */ stmt->concurrent); /* concurrent */ } diff --git a/src/include/catalog/index.h b/src/include/catalog/index.h index d336576..3b019cb 100644 --- a/src/include/catalog/index.h +++ b/src/include/catalog/index.h @@ -44,6 +44,7 @@ extern Oid index_create(Oid heapRelationId, bool initdeferred, bool allow_system_table_mods, bool skip_build, + bool index_exists, bool concurrent); extern void index_drop(Oid indexId); diff --git a/src/include/commands/defrem.h b/src/include/commands/defrem.h index 7edda97..c774710 100644 --- a/src/include/commands/defrem.h +++ b/src/include/commands/defrem.h @@ -35,6 +35,7 @@ extern void DefineIndex(RangeVar *heapRelation, bool is_alter_table, bool check_rights, bool skip_build, + bool index_exists, bool quiet, bool concurrent); extern void ReindexIndex(RangeVar *indexRelation);