From: Alvaro Herrera <alvherre@alvh.no-ip.org>
Date: Wed, 12 Sep 2007 15:16:23 +0000 (+0000)
Subject: Fix the database-wide version of CLUSTER to silently skip temp tables of
X-Git-Url: http://waps.l3s.uni-hannover.de/gitweb/?a=commitdiff_plain;h=17e22cf7e10a14c014fd3d06a6ed362031b3d946;p=users%2Fbernd%2Fpostgres.git

Fix the database-wide version of CLUSTER to silently skip temp tables of
remote sessions, instead of erroring out in the middle of the operation.

This is a backpatch of a previous fix applied to CLUSTER to HEAD and 8.2, all
the way back that it is relevant to.
---

diff --git a/src/backend/commands/cluster.c b/src/backend/commands/cluster.c
index 4f95dd1b2b..1b724018d7 100644
--- a/src/backend/commands/cluster.c
+++ b/src/backend/commands/cluster.c
@@ -291,6 +291,18 @@ cluster_rel(RelToCluster *rvtc, bool recheck)
 	 */
 	OldHeap = heap_open(rvtc->tableOid, AccessExclusiveLock);
 
+	/*
+	 * Don't allow cluster on temp tables of other backends ... their
+	 * local buffer manager is not going to cope.  In the recheck case,
+	 * silently skip it.  Otherwise continue -- there is a hard error
+	 * in check_index_is_clusterable.
+	 */
+	if (recheck && isOtherTempNamespace(RelationGetNamespace(OldHeap)))
+	{
+		heap_close(OldHeap, AccessExclusiveLock);
+		return;
+	}
+
 	/* Check index is valid to cluster on */
 	check_index_is_clusterable(OldHeap, rvtc->indexOid);
 
@@ -605,32 +617,80 @@ copy_heap_data(Oid OIDNewHeap, Oid OIDOldHeap, Oid OIDOldIndex)
 	Relation	NewHeap,
 				OldHeap,
 				OldIndex;
+	TupleDesc	oldTupDesc;
+	TupleDesc	newTupDesc;
+	int			natts;
+	Datum	   *values;
+	char	   *nulls;
 	IndexScanDesc scan;
 	HeapTuple	tuple;
 
 	/*
-	 * Open the relations I need. Scan through the OldHeap on the OldIndex
-	 * and insert each tuple into the NewHeap.
+	 * Open the relations we need.
 	 */
 	NewHeap = heap_open(OIDNewHeap, AccessExclusiveLock);
 	OldHeap = heap_open(OIDOldHeap, AccessExclusiveLock);
 	OldIndex = index_open(OIDOldIndex);
 
+	/*
+	 * Their tuple descriptors should be exactly alike, but here we only
+	 * need assume that they have the same number of columns.
+	 */
+	oldTupDesc = RelationGetDescr(OldHeap);
+	newTupDesc = RelationGetDescr(NewHeap);
+	Assert(newTupDesc->natts == oldTupDesc->natts);
+
+	/* Preallocate values/nulls arrays */
+	natts = newTupDesc->natts;
+	values = (Datum *) palloc0(natts * sizeof(Datum));
+	nulls = (char *) palloc(natts * sizeof(char));
+	memset(nulls, 'n', natts * sizeof(char));
+
+	/*
+	 * Scan through the OldHeap on the OldIndex and copy each tuple into the
+	 * NewHeap.
+	 */
 	scan = index_beginscan(OldHeap, OldIndex, SnapshotNow, 0, (ScanKey) NULL);
 
 	while ((tuple = index_getnext(scan, ForwardScanDirection)) != NULL)
 	{
 		/*
-		 * We must copy the tuple because heap_insert() will overwrite the
-		 * commit-status fields of the tuple it's handed, and the
-		 * retrieved tuple will actually be in a disk buffer!  Thus, the
-		 * source relation would get trashed, which is bad news if we
-		 * abort later on.	(This was a bug in releases thru 7.0)
+		 * We cannot simply pass the tuple to heap_insert(), for several
+		 * reasons:
+		 *
+		 * 1. heap_insert() will overwrite the commit-status fields of the
+		 * tuple it's handed.  This would trash the source relation, which is
+		 * bad news if we abort later on.  (This was a bug in releases thru
+		 * 7.0)
+		 *
+		 * 2. We'd like to squeeze out the values of any dropped columns,
+		 * both to save space and to ensure we have no corner-case failures.
+		 * (It's possible for example that the new table hasn't got a TOAST
+		 * table and so is unable to store any large values of dropped cols.)
 		 *
-		 * Note that the copied tuple will have the original OID, if any, so
-		 * this does preserve OIDs.
+		 * 3. The tuple might not even be legal for the new table; this is
+		 * currently only known to happen as an after-effect of ALTER TABLE
+		 * SET WITHOUT OIDS.
+		 *
+		 * So, we must reconstruct the tuple from component Datums.
 		 */
-		HeapTuple	copiedTuple = heap_copytuple(tuple);
+		HeapTuple	copiedTuple;
+		int			i;
+
+		heap_deformtuple(tuple, oldTupDesc, values, nulls);
+
+		/* Be sure to null out any dropped columns */
+		for (i = 0; i < natts; i++)
+		{
+			if (newTupDesc->attrs[i]->attisdropped)
+				nulls[i] = 'n';
+		}
+
+		copiedTuple = heap_formtuple(newTupDesc, values, nulls);
+
+		/* Preserve OID, if any */
+		if (NewHeap->rd_rel->relhasoids)
+			HeapTupleSetOid(copiedTuple, HeapTupleGetOid(tuple));
 
 		simple_heap_insert(NewHeap, copiedTuple);
 
@@ -641,6 +701,9 @@ copy_heap_data(Oid OIDNewHeap, Oid OIDOldHeap, Oid OIDOldIndex)
 
 	index_endscan(scan);
 
+	pfree(values);
+	pfree(nulls);
+
 	index_close(OldIndex);
 	heap_close(OldHeap, NoLock);
 	heap_close(NewHeap, NoLock);