Index: doc/src/sgml/ref/cluster.sgml
===================================================================
RCS file: /projects/cvsroot/pgsql-server/doc/src/sgml/ref/cluster.sgml,v
retrieving revision 1.19
diff -c -r1.19 cluster.sgml
*** doc/src/sgml/ref/cluster.sgml	2002/08/11 02:43:57	1.19
--- doc/src/sgml/ref/cluster.sgml	2002/09/02 05:14:11
***************
*** 21,27 ****
     <date>1999-07-20</date>
    </refsynopsisdivinfo>
    <synopsis>
! CLUSTER <replaceable class="PARAMETER">indexname</replaceable> ON <replaceable class="PARAMETER">tablename</replaceable>
    </synopsis>
  
    <refsect2 id="R2-SQL-CLUSTER-1">
--- 21,27 ----
     <date>1999-07-20</date>
    </refsynopsisdivinfo>
    <synopsis>
! CLUSTER [ <replaceable class="PARAMETER">indexname</replaceable> ON <replaceable class="PARAMETER">tablename</replaceable> ]
    </synopsis>
  
    <refsect2 id="R2-SQL-CLUSTER-1">
***************
*** 104,109 ****
--- 104,117 ----
     periodically re-cluster by issuing the command again.
    </para>
  
+   <para>
+    When a table is clustered, <productname>PostgreSQL</productname>
+    remembers on which index it was clustered.  In calls to
+    <command>CLUSTER</command> without parameters, all the tables
+    that have been previously clustered will be clustered again on
+    the same indexes, using the saved information.
+   </para>
+ 
    <refsect2 id="R2-SQL-CLUSTER-3">
     <refsect2info>
      <date>1998-09-08</date>
***************
*** 140,148 ****
      sizes.
     </para>
  
     <para>
!     CLUSTER preserves GRANT, inheritance, index, foreign key, and other
!     ancillary information about the table.
     </para>
  
     <para>
--- 148,163 ----
      sizes.
     </para>
  
+    <para>
+ 	<command>CLUSTER</command> preserves GRANT, inheritance, index, foreign
+ 	key, and other ancillary information about the table.
+    </para>
+ 
     <para>
! 	Because <command>CLUSTER</command> remembers the clustering information,
! 	one can cluster the tables one wants clustered the first time manually, and
! 	setup a timed event similar to <command>VACUUM</command> so that the tables
! 	are periodically and automatically clustered.
     </para>
  
     <para>
Index: src/backend/commands/cluster.c
===================================================================
RCS file: /projects/cvsroot/pgsql-server/src/backend/commands/cluster.c,v
retrieving revision 1.87
diff -c -r1.87 cluster.c
*** src/backend/commands/cluster.c	2002/08/27 03:38:27	1.87
--- src/backend/commands/cluster.c	2002/09/02 05:14:13
***************
*** 25,33 ****
--- 25,35 ----
  #include "catalog/index.h"
  #include "catalog/indexing.h"
  #include "catalog/catname.h"
+ #include "catalog/namespace.h"
  #include "commands/cluster.h"
  #include "commands/tablecmds.h"
  #include "miscadmin.h"
+ #include "utils/acl.h"
  #include "utils/fmgroids.h"
  #include "utils/lsyscache.h"
  #include "utils/syscache.h"
***************
*** 48,59 ****
--- 50,76 ----
  	bool		isclustered;
  } IndexAttrs;
  
+ /* This struct is used to pass around the information on tables to be
+  * clustered. We need this so we can make a list of them when invoked without
+  * a specific table/index pair.
+  */
+ typedef struct
+ {
+ 	Oid		tableOid;
+ 	Oid		indexOid;
+ 	bool	isPrevious;
+ } relToCluster;
+ 
  static Oid	make_new_heap(Oid OIDOldHeap, const char *NewName);
  static void copy_heap_data(Oid OIDNewHeap, Oid OIDOldHeap, Oid OIDOldIndex);
  static List *get_indexattr_list(Relation OldHeap, Oid OldIndex);
  static void recreate_indexattr(Oid OIDOldHeap, List *indexes);
  static void swap_relfilenodes(Oid r1, Oid r2);
+ static void cluster_rel(relToCluster *rv);
+ static bool check_cluster_ownership(Oid relOid);
+ static List *get_tables_to_cluster(Oid owner);
  
+ static MemoryContext cluster_context = NULL;
  
  /*
   * cluster
***************
*** 69,111 ****
   * the new table, it's better to create the indexes afterwards than to fill
   * them incrementally while we load the table.
   *
!  * Permissions checks were done already.
   */
  void
! cluster(RangeVar *oldrelation, char *oldindexname)
  {
! 	Oid			OIDOldHeap,
! 				OIDOldIndex,
! 				OIDNewHeap;
  	Relation	OldHeap,
  				OldIndex;
  	char		NewHeapName[NAMEDATALEN];
  	ObjectAddress object;
  	List	   *indexes;
  
  	/*
  	 * We grab exclusive access to the target rel and index for the
  	 * duration of the transaction.
  	 */
! 	OldHeap = heap_openrv(oldrelation, AccessExclusiveLock);
! 	OIDOldHeap = RelationGetRelid(OldHeap);
  
! 	/*
! 	 * The index is expected to be in the same namespace as the relation.
! 	 */
! 	OIDOldIndex = get_relname_relid(oldindexname,
! 									RelationGetNamespace(OldHeap));
! 	if (!OidIsValid(OIDOldIndex))
! 		elog(ERROR, "CLUSTER: cannot find index \"%s\" for table \"%s\"",
! 			 oldindexname, RelationGetRelationName(OldHeap));
! 	OldIndex = index_open(OIDOldIndex);
  	LockRelation(OldIndex, AccessExclusiveLock);
  
  	/*
  	 * Check that index is in fact an index on the given relation
  	 */
  	if (OldIndex->rd_index == NULL ||
! 		OldIndex->rd_index->indrelid != OIDOldHeap)
  		elog(ERROR, "CLUSTER: \"%s\" is not an index for table \"%s\"",
  			 RelationGetRelationName(OldIndex),
  			 RelationGetRelationName(OldHeap));
--- 86,161 ----
   * the new table, it's better to create the indexes afterwards than to fill
   * them incrementally while we load the table.
   *
!  * Since we open a new transaction, we have to check that the relation
!  * still is what we think it is.
   */
  void
! cluster_rel(relToCluster *rvtc)
  {
! 	Oid			OIDNewHeap;
  	Relation	OldHeap,
  				OldIndex;
  	char		NewHeapName[NAMEDATALEN];
  	ObjectAddress object;
  	List	   *indexes;
  
+ 	/* Start a new transaction for this relation. Every normal
+ 	 * return of this function should leave this transaction open
+ 	 * for future work.
+ 	 */
+ 	StartTransactionCommand();
+ 
+ 	/* Check for user-requested abort. */
+ 	CHECK_FOR_INTERRUPTS();
+ 
+ 	/* Check if the relation and index still exist before opening them
+ 	 */
+ 	if (!SearchSysCacheExists(RELOID,
+ 							  ObjectIdGetDatum(rvtc->tableOid),
+ 							  0, 0, 0) ||
+ 			!SearchSysCacheExists(RELOID,
+ 								  ObjectIdGetDatum(rvtc->indexOid),
+ 								  0, 0, 0))
+ 		return;
+ 
+ 	/* Check that the user still owns the relation */
+ 	if (!check_cluster_ownership(rvtc->tableOid))
+ 		return;
+ 
+ 	/* Check that the index is still the one with indisclustered set.
+ 	 * If this is a standalone cluster, skip this test.
+ 	 */
+ 	if (rvtc->isPrevious)
+ 	{
+ 		HeapTuple		tuple;
+ 		Form_pg_index	indexForm;
+ 
+ 		tuple = SearchSysCache(INDEXRELID,
+ 							   ObjectIdGetDatum(rvtc->indexOid),
+ 							   0, 0, 0);
+ 		indexForm = (Form_pg_index) GETSTRUCT(tuple);
+ 		if (!indexForm->indisclustered)
+ 		{
+ 			ReleaseSysCache(tuple);
+ 			return;
+ 		}
+ 		ReleaseSysCache(tuple);
+ 	}
+ 
  	/*
  	 * We grab exclusive access to the target rel and index for the
  	 * duration of the transaction.
  	 */
! 	OldHeap = heap_open(rvtc->tableOid, AccessExclusiveLock);
  
! 	OldIndex = index_open(rvtc->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)
  		elog(ERROR, "CLUSTER: \"%s\" is not an index for table \"%s\"",
  			 RelationGetRelationName(OldIndex),
  			 RelationGetRelationName(OldHeap));
***************
*** 122,128 ****
  			 RelationGetRelationName(OldHeap));
  
  	/* Save the information of all indexes on the relation. */
! 	indexes = get_indexattr_list(OldHeap, OIDOldIndex);
  
  	/* Drop relcache refcnts, but do NOT give up the locks */
  	index_close(OldIndex);
--- 172,178 ----
  			 RelationGetRelationName(OldHeap));
  
  	/* Save the information of all indexes on the relation. */
! 	indexes = get_indexattr_list(OldHeap, rvtc->indexOid);
  
  	/* Drop relcache refcnts, but do NOT give up the locks */
  	index_close(OldIndex);
***************
*** 135,156 ****
  	 * particular, we can't create the new heap in a different namespace from
  	 * the old, or we will have problems with the TEMP status of temp tables.
  	 */
! 	snprintf(NewHeapName, NAMEDATALEN, "pg_temp_%u", OIDOldHeap);
  
! 	OIDNewHeap = make_new_heap(OIDOldHeap, NewHeapName);
  
  	/* We don't need CommandCounterIncrement() because make_new_heap did it. */
  
  	/*
  	 * Copy the heap data into the new table in the desired order.
  	 */
! 	copy_heap_data(OIDNewHeap, OIDOldHeap, OIDOldIndex);
  
  	/* To make the new heap's data visible (probably not needed?). */
  	CommandCounterIncrement();
  
  	/* Swap the relfilenodes of the old and new heaps. */
! 	swap_relfilenodes(OIDOldHeap, OIDNewHeap);
  
  	CommandCounterIncrement();
  
--- 185,206 ----
  	 * particular, we can't create the new heap in a different namespace from
  	 * the old, or we will have problems with the TEMP status of temp tables.
  	 */
! 	snprintf(NewHeapName, NAMEDATALEN, "pg_temp_%u", rvtc->tableOid);
  
! 	OIDNewHeap = make_new_heap(rvtc->tableOid, NewHeapName);
  
  	/* We don't need CommandCounterIncrement() because make_new_heap did it. */
  
  	/*
  	 * Copy the heap data into the new table in the desired order.
  	 */
! 	copy_heap_data(OIDNewHeap, rvtc->tableOid, rvtc->indexOid);
  
  	/* To make the new heap's data visible (probably not needed?). */
  	CommandCounterIncrement();
  
  	/* Swap the relfilenodes of the old and new heaps. */
! 	swap_relfilenodes(rvtc->tableOid, OIDNewHeap);
  
  	CommandCounterIncrement();
  
***************
*** 171,177 ****
  	 * Recreate each index on the relation.  We do not need
  	 * CommandCounterIncrement() because recreate_indexattr does it.
  	 */
! 	recreate_indexattr(OIDOldHeap, indexes);
  }
  
  /*
--- 221,227 ----
  	 * Recreate each index on the relation.  We do not need
  	 * CommandCounterIncrement() because recreate_indexattr does it.
  	 */
! 	recreate_indexattr(rvtc->tableOid, indexes);
  }
  
  /*
***************
*** 549,552 ****
--- 599,790 ----
  	heap_freetuple(reltup2);
  
  	heap_close(relRelation, RowExclusiveLock);
+ }
+ 
+ /* This cluster code allows for clustering multiple tables at once. Because
+  * of this, we cannot just run everything on a single transaction, or we
+  * would be forced to acquire exclusive locks on all the tables being
+  * clustered.  To solve this we follow a similar strategy to VACUUM code,
+  * clustering each relation in a separate transaction. For this to work,
+  * we need to:
+  *  - provide a separate memory context so that we can pass information in
+  *    a way that trascends transactions
+  *  - start a new transaction every time a new relation is clustered
+  *  - check for validity of the information on to-be-clustered relations,
+  *    as someone might have deleted a relation behind our back, or
+  *    clustered one on a different index
+  *  - end a new transaction
+  *
+  *  This adds some overhead for the single-relation clustering operation.
+  *
+  */
+ void
+ cluster(ClusterStmt *stmt)
+ {
+ 
+ 	/*
+ 	 * We cannot run CLUSTER inside a user transaction block; if we were inside
+ 	 * a transaction, then our commit- and start-transaction-command calls
+ 	 * would not have the intended effect!  Furthermore, the forced commit that
+ 	 * occurs before truncating the relation's file would have the effect of
+ 	 * committing the rest of the user's transaction too, which would certainly
+ 	 * not be the desired behavior.
+ 	 */
+ 	if (IsTransactionBlock())
+ 		elog(ERROR, "CLUSTER cannot run inside a BEGIN/END block");
+ 
+ 	/* Running CLUSTER from a function would free the function context */
+ 	if (!MemoryContextContains(QueryContext, stmt))
+ 		elog(ERROR, "CLUSTER cannot be called from a function");
+ 
+ 	if (stmt->relation != NULL)
+ 	{
+ 		Oid				indexOid,
+ 						tableOid;
+ 		relToCluster	rvtc;
+ 		HeapTuple		tuple;
+ 
+ 		tableOid = RangeVarGetRelid(stmt->relation, false);
+ 		if (!check_cluster_ownership(tableOid))
+ 			elog(ERROR, "CLUSTER: You do not own relation %s",
+ 					stmt->relation->relname);
+ 
+ 		/* The index is expected to be in the same namespace as the
+ 		 * relation. */
+ 		tuple = SearchSysCache(RELOID,
+ 							   ObjectIdGetDatum(tableOid),
+ 							   0, 0, 0);
+ 		if (!HeapTupleIsValid(tuple))
+ 			elog(ERROR, "Cache lookup failed for relation %u", tableOid);
+ 		indexOid = get_relname_relid(stmt->indexname,
+ 				                     ((Form_pg_class) GETSTRUCT(tuple))->relnamespace);
+ 		ReleaseSysCache(tuple);
+ 
+ 		/* XXX The namespace should be reported as well */
+ 		if (!OidIsValid(indexOid))
+ 			elog(ERROR, "CLUSTER: cannot find index \"%s\" for table \"%s\"",
+ 					stmt->indexname, stmt->relation->relname);
+ 		rvtc.tableOid = tableOid;
+ 		rvtc.indexOid = indexOid;
+ 		rvtc.isPrevious = false;
+ 
+ 		/* End the current transaction so the cluster_rel logic transaction
+ 		 * management can cope with this.  This is needed to match the behavior
+ 		 * needed by multi-relation clustering.
+ 		 */
+ 		CommitTransactionCommand();
+ 
+ 		/* Do the job */
+ 		cluster_rel(&rvtc);
+ 	}
+ 	else
+ 	{
+ 		relToCluster	*rvtc;
+ 		List			*rv,
+ 						*rvs;
+ 
+ 		/* Create special memory context for cross-transaction storage.
+ 		 *
+ 		 * Since it is a child of QueryContext, it will go away even in case
+ 		 * of error.
+ 		 */
+ 		cluster_context = AllocSetContextCreate(QueryContext,
+ 										  "Cluster",
+ 										  ALLOCSET_DEFAULT_MINSIZE,
+ 										  ALLOCSET_DEFAULT_INITSIZE,
+ 										  ALLOCSET_DEFAULT_MAXSIZE);
+ 
+ 		/* Build the list of relations to cluster.  Note that this lives in
+ 		 * cluster_context
+ 		 */
+ 		rvs = get_tables_to_cluster(GetUserId());
+ 
+ 		/* Ok, now that we've got them all, cluster them one by one */
+ 		foreach (rv, rvs)
+ 		{
+ 			rvtc = (relToCluster *)lfirst(rv);
+ 
+ 			/* End the current transaction so that cluster_rel can start
+ 			 * a new one for each relation.
+ 			 */
+ 			CommitTransactionCommand();
+ 
+ 			cluster_rel(rvtc);
+ 		}
+ 	}
+ 
+ 	/* The last cluster_rel left us with an open TransactionCommand */
+ 
+ 	/* Clean up working storage */
+ 	if (stmt->relation == NULL)
+ 	{
+ 		MemoryContextDelete(cluster_context);
+ 		cluster_context = NULL;
+ 	}
+ }
+ 
+ /* Checks if the user owns the relation. Superusers
+  * are allowed to cluster any table.
+  */
+ bool
+ check_cluster_ownership(Oid relOid)
+ {
+ 	/* Superusers bypass this check */
+ 	return pg_class_ownercheck(relOid, GetUserId());
+ }
+ 
+ /* Get a list of tables that the current user owns and
+  * have indisclustered set.  Return the list in a List * of rvsToCluster
+  * with the tableOid and the indexOid on which the table is already 
+  * clustered.
+  */
+ List *
+ get_tables_to_cluster(Oid owner)
+ {
+ 	Relation		indRelation;
+ 	HeapScanDesc	scan;
+ 	ScanKeyData		entry;
+ 	HeapTuple		indexTuple;
+ 	Form_pg_index	index;
+ 	relToCluster   *rvtc;
+ 	List		   *rvs = NIL;
+ 
+ 	/* Get all indexes that have indisclustered set.
+ 	 * System relations or nailed-in relations cannot ever have indisclustered
+ 	 * set, because CLUSTER will refuse to set it when called with one of them
+ 	 * as argument.
+ 	 */
+ 	indRelation = relation_openr(IndexRelationName, RowExclusiveLock);
+ 	ScanKeyEntryInitialize(&entry, 0, Anum_pg_index_indisclustered,
+ 						   F_BOOLEQ, true);
+ 	scan = heap_beginscan(indRelation, SnapshotNow, 1, &entry);
+ 	while ((indexTuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
+ 	{
+ 		MemoryContext	old_context = NULL;
+ 
+ 		index = (Form_pg_index) GETSTRUCT(indexTuple);
+ 		if (!check_cluster_ownership(index->indrelid))
+ 			continue;
+ 
+ 		/* We have to build the struct in a different memory context so
+ 		 * it will survive the cross-transaction processing
+ 		 */
+ 
+ 		old_context = MemoryContextSwitchTo(cluster_context);
+ 
+ 		rvtc = (relToCluster *)palloc(sizeof(relToCluster));
+ 		rvtc->indexOid = index->indexrelid;
+ 		rvtc->tableOid = index->indrelid;
+ 		rvtc->isPrevious = true;
+ 		rvs = lcons((void *)rvtc, rvs);
+ 
+ 		MemoryContextSwitchTo(old_context);
+ 	}
+ 	heap_endscan(scan);
+ 
+ 	/* Release the lock on pg_index. We will check again later
+ 	 * the indexes anyway.
+ 	 */
+ 	relation_close(indRelation, RowExclusiveLock);
+ 	return rvs;
  }
Index: src/backend/parser/gram.y
===================================================================
RCS file: /projects/cvsroot/pgsql-server/src/backend/parser/gram.y,v
retrieving revision 2.364
diff -c -r2.364 gram.y
*** src/backend/parser/gram.y	2002/08/29 00:17:04	2.364
--- src/backend/parser/gram.y	2002/09/02 05:14:31
***************
*** 3740,3746 ****
  /*****************************************************************************
   *
   *		QUERY:
!  *				cluster <index_name> on <qualified_name>
   *
   *****************************************************************************/
  
--- 3740,3746 ----
  /*****************************************************************************
   *
   *		QUERY:
!  *				cluster [ <index_name> on <qualified_name> ]
   *
   *****************************************************************************/
  
***************
*** 3750,3755 ****
--- 3750,3762 ----
  				   ClusterStmt *n = makeNode(ClusterStmt);
  				   n->relation = $4;
  				   n->indexname = $2;
+ 				   $$ = (Node*)n;
+ 				}
+ 			| CLUSTER
+ 			    {
+ 				   ClusterStmt *n = makeNode(ClusterStmt);
+ 				   n->relation = NULL;
+ 				   n->indexname = NULL;
  				   $$ = (Node*)n;
  				}
  		;
Index: src/backend/tcop/utility.c
===================================================================
RCS file: /projects/cvsroot/pgsql-server/src/backend/tcop/utility.c,v
retrieving revision 1.175
diff -c -r1.175 utility.c
*** src/backend/tcop/utility.c	2002/08/30 19:23:20	1.175
--- src/backend/tcop/utility.c	2002/09/02 05:14:33
***************
*** 695,703 ****
  			{
  				ClusterStmt *stmt = (ClusterStmt *) parsetree;
  
! 				CheckOwnership(stmt->relation, true);
! 
! 				cluster(stmt->relation, stmt->indexname);
  			}
  			break;
  
--- 695,701 ----
  			{
  				ClusterStmt *stmt = (ClusterStmt *) parsetree;
  
! 				cluster(stmt);
  			}
  			break;
  
Index: src/include/commands/cluster.h
===================================================================
RCS file: /projects/cvsroot/pgsql-server/src/include/commands/cluster.h,v
retrieving revision 1.15
diff -c -r1.15 cluster.h
*** src/include/commands/cluster.h	2002/08/10 21:00:34	1.15
--- src/include/commands/cluster.h	2002/09/02 05:14:37
***************
*** 13,21 ****
  #ifndef CLUSTER_H
  #define CLUSTER_H
  
  /*
   * functions
   */
! extern void cluster(RangeVar *oldrelation, char *oldindexname);
  
  #endif   /* CLUSTER_H */
--- 13,22 ----
  #ifndef CLUSTER_H
  #define CLUSTER_H
  
+ #include <nodes/parsenodes.h>
  /*
   * functions
   */
! extern void cluster(ClusterStmt *stmt);
  
  #endif   /* CLUSTER_H */
Index: src/include/nodes/parsenodes.h
===================================================================
RCS file: /projects/cvsroot/pgsql-server/src/include/nodes/parsenodes.h,v
retrieving revision 1.203
diff -c -r1.203 parsenodes.h
*** src/include/nodes/parsenodes.h	2002/08/30 19:23:20	1.203
--- src/include/nodes/parsenodes.h	2002/09/02 05:14:40
***************
*** 1466,1472 ****
  typedef struct ClusterStmt
  {
  	NodeTag		type;
! 	RangeVar   *relation;		/* relation being indexed */
  	char	   *indexname;		/* original index defined */
  } ClusterStmt;
  
--- 1466,1472 ----
  typedef struct ClusterStmt
  {
  	NodeTag		type;
! 	RangeVar   *relation;		/* relation being indexed, or NULL if all */
  	char	   *indexname;		/* original index defined */
  } ClusterStmt;
  
Index: src/test/regress/expected/cluster.out
===================================================================
RCS file: /projects/cvsroot/pgsql-server/src/test/regress/expected/cluster.out,v
retrieving revision 1.3
diff -c -r1.3 cluster.out
*** src/test/regress/expected/cluster.out	2002/08/27 03:38:27	1.3
--- src/test/regress/expected/cluster.out	2002/09/02 05:14:45
***************
*** 285,287 ****
--- 285,351 ----
   clstr_tst_c
  (1 row)
  
+ -- Verify that clustering all tables does in fact cluster the right ones
+ CREATE USER clstr_user;
+ CREATE TABLE clstr_1 (a INT PRIMARY KEY);
+ NOTICE:  CREATE TABLE / PRIMARY KEY will create implicit index 'clstr_1_pkey' for table 'clstr_1'
+ CREATE TABLE clstr_2 (a INT PRIMARY KEY);
+ NOTICE:  CREATE TABLE / PRIMARY KEY will create implicit index 'clstr_2_pkey' for table 'clstr_2'
+ CREATE TABLE clstr_3 (a INT PRIMARY KEY);
+ NOTICE:  CREATE TABLE / PRIMARY KEY will create implicit index 'clstr_3_pkey' for table 'clstr_3'
+ ALTER TABLE clstr_1 OWNER TO clstr_user;
+ ALTER TABLE clstr_3 OWNER TO clstr_user;
+ GRANT SELECT ON clstr_2 TO clstr_user;
+ INSERT INTO clstr_1 VALUES (2);
+ INSERT INTO clstr_1 VALUES (1);
+ INSERT INTO clstr_2 VALUES (2);
+ INSERT INTO clstr_2 VALUES (1);
+ INSERT INTO clstr_3 VALUES (2);
+ INSERT INTO clstr_3 VALUES (1);
+ CLUSTER clstr_1_pkey ON clstr_1;
+ CLUSTER clstr_2_pkey ON clstr_2;
+ SELECT * FROM clstr_1 UNION ALL
+   SELECT * FROM clstr_2 UNION ALL
+   SELECT * FROM clstr_3;
+  a 
+ ---
+  1
+  2
+  1
+  2
+  2
+  1
+ (6 rows)
+ 
+ -- revert to the original state
+ DELETE FROM clstr_1;
+ DELETE FROM clstr_2;
+ DELETE FROM clstr_3;
+ INSERT INTO clstr_1 VALUES (2);
+ INSERT INTO clstr_1 VALUES (1);
+ INSERT INTO clstr_2 VALUES (2);
+ INSERT INTO clstr_2 VALUES (1);
+ INSERT INTO clstr_3 VALUES (2);
+ INSERT INTO clstr_3 VALUES (1);
+ -- this user can only cluster clstr_1 and clstr_3, but the latter
+ -- has not been clustered
+ SET SESSION AUTHORIZATION clstr_user;
+ CLUSTER;
+ SELECT * FROM clstr_1 UNION ALL
+   SELECT * FROM clstr_2 UNION ALL
+   SELECT * FROM clstr_3;
+  a 
+ ---
+  1
+  2
+  2
+  1
+  2
+  1
+ (6 rows)
+ 
+ -- clean up
+ \c -
+ DROP TABLE clstr_1;
+ DROP TABLE clstr_3;
+ DROP USER clstr_user;
Index: src/test/regress/sql/cluster.sql
===================================================================
RCS file: /projects/cvsroot/pgsql-server/src/test/regress/sql/cluster.sql,v
retrieving revision 1.4
diff -c -r1.4 cluster.sql
*** src/test/regress/sql/cluster.sql	2002/08/27 03:38:28	1.4
--- src/test/regress/sql/cluster.sql	2002/09/02 05:14:45
***************
*** 86,88 ****
--- 86,132 ----
  	AND indrelid=pg_class_2.oid
  	AND pg_class_2.relname = 'clstr_tst'
  	AND indisclustered;
+ 
+ -- Verify that clustering all tables does in fact cluster the right ones
+ CREATE USER clstr_user;
+ CREATE TABLE clstr_1 (a INT PRIMARY KEY);
+ CREATE TABLE clstr_2 (a INT PRIMARY KEY);
+ CREATE TABLE clstr_3 (a INT PRIMARY KEY);
+ ALTER TABLE clstr_1 OWNER TO clstr_user;
+ ALTER TABLE clstr_3 OWNER TO clstr_user;
+ GRANT SELECT ON clstr_2 TO clstr_user;
+ INSERT INTO clstr_1 VALUES (2);
+ INSERT INTO clstr_1 VALUES (1);
+ INSERT INTO clstr_2 VALUES (2);
+ INSERT INTO clstr_2 VALUES (1);
+ INSERT INTO clstr_3 VALUES (2);
+ INSERT INTO clstr_3 VALUES (1);
+ CLUSTER clstr_1_pkey ON clstr_1;
+ CLUSTER clstr_2_pkey ON clstr_2;
+ SELECT * FROM clstr_1 UNION ALL
+   SELECT * FROM clstr_2 UNION ALL
+   SELECT * FROM clstr_3;
+ 
+ -- revert to the original state
+ DELETE FROM clstr_1;
+ DELETE FROM clstr_2;
+ DELETE FROM clstr_3;
+ INSERT INTO clstr_1 VALUES (2);
+ INSERT INTO clstr_1 VALUES (1);
+ INSERT INTO clstr_2 VALUES (2);
+ INSERT INTO clstr_2 VALUES (1);
+ INSERT INTO clstr_3 VALUES (2);
+ INSERT INTO clstr_3 VALUES (1);
+ 
+ -- this user can only cluster clstr_1 and clstr_3, but the latter
+ -- has not been clustered
+ SET SESSION AUTHORIZATION clstr_user;
+ CLUSTER;
+ SELECT * FROM clstr_1 UNION ALL
+   SELECT * FROM clstr_2 UNION ALL
+   SELECT * FROM clstr_3;
+ -- clean up
+ \c -
+ DROP TABLE clstr_1;
+ DROP TABLE clstr_3;
+ DROP USER clstr_user;
