cluster-without-index_20091023.patch

diff -cprN head/src/backend/commands/cluster.c work/src/backend/commands/cluster.c
*** head/src/backend/commands/cluster.c	2009-10-06 04:24:36.000000000 +0900
--- work/src/backend/commands/cluster.c	2009-10-23 11:43:39.105505012 +0900
*************** cluster(ClusterStmt *stmt, bool isTopLev
*** 122,128 ****
  					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
  			   errmsg("cannot cluster temporary tables of other sessions")));
  
! 		if (stmt->indexname == NULL)
  		{
  			ListCell   *index;
  
--- 122,131 ----
  					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
  			   errmsg("cannot cluster temporary tables of other sessions")));
  
! 		if (stmt->without_index)
! 		{
! 		}
! 		else if (stmt->indexname == NULL)
  		{
  			ListCell   *index;
  
*************** cluster_rel(RelToCluster *rvtc, bool rec
*** 308,347 ****
  			return;
  		}
  
! 		/*
! 		 * Check that the index still exists
! 		 */
! 		if (!SearchSysCacheExists(RELOID,
! 								  ObjectIdGetDatum(rvtc->indexOid),
! 								  0, 0, 0))
  		{
! 			relation_close(OldHeap, AccessExclusiveLock);
! 			return;
! 		}
  
! 		/*
! 		 * Check that the index is still the one with indisclustered set.
! 		 */
! 		tuple = SearchSysCache(INDEXRELID,
! 							   ObjectIdGetDatum(rvtc->indexOid),
! 							   0, 0, 0);
! 		if (!HeapTupleIsValid(tuple))	/* probably can't happen */
! 		{
! 			relation_close(OldHeap, AccessExclusiveLock);
! 			return;
! 		}
! 		indexForm = (Form_pg_index) GETSTRUCT(tuple);
! 		if (!indexForm->indisclustered)
! 		{
  			ReleaseSysCache(tuple);
- 			relation_close(OldHeap, AccessExclusiveLock);
- 			return;
  		}
- 		ReleaseSysCache(tuple);
  	}
  
  	/* Check index is valid to cluster on */
! 	check_index_is_clusterable(OldHeap, rvtc->indexOid, recheck);
  
  	/* rebuild_relation does all the dirty work */
  	ereport(verbose ? INFO : DEBUG2,
--- 311,350 ----
  			return;
  		}
  
! 		if (OidIsValid(rvtc->indexOid))
  		{
! 			/* Check that the index still exists */
! 			if (!SearchSysCacheExists(RELOID,
! 									  ObjectIdGetDatum(rvtc->indexOid),
! 									  0, 0, 0))
! 			{
! 				relation_close(OldHeap, AccessExclusiveLock);
! 				return;
! 			}
  
! 			/* Check that the index is still the one with indisclustered set. */
! 			tuple = SearchSysCache(INDEXRELID,
! 								   ObjectIdGetDatum(rvtc->indexOid),
! 								   0, 0, 0);
! 			if (!HeapTupleIsValid(tuple))	/* probably can't happen */
! 			{
! 				relation_close(OldHeap, AccessExclusiveLock);
! 				return;
! 			}
! 			indexForm = (Form_pg_index) GETSTRUCT(tuple);
! 			if (!indexForm->indisclustered)
! 			{
! 				ReleaseSysCache(tuple);
! 				relation_close(OldHeap, AccessExclusiveLock);
! 				return;
! 			}
  			ReleaseSysCache(tuple);
  		}
  	}
  
  	/* Check index is valid to cluster on */
! 	if (OidIsValid(rvtc->indexOid))
! 		check_index_is_clusterable(OldHeap, rvtc->indexOid, recheck);
  
  	/* rebuild_relation does all the dirty work */
  	ereport(verbose ? INFO : DEBUG2,
*************** rebuild_relation(Relation OldHeap, Oid i
*** 576,582 ****
  	Relation	newrel;
  
  	/* Mark the correct index as clustered */
! 	mark_index_clustered(OldHeap, indexOid);
  
  	/* Close relcache entry, but keep lock until transaction commit */
  	heap_close(OldHeap, NoLock);
--- 579,586 ----
  	Relation	newrel;
  
  	/* Mark the correct index as clustered */
! 	if (OidIsValid(indexOid))
! 		mark_index_clustered(OldHeap, indexOid);
  
  	/* Close relcache entry, but keep lock until transaction commit */
  	heap_close(OldHeap, NoLock);
*************** make_new_heap(Oid OIDOldHeap, const char
*** 753,758 ****
--- 757,834 ----
  	return OIDNewHeap;
  }
  
+ typedef struct ScanDescData
+ {
+ 	bool	isIndexScan;
+ 	void   *desc;
+ } ScanDescData;
+ 
+ typedef struct ScanDescData *ScanDesc;
+ 
+ static ScanDesc
+ genam_beginscan(Relation heapRelation,
+ 				Relation indexRelation,
+ 				Snapshot snapshot,
+ 				int nkeys, ScanKey key)
+ {
+ 	ScanDesc scan = palloc(sizeof(ScanDescData));
+ 
+ 	if (indexRelation)
+ 	{
+ 		scan->isIndexScan = true;
+ 		scan->desc = index_beginscan(heapRelation, indexRelation,
+ 									 snapshot, nkeys, key);
+ 	}
+ 	else
+ 	{
+ 		/* Do not allow syncscan not to change physical order of tuples. */
+ 		scan->isIndexScan = false;
+ 		scan->desc = heap_beginscan_strat(heapRelation,
+ 										  snapshot, nkeys, key, true, false);
+ 	}
+ 
+ 	return scan;
+ }
+ 
+ static void
+ genam_endscan(ScanDesc scan)
+ {
+ 	if (scan->isIndexScan)
+ 		index_endscan((IndexScanDesc) scan->desc);
+ 	else
+ 		heap_endscan((HeapScanDesc) scan->desc);
+ 	pfree(scan);
+ }
+ 
+ static HeapTuple
+ genam_getnext(ScanDesc scan, ScanDirection direction, Buffer *buffer)
+ {
+ 	HeapTuple	tuple;
+ 
+ 	if (scan->isIndexScan)
+ 	{
+ 		IndexScanDesc	scandesc = (IndexScanDesc) scan->desc;
+ 
+ 		tuple = index_getnext(scandesc, direction);
+ 
+ 		/* Since we used no scan keys, should never need to recheck */
+ 		if (scandesc->xs_recheck)
+ 			elog(ERROR, "CLUSTER does not support lossy index conditions");
+ 
+ 		*buffer = scandesc->xs_cbuf;
+ 	}
+ 	else
+ 	{
+ 		HeapScanDesc	scandesc = (HeapScanDesc) scan->desc;
+ 
+ 		tuple = heap_getnext(scandesc, direction);
+ 
+ 		*buffer = scandesc->rs_cbuf;
+ 	}
+ 
+ 	return tuple;
+ }
+ 
  /*
   * Do the physical copying of heap data.  Returns the TransactionId used as
   * freeze cutoff point for the tuples.
*************** copy_heap_data(Oid OIDNewHeap, Oid OIDOl
*** 768,775 ****
  	int			natts;
  	Datum	   *values;
  	bool	   *isnull;
! 	IndexScanDesc scan;
  	HeapTuple	tuple;
  	bool		use_wal;
  	TransactionId OldestXmin;
  	TransactionId FreezeXid;
--- 844,852 ----
  	int			natts;
  	Datum	   *values;
  	bool	   *isnull;
! 	ScanDesc	scan;
  	HeapTuple	tuple;
+ 	Buffer		buf;
  	bool		use_wal;
  	TransactionId OldestXmin;
  	TransactionId FreezeXid;
*************** copy_heap_data(Oid OIDNewHeap, Oid OIDOl
*** 780,786 ****
  	 */
  	NewHeap = heap_open(OIDNewHeap, AccessExclusiveLock);
  	OldHeap = heap_open(OIDOldHeap, AccessExclusiveLock);
! 	OldIndex = index_open(OIDOldIndex, AccessExclusiveLock);
  
  	/*
  	 * Their tuple descriptors should be exactly alike, but here we only need
--- 857,866 ----
  	 */
  	NewHeap = heap_open(OIDNewHeap, AccessExclusiveLock);
  	OldHeap = heap_open(OIDOldHeap, AccessExclusiveLock);
! 	if (OidIsValid(OIDOldIndex))
! 		OldIndex = index_open(OIDOldIndex, AccessExclusiveLock);
! 	else
! 		OldIndex = NULL;
  
  	/*
  	 * Their tuple descriptors should be exactly alike, but here we only need
*************** copy_heap_data(Oid OIDNewHeap, Oid OIDOl
*** 828,837 ****
  	 * copied, we scan with SnapshotAny and use HeapTupleSatisfiesVacuum for
  	 * the visibility test.
  	 */
! 	scan = index_beginscan(OldHeap, OldIndex,
! 						   SnapshotAny, 0, (ScanKey) NULL);
  
! 	while ((tuple = index_getnext(scan, ForwardScanDirection)) != NULL)
  	{
  		HeapTuple	copiedTuple;
  		bool		isdead;
--- 908,916 ----
  	 * copied, we scan with SnapshotAny and use HeapTupleSatisfiesVacuum for
  	 * the visibility test.
  	 */
! 	scan = genam_beginscan(OldHeap, OldIndex, SnapshotAny, 0, (ScanKey) NULL);
  
! 	while ((tuple = genam_getnext(scan, ForwardScanDirection, &buf)) != NULL)
  	{
  		HeapTuple	copiedTuple;
  		bool		isdead;
*************** copy_heap_data(Oid OIDNewHeap, Oid OIDOl
*** 839,852 ****
  
  		CHECK_FOR_INTERRUPTS();
  
! 		/* Since we used no scan keys, should never need to recheck */
! 		if (scan->xs_recheck)
! 			elog(ERROR, "CLUSTER does not support lossy index conditions");
  
! 		LockBuffer(scan->xs_cbuf, BUFFER_LOCK_SHARE);
! 
! 		switch (HeapTupleSatisfiesVacuum(tuple->t_data, OldestXmin,
! 										 scan->xs_cbuf))
  		{
  			case HEAPTUPLE_DEAD:
  				/* Definitely dead */
--- 918,926 ----
  
  		CHECK_FOR_INTERRUPTS();
  
! 		LockBuffer(buf, BUFFER_LOCK_SHARE);
  
! 		switch (HeapTupleSatisfiesVacuum(tuple->t_data, OldestXmin, buf))
  		{
  			case HEAPTUPLE_DEAD:
  				/* Definitely dead */
*************** copy_heap_data(Oid OIDNewHeap, Oid OIDOl
*** 888,894 ****
  				break;
  		}
  
! 		LockBuffer(scan->xs_cbuf, BUFFER_LOCK_UNLOCK);
  
  		if (isdead)
  		{
--- 962,968 ----
  				break;
  		}
  
! 		LockBuffer(buf, BUFFER_LOCK_UNLOCK);
  
  		if (isdead)
  		{
*************** copy_heap_data(Oid OIDNewHeap, Oid OIDOl
*** 932,938 ****
  		heap_freetuple(copiedTuple);
  	}
  
! 	index_endscan(scan);
  
  	/* Write out any remaining tuples, and fsync if needed */
  	end_heap_rewrite(rwstate);
--- 1006,1012 ----
  		heap_freetuple(copiedTuple);
  	}
  
! 	genam_endscan(scan);
  
  	/* Write out any remaining tuples, and fsync if needed */
  	end_heap_rewrite(rwstate);
*************** copy_heap_data(Oid OIDNewHeap, Oid OIDOl
*** 940,946 ****
  	pfree(values);
  	pfree(isnull);
  
! 	index_close(OldIndex, NoLock);
  	heap_close(OldHeap, NoLock);
  	heap_close(NewHeap, NoLock);
  
--- 1014,1021 ----
  	pfree(values);
  	pfree(isnull);
  
! 	if (OldIndex != NULL)
! 		index_close(OldIndex, NoLock);
  	heap_close(OldHeap, NoLock);
  	heap_close(NewHeap, NoLock);
  
diff -cprN head/src/backend/nodes/copyfuncs.c work/src/backend/nodes/copyfuncs.c
*** head/src/backend/nodes/copyfuncs.c	2009-10-15 07:14:21.000000000 +0900
--- work/src/backend/nodes/copyfuncs.c	2009-10-23 11:12:41.009498108 +0900
*************** _copyClusterStmt(ClusterStmt *from)
*** 2450,2455 ****
--- 2450,2456 ----
  	COPY_NODE_FIELD(relation);
  	COPY_STRING_FIELD(indexname);
  	COPY_SCALAR_FIELD(verbose);
+ 	COPY_SCALAR_FIELD(without_index);
  
  	return newnode;
  }
diff -cprN head/src/backend/nodes/equalfuncs.c work/src/backend/nodes/equalfuncs.c
*** head/src/backend/nodes/equalfuncs.c	2009-10-15 07:14:21.000000000 +0900
--- work/src/backend/nodes/equalfuncs.c	2009-10-23 11:12:52.845483120 +0900
*************** _equalClusterStmt(ClusterStmt *a, Cluste
*** 1073,1078 ****
--- 1073,1079 ----
  	COMPARE_NODE_FIELD(relation);
  	COMPARE_STRING_FIELD(indexname);
  	COMPARE_SCALAR_FIELD(verbose);
+ 	COMPARE_SCALAR_FIELD(without_index);
  
  	return true;
  }
diff -cprN head/src/backend/parser/gram.y work/src/backend/parser/gram.y
*** head/src/backend/parser/gram.y	2009-10-15 07:14:22.000000000 +0900
--- work/src/backend/parser/gram.y	2009-10-23 11:12:29.632497726 +0900
*************** ClusterStmt:
*** 6587,6592 ****
--- 6587,6602 ----
  				   n->relation = $3;
  				   n->indexname = $4;
  				   n->verbose = $2;
+ 				   n->without_index = false;
+ 				   $$ = (Node*)n;
+ 				}
+ 			| CLUSTER opt_verbose qualified_name WITHOUT INDEX
+ 				{
+ 			       ClusterStmt *n = makeNode(ClusterStmt);
+ 				   n->relation = $3;
+ 				   n->indexname = NULL;
+ 				   n->verbose = $2;
+ 				   n->without_index = true;
  				   $$ = (Node*)n;
  				}
  			| CLUSTER opt_verbose
*************** ClusterStmt:
*** 6595,6600 ****
--- 6605,6611 ----
  				   n->relation = NULL;
  				   n->indexname = NULL;
  				   n->verbose = $2;
+ 				   n->without_index = false;
  				   $$ = (Node*)n;
  				}
  			/* kept for pre-8.3 compatibility */
*************** ClusterStmt:
*** 6604,6609 ****
--- 6615,6621 ----
  				   n->relation = $5;
  				   n->indexname = $3;
  				   n->verbose = $2;
+ 				   n->without_index = false;
  				   $$ = (Node*)n;
  				}
  		;
diff -cprN head/src/include/nodes/parsenodes.h work/src/include/nodes/parsenodes.h
*** head/src/include/nodes/parsenodes.h	2009-10-15 07:14:24.000000000 +0900
--- work/src/include/nodes/parsenodes.h	2009-10-23 11:09:15.911477359 +0900
*************** typedef struct ClusterStmt
*** 2206,2211 ****
--- 2206,2212 ----
  	RangeVar   *relation;		/* relation being indexed, or NULL if all */
  	char	   *indexname;		/* original index defined */
  	bool		verbose;		/* print progress info */
+ 	bool		without_index;	/* no index */
  } ClusterStmt;
  
  /* ----------------------