Make ALTER COLUMN TYPE preserve clustered status for indexes it doesn't

modify.  Also fix a passel of problems with ALTER TABLE CLUSTER ON:
failure to check that the index is safe to cluster on (or even belongs
to the indicated rel, or even exists), and failure to broadcast a relcache
flush event when changing an index's state.

diff --git a/src/backend/commands/cluster.c b/src/backend/commands/cluster.c
index b4bf08006ebb31cdee59ba08441c0856b7aa4559..ffd2aff2364543661a65e73d34c51a7355963266 100644 (file)
--- a/src/backend/commands/cluster.c
+++ b/src/backend/commands/cluster.c
@@ -11,7 +11,7 @@
  *
  *
  * IDENTIFICATION
- *   $PostgreSQL: pgsql/src/backend/commands/cluster.c,v 1.121 2004/05/05 04:48:45 tgl Exp $
+ *   $PostgreSQL: pgsql/src/backend/commands/cluster.c,v 1.122 2004/05/06 16:10:57 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -31,6 +31,7 @@
 #include "miscadmin.h"
 #include "utils/acl.h"
 #include "utils/fmgroids.h"
+#include "utils/inval.h"
 #include "utils/lsyscache.h"
 #include "utils/syscache.h"
 #include "utils/relcache.h"
@@ -48,7 +49,6 @@ typedef struct
    IndexInfo  *indexInfo;
    Oid         accessMethodOID;
    Oid        *classOID;
-   bool        isclustered;
 } IndexAttrs;
 
 /*
@@ -246,8 +246,7 @@ cluster(ClusterStmt *stmt)
 static void
 cluster_rel(RelToCluster *rvtc, bool recheck)
 {
-   Relation    OldHeap,
-               OldIndex;
+   Relation    OldHeap;
 
    /* Check for user-requested abort. */
    CHECK_FOR_INTERRUPTS();
@@ -300,18 +299,41 @@ cluster_rel(RelToCluster *rvtc, bool recheck)
    /*
     * We grab exclusive access to the target rel and index for the
     * duration of the transaction.  (This is redundant for the single-
-    * transaction case, since cluster() already did it.)
+    * transaction case, since cluster() already did it.)  The index
+    * lock is taken inside check_index_is_clusterable.
     */
    OldHeap = heap_open(rvtc->tableOid, AccessExclusiveLock);
 
-   OldIndex = index_open(rvtc->indexOid);
+   /* Check index is valid to cluster on */
+   check_index_is_clusterable(OldHeap, rvtc->indexOid);
+
+   /* rebuild_relation does all the dirty work */
+   rebuild_relation(OldHeap, rvtc->indexOid);
+
+   /* NB: rebuild_relation does heap_close() on OldHeap */
+}
+
+/*
+ * Verify that the specified index is a legitimate index to cluster on
+ *
+ * Side effect: obtains exclusive lock on the index.  The caller should
+ * already have exclusive lock on the table, so the index lock is likely
+ * redundant, but it seems best to grab it anyway to ensure the index
+ * definition can't change under us.
+ */
+void
+check_index_is_clusterable(Relation OldHeap, Oid indexOid)
+{
+   Relation    OldIndex;
+
+   OldIndex = index_open(indexOid);
    LockRelation(OldIndex, AccessExclusiveLock);
 
    /*
     * Check that index is in fact an index on the given relation
     */
    if (OldIndex->rd_index == NULL ||
-       OldIndex->rd_index->indrelid != rvtc->tableOid)
+       OldIndex->rd_index->indrelid != RelationGetRelid(OldHeap))
        ereport(ERROR,
                (errcode(ERRCODE_WRONG_OBJECT_TYPE),
                 errmsg("\"%s\" is not an index for table \"%s\"",
@@ -386,11 +408,6 @@ cluster_rel(RelToCluster *rvtc, bool recheck)
 
    /* Drop relcache refcnt on OldIndex, but keep lock */
    index_close(OldIndex);
-
-   /* rebuild_relation does all the dirty work */
-   rebuild_relation(OldHeap, rvtc->indexOid);
-
-   /* NB: rebuild_relation does heap_close() on OldHeap */
 }
 
 /*
@@ -410,13 +427,14 @@ void
 rebuild_relation(Relation OldHeap, Oid indexOid)
 {
    Oid         tableOid = RelationGetRelid(OldHeap);
+   Oid         oldClusterIndex;
    List       *indexes;
    Oid         OIDNewHeap;
    char        NewHeapName[NAMEDATALEN];
    ObjectAddress object;
 
    /* Save the information about all indexes on the relation. */
-   indexes = get_indexattr_list(OldHeap, indexOid);
+   indexes = get_indexattr_list(OldHeap, &oldClusterIndex);
 
    /* Close relcache entry, but keep lock until transaction commit */
    heap_close(OldHeap, NoLock);
@@ -469,7 +487,8 @@ rebuild_relation(Relation OldHeap, Oid indexOid)
     * Recreate each index on the relation.  We do not need
     * CommandCounterIncrement() because rebuild_indexes does it.
     */
-   rebuild_indexes(tableOid, indexes);
+   rebuild_indexes(tableOid, indexes,
+                   (OidIsValid(indexOid) ? indexOid : oldClusterIndex));
 }
 
 /*
@@ -572,14 +591,18 @@ copy_heap_data(Oid OIDNewHeap, Oid OIDOldHeap, Oid OIDOldIndex)
 
 /*
  * Get the necessary info about the indexes of the relation and
- * return a list of IndexAttrs structures.
+ * return a list of IndexAttrs structures.  Also, *OldClusterIndex
+ * is set to the OID of the existing clustered index, or InvalidOid
+ * if there is none.
  */
 List *
-get_indexattr_list(Relation OldHeap, Oid OldIndex)
+get_indexattr_list(Relation OldHeap, Oid *OldClusterIndex)
 {
    List       *indexes = NIL;
    List       *indlist;
 
+   *OldClusterIndex = InvalidOid;
+
    /* Ask the relcache to produce a list of the indexes of the old rel */
    foreach(indlist, RelationGetIndexList(OldHeap))
    {
@@ -598,16 +621,12 @@ get_indexattr_list(Relation OldHeap, Oid OldIndex)
            palloc(sizeof(Oid) * attrs->indexInfo->ii_NumIndexAttrs);
        memcpy(attrs->classOID, oldIndex->rd_index->indclass,
               sizeof(Oid) * attrs->indexInfo->ii_NumIndexAttrs);
-       /* We adjust the isclustered attribute to correct new state */
-       attrs->isclustered = (indexOID == OldIndex);
+       if (oldIndex->rd_index->indisclustered)
+           *OldClusterIndex = indexOID;
 
        index_close(oldIndex);
 
-       /*
-        * Cons the gathered data into the list.  We do not care about
-        * ordering, and this is more efficient than append.
-        */
-       indexes = lcons(attrs, indexes);
+       indexes = lappend(indexes, attrs);
    }
 
    return indexes;
@@ -616,17 +635,22 @@ get_indexattr_list(Relation OldHeap, Oid OldIndex)
 /*
  * Create new indexes and swap the filenodes with old indexes. Then drop
  * the new index (carrying the old index filenode along).
+ *
+ * OIDClusterIndex is the OID of the index to be marked as clustered, or
+ * InvalidOid if none should be marked clustered.
  */
 void
-rebuild_indexes(Oid OIDOldHeap, List *indexes)
+rebuild_indexes(Oid OIDOldHeap, List *indexes, Oid OIDClusterIndex)
 {
    List       *elem;
 
    foreach(elem, indexes)
    {
        IndexAttrs *attrs = (IndexAttrs *) lfirst(elem);
+       Oid         oldIndexOID = attrs->indexOID;
        Oid         newIndexOID;
        char        newIndexName[NAMEDATALEN];
+       bool        isclustered;
        ObjectAddress object;
        Form_pg_index index;
        HeapTuple   tuple;
@@ -634,7 +658,7 @@ rebuild_indexes(Oid OIDOldHeap, List *indexes)
 
        /* Create the new index under a temporary name */
        snprintf(newIndexName, sizeof(newIndexName),
-                "pg_temp_%u", attrs->indexOID);
+                "pg_temp_%u", oldIndexOID);
 
        /*
         * The new index will have primary and constraint status set to
@@ -654,26 +678,30 @@ rebuild_indexes(Oid OIDOldHeap, List *indexes)
        CommandCounterIncrement();
 
        /* Swap the filenodes. */
-       swap_relfilenodes(attrs->indexOID, newIndexOID);
+       swap_relfilenodes(oldIndexOID, newIndexOID);
 
        CommandCounterIncrement();
 
        /*
         * Make sure that indisclustered is correct: it should be set only
-        * for the index we just clustered on.
+        * for the index specified by the caller.
         */
+       isclustered = (oldIndexOID == OIDClusterIndex);
+
        pg_index = heap_openr(IndexRelationName, RowExclusiveLock);
        tuple = SearchSysCacheCopy(INDEXRELID,
-                                  ObjectIdGetDatum(attrs->indexOID),
+                                  ObjectIdGetDatum(oldIndexOID),
                                   0, 0, 0);
        if (!HeapTupleIsValid(tuple))
-           elog(ERROR, "cache lookup failed for index %u", attrs->indexOID);
+           elog(ERROR, "cache lookup failed for index %u", oldIndexOID);
        index = (Form_pg_index) GETSTRUCT(tuple);
-       if (index->indisclustered != attrs->isclustered)
+       if (index->indisclustered != isclustered)
        {
-           index->indisclustered = attrs->isclustered;
+           index->indisclustered = isclustered;
            simple_heap_update(pg_index, &tuple->t_self, tuple);
            CatalogUpdateIndexes(pg_index, tuple);
+           /* Ensure we see the update in the index's relcache entry */
+           CacheInvalidateRelcacheByRelid(oldIndexOID);
        }
        heap_freetuple(tuple);
        heap_close(pg_index, RowExclusiveLock);

diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index a9c307b80c30f86ebc66e28bc9e9a4c3781caec5..f19645a82604e664573efa8b8f3729fffe595a26 100644 (file)
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *   $PostgreSQL: pgsql/src/backend/commands/tablecmds.c,v 1.103 2004/05/05 04:48:45 tgl Exp $
+ *   $PostgreSQL: pgsql/src/backend/commands/tablecmds.c,v 1.104 2004/05/06 16:10:57 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -2099,12 +2099,13 @@ ATRewriteTables(List **wqueue)
            Oid         OIDNewHeap;
            char        NewHeapName[NAMEDATALEN];
            List       *indexes;
+           Oid         oldClusterIndex;
            Relation    OldHeap;
            ObjectAddress   object;
 
            /* Save the information about all indexes on the relation. */
            OldHeap = heap_open(tab->relid, NoLock);
-           indexes = get_indexattr_list(OldHeap, InvalidOid);
+           indexes = get_indexattr_list(OldHeap, &oldClusterIndex);
            heap_close(OldHeap, NoLock);
 
            /*
@@ -2148,7 +2149,7 @@ ATRewriteTables(List **wqueue)
             * Rebuild each index on the relation.  We do not need
             * CommandCounterIncrement() because rebuild_indexes does it.
             */
-           rebuild_indexes(tab->relid, indexes);
+           rebuild_indexes(tab->relid, indexes, oldClusterIndex);
        }
        else
        {
@@ -5004,6 +5005,9 @@ ATExecClusterOn(Relation rel, const char *indexName)
                 errmsg("index \"%s\" for table \"%s\" does not exist",
                        indexName, RelationGetRelationName(rel))));
 
+   /* Check index is valid to cluster on */
+   check_index_is_clusterable(rel, indexOid);
+
    indexTuple = SearchSysCache(INDEXRELID,
                                ObjectIdGetDatum(indexOid),
                                0, 0, 0);
@@ -5050,12 +5054,16 @@ ATExecClusterOn(Relation rel, const char *indexName)
            idxForm->indisclustered = false;
            simple_heap_update(pg_index, &idxtuple->t_self, idxtuple);
            CatalogUpdateIndexes(pg_index, idxtuple);
+           /* Ensure we see the update in the index's relcache entry */
+           CacheInvalidateRelcacheByRelid(indexOid);
        }
        else if (idxForm->indexrelid == indexForm->indexrelid)
        {
            idxForm->indisclustered = true;
            simple_heap_update(pg_index, &idxtuple->t_self, idxtuple);
            CatalogUpdateIndexes(pg_index, idxtuple);
+           /* Ensure we see the update in the index's relcache entry */
+           CacheInvalidateRelcacheByRelid(indexOid);
        }
        heap_freetuple(idxtuple);
    }

diff --git a/src/backend/utils/cache/inval.c b/src/backend/utils/cache/inval.c
index 3364322dd5b89eff163493c56642475b0f016eb0..7b00f0531e2c56f221e2afdbbe7b50f58d964142 100644 (file)
--- a/src/backend/utils/cache/inval.c
+++ b/src/backend/utils/cache/inval.c
@@ -74,7 +74,7 @@
  * Portions Copyright (c) 1994, Regents of the University of California
  *
  * IDENTIFICATION
- *   $PostgreSQL: pgsql/src/backend/utils/cache/inval.c,v 1.60 2004/02/10 01:55:26 tgl Exp $
+ *   $PostgreSQL: pgsql/src/backend/utils/cache/inval.c,v 1.61 2004/05/06 16:10:57 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -88,6 +88,7 @@
 #include "utils/inval.h"
 #include "utils/memutils.h"
 #include "utils/relcache.h"
+#include "utils/syscache.h"
 
 
 /*
@@ -765,6 +766,26 @@ CacheInvalidateRelcacheByTuple(HeapTuple classTuple)
    RegisterRelcacheInvalidation(databaseId, relationId, rnode);
 }
 
+/*
+ * CacheInvalidateRelcacheByRelid
+ *     As above, but relation is identified by passing its OID.
+ *     This is the least efficient of the three options; use one of
+ *     the above routines if you have a Relation or pg_class tuple.
+ */
+void
+CacheInvalidateRelcacheByRelid(Oid relid)
+{
+   HeapTuple   tup;
+
+   tup = SearchSysCache(RELOID,
+                        ObjectIdGetDatum(relid),
+                        0, 0, 0);
+   if (!HeapTupleIsValid(tup))
+       elog(ERROR, "cache lookup failed for relation %u", relid);
+   CacheInvalidateRelcacheByTuple(tup);
+   ReleaseSysCache(tup);
+}
+
 /*
  * CacheRegisterSyscacheCallback
  *     Register the specified function to be called for all future

diff --git a/src/include/commands/cluster.h b/src/include/commands/cluster.h
index 5474fdd54908cd85446a14c69d802937d36c936e..d97be66d73a731736ddd2b435045894f2260caad 100644 (file)
--- a/src/include/commands/cluster.h
+++ b/src/include/commands/cluster.h
@@ -6,7 +6,7 @@
  * Portions Copyright (c) 1996-2003, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994-5, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/include/commands/cluster.h,v 1.21 2004/05/05 04:48:47 tgl Exp $
+ * $PostgreSQL: pgsql/src/include/commands/cluster.h,v 1.22 2004/05/06 16:10:57 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -19,10 +19,12 @@
 
 extern void cluster(ClusterStmt *stmt);
 
+extern void check_index_is_clusterable(Relation OldHeap, Oid indexOid);
 extern void rebuild_relation(Relation OldHeap, Oid indexOid);
 extern Oid make_new_heap(Oid OIDOldHeap, const char *NewName);
-extern List *get_indexattr_list(Relation OldHeap, Oid OldIndex);
-extern void rebuild_indexes(Oid OIDOldHeap, List *indexes);
+extern List *get_indexattr_list(Relation OldHeap, Oid *OldClusterIndex);
+extern void rebuild_indexes(Oid OIDOldHeap, List *indexes,
+                           Oid OIDClusterIndex);
 extern void swap_relfilenodes(Oid r1, Oid r2);
 
 #endif   /* CLUSTER_H */

diff --git a/src/include/utils/inval.h b/src/include/utils/inval.h
index e7052726f27c9772b5f714b5fe429bc525329e13..a2bad9cd06ce4483e73318ee95d7b932b96192f3 100644 (file)
--- a/src/include/utils/inval.h
+++ b/src/include/utils/inval.h
@@ -7,7 +7,7 @@
  * Portions Copyright (c) 1996-2003, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/include/utils/inval.h,v 1.30 2004/02/10 01:55:26 tgl Exp $
+ * $PostgreSQL: pgsql/src/include/utils/inval.h,v 1.31 2004/05/06 16:10:57 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -32,6 +32,8 @@ extern void CacheInvalidateRelcache(Relation relation);
 
 extern void CacheInvalidateRelcacheByTuple(HeapTuple classTuple);
 
+extern void CacheInvalidateRelcacheByRelid(Oid relid);
+
 extern void CacheRegisterSyscacheCallback(int cacheid,
                              CacheCallbackFunction func,
                              Datum arg);