Repair CLUSTER failure after ALTER TABLE SET WITHOUT OIDS. Turns out

there are corner cases involving dropping toasted columns in which the
previous coding would fail, too: the new version of the table might not
have any TOAST table, but we'd still propagate possibly-wide values of
dropped columns forward.

diff --git a/src/backend/commands/cluster.c b/src/backend/commands/cluster.c
index 1e61af151f016f52eca365c4e5b3c41ed7079f35..17fcbb3afba01cc51d3bbb8d51012dc4d2844c3f 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.131 2004/12/31 21:59:41 pgsql Exp $
+ *   $PostgreSQL: pgsql/src/backend/commands/cluster.c,v 1.132 2005/02/06 20:19:08 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -605,32 +605,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 +689,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);