diff --git a/src/java/org/apache/poi/hssf/model/Sheet.java b/src/java/org/apache/poi/hssf/model/Sheet.java
index aad645049..11ee42b6a 100644
--- a/src/java/org/apache/poi/hssf/model/Sheet.java
+++ b/src/java/org/apache/poi/hssf/model/Sheet.java
@@ -77,7 +77,6 @@ import org.apache.poi.hssf.record.aggregates.CFRecordsAggregate;
 import org.apache.poi.hssf.record.aggregates.ColumnInfoRecordsAggregate;
 import org.apache.poi.hssf.record.aggregates.ConditionalFormattingTable;
 import org.apache.poi.hssf.record.aggregates.DataValidityTable;
-import org.apache.poi.hssf.record.aggregates.FormulaRecordAggregate;
 import org.apache.poi.hssf.record.aggregates.MergedCellsTable;
 import org.apache.poi.hssf.record.aggregates.RecordAggregate;
 import org.apache.poi.hssf.record.aggregates.RowRecordsAggregate;
@@ -389,6 +388,7 @@ public final class Sheet implements Model {
             _destList.add(r.clone());
         }
     }
+
     /**
      * Clones the low level records of this sheet and returns the new sheet instance.
      * This method is implemented by adding methods for deep cloning to all records that
@@ -396,53 +396,18 @@ public final class Sheet implements Model {
      * When adding a new record, implement a public clone method if and only if the record
      * belongs to a sheet.
      */
-    public Sheet cloneSheet()
-    {
-      ArrayList clonedRecords = new ArrayList(this.records.size());
-      for (int i=0; i<this.records.size();i++) {
-        RecordBase rb = (RecordBase) this.records.get(i);
-        if (rb instanceof RecordAggregate) {
-            ((RecordAggregate)rb).visitContainedRecords(new RecordCloner(clonedRecords));
-            // TODO - make sure this logic works for the other RecordAggregates
-            continue;
+    public Sheet cloneSheet() {
+        ArrayList clonedRecords = new ArrayList(this.records.size());
+        for (int i = 0; i < this.records.size(); i++) {
+            RecordBase rb = (RecordBase) this.records.get(i);
+            if (rb instanceof RecordAggregate) {
+                ((RecordAggregate) rb).visitContainedRecords(new RecordCloner(clonedRecords));
+                continue;
+            }
+            Record rec = (Record) ((Record) rb).clone();
+            clonedRecords.add(rec);
         }
-        Record rec = (Record)((Record)rb).clone();
-        //Need to pull out the Row record and the Value records from their
-        //Aggregates.
-        //This is probably the best way to do it since we probably dont want the createSheet
-        //To cater for these artificial Record types
-        if (rec instanceof RowRecordsAggregate) {
-          RowRecordsAggregate rrAgg = (RowRecordsAggregate)rec;
-          for (Iterator rowIter = rrAgg.getAllRecordsIterator();rowIter.hasNext();) {
-            Record valRec = (Record)rowIter.next();
-
-            if (valRec instanceof FormulaRecordAggregate) {
-                FormulaRecordAggregate fmAgg = (FormulaRecordAggregate)valRec;
-                Record fmAggRec = fmAgg.getFormulaRecord();
-                if (fmAggRec != null) {
-                    clonedRecords.add(fmAggRec);
-                }
-                fmAggRec =   fmAgg.getStringRecord();
-                if (fmAggRec != null) {
-                    clonedRecords.add(fmAggRec);
-                }
-              } else {
-                clonedRecords.add(valRec);
-              }
-          }
-        } else if (rec instanceof FormulaRecordAggregate) {  //Is this required now??
-          FormulaRecordAggregate fmAgg = (FormulaRecordAggregate)rec;
-          Record fmAggRec = fmAgg.getFormulaRecord();
-          if (fmAggRec != null)
-            clonedRecords.add(fmAggRec);
-          fmAggRec =   fmAgg.getStringRecord();
-          if (fmAggRec != null)
-            clonedRecords.add(fmAggRec);
-        } else {
-          clonedRecords.add(rec);
-        }
-      }
-      return createSheet(clonedRecords, 0, 0);
+        return createSheet(clonedRecords, 0, 0);
     }
 
 
@@ -856,11 +821,12 @@ public final class Sheet implements Model {
         {
             d.setFirstRow(row.getRowNumber());
         }
-        //IndexRecord index = null;
-         //If the row exists remove it, so that any cells attached to the row are removed
-         RowRecord existingRow = _rowsAggregate.getRow(row.getRowNumber());
-         if (existingRow != null)
-           _rowsAggregate.removeRow(existingRow);
+
+        //If the row exists remove it, so that any cells attached to the row are removed
+        RowRecord existingRow = _rowsAggregate.getRow(row.getRowNumber());
+        if (existingRow != null) {
+            _rowsAggregate.removeRow(existingRow);
+        }
 
         _rowsAggregate.insertRow(row);
 
@@ -1508,6 +1474,11 @@ public final class Sheet implements Model {
             }
             retval += record.getRecordSize();
         }
+        // add space for IndexRecord if needed
+        if (_rowsAggregate != null) {
+            // rowsAggregate knows how to make the index record
+            retval += IndexRecord.getRecordSizeForBlockCount(_rowsAggregate.getRowBlockCount());
+        }
         // Add space for UncalcedRecord
         if (_isUncalced) {
             retval += UncalcedRecord.getStaticRecordSize();
diff --git a/src/java/org/apache/poi/hssf/record/DBCellRecord.java b/src/java/org/apache/poi/hssf/record/DBCellRecord.java
index 1da6b82c7..d14e689ed 100644
--- a/src/java/org/apache/poi/hssf/record/DBCellRecord.java
+++ b/src/java/org/apache/poi/hssf/record/DBCellRecord.java
@@ -35,6 +35,7 @@ public final class DBCellRecord extends Record {
 
     public DBCellRecord()
     {
+        field_2_cell_offsets = new short[0];
     }
 
     /**
@@ -185,4 +186,9 @@ public final class DBCellRecord extends Record {
     {
         return true;
     }
+    public Object clone() {
+        // TODO - make immutable.
+        // this should be safe because only the instantiating code mutates these objects
+        return this;
+    }
 }
diff --git a/src/java/org/apache/poi/hssf/record/aggregates/RecordAggregate.java b/src/java/org/apache/poi/hssf/record/aggregates/RecordAggregate.java
index ce0bf8945..f9cbcd777 100644
--- a/src/java/org/apache/poi/hssf/record/aggregates/RecordAggregate.java
+++ b/src/java/org/apache/poi/hssf/record/aggregates/RecordAggregate.java
@@ -54,6 +54,10 @@ public abstract class RecordAggregate extends RecordBase {
 	}
 	
 	public interface RecordVisitor {
+		/**
+		 * Implementors may call non-mutating methods on Record r.
+		 * @param r must not be <code>null</code>
+		 */
 		void visitRecord(Record r);
 	}
 	
diff --git a/src/java/org/apache/poi/hssf/record/aggregates/RowRecordsAggregate.java b/src/java/org/apache/poi/hssf/record/aggregates/RowRecordsAggregate.java
index 91028e360..c9a302b61 100644
--- a/src/java/org/apache/poi/hssf/record/aggregates/RowRecordsAggregate.java
+++ b/src/java/org/apache/poi/hssf/record/aggregates/RowRecordsAggregate.java
@@ -27,8 +27,6 @@ import org.apache.poi.hssf.record.CellValueRecordInterface;
 import org.apache.poi.hssf.record.DBCellRecord;
 import org.apache.poi.hssf.record.IndexRecord;
 import org.apache.poi.hssf.record.Record;
-import org.apache.poi.hssf.record.RecordBase;
-import org.apache.poi.hssf.record.RecordInputStream;
 import org.apache.poi.hssf.record.RowRecord;
 
 /**
@@ -36,7 +34,7 @@ import org.apache.poi.hssf.record.RowRecord;
  * @author  andy
  * @author Jason Height (jheight at chariot dot net dot au)
  */
-public final class RowRecordsAggregate extends Record {
+public final class RowRecordsAggregate extends RecordAggregate {
     private int _firstrow = -1;
     private int _lastrow  = -1;
     private final Map _rowRecords;
@@ -162,128 +160,64 @@ public final class RowRecordsAggregate extends Record {
       }
       return row.getRowNumber();
     }
-
-
-    /** Serializes a block of the rows */
-    private int serializeRowBlock(final int block, final int offset, byte[] data) {
-      final int startIndex = block*DBCellRecord.BLOCK_SIZE;
-      final int endIndex = startIndex + DBCellRecord.BLOCK_SIZE;
-
-      Iterator rowIterator = _rowRecords.values().iterator();
-      int pos = offset;
-
-      //Given that we basically iterate through the rows in order,
-      //For a performance improvement, it would be better to return an instance of
-      //an iterator and use that instance throughout, rather than recreating one and
-      //having to move it to the right position.
-      int i=0;
-      for (;i<startIndex;i++)
-        rowIterator.next();
-      while(rowIterator.hasNext() && (i++ < endIndex)) {
-        RowRecord row = (RowRecord)rowIterator.next();
-        pos += row.serialize(pos, data);
-      }
-      return pos - offset;
-    }
     
+    private int visitRowRecordsForBlock(int blockIndex, RecordVisitor rv) {
+        final int startIndex = blockIndex*DBCellRecord.BLOCK_SIZE;
+        final int endIndex = startIndex + DBCellRecord.BLOCK_SIZE;
 
-    /**
-     * called by the class that is responsible for writing this sucker.
-     * Subclasses should implement this so that their data is passed back in a
-     * byte array.
-     *
-     * @param offset    offset to begin writing at
-     * @param data      byte array containing instance data
-     * @return number of bytes written
-     */
-    public int serialize(int offset, byte [] data) {
+        Iterator rowIterator = _rowRecords.values().iterator();
+
+        //Given that we basically iterate through the rows in order,
+        //For a performance improvement, it would be better to return an instance of
+        //an iterator and use that instance throughout, rather than recreating one and
+        //having to move it to the right position.
+        int i=0;
+        for (;i<startIndex;i++)
+          rowIterator.next();
+        int result = 0;
+        while(rowIterator.hasNext() && (i++ < endIndex)) {
+          Record rec = (Record)rowIterator.next();
+          result += rec.getRecordSize();
+          rv.visitRecord(rec);
+        }
+        return result;
+      }
+    
+    public void visitContainedRecords(RecordVisitor rv) {
         ValueRecordsAggregate cells = _valuesAgg;
-        int pos = offset;
-
+       
         //DBCells are serialized before row records.
         final int blockCount = getRowBlockCount();
-        for (int block=0;block<blockCount;block++) {
-          //Serialize a block of rows.
-          //Hold onto the position of the first row in the block
-          final int rowStartPos = pos;
-          //Hold onto the size of this block that was serialized
-          final int rowBlockSize = serializeRowBlock(block, pos, data);
-          pos += rowBlockSize;
-          //Serialize a block of cells for those rows
-          final int startRowNumber = getStartRowNumberForBlock(block);
-          final int endRowNumber = getEndRowNumberForBlock(block);
-          DBCellRecord cellRecord = new DBCellRecord();
-          //Note: Cell references start from the second row...
-          int cellRefOffset = (rowBlockSize-RowRecord.ENCODED_SIZE);
-          for (int row=startRowNumber;row<=endRowNumber;row++) {
-            if (null != cells && cells.rowHasCells(row)) {
-              final int rowCellSize = cells.serializeCellRow(row, pos, data);
-              pos += rowCellSize;
-              //Add the offset to the first cell for the row into the DBCellRecord.
-              cellRecord.addCellOffset((short)cellRefOffset);
-              cellRefOffset = rowCellSize;
+        for (int blockIndex = 0; blockIndex < blockCount; blockIndex++) {
+            // Serialize a block of rows.
+            // Hold onto the position of the first row in the block
+            int pos=0;
+            // Hold onto the size of this block that was serialized
+            final int rowBlockSize = visitRowRecordsForBlock(blockIndex, rv);
+            pos += rowBlockSize;
+            // Serialize a block of cells for those rows
+            final int startRowNumber = getStartRowNumberForBlock(blockIndex);
+            final int endRowNumber = getEndRowNumberForBlock(blockIndex);
+            DBCellRecord cellRecord = new DBCellRecord();
+            // Note: Cell references start from the second row...
+            int cellRefOffset = (rowBlockSize - RowRecord.ENCODED_SIZE);
+            for (int row = startRowNumber; row <= endRowNumber; row++) {
+                if (cells.rowHasCells(row)) {
+                    final int rowCellSize = cells.visitCellsForRow(row, rv);
+                    pos += rowCellSize;
+                    // Add the offset to the first cell for the row into the
+                    // DBCellRecord.
+                    cellRecord.addCellOffset((short) cellRefOffset);
+                    cellRefOffset = rowCellSize;
+                }
             }
-          }
-          //Calculate Offset from the start of a DBCellRecord to the first Row
-          cellRecord.setRowOffset(pos - rowStartPos);
-          pos += cellRecord.serialize(pos, data);
-
+            // Calculate Offset from the start of a DBCellRecord to the first Row
+            cellRecord.setRowOffset(pos);
+            rv.visitRecord(cellRecord);
         }
-        return pos - offset;
     }
 
-    /**
-     * You never fill an aggregate
-     */
-    protected void fillFields(RecordInputStream in)
-    {
-    }
-
-    /**
-     * called by constructor, should throw runtime exception in the event of a
-     * record passed with a differing ID.
-     *
-     * @param id alleged id for this record
-     */
-
-    protected void validateSid(short id)
-    {
-    }
-
-    /**
-     * return the non static version of the id for this record.
-     */
-
-    public short getSid()
-    {
-        return -1000;
-    }
-
-    public int getRecordSize() {
-
-        int retval = this._rowRecords.size() * RowRecord.ENCODED_SIZE;
-        
-        for (Iterator itr = _valuesAgg.getIterator(); itr.hasNext();) {
-            RecordBase record = (RecordBase) itr.next();
-            retval += record.getRecordSize();
-        }
-
-        // Add space for the IndexRecord and DBCell records
-        final int nBlocks = getRowBlockCount();
-        int nRows = 0;
-        for (Iterator itr = getIterator(); itr.hasNext();) {
-            RowRecord row = (RowRecord)itr.next();
-            if (_valuesAgg.rowHasCells(row.getRowNumber())) {
-                nRows++;
-            }
-        }
-        retval += IndexRecord.getRecordSizeForBlockCount(nBlocks);
-        retval += DBCellRecord.calculateSizeOfRecords(nBlocks, nRows);
-        return retval;
-    }
-
-    public Iterator getIterator()
-    {
+    public Iterator getIterator() {
         return _rowRecords.values().iterator();
     }
     
@@ -297,23 +231,6 @@ public final class RowRecordsAggregate extends Record {
         }
         return result.iterator();
     }
-    /**
-     * Performs a deep clone of the record
-     */
-    public Object clone()
-    {
-        TreeMap rows = new TreeMap();
-
-        for ( Iterator rowIter = getIterator(); rowIter.hasNext(); )
-        {
-            //return the cloned Row Record & insert
-            RowRecord row = (RowRecord) ( (RowRecord) rowIter.next() ).clone();
-            rows.put(row, row);
-        }
-        ValueRecordsAggregate valuesAgg = (ValueRecordsAggregate) _valuesAgg.clone();
-        return new RowRecordsAggregate(rows, valuesAgg);
-    }
-
 
     public int findStartOfRowOutlineGroup(int row)
     {
diff --git a/src/java/org/apache/poi/hssf/record/aggregates/ValueRecordsAggregate.java b/src/java/org/apache/poi/hssf/record/aggregates/ValueRecordsAggregate.java
index 8ba786a38..8d5764187 100644
--- a/src/java/org/apache/poi/hssf/record/aggregates/ValueRecordsAggregate.java
+++ b/src/java/org/apache/poi/hssf/record/aggregates/ValueRecordsAggregate.java
@@ -28,6 +28,7 @@ import org.apache.poi.hssf.record.Record;
 import org.apache.poi.hssf.record.SharedFormulaRecord;
 import org.apache.poi.hssf.record.StringRecord;
 import org.apache.poi.hssf.record.UnknownRecord;
+import org.apache.poi.hssf.record.aggregates.RecordAggregate.RecordVisitor;
 
 /**
  *
@@ -87,19 +88,19 @@ public final class ValueRecordsAggregate {
 
     public void removeCell(CellValueRecordInterface cell) {
         if (cell == null) {
-        	throw new IllegalArgumentException("cell must not be null");
+            throw new IllegalArgumentException("cell must not be null");
         }
         int row = cell.getRow();
         if (row >= records.length) {
-        	throw new RuntimeException("cell row is out of range");
+            throw new RuntimeException("cell row is out of range");
         }
         CellValueRecordInterface[] rowCells = records[row];
         if (rowCells == null) {
-        	throw new RuntimeException("cell row is already empty");
+            throw new RuntimeException("cell row is already empty");
         }
         short column = cell.getColumn();
         if (column >= rowCells.length) {
-        	throw new RuntimeException("cell column is out of range");
+            throw new RuntimeException("cell column is out of range");
         }
         rowCells[column] = null;
     }
@@ -266,6 +267,35 @@ public final class ValueRecordsAggregate {
         }
         return pos - offset;
     }
+    
+    public int visitCellsForRow(int rowIndex, RecordVisitor rv) {
+        int result = 0;
+        CellValueRecordInterface[] cellRecs = records[rowIndex];
+        if (cellRecs != null) {
+            for (int i = 0; i < cellRecs.length; i++) {
+                CellValueRecordInterface cvr = cellRecs[i];
+                if (cvr == null) {
+                    continue;
+                }
+                if (cvr instanceof FormulaRecordAggregate) {
+                    FormulaRecordAggregate fmAgg = (FormulaRecordAggregate) cvr;
+                    Record fmAggRec = fmAgg.getFormulaRecord();
+                    rv.visitRecord(fmAggRec);
+                    result += fmAggRec.getRecordSize();
+                    fmAggRec = fmAgg.getStringRecord();
+                    if (fmAggRec != null) {
+                        rv.visitRecord(fmAggRec);
+                        result += fmAggRec.getRecordSize();
+                    }
+                } else {
+                    Record rec = (Record) cvr;
+                    rv.visitRecord(rec);
+                    result += rec.getRecordSize();
+                }
+            }
+        }
+        return result;
+    }
 
     public CellValueRecordInterface[] getValueRecords() {
         List temp = new ArrayList();
diff --git a/src/java/org/apache/poi/hssf/usermodel/HSSFWorkbook.java b/src/java/org/apache/poi/hssf/usermodel/HSSFWorkbook.java
index 13f46255c..5966c0d80 100644
--- a/src/java/org/apache/poi/hssf/usermodel/HSSFWorkbook.java
+++ b/src/java/org/apache/poi/hssf/usermodel/HSSFWorkbook.java
@@ -256,21 +256,11 @@ public class HSSFWorkbook extends POIDocument
 
         // convert all LabelRecord records to LabelSSTRecord
         convertLabelRecords(records, recOffset);
-        while (recOffset < records.size())
-        {
+        while (recOffset < records.size()) {
             Sheet sheet = Sheet.createSheet(records, sheetNum++, recOffset );
 
-            recOffset = sheet.getEofLoc()+1;
-            if (recOffset == 1)
-            {
-                break;
-            }
-
-            HSSFSheet hsheet = new HSSFSheet(this, sheet);
-
-            _sheets.add(hsheet);
-
-            // workbook.setSheetName(sheets.size() -1, "Sheet"+sheets.size());
+            recOffset = sheet.getEofLoc()+1; // TODO - use better technique to keep track of the used records
+            _sheets.add(new HSSFSheet(this, sheet));
         }
 
         for (int i = 0 ; i < workbook.getNumNames() ; ++i){
diff --git a/src/testcases/org/apache/poi/hssf/usermodel/TestWorkbook.java b/src/testcases/org/apache/poi/hssf/usermodel/TestWorkbook.java
index 09b93cf87..ebbe6cbc0 100644
--- a/src/testcases/org/apache/poi/hssf/usermodel/TestWorkbook.java
+++ b/src/testcases/org/apache/poi/hssf/usermodel/TestWorkbook.java
@@ -17,6 +17,7 @@
 
 package org.apache.poi.hssf.usermodel;
 
+import java.io.ByteArrayInputStream;
 import java.io.File;
 import java.io.FileInputStream;
 import java.io.FileOutputStream;
@@ -26,6 +27,8 @@ import java.util.Iterator;
 import junit.framework.TestCase;
 
 import org.apache.poi.hssf.HSSFTestDataSamples;
+import org.apache.poi.hssf.eventmodel.ERFListener;
+import org.apache.poi.hssf.eventmodel.EventRecordFactory;
 import org.apache.poi.hssf.model.Workbook;
 import org.apache.poi.hssf.record.BackupRecord;
 import org.apache.poi.hssf.record.LabelSSTRecord;
@@ -453,13 +456,13 @@ public final class TestWorkbook extends TestCase {
     }
 
     private static void confirmRegion(CellRangeAddress ra, CellRangeAddress rb) {
-		assertEquals(ra.getFirstRow(), rb.getFirstRow());
-		assertEquals(ra.getLastRow(), rb.getLastRow());
-		assertEquals(ra.getFirstColumn(), rb.getFirstColumn());
-		assertEquals(ra.getLastColumn(), rb.getLastColumn());
-	}
+        assertEquals(ra.getFirstRow(), rb.getFirstRow());
+        assertEquals(ra.getLastRow(), rb.getLastRow());
+        assertEquals(ra.getFirstColumn(), rb.getFirstColumn());
+        assertEquals(ra.getLastColumn(), rb.getLastColumn());
+    }
 
-	/**
+    /**
      * Test the backup field gets set as expected.
      */
 
@@ -476,38 +479,44 @@ public final class TestWorkbook extends TestCase {
         assertEquals(1, record.getBackup());
     }
 
+    private static final class RecordCounter implements ERFListener {
+        private int _count;
+
+        public RecordCounter() {
+            _count=0;
+        }
+        public int getCount() {
+            return _count;
+        }
+        public boolean processRecord(Record rec) {
+            _count++;
+            return true;
+        }
+    }
+    
     /**
      * This tests is for bug [ #506658 ] Repeating output.
      *
      * We need to make sure only one LabelSSTRecord is produced.
      */
-
     public void testRepeatingBug()
         throws Exception
     {
         HSSFWorkbook workbook = new HSSFWorkbook();
         HSSFSheet    sheet    = workbook.createSheet("Design Variants");
-        HSSFRow      row      = sheet.createRow(( short ) 2);
-        HSSFCell     cell     = row.createCell(( short ) 1);
+        HSSFRow      row      = sheet.createRow(2);
+        HSSFCell     cell     = row.createCell(1);
 
         cell.setCellValue(new HSSFRichTextString("Class"));
-        cell = row.createCell(( short ) 2);
+        cell = row.createCell(2);
 
-        // workbook.write(new FileOutputStream("/a2.xls"));
-        RowRecordsAggregate rra = (RowRecordsAggregate) 
-            sheet.getSheet().findFirstRecordBySid((short)-1000);
+        byte[] data = new byte[sheet.getSheet().getSize()];
+        sheet.getSheet().serialize(0, data);
+        RecordCounter rc = new RecordCounter();
+        EventRecordFactory erf = new EventRecordFactory(rc, new short[] { LabelSSTRecord.sid, });
+        erf.processRecords(new ByteArrayInputStream(data));
         
-        int                   sstRecords     = 0;
-        Iterator              iterator       = rra.getAllRecordsIterator();
-
-        while (iterator.hasNext())
-        {
-            if ((( Record ) iterator.next()).getSid() == LabelSSTRecord.sid)
-            {
-                sstRecords++;
-            }
-        }
-        assertEquals(1, sstRecords);
+        assertEquals(1, rc.getCount());
     }