diff --git a/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/edu/uci/ics/hyracks/tests/btree/BTreeOperatorsTest.java b/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/edu/uci/ics/hyracks/tests/btree/BTreeOperatorsTest.java
index 49ed678..daa2425 100644
--- a/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/edu/uci/ics/hyracks/tests/btree/BTreeOperatorsTest.java
+++ b/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/edu/uci/ics/hyracks/tests/btree/BTreeOperatorsTest.java
@@ -15,8 +15,10 @@
 
 package edu.uci.ics.hyracks.tests.btree;
 
+import java.io.DataOutputStream;
 import java.io.File;
 import java.io.RandomAccessFile;
+import java.nio.ByteBuffer;
 
 import org.junit.Test;
 
@@ -24,12 +26,17 @@
 import edu.uci.ics.hyracks.api.constraints.ExplicitPartitionConstraint;
 import edu.uci.ics.hyracks.api.constraints.LocationConstraint;
 import edu.uci.ics.hyracks.api.constraints.PartitionConstraint;
+import edu.uci.ics.hyracks.api.dataflow.IConnectorDescriptor;
 import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparator;
 import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparatorFactory;
 import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
 import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
 import edu.uci.ics.hyracks.api.job.JobSpecification;
+import edu.uci.ics.hyracks.dataflow.common.comm.io.ByteArrayAccessibleOutputStream;
+import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
+import edu.uci.ics.hyracks.dataflow.common.data.comparators.IntegerBinaryComparatorFactory;
 import edu.uci.ics.hyracks.dataflow.common.data.comparators.UTF8StringBinaryComparatorFactory;
+import edu.uci.ics.hyracks.dataflow.common.data.marshalling.IntegerSerializerDeserializer;
 import edu.uci.ics.hyracks.dataflow.common.data.marshalling.UTF8StringSerializerDeserializer;
 import edu.uci.ics.hyracks.dataflow.common.data.parsers.IValueParserFactory;
 import edu.uci.ics.hyracks.dataflow.common.data.parsers.UTF8StringParserFactory;
@@ -40,17 +47,18 @@
 import edu.uci.ics.hyracks.dataflow.std.file.FileSplit;
 import edu.uci.ics.hyracks.dataflow.std.file.IFileSplitProvider;
 import edu.uci.ics.hyracks.dataflow.std.misc.NullSinkOperatorDescriptor;
+import edu.uci.ics.hyracks.dataflow.std.misc.PrinterOperatorDescriptor;
 import edu.uci.ics.hyracks.dataflow.std.sort.InMemorySortOperatorDescriptor;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeCursor;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeInteriorFrameFactory;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeLeafFrameFactory;
 import edu.uci.ics.hyracks.storage.am.btree.api.IFieldAccessor;
 import edu.uci.ics.hyracks.storage.am.btree.api.IFieldAccessorFactory;
-import edu.uci.ics.hyracks.storage.am.btree.api.IFieldIterator;
 import edu.uci.ics.hyracks.storage.am.btree.dataflow.BTreeBulkLoadOperatorDescriptor;
 import edu.uci.ics.hyracks.storage.am.btree.dataflow.BTreeInsertUpdateDeleteOperatorDescriptor;
 import edu.uci.ics.hyracks.storage.am.btree.dataflow.BTreeRegistry;
 import edu.uci.ics.hyracks.storage.am.btree.dataflow.BTreeRegistryProvider;
+import edu.uci.ics.hyracks.storage.am.btree.dataflow.BTreeSearchOperatorDescriptor;
 import edu.uci.ics.hyracks.storage.am.btree.dataflow.BufferCacheProvider;
 import edu.uci.ics.hyracks.storage.am.btree.dataflow.IBTreeRegistryProvider;
 import edu.uci.ics.hyracks.storage.am.btree.dataflow.IBufferCacheProvider;
@@ -62,6 +70,8 @@
 import edu.uci.ics.hyracks.storage.am.btree.impls.MultiComparator;
 import edu.uci.ics.hyracks.storage.am.btree.impls.RangePredicate;
 import edu.uci.ics.hyracks.storage.am.btree.impls.RangeSearchCursor;
+import edu.uci.ics.hyracks.storage.am.btree.impls.SelfDescTupleReference;
+import edu.uci.ics.hyracks.storage.am.btree.types.Int32AccessorFactory;
 import edu.uci.ics.hyracks.storage.am.btree.types.UTF8StringAccessorFactory;
 import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
 import edu.uci.ics.hyracks.storage.common.file.FileInfo;
@@ -150,8 +160,8 @@
         try {
         	while (scanCursor.hasNext()) {
         		scanCursor.next();
-        		IFieldIterator fieldIter = scanCursor.getFieldIterator();                                
-                String rec = cmp.printRecord(fieldIter);
+        		ITupleReference frameTuple = scanCursor.getTuple();                                
+                String rec = cmp.printTuple(frameTuple, ordersDesc.getFields());
                 System.out.println(rec);
         	}
         } catch (Exception e) {
@@ -160,7 +170,7 @@
         	scanCursor.close();
         }                        
 	}
-		
+	
 	/*
 	@Test
 	public void btreeSearchTest() throws Exception {
@@ -175,11 +185,24 @@
 		IFieldAccessorFactory[] fieldAccessorFactories = new IFieldAccessorFactory[2];
 		fieldAccessorFactories[0] = new Int32AccessorFactory(); // key
 		fieldAccessorFactories[1] = new Int32AccessorFactory(); // value
-		
+						
 		int keyLen = 1;
 		IBinaryComparatorFactory[] comparatorFactories = new IBinaryComparatorFactory[keyLen];
 		comparatorFactories[0] = IntegerBinaryComparatorFactory.INSTANCE;	
 		
+		// construct a multicomparator from the factories (only for printing purposes)                
+        IFieldAccessor[] fields = new IFieldAccessor[fieldAccessorFactories.length];
+    	for(int i = 0; i < fieldAccessorFactories.length; i++) {
+    		fields[i] = fieldAccessorFactories[i].getFieldAccessor();
+    	}
+    	
+    	IBinaryComparator[] comparators = new IBinaryComparator[comparatorFactories.length];
+    	for(int i = 0; i < comparatorFactories.length; i++) {
+    		comparators[i] = comparatorFactories[i].createBinaryComparator();
+    	}
+    	
+        MultiComparator cmp = new MultiComparator(comparators, fields);
+		
 		ByteArrayAccessibleOutputStream lkbaaos = new ByteArrayAccessibleOutputStream();
 		DataOutputStream lkdos = new DataOutputStream(lkbaaos);    	    	    	
 		IntegerSerializerDeserializer.INSTANCE.serialize(-1000, lkdos);
@@ -188,8 +211,15 @@
 		DataOutputStream hkdos = new DataOutputStream(hkbaaos);    	    	    	
 		IntegerSerializerDeserializer.INSTANCE.serialize(1000, hkdos);
 				
-		byte[] lowKey = lkbaaos.toByteArray();
-		byte[] highKey = hkbaaos.toByteArray();
+		byte[] lowKeyBytes = lkbaaos.toByteArray();
+        ByteBuffer lowKeyBuf = ByteBuffer.wrap(lowKeyBytes);
+        SelfDescTupleReference lowKey = new SelfDescTupleReference(cmp.getFields());
+        lowKey.reset(lowKeyBuf, 0);
+        
+        byte[] highKeyBytes = hkbaaos.toByteArray();
+        ByteBuffer highKeyBuf = ByteBuffer.wrap(highKeyBytes);
+        SelfDescTupleReference highKey = new SelfDescTupleReference(cmp.getFields());
+        highKey.reset(highKeyBuf, 0);
 		
 		IBufferCacheProvider bufferCacheProvider = new BufferCacheProvider();
 		IBTreeRegistryProvider btreeRegistryProvider = new BTreeRegistryProvider();
@@ -214,7 +244,7 @@
         spec.addRoot(printer);
         runTest(spec);
     }
-    */
+	*/
 	
 	@Test
 	public void insertTest() throws Exception {
@@ -379,9 +409,9 @@
         try {
         	while (scanCursorA.hasNext()) {
         		scanCursorA.next();
-        		IFieldIterator fieldIter = scanCursorA.getFieldIterator();                                
-                String rec = primaryCmp.printRecord(fieldIter);
-                System.out.println(rec);
+        		ITupleReference frameTuple = scanCursorA.getTuple();                                
+                String rec = primaryCmp.printTuple(frameTuple, ordersDesc.getFields());
+                System.out.println(rec);        		
         	}
         } catch (Exception e) {
         	e.printStackTrace();
@@ -398,9 +428,9 @@
         try {
         	while (scanCursorB.hasNext()) {
         		scanCursorB.next();
-        		IFieldIterator fieldIter = scanCursorB.getFieldIterator();                                
-                String rec = secondaryCmp.printRecord(fieldIter);
-                System.out.println(rec);
+        		ITupleReference frameTuple = scanCursorB.getTuple();                                
+                String rec = secondaryCmp.printTuple(frameTuple, ordersDesc.getFields());
+                System.out.println(rec);   
         	}
         } catch (Exception e) {
         	e.printStackTrace();
@@ -417,8 +447,8 @@
         try {
         	while (scanCursorC.hasNext()) {
         		scanCursorC.next();
-        		IFieldIterator fieldIter = scanCursorC.getFieldIterator();                                
-                String rec = secondaryCmp.printRecord(fieldIter);
+        		ITupleReference frameTuple = scanCursorC.getTuple();                                
+                String rec = secondaryCmp.printTuple(frameTuple, ordersDesc.getFields());
                 System.out.println(rec);
         	}
         } catch (Exception e) {
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IBTreeCursor.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IBTreeCursor.java
index 74e6478..47e2fb5 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IBTreeCursor.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IBTreeCursor.java
@@ -15,6 +15,7 @@
 
 package edu.uci.ics.hyracks.storage.am.btree.api;
 
+import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
 import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
 import edu.uci.ics.hyracks.storage.common.buffercache.ICachedPage;
 
@@ -27,5 +28,5 @@
 	public void close() throws Exception;
 	public void setBufferCache(IBufferCache bufferCache);
 	public void setFileId(int fileId);
-	public IFieldIterator getFieldIterator();
+	public ITupleReference getTuple();
 }
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IBTreeFrame.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IBTreeFrame.java
index 511d808..b0ad734 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IBTreeFrame.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IBTreeFrame.java
@@ -37,13 +37,13 @@
 	
 	public void initBuffer(byte level);
 	
-	public int getNumRecords();
+	public int getTupleCount();
 		
 	// assumption: page must be write-latched at this point
 	public SpaceStatus hasSpaceInsert(ITupleReference tuple, MultiComparator cmp);
 	public SpaceStatus hasSpaceUpdate(int rid, ITupleReference tuple, MultiComparator cmp);
 	
-	public int getRecordOffset(int slotNum);
+	public int getTupleOffset(int slotNum);
 	
 	public int getTotalFreeSpace();	
 	
@@ -52,10 +52,10 @@
 	
 	// for debugging
 	public void printHeader();
-	public String printKeys(MultiComparator cmp);
+	public String printKeys(MultiComparator cmp, int fieldCount);
 	
 	
-	// TODO; what if records more than half-page size?
+	// TODO; what if tuples more than half-page size?
 	public int split(IBTreeFrame rightFrame, ITupleReference tuple, MultiComparator cmp, SplitKey splitKey) throws Exception;		
 	
 	// TODO: check if we do something nicer than returning object
@@ -73,10 +73,10 @@
 	
 	public int getSlotSize();
 	
-	public IFieldIterator createFieldIterator();
+	public IBTreeTupleReference createTupleReference();
 	
-	// TODO: should be removed after new record format		
-	public void setPageTupleFields(IFieldAccessor[] fields);
+	// TODO: should be removed after new tuple format		
+	public void setPageTupleFieldCount(int fieldCount);
 	
 	// for debugging
 	public int getFreeSpaceOff();
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IBTreeInteriorFrame.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IBTreeInteriorFrame.java
index ede5adf..ee6e94b 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IBTreeInteriorFrame.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IBTreeInteriorFrame.java
@@ -19,7 +19,6 @@
 import edu.uci.ics.hyracks.storage.am.btree.impls.RangePredicate;
 
 public interface IBTreeInteriorFrame extends IBTreeFrame {
-	//public int getChildPageId(IFieldAccessor[] fields, MultiComparator cmp);
 	public int getChildPageId(RangePredicate pred, MultiComparator srcCmp);
 	public int getLeftmostChildPageId(MultiComparator cmp);
 	public int getRightmostChildPageId(MultiComparator cmp);
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IBTreeTupleReference.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IBTreeTupleReference.java
new file mode 100644
index 0000000..4e82bed
--- /dev/null
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IBTreeTupleReference.java
@@ -0,0 +1,11 @@
+package edu.uci.ics.hyracks.storage.am.btree.api;
+
+import java.nio.ByteBuffer;
+
+import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
+
+public interface IBTreeTupleReference extends ITupleReference {
+	public void setFieldCount(int fieldCount);	
+	public void resetByOffset(ByteBuffer buf, int tupleStartOffset);
+	public void resetByTupleIndex(IBTreeFrame frame, int tupleIndex);
+}
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IFieldIterator.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IFieldIterator.java
deleted file mode 100644
index 65702b2..0000000
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IFieldIterator.java
+++ /dev/null
@@ -1,40 +0,0 @@
-/*
- * Copyright 2009-2010 by The Regents of the University of California
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * you may obtain a copy of the License from
- * 
- *     http://www.apache.org/licenses/LICENSE-2.0
- * 
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package edu.uci.ics.hyracks.storage.am.btree.api;
-
-import java.nio.ByteBuffer;
-
-public interface IFieldIterator {
-	public void setFrame(IBTreeFrame frame);
-
-    public void setFields(IFieldAccessor[] fields);
-    
-    public void openRecSlotNum(int recSlotNum);
-    
-    public void openRecSlotOff(int recSlotOff);
-    
-    public void reset();
-    
-    public void nextField();
-    
-    public int getFieldOff();
-    
-    public int getFieldSize();
-    
-    public int copyFields(int startField, int endField, byte[] dest, int destOff);   
-    
-    public ByteBuffer getBuffer();
-}
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IPrefixSlotManager.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IPrefixSlotManager.java
index 1b72b0f..3f74077 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IPrefixSlotManager.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IPrefixSlotManager.java
@@ -20,23 +20,23 @@
 import edu.uci.ics.hyracks.storage.am.btree.impls.MultiComparator;
 
 // a slot consists of two fields:
-// first field is 1 byte, it indicates the slot number of a prefix record
+// first field is 1 byte, it indicates the slot number of a prefix tuple
 // we call the first field prefixSlotOff
-// second field is 3 bytes, it points to the start offset of a record
-// we call the second field recOff
+// second field is 3 bytes, it points to the start offset of a tuple
+// we call the second field tupleOff
 
 // we distinguish between two slot types:
-// prefix slots that point to prefix records, 
-// a frame is assumed to have a field numPrefixRecords
-// record slots that point to data records
-// a frame is assumed to have a field numRecords
-// a record slot contains a record pointer and a pointer to a prefix slot (prefix slot number) 
+// prefix slots that point to prefix tuples, 
+// a frame is assumed to have a field numPrefixTuples
+// tuple slots that point to data tuples
+// a frame is assumed to have a field numTuples
+// a tuple slot contains a tuple pointer and a pointer to a prefix slot (prefix slot number) 
 
 // INSERT procedure
-// a record insertion may use an existing prefix record 
-// a record insertion may never create a new prefix record
+// a tuple insertion may use an existing prefix tuple 
+// a tuple insertion may never create a new prefix tuple
 // modifying the prefix slots would be extremely expensive because: 
-// potentially all records slots would have to change their prefix slot pointers
+// potentially all tuples slots would have to change their prefix slot pointers
 // all prefixes are recomputed during a reorg or compaction
 
 public interface IPrefixSlotManager {
@@ -46,19 +46,19 @@
 	public int decodeSecondSlotField(int slot);		
 	public int encodeSlotFields(int firstField, int secondField);
 	
-	public int findSlot(ITupleReference tuple, MultiComparator multiCmp, boolean exact);
-	public int insertSlot(int slot, int recOff);
-					
-	// returns prefix slot number, returns RECORD_UNCOMPRESSED if none found
-	public int findPrefix(ITupleReference tuple, MultiComparator multiCmp);
+	public int findSlot(ITupleReference tuple, IBTreeTupleReference frameTuple, IBTreeTupleReference framePrefixTuple, MultiComparator multiCmp, boolean exact);
+	public int insertSlot(int slot, int tupleOff);
 	
-	public int getRecSlotStartOff();
-	public int getRecSlotEndOff();
+	// returns prefix slot number, returns TUPLE_UNCOMPRESSED if none found
+	public int findPrefix(ITupleReference tuple, IBTreeTupleReference framePrefixTuple, MultiComparator multiCmp);
+	
+	public int getTupleSlotStartOff();
+	public int getTupleSlotEndOff();
 	
 	public int getPrefixSlotStartOff();
 	public int getPrefixSlotEndOff();
 	
-	public int getRecSlotOff(int slotNum);
+	public int getTupleSlotOff(int slotNum);
 	public int getPrefixSlotOff(int slotNum);	
 		
 	public int getSlotSize();		
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/ISlotManager.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/ISlotManager.java
index 04fb2c1..e289c73 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/ISlotManager.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/ISlotManager.java
@@ -21,13 +21,13 @@
 public interface ISlotManager {
 	public void setFrame(IBTreeFrame frame);
 	
-	public int findSlot(ITupleReference tuple, MultiComparator multiCmp, boolean exact);
-	public int insertSlot(int slotOff, int recOff);
+	public int findSlot(ITupleReference tuple, IBTreeTupleReference pageTuple, MultiComparator multiCmp, boolean exact);
+	public int insertSlot(int slotOff, int tupleOff);
 	
 	public int getSlotStartOff();
 	public int getSlotEndOff();
 	
-	public int getRecOff(int slotOff);		
+	public int getTupleOff(int slotOff);		
 	public void setSlot(int slotOff, int value);	
 	
 	public int getSlotOff(int slotNum);
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/ITupleWriter.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/ITupleWriter.java
new file mode 100644
index 0000000..52c3464
--- /dev/null
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/ITupleWriter.java
@@ -0,0 +1,17 @@
+package edu.uci.ics.hyracks.storage.am.btree.api;
+
+import java.nio.ByteBuffer;
+
+import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
+
+public interface ITupleWriter {	
+	public int writeTuple(ITupleReference tuple, ByteBuffer targetBuf, int targetOff);
+	public int bytesRequired(ITupleReference tuple);
+	
+	public int writeTupleFields(ITupleReference tuple, int startField, int numFields, ByteBuffer targetBuf, int targetOff);
+	public int bytesRequired(ITupleReference tuple, int startField, int numFields);
+	
+	// return a tuplereference instance that can read the tuple written by this writer
+	// the main idea is that the format of the written tuple may not be the same as the format written by this writer
+	public ITupleReference createTupleReference();
+}
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/compressors/FieldPrefixCompressor.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/compressors/FieldPrefixCompressor.java
index 08dca87..cd637ea 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/compressors/FieldPrefixCompressor.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/compressors/FieldPrefixCompressor.java
@@ -25,16 +25,17 @@
 import edu.uci.ics.hyracks.storage.am.btree.api.IFrameCompressor;
 import edu.uci.ics.hyracks.storage.am.btree.api.IPrefixSlotManager;
 import edu.uci.ics.hyracks.storage.am.btree.frames.FieldPrefixNSMLeafFrame;
-import edu.uci.ics.hyracks.storage.am.btree.impls.FieldPrefixFieldIterator;
 import edu.uci.ics.hyracks.storage.am.btree.impls.FieldPrefixSlotManager;
+import edu.uci.ics.hyracks.storage.am.btree.impls.FieldPrefixTupleReference;
 import edu.uci.ics.hyracks.storage.am.btree.impls.MultiComparator;
+import edu.uci.ics.hyracks.storage.am.btree.impls.SimpleTupleWriter;
 
 public class FieldPrefixCompressor implements IFrameCompressor {
 	
-	// minimum ratio of uncompressed records to total records to consider re-compression
+	// minimum ratio of uncompressed tuples to total tuple to consider re-compression
 	private float ratioThreshold;
 	
-	// minimum number of records matching field prefixes to consider compressing them 
+	// minimum number of tuple matching field prefixes to consider compressing them 
 	private int occurrenceThreshold;
 	
 	public FieldPrefixCompressor(float ratioThreshold, int occurrenceThreshold) {
@@ -44,16 +45,16 @@
 	
 	@Override
 	public boolean compress(FieldPrefixNSMLeafFrame frame, MultiComparator cmp) throws Exception {		
-		int numRecords = frame.getNumRecords();
-    	if(numRecords <= 0) {
-            frame.setNumPrefixRecords(0);
+		int tupleCount = frame.getTupleCount();
+    	if(tupleCount <= 0) {
+            frame.setPrefixTupleCount(0);
             frame.setFreeSpaceOff(frame.getOrigFreeSpaceOff());
             frame.setTotalFreeSpace(frame.getOrigTotalFreeSpace());            
             return false;
         }
     	    	
-    	int numUncompressedRecords = frame.getNumUncompressedRecords();
-    	float ratio = (float)numUncompressedRecords / (float)numRecords;    	
+    	int uncompressedTupleCount = frame.getUncompressedTupleCount();
+    	float ratio = (float)uncompressedTupleCount / (float)tupleCount;    	
     	if(ratio < ratioThreshold) return false;    	
     	
         IBinaryComparator[] cmps = cmp.getComparators();
@@ -67,7 +68,7 @@
         ArrayList<KeyPartition> keyPartitions = getKeyPartitions(frame, cmp, occurrenceThreshold);
         if(keyPartitions.size() == 0) return false;        
         
-        // for each keyPartition, determine the best prefix length for compression, and count how many prefix records we would need in total
+        // for each keyPartition, determine the best prefix length for compression, and count how many prefix tuple we would need in total
         int totalSlotsNeeded = 0;
         int totalPrefixBytes = 0;
         for(KeyPartition kp : keyPartitions) {        	
@@ -128,119 +129,126 @@
         
         //System.out.println("TOTALPREFIXBYTES: " + totalPrefixBytes);
         
-        int[] newRecordSlots = new int[numRecords];
+        int[] newTupleSlots = new int[tupleCount];
         
         // WARNING: our hope is that compression is infrequent
-        // here we allocate a big chunk of memory to temporary hold the new, re-compressed records
+        // here we allocate a big chunk of memory to temporary hold the new, re-compressed tuple
         // in general it is very hard to avoid this step        
         int prefixFreeSpace = frame.getOrigFreeSpaceOff();
-        int recordFreeSpace = prefixFreeSpace + totalPrefixBytes;
+        int tupleFreeSpace = prefixFreeSpace + totalPrefixBytes;
         byte[] buffer = new byte[buf.capacity()];        
+        ByteBuffer byteBuffer = ByteBuffer.wrap(buffer);
         
         // perform compression, and reorg
         // we assume that the keyPartitions are sorted by the prefixes (i.e., in the logical target order)
         int kpIndex = 0;
-        int recSlotNum = 0;
-        int prefixSlotNum = 0;
-        numUncompressedRecords = 0;
-        FieldPrefixFieldIterator recToWrite = new FieldPrefixFieldIterator(fields, frame);
-        while(recSlotNum < numRecords) {           
+        int tupleIndex = 0;
+        int prefixTupleIndex = 0;
+        uncompressedTupleCount = 0;
+        
+        FieldPrefixTupleReference tupleToWrite = new FieldPrefixTupleReference();
+        tupleToWrite.setFieldCount(fields.length);
+        
+        SimpleTupleWriter tupleWriter = new SimpleTupleWriter();
+        
+        while(tupleIndex < tupleCount) {           
             if(kpIndex < keyPartitions.size()) {
             	
             	// beginning of keyPartition found, compress entire keyPartition
-            	if(recSlotNum == keyPartitions.get(kpIndex).firstRecSlotNum) {            		
+            	if(tupleIndex == keyPartitions.get(kpIndex).firstTupleIndex) {            		
             		
             		// number of fields we decided to use for compression of this keyPartition
             		int numFieldsToCompress = keyPartitions.get(kpIndex).maxPmiIndex + 1;            		
-            		int segmentStart = keyPartitions.get(kpIndex).firstRecSlotNum;
-            		int recordsInSegment = 1;
+            		int segmentStart = keyPartitions.get(kpIndex).firstTupleIndex;
+            		int tuplesInSegment = 1;
             		
             		//System.out.println("PROCESSING KEYPARTITION: " + kpIndex + " RANGE: " + keyPartitions.get(kpIndex).firstRecSlotNum + " " + keyPartitions.get(kpIndex).lastRecSlotNum + " FIELDSTOCOMPRESS: " + numFieldsToCompress);
             		
-            		FieldPrefixFieldIterator prevRec = new FieldPrefixFieldIterator(fields, frame);        
-                    FieldPrefixFieldIterator rec = new FieldPrefixFieldIterator(fields, frame);
-                    
-                    for(int i = recSlotNum + 1; i <= keyPartitions.get(kpIndex).lastRecSlotNum; i++) {
-                    	prevRec.openRecSlotNum(i - 1);
-                        rec.openRecSlotNum(i);
-                        
-                        // check if records match in numFieldsToCompress of their first fields
+            		FieldPrefixTupleReference prevTuple = new FieldPrefixTupleReference();
+            		prevTuple.setFieldCount(fields.length);
+            		
+            		FieldPrefixTupleReference tuple = new FieldPrefixTupleReference();
+            		tuple.setFieldCount(fields.length);
+            		            	
+                    for(int i = tupleIndex + 1; i <= keyPartitions.get(kpIndex).lastTupleIndex; i++) {
+                    	prevTuple.resetByTupleIndex(frame, i - 1);
+                    	tuple.resetByTupleIndex(frame, i);
+                    	                        
+                        // check if tuples match in numFieldsToCompress of their first fields
                         int prefixFieldsMatch = 0;
                         for(int j = 0; j < numFieldsToCompress; j++) {                                                    
-                            if(cmps[j].compare(pageArray, prevRec.getFieldOff(), prevRec.getFieldSize(), pageArray, rec.getFieldOff(), rec.getFieldSize()) == 0) prefixFieldsMatch++;
-                            else break;
-                            prevRec.nextField();
-                            rec.nextField();
+                            if(cmps[j].compare(pageArray, prevTuple.getFieldStart(j), prevTuple.getFieldLength(j), pageArray, tuple.getFieldStart(j), tuple.getFieldLength(j)) == 0) prefixFieldsMatch++;
+                            else break;                          
                         }
                                                                   	                       
-                        // the two records must match in exactly the number of fields we decided to compress for this keyPartition
+                        // the two tuples must match in exactly the number of fields we decided to compress for this keyPartition
                         int processSegments = 0;
-                        if(prefixFieldsMatch == numFieldsToCompress) recordsInSegment++;                        
+                        if(prefixFieldsMatch == numFieldsToCompress) tuplesInSegment++;                        
                         else processSegments++;
 
-                        if(i == keyPartitions.get(kpIndex).lastRecSlotNum) processSegments++;
+                        if(i == keyPartitions.get(kpIndex).lastTupleIndex) processSegments++;
                         
                         for(int r = 0; r < processSegments; r++) {
                         	// compress current segment and then start new segment                       
-                        	if(recordsInSegment < occurrenceThreshold || numFieldsToCompress <= 0) {
-                        		// segment does not have at least occurrenceThreshold records, so write records uncompressed
-                        		for(int j = 0; j < recordsInSegment; j++) {
+                        	if(tuplesInSegment < occurrenceThreshold || numFieldsToCompress <= 0) {
+                        		// segment does not have at least occurrenceThreshold tuples, so write tuples uncompressed
+                        		for(int j = 0; j < tuplesInSegment; j++) {
                         		    int slotNum = segmentStart + j;
-                        		    recToWrite.openRecSlotNum(slotNum);
-                        		    newRecordSlots[numRecords - 1 - slotNum] = slotManager.encodeSlotFields(FieldPrefixSlotManager.RECORD_UNCOMPRESSED, recordFreeSpace);
-                        		    recordFreeSpace += recToWrite.copyFields(0, fields.length - 1, buffer, recordFreeSpace);
+                        		    tupleToWrite.resetByTupleIndex(frame, slotNum); 
+                        		    newTupleSlots[tupleCount - 1 - slotNum] = slotManager.encodeSlotFields(FieldPrefixSlotManager.TUPLE_UNCOMPRESSED, tupleFreeSpace);
+                        		    tupleFreeSpace += tupleWriter.writeTuple(tupleToWrite, byteBuffer, tupleFreeSpace);                        		    
                         		}
-                        		numUncompressedRecords += recordsInSegment;
+                        		uncompressedTupleCount += tuplesInSegment;
                         	}
                         	else {
-                        	    // segment has enough records, compress segment
-                        		// extract prefix, write prefix record to buffer, and set prefix slot                        	    
-                        		newPrefixSlots[newPrefixSlots.length - 1 - prefixSlotNum] = slotManager.encodeSlotFields(numFieldsToCompress, prefixFreeSpace);
+                        	    // segment has enough tuples, compress segment
+                        		// extract prefix, write prefix tuple to buffer, and set prefix slot                        	    
+                        		newPrefixSlots[newPrefixSlots.length - 1 - prefixTupleIndex] = slotManager.encodeSlotFields(numFieldsToCompress, prefixFreeSpace);
                         	    //int tmp = freeSpace;
                         	    //prevRec.reset();
-                        	    //System.out.println("SOURCE CONTENTS: " + buf.getInt(prevRec.getFieldOff()) + " " + buf.getInt(prevRec.getFieldOff()+4));
-                        	    prefixFreeSpace += prevRec.copyFields(0, numFieldsToCompress - 1, buffer, prefixFreeSpace);            	                            	                           	    
+                        	    //System.out.println("SOURCE CONTENTS: " + buf.getInt(prevRec.getFieldOff()) + " " + buf.getInt(prevRec.getFieldOff()+4));                        	    
+                        		prefixFreeSpace += tupleWriter.writeTupleFields(prevTuple, 0, numFieldsToCompress, byteBuffer, prefixFreeSpace);            		                        		                      	                           	    
                         	    //System.out.println("WRITING PREFIX RECORD " + prefixSlotNum + " AT " + tmp + " " + freeSpace);
                         	    //System.out.print("CONTENTS: ");
                         	    //for(int x = 0; x < numFieldsToCompress; x++) System.out.print(buf.getInt(tmp + x*4) + " ");
                         	    //System.out.println();
                         	    
-                        		// truncate records, write them to buffer, and set record slots
-                        	    for(int j = 0; j < recordsInSegment; j++) {
-                        	        int slotNum = segmentStart + j;
-                                    recToWrite.openRecSlotNum(slotNum);
-                                    newRecordSlots[numRecords - 1 - slotNum] = slotManager.encodeSlotFields(prefixSlotNum, recordFreeSpace);
-                                    recordFreeSpace += recToWrite.copyFields(numFieldsToCompress, fields.length - 1, buffer, recordFreeSpace);                              
+                        		// truncate tuples, write them to buffer, and set tuple slots
+                        	    for(int j = 0; j < tuplesInSegment; j++) {
+                        	        int currTupleIndex = segmentStart + j;
+                                    tupleToWrite.resetByTupleIndex(frame, currTupleIndex);
+                                    newTupleSlots[tupleCount - 1 - currTupleIndex] = slotManager.encodeSlotFields(prefixTupleIndex, tupleFreeSpace);                                    
+                                    tupleFreeSpace += tupleWriter.writeTupleFields(tupleToWrite, numFieldsToCompress, fields.length - numFieldsToCompress, byteBuffer, tupleFreeSpace);
                                 }
                         	    
-                        	    prefixSlotNum++;
+                        	    prefixTupleIndex++;
                         	}
                         	
                         	// begin new segment
                         	segmentStart = i;
-                        	recordsInSegment = 1;               	
+                        	tuplesInSegment = 1;               	
                         }                   
                     }
                     
-                    recSlotNum = keyPartitions.get(kpIndex).lastRecSlotNum;
+                    tupleIndex = keyPartitions.get(kpIndex).lastTupleIndex;
                 	kpIndex++; 	
                 }
             	else {
-                    // just write the record uncompressed
-            	    recToWrite.openRecSlotNum(recSlotNum);
-            	    newRecordSlots[numRecords - 1 - recSlotNum] = slotManager.encodeSlotFields(FieldPrefixSlotManager.RECORD_UNCOMPRESSED, recordFreeSpace);
-            	    recordFreeSpace += recToWrite.copyFields(0, fields.length - 1, buffer, recordFreeSpace);
-            	    numUncompressedRecords++;
+                    // just write the tuple uncompressed
+            		tupleToWrite.resetByTupleIndex(frame, tupleIndex);
+            	    newTupleSlots[tupleCount - 1 - tupleIndex] = slotManager.encodeSlotFields(FieldPrefixSlotManager.TUPLE_UNCOMPRESSED, tupleFreeSpace);
+            	    tupleFreeSpace += tupleWriter.writeTuple(tupleToWrite, byteBuffer, tupleFreeSpace);
+            	    uncompressedTupleCount++;
             	}
             }
             else {
-                // just write the record uncompressed
-                recToWrite.openRecSlotNum(recSlotNum);
-                newRecordSlots[numRecords - 1 - recSlotNum] = slotManager.encodeSlotFields(FieldPrefixSlotManager.RECORD_UNCOMPRESSED, recordFreeSpace);
-                recordFreeSpace += recToWrite.copyFields(0, fields.length - 1, buffer, recordFreeSpace);
-                numUncompressedRecords++;
+                // just write the tuple uncompressed
+            	tupleToWrite.resetByTupleIndex(frame, tupleIndex);
+                newTupleSlots[tupleCount - 1 - tupleIndex] = slotManager.encodeSlotFields(FieldPrefixSlotManager.TUPLE_UNCOMPRESSED, tupleFreeSpace);
+                tupleFreeSpace += tupleWriter.writeTuple(tupleToWrite, byteBuffer, tupleFreeSpace);
+                uncompressedTupleCount++;
             }   
-            recSlotNum++;
+            tupleIndex++;
         }            
         
         // sanity check to see if we have written exactly as many prefix bytes as computed before
@@ -251,12 +259,12 @@
         // in some rare instances our procedure could even increase the space requirement which is very dangerous
         // this can happen to to the greedy solution of the knapsack-like problem
         // therefore, we check if the new space exceeds the page size to avoid the only danger of an increasing space
-        int totalSpace = recordFreeSpace + newRecordSlots.length * slotManager.getSlotSize() + newPrefixSlots.length * slotManager.getSlotSize();
+        int totalSpace = tupleFreeSpace + newTupleSlots.length * slotManager.getSlotSize() + newPrefixSlots.length * slotManager.getSlotSize();
         if(totalSpace > buf.capacity()) return false; // just leave the page as is
         
-        // copy new records and new slots into original page
+        // copy new tuple and new slots into original page
         int freeSpaceAfterInit = frame.getOrigFreeSpaceOff();
-        System.arraycopy(buffer, freeSpaceAfterInit, pageArray, freeSpaceAfterInit, recordFreeSpace - freeSpaceAfterInit);
+        System.arraycopy(buffer, freeSpaceAfterInit, pageArray, freeSpaceAfterInit, tupleFreeSpace - freeSpaceAfterInit);
         
         // copy prefix slots
         int slotOffRunner = buf.capacity() - slotManager.getSlotSize();
@@ -265,9 +273,9 @@
             slotOffRunner -= slotManager.getSlotSize();
         }
         
-        // copy record slots
-        for(int i = 0; i < newRecordSlots.length; i++) {
-            buf.putInt(slotOffRunner, newRecordSlots[newRecordSlots.length - 1 - i]);
+        // copy tuple slots
+        for(int i = 0; i < newTupleSlots.length; i++) {
+            buf.putInt(slotOffRunner, newTupleSlots[newTupleSlots.length - 1 - i]);
             slotOffRunner -= slotManager.getSlotSize();
         }
         
@@ -284,21 +292,21 @@
 //        System.out.println("RECORDS: " + newRecordSlots.length);
         
         // update space fields, TODO: we need to update more fields 
-        frame.setFreeSpaceOff(recordFreeSpace);
-        frame.setNumPrefixRecords(newPrefixSlots.length);
-        frame.setNumUncompressedRecords(numUncompressedRecords);
-        int totalFreeSpace = buf.capacity() - recordFreeSpace - ((newRecordSlots.length + newPrefixSlots.length) * slotManager.getSlotSize());
+        frame.setFreeSpaceOff(tupleFreeSpace);
+        frame.setPrefixTupleCount(newPrefixSlots.length);
+        frame.setUncompressedTupleCount(uncompressedTupleCount);
+        int totalFreeSpace = buf.capacity() - tupleFreeSpace - ((newTupleSlots.length + newPrefixSlots.length) * slotManager.getSlotSize());
         frame.setTotalFreeSpace(totalFreeSpace);
         
         return true;
     }
 	
-    // we perform an analysis pass over the records to determine the costs and benefits of different compression options
-    // a "keypartition" is a range of records that has an identical first field
+    // we perform an analysis pass over the tuples to determine the costs and benefits of different compression options
+    // a "keypartition" is a range of tuples that has an identical first field
     // for each keypartition we chose a prefix length to use for compression
-    // i.e., all records in a keypartition will be compressed based on the same prefix length (number of fields)
+    // i.e., all tuples in a keypartition will be compressed based on the same prefix length (number of fields)
     // the prefix length may be different for different keypartitions
-    // the occurrenceThreshold determines the minimum number of records that must share a common prefix in order for us to consider compressing them        
+    // the occurrenceThreshold determines the minimum number of tuples that must share a common prefix in order for us to consider compressing them        
     private ArrayList<KeyPartition> getKeyPartitions(FieldPrefixNSMLeafFrame frame, MultiComparator cmp, int occurrenceThreshold) {        
     	IBinaryComparator[] cmps = cmp.getComparators();
         IFieldAccessor[] fields = cmp.getFields();
@@ -312,47 +320,49 @@
         KeyPartition kp = new KeyPartition(maxCmps);        
         keyPartitions.add(kp);
         
-        FieldPrefixFieldIterator prevRec = new FieldPrefixFieldIterator(fields, frame);        
-        FieldPrefixFieldIterator rec = new FieldPrefixFieldIterator(fields, frame);
-        
-        kp.firstRecSlotNum = 0;        
-        int numRecords = frame.getNumRecords();
-        for(int i = 1; i < numRecords; i++) {        	        	
-        	prevRec.openRecSlotNum(i-1);
-        	rec.openRecSlotNum(i);
-        	
+        FieldPrefixTupleReference prevTuple = new FieldPrefixTupleReference();
+		prevTuple.setFieldCount(fields.length);
+		
+		FieldPrefixTupleReference tuple = new FieldPrefixTupleReference();
+		tuple.setFieldCount(fields.length);
+		
+		SimpleTupleWriter tupleWriter = new SimpleTupleWriter();
+		
+        kp.firstTupleIndex = 0;        
+        int tupleCount = frame.getTupleCount();
+        for(int i = 1; i < tupleCount; i++) {        	        	
+        	prevTuple.resetByTupleIndex(frame, i - 1);
+        	tuple.resetByTupleIndex(frame, i);
+        	        	
             //System.out.println("BEFORE RECORD: " + i + " " + rec.recSlotOff + " " + rec.recOff);
             //kp.print();
             
-            int prefixFieldsMatch = 0;            
-            int prefixBytes = 0; // counts the bytes of the common prefix fields
-            
+            int prefixFieldsMatch = 0; 
             for(int j = 0; j < maxCmps; j++) {                
             	            	
-            	if(cmps[j].compare(pageArray, prevRec.getFieldOff(), prevRec.getFieldSize(), pageArray, rec.getFieldOff(), prevRec.getFieldSize()) == 0) {
+            	if(cmps[j].compare(pageArray, prevTuple.getFieldStart(j), prevTuple.getFieldLength(j), pageArray, tuple.getFieldStart(j), prevTuple.getFieldLength(j)) == 0) {
                     prefixFieldsMatch++;
-                    kp.pmi[j].matches++;                     
-                    prefixBytes += rec.getFieldSize();            
+                    kp.pmi[j].matches++;                    
                     
+                    int prefixBytes = tupleWriter.bytesRequired(tuple, 0, prefixFieldsMatch);
+                    int spaceBenefit = tupleWriter.bytesRequired(tuple) - tupleWriter.bytesRequired(tuple, prefixFieldsMatch, tuple.getFieldCount() - prefixFieldsMatch);
+                                        
                     if(kp.pmi[j].matches == occurrenceThreshold) {
                         // if we compress this prefix, we pay the cost of storing it once, plus the size for one prefix slot
                         kp.pmi[j].prefixBytes += prefixBytes;
                         kp.pmi[j].spaceCost += prefixBytes + slotManager.getSlotSize();
                         kp.pmi[j].prefixSlotsNeeded++;
-                        kp.pmi[j].spaceBenefit += occurrenceThreshold*prefixBytes;                        
+                        kp.pmi[j].spaceBenefit += occurrenceThreshold * spaceBenefit;                        
                     }
                     else if(kp.pmi[j].matches > occurrenceThreshold) {
-                        // we are beyond the occurrence threshold, every additional record with a matching prefix increases the benefit 
-                        kp.pmi[j].spaceBenefit += prefixBytes;
+                        // we are beyond the occurrence threshold, every additional tuple with a matching prefix increases the benefit 
+                        kp.pmi[j].spaceBenefit += spaceBenefit;
                     }
                 }
                 else {
-                    kp.pmi[j].matches = 1;     
+                    kp.pmi[j].matches = 1;
                     break;
-                }         
-                
-                prevRec.nextField();
-                rec.nextField();
+                }                              
             }
             
             //System.out.println();
@@ -363,19 +373,19 @@
             // this means not even the first field matched, so we start to consider a new "key partition"
             if(maxCmps > 0 && prefixFieldsMatch == 0) {                
             	//System.out.println("NEW KEY PARTITION");            	
-            	kp.lastRecSlotNum = i-1;
+            	kp.lastTupleIndex = i-1;
             	
-                // remove keyPartitions that don't have enough records
-                if((kp.lastRecSlotNum - kp.firstRecSlotNum) + 1 < occurrenceThreshold) keyPartitions.remove(keyPartitions.size() - 1);
+                // remove keyPartitions that don't have enough tuples
+                if((kp.lastTupleIndex - kp.firstTupleIndex) + 1 < occurrenceThreshold) keyPartitions.remove(keyPartitions.size() - 1);
                                 
             	kp = new KeyPartition(maxCmps);
                 keyPartitions.add(kp);
-                kp.firstRecSlotNum = i;
+                kp.firstTupleIndex = i;
             }         
         }   
-        kp.lastRecSlotNum = numRecords - 1;
-        // remove keyPartitions that don't have enough records
-        if((kp.lastRecSlotNum - kp.firstRecSlotNum) + 1 < occurrenceThreshold) keyPartitions.remove(keyPartitions.size() - 1);
+        kp.lastTupleIndex = tupleCount - 1;
+        // remove keyPartitions that don't have enough tuples
+        if((kp.lastTupleIndex - kp.firstTupleIndex) + 1 < occurrenceThreshold) keyPartitions.remove(keyPartitions.size() - 1);
         
         return keyPartitions;
     }
@@ -390,8 +400,8 @@
     }
     
     private class KeyPartition {        
-        public int firstRecSlotNum;
-        public int lastRecSlotNum;
+        public int firstTupleIndex;
+        public int lastTupleIndex;
         public PrefixMatchInfo[] pmi;
         
         public int maxBenefitMinusCost = 0;
@@ -426,7 +436,7 @@
     private class SortByOriginalRank implements Comparator<KeyPartition>{            
         @Override
         public int compare(KeyPartition a, KeyPartition b) {
-            return a.firstRecSlotNum - b.firstRecSlotNum;
+            return a.firstTupleIndex - b.firstTupleIndex;
         }
     }
 }
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeBulkLoadOperatorNodePushable.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeBulkLoadOperatorNodePushable.java
index 9120f15..39bf987 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeBulkLoadOperatorNodePushable.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeBulkLoadOperatorNodePushable.java
@@ -61,7 +61,7 @@
         for (int i = 0; i < tupleCount; i++) {
             tuple.reset(accessor, i);
             try {
-                btreeOpHelper.getBTree().bulkLoadAddRecord(bulkLoadCtx, tuple);
+                btreeOpHelper.getBTree().bulkLoadAddTuple(bulkLoadCtx, tuple);
             } catch (Exception e) {
                 e.printStackTrace();
             }
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeDiskOrderScanOperatorNodePushable.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeDiskOrderScanOperatorNodePushable.java
index 0c3b8d5..bcfcd08 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeDiskOrderScanOperatorNodePushable.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeDiskOrderScanOperatorNodePushable.java
@@ -23,10 +23,10 @@
 import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
 import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAppender;
 import edu.uci.ics.hyracks.dataflow.common.comm.util.FrameUtils;
+import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
 import edu.uci.ics.hyracks.dataflow.std.base.AbstractUnaryOutputSourceOperatorNodePushable;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeLeafFrame;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeMetaDataFrame;
-import edu.uci.ics.hyracks.storage.am.btree.api.IFieldIterator;
 import edu.uci.ics.hyracks.storage.am.btree.frames.MetaDataFrame;
 import edu.uci.ics.hyracks.storage.am.btree.impls.DiskOrderScanCursor;
 import edu.uci.ics.hyracks.storage.am.btree.impls.MultiComparator;
@@ -67,10 +67,9 @@
                 tb.reset();
                 cursor.next();
 
-                IFieldIterator fieldIter = cursor.getFieldIterator();
-                for (int i = 0; i < cmp.getFields().length; i++) {
-                    int fieldLen = fieldIter.getFieldSize();
-                    dos.write(fieldIter.getBuffer().array(), fieldIter.getFieldOff(), fieldLen);
+                ITupleReference frameTuple = cursor.getTuple();                
+                for (int i = 0; i < frameTuple.getFieldCount(); i++) {
+                    dos.write(frameTuple.getFieldData(i), frameTuple.getFieldStart(i), frameTuple.getFieldLength(i));
                     tb.addFieldEndOffset();
                 }
 
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeSearchOperatorNodePushable.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeSearchOperatorNodePushable.java
index 2b78993..84b8113 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeSearchOperatorNodePushable.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeSearchOperatorNodePushable.java
@@ -29,7 +29,6 @@
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeCursor;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeInteriorFrame;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeLeafFrame;
-import edu.uci.ics.hyracks.storage.am.btree.api.IFieldIterator;
 import edu.uci.ics.hyracks.storage.am.btree.impls.BTree;
 import edu.uci.ics.hyracks.storage.am.btree.impls.MultiComparator;
 import edu.uci.ics.hyracks.storage.am.btree.impls.RangePredicate;
@@ -93,11 +92,10 @@
             while (cursor.hasNext()) {
                 tb.reset();
                 cursor.next();
-
-                IFieldIterator fieldIter = cursor.getFieldIterator();
-                for (int i = 0; i < cmp.getFields().length; i++) {
-                    int fieldLen = fieldIter.getFieldSize();
-                    dos.write(fieldIter.getBuffer().array(), fieldIter.getFieldOff(), fieldLen);
+                
+                ITupleReference frameTuple = cursor.getTuple();                
+                for (int i = 0; i < frameTuple.getFieldCount(); i++) {
+                    dos.write(frameTuple.getFieldData(i), frameTuple.getFieldStart(i), frameTuple.getFieldLength(i));
                     tb.addFieldEndOffset();
                 }
 
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/FieldPrefixNSMLeafFrame.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/FieldPrefixNSMLeafFrame.java
index 6c05eb6..98dbf4c 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/FieldPrefixNSMLeafFrame.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/FieldPrefixNSMLeafFrame.java
@@ -22,19 +22,18 @@
 import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeFrame;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeLeafFrame;
-import edu.uci.ics.hyracks.storage.am.btree.api.IFieldAccessor;
-import edu.uci.ics.hyracks.storage.am.btree.api.IFieldIterator;
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeTupleReference;
 import edu.uci.ics.hyracks.storage.am.btree.api.IFrameCompressor;
 import edu.uci.ics.hyracks.storage.am.btree.api.IPrefixSlotManager;
 import edu.uci.ics.hyracks.storage.am.btree.api.ISlotManager;
 import edu.uci.ics.hyracks.storage.am.btree.compressors.FieldPrefixCompressor;
 import edu.uci.ics.hyracks.storage.am.btree.impls.BTreeException;
-import edu.uci.ics.hyracks.storage.am.btree.impls.FieldPrefixFieldIterator;
-import edu.uci.ics.hyracks.storage.am.btree.impls.FieldPrefixPrefixTuple;
+import edu.uci.ics.hyracks.storage.am.btree.impls.FieldPrefixPrefixTupleReference;
 import edu.uci.ics.hyracks.storage.am.btree.impls.FieldPrefixSlotManager;
-import edu.uci.ics.hyracks.storage.am.btree.impls.FieldPrefixTuple;
+import edu.uci.ics.hyracks.storage.am.btree.impls.FieldPrefixTupleReference;
 import edu.uci.ics.hyracks.storage.am.btree.impls.MultiComparator;
-import edu.uci.ics.hyracks.storage.am.btree.impls.SlotOffRecOff;
+import edu.uci.ics.hyracks.storage.am.btree.impls.SimpleTupleWriter;
+import edu.uci.ics.hyracks.storage.am.btree.impls.SlotOffTupleOff;
 import edu.uci.ics.hyracks.storage.am.btree.impls.SpaceStatus;
 import edu.uci.ics.hyracks.storage.am.btree.impls.SplitKey;
 import edu.uci.ics.hyracks.storage.common.buffercache.ICachedPage;
@@ -42,15 +41,15 @@
 public class FieldPrefixNSMLeafFrame implements IBTreeLeafFrame {
 	
     protected static final int pageLsnOff = 0;                              // 0
-    protected static final int numRecordsOff = pageLsnOff + 4;              // 4    
-    protected static final int freeSpaceOff = numRecordsOff + 4;      		// 8
+    protected static final int tupleCountOff = pageLsnOff + 4;              // 4    
+    protected static final int freeSpaceOff = tupleCountOff + 4;      		// 8
     protected static final int totalFreeSpaceOff = freeSpaceOff + 4;        // 12	
 	protected static final int levelOff = totalFreeSpaceOff + 4;         	// 16
 	protected static final int smFlagOff = levelOff + 1;                   	// 17
-	protected static final int numUncompressedRecordsOff = smFlagOff + 1;				// 18
-	protected static final int numPrefixRecordsOff = numUncompressedRecordsOff + 4; 		// 21
+	protected static final int uncompressedTupleCountOff = smFlagOff + 1;				// 18
+	protected static final int prefixTupleCountOff = uncompressedTupleCountOff + 4; 		// 21
 	
-	protected static final int prevLeafOff = numPrefixRecordsOff + 4;		// 22
+	protected static final int prevLeafOff = prefixTupleCountOff + 4;		// 22
 	protected static final int nextLeafOff = prevLeafOff + 4;				// 26
 	
 	protected ICachedPage page = null;
@@ -58,9 +57,11 @@
     public IFrameCompressor compressor;
     public IPrefixSlotManager slotManager; // TODO: should be protected, but will trigger some refactoring
     
-    private FieldPrefixTuple pageTuple = new FieldPrefixTuple();
-    private FieldPrefixPrefixTuple pagePrefixTuple = new FieldPrefixPrefixTuple(); 
+    private SimpleTupleWriter tupleWriter = new SimpleTupleWriter();
     
+    private FieldPrefixTupleReference frameTuple = new FieldPrefixTupleReference();            
+    private FieldPrefixPrefixTupleReference framePrefixTuple = new FieldPrefixPrefixTupleReference();
+          
     public FieldPrefixNSMLeafFrame() {
         this.slotManager = new FieldPrefixSlotManager();
         this.compressor = new FieldPrefixCompressor(0.001f, 2);        
@@ -89,176 +90,137 @@
     }
     
     // assumptions: 
-    // 1. prefix records are stored contiguously
-    // 2. prefix records are located before records (physically on the page)
-    // this procedure will not move prefix records
+    // 1. prefix tuple are stored contiguously
+    // 2. prefix tuple are located before tuples (physically on the page)
+    // 3. prefix tuple are sorted (last prefix tuple is at highest offset)
+    // this procedure will not move prefix tuples
     @Override
     public void compact(MultiComparator cmp) {
         resetSpaceParams();
-                        
-        int numRecords = buf.getInt(numRecordsOff);
-        byte[] data = buf.array();
+        
+        frameTuple.setFieldCount(cmp.getFields().length);
+        
+        int tupleCount = buf.getInt(tupleCountOff);
         
         // determine start of target free space (depends on assumptions stated above)
         int freeSpace = buf.getInt(freeSpaceOff);        
-        int numPrefixRecords = buf.getInt(numPrefixRecordsOff);
-        if(numPrefixRecords > 0) {
-        	int prefixFields = 0;
-        	for(int i = 0; i < numPrefixRecords; i++) {
-        		int prefixSlotOff = slotManager.getPrefixSlotOff(i);
-        		int prefixSlot = buf.getInt(prefixSlotOff);
-        		int prefixRecOff = slotManager.decodeSecondSlotField(prefixSlot);
-        		if(prefixRecOff >= freeSpace) {
-        			freeSpace = prefixRecOff;
-        			prefixFields = slotManager.decodeFirstSlotField(prefixSlot);
-        		}
+        int prefixTupleCount = buf.getInt(prefixTupleCountOff);
+        if(prefixTupleCount > 0) {        	
+        	
+        	// debug
+        	int max = 0;
+        	for(int i = 0; i < prefixTupleCount; i++) {
+        		framePrefixTuple.resetByTupleIndex(this, i);
+        		int end = framePrefixTuple.getFieldStart(framePrefixTuple.getFieldCount()-1) + framePrefixTuple.getFieldLength(framePrefixTuple.getFieldCount()-1);
+        		if(end > max) max = end;
         	}
-        	for(int i = 0; i < prefixFields; i++) {
-        		freeSpace += cmp.getFields()[i].getLength(data, freeSpace);
-        	}
+        	        	
+        	framePrefixTuple.resetByTupleIndex(this, prefixTupleCount - 1);
+        	freeSpace = framePrefixTuple.getFieldStart(framePrefixTuple.getFieldCount()-1) + framePrefixTuple.getFieldLength(framePrefixTuple.getFieldCount()-1);        	
         }
-        
-        ArrayList<SlotOffRecOff> sortedRecOffs = new ArrayList<SlotOffRecOff>();
-        sortedRecOffs.ensureCapacity(numRecords);
-        for(int i = 0; i < numRecords; i++) {           
-            int recSlotOff = slotManager.getRecSlotOff(i);
-            int recSlot = buf.getInt(recSlotOff);
-            int recOff = slotManager.decodeSecondSlotField(recSlot);
-            sortedRecOffs.add(new SlotOffRecOff(recSlotOff, recOff));
+
+        ArrayList<SlotOffTupleOff> sortedTupleOffs = new ArrayList<SlotOffTupleOff>();
+        sortedTupleOffs.ensureCapacity(tupleCount);
+        for(int i = 0; i < tupleCount; i++) {           
+            int tupleSlotOff = slotManager.getTupleSlotOff(i);
+            int tupleSlot = buf.getInt(tupleSlotOff);
+            int tupleOff = slotManager.decodeSecondSlotField(tupleSlot);
+            sortedTupleOffs.add(new SlotOffTupleOff(i, tupleSlotOff, tupleOff));                        
+            
         }
-        Collections.sort(sortedRecOffs);
-        
-        for(int i = 0; i < sortedRecOffs.size(); i++) {                    	
-        	int recOff = sortedRecOffs.get(i).recOff;
-            int recSlot = buf.getInt(sortedRecOffs.get(i).slotOff);
-            int prefixSlotNum = slotManager.decodeFirstSlotField(recSlot);            
-            
-            int fieldStart = 0;
-            if(prefixSlotNum != FieldPrefixSlotManager.RECORD_UNCOMPRESSED) {
-            	int prefixSlotOff = slotManager.getPrefixSlotOff(prefixSlotNum);
-                int prefixSlot = buf.getInt(prefixSlotOff);
-                fieldStart = slotManager.decodeFirstSlotField(prefixSlot);                                
-            }
-            
-            int recRunner = recOff;
-            for(int j = fieldStart; j < cmp.getFields().length; j++) {
-            	recRunner += cmp.getFields()[j].getLength(data, recRunner);
-            }
-            int recSize = recRunner - recOff;
+        Collections.sort(sortedTupleOffs);
+               
+        for(int i = 0; i < sortedTupleOffs.size(); i++) {                    	
+        	int tupleOff = sortedTupleOffs.get(i).tupleOff;
+        	int tupleSlot = buf.getInt(sortedTupleOffs.get(i).slotOff);
+            int prefixSlotNum = slotManager.decodeFirstSlotField(tupleSlot);            
                         
-            System.arraycopy(data, recOff, data, freeSpace, recSize);
-            slotManager.setSlot(sortedRecOffs.get(i).slotOff, slotManager.encodeSlotFields(prefixSlotNum, freeSpace));
-            freeSpace += recSize;
+            frameTuple.resetByTupleIndex(this, sortedTupleOffs.get(i).tupleIndex);
+            int tupleEndOff = frameTuple.getFieldStart(frameTuple.getFieldCount()-1) + frameTuple.getFieldLength(frameTuple.getFieldCount()-1);
+            int tupleLength = tupleEndOff - tupleOff;
+            System.arraycopy(buf.array(), tupleOff, buf.array(), freeSpace, tupleLength);
+                                    
+            slotManager.setSlot(sortedTupleOffs.get(i).slotOff, slotManager.encodeSlotFields(prefixSlotNum, freeSpace));
+            freeSpace += tupleLength;
         }
         
         buf.putInt(freeSpaceOff, freeSpace);
-        int totalFreeSpace = buf.capacity() - buf.getInt(freeSpaceOff) - ((buf.getInt(numRecordsOff) + buf.getInt(numPrefixRecordsOff)) * slotManager.getSlotSize());        
+        int totalFreeSpace = buf.capacity() - buf.getInt(freeSpaceOff) - ((buf.getInt(tupleCountOff) + buf.getInt(prefixTupleCountOff)) * slotManager.getSlotSize());        
         buf.putInt(totalFreeSpaceOff, totalFreeSpace);
     }
     
     @Override
     public void delete(ITupleReference tuple, MultiComparator cmp, boolean exactDelete) throws Exception {        
-        int slot = slotManager.findSlot(tuple, cmp, true);
-        int recSlotNum = slotManager.decodeSecondSlotField(slot);
-        if(recSlotNum == FieldPrefixSlotManager.GREATEST_SLOT) {
+        int slot = slotManager.findSlot(tuple, frameTuple, framePrefixTuple, cmp, true);
+        int tupleIndex = slotManager.decodeSecondSlotField(slot);
+        if(tupleIndex == FieldPrefixSlotManager.GREATEST_SLOT) {
             throw new BTreeException("Key to be deleted does not exist.");   
         }
         else {
             int prefixSlotNum = slotManager.decodeFirstSlotField(slot);            
-            int recSlotOff = slotManager.getRecSlotOff(recSlotNum);
+            int tupleSlotOff = slotManager.getTupleSlotOff(tupleIndex);
             
-            if(exactDelete) {                                    
-                pageTuple.setFields(cmp.getFields());
-                pageTuple.setFrame(this);
-                pageTuple.openRecSlotNum(recSlotNum);
+            if(exactDelete) {                
+                frameTuple.setFieldCount(cmp.getFields().length);
+                frameTuple.resetByTupleIndex(this, tupleIndex);
                 
-                int comparison = cmp.fieldRangeCompare(tuple, pageTuple, cmp.getKeyLength()-1, cmp.getFields().length - cmp.getKeyLength());
+                int comparison = cmp.fieldRangeCompare(tuple, frameTuple, cmp.getKeyLength()-1, cmp.getFields().length - cmp.getKeyLength());
                 if(comparison != 0) {
-                	throw new BTreeException("Cannot delete record. Byte-by-byte comparison failed to prove equality.");
+                	throw new BTreeException("Cannot delete tuple. Byte-by-byte comparison failed to prove equality.");
                 }                                  
             }
             
             // perform deletion (we just do a memcpy to overwrite the slot)
-            int slotEndOff = slotManager.getRecSlotEndOff();
-            int length = recSlotOff - slotEndOff;
+            int slotEndOff = slotManager.getTupleSlotEndOff();
+            int length = tupleSlotOff - slotEndOff;
             System.arraycopy(buf.array(), slotEndOff, buf.array(), slotEndOff + slotManager.getSlotSize(), length);
             
-            // maintain space information, get size of record suffix (suffix could be entire record)
-            int recSize = 0;                        
+            // maintain space information, get size of tuple suffix (suffix could be entire tuple)
+            int tupleSize = 0;                        
             int suffixFieldStart = 0;
-            FieldPrefixFieldIterator fieldIter = new FieldPrefixFieldIterator(cmp.getFields(), this);                 
-            fieldIter.openRecSlotOff(recSlotOff);
-            if(prefixSlotNum == FieldPrefixSlotManager.RECORD_UNCOMPRESSED) {
+            if(prefixSlotNum == FieldPrefixSlotManager.TUPLE_UNCOMPRESSED) {
                 suffixFieldStart = 0;
-                buf.putInt(numUncompressedRecordsOff, buf.getInt(numUncompressedRecordsOff) - 1);
+                buf.putInt(uncompressedTupleCountOff, buf.getInt(uncompressedTupleCountOff) - 1);
             }
             else {                
                 int prefixSlot = buf.getInt(slotManager.getPrefixSlotOff(prefixSlotNum));
                 suffixFieldStart = slotManager.decodeFirstSlotField(prefixSlot); 
             }
             
-            for(int i = 0; i < suffixFieldStart; i++) {
-                fieldIter.nextField();
-            }
-            for(int i = suffixFieldStart; i < cmp.getFields().length; i++) {
-                recSize += cmp.getFields()[i].getLength(buf.array(), fieldIter.getFieldOff());
-                fieldIter.nextField();
-            }
+            frameTuple.resetByTupleIndex(this, tupleIndex);
+            tupleSize = tupleWriter.bytesRequired(frameTuple, suffixFieldStart, frameTuple.getFieldCount() - suffixFieldStart);
             
-            buf.putInt(numRecordsOff, buf.getInt(numRecordsOff) - 1);
-            buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) + recSize + slotManager.getSlotSize());            
+            buf.putInt(tupleCountOff, buf.getInt(tupleCountOff) - 1);
+            buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) + tupleSize + slotManager.getSlotSize());            
         }    
     }
     
     @Override
     public SpaceStatus hasSpaceInsert(ITupleReference tuple, MultiComparator cmp) {                    	
-    	int freeContiguous = buf.capacity() - buf.getInt(freeSpaceOff) - ((buf.getInt(numRecordsOff) + buf.getInt(numPrefixRecordsOff)) * slotManager.getSlotSize());     	    	                        
+    	int freeContiguous = buf.capacity() - buf.getInt(freeSpaceOff) - ((buf.getInt(tupleCountOff) + buf.getInt(prefixTupleCountOff)) * slotManager.getSlotSize());     	    	                        
         
-    	int tupleSpace = spaceNeededForTuple(tuple);
+    	int bytesRequired = tupleWriter.bytesRequired(tuple);
     	
-        // see if the record would fit uncompressed
-        if(tupleSpace + slotManager.getSlotSize() <= freeContiguous) return SpaceStatus.SUFFICIENT_CONTIGUOUS_SPACE;
+        // see if the tuple would fit uncompressed
+        if(bytesRequired + slotManager.getSlotSize() <= freeContiguous) return SpaceStatus.SUFFICIENT_CONTIGUOUS_SPACE;
         
-        // see if record would fit into remaining space after compaction
-        if(tupleSpace + slotManager.getSlotSize() <= buf.getInt(totalFreeSpaceOff)) return SpaceStatus.SUFFICIENT_SPACE;
+        // see if tuple would fit into remaining space after compaction
+        if(bytesRequired + slotManager.getSlotSize() <= buf.getInt(totalFreeSpaceOff)) return SpaceStatus.SUFFICIENT_SPACE;
         
-        // see if the record matches a prefix and will fit after truncating the prefix
-        int prefixSlotNum = slotManager.findPrefix(tuple, cmp);
-        if(prefixSlotNum != FieldPrefixSlotManager.RECORD_UNCOMPRESSED) {
+        // see if the tuple matches a prefix and will fit after truncating the prefix
+        int prefixSlotNum = slotManager.findPrefix(tuple, framePrefixTuple, cmp);
+        if(prefixSlotNum != FieldPrefixSlotManager.TUPLE_UNCOMPRESSED) {
         	int prefixSlotOff = slotManager.getPrefixSlotOff(prefixSlotNum);
         	int prefixSlot = buf.getInt(prefixSlotOff);
         	int numPrefixFields = slotManager.decodeFirstSlotField(prefixSlot);
         	
-        	// TODO relies on length being stored in serialized field
-        	int recRunner = 0;
-        	for(int i = 0; i < numPrefixFields; i++) {
-        		recRunner += cmp.getFields()[i].getLength(tuple.getFieldData(i), recRunner);
-        	}
-        	int compressedSize = tupleSpace - recRunner;
+        	int compressedSize = tupleWriter.bytesRequired(tuple, numPrefixFields, tuple.getFieldCount() - numPrefixFields);        	        	
         	if(compressedSize + slotManager.getSlotSize() <= freeContiguous) return SpaceStatus.SUFFICIENT_CONTIGUOUS_SPACE;        	
         }
         
         return SpaceStatus.INSUFFICIENT_SPACE;
     }
-    
-    // TODO: need to change these methods
-	protected int writeTupleFields(ITupleReference src, int numFields, ByteBuffer targetBuf, int targetOff) {
-		int runner = targetOff;		
-		for(int i = 0; i < numFields; i++) {
-			System.arraycopy(src.getFieldData(i), src.getFieldStart(i), targetBuf.array(), runner, src.getFieldLength(i));
-			runner += src.getFieldLength(i);
-		}
-		return runner - targetOff;
-	}
-	
-	protected int spaceNeededForTuple(ITupleReference src) {
-		int space = 0;
-		for(int i = 0; i < src.getFieldCount(); i++) {
-			space += src.getFieldLength(i);
-		}
-		return space;
-	}
-    
+        
     @Override
     public SpaceStatus hasSpaceUpdate(int rid, ITupleReference tuple, MultiComparator cmp) {
         // TODO Auto-generated method stub
@@ -273,10 +235,10 @@
     @Override
     public void initBuffer(byte level) {        
         buf.putInt(pageLsnOff, 0); // TODO: might to set to a different lsn during creation
-        buf.putInt(numRecordsOff, 0);   
+        buf.putInt(tupleCountOff, 0);   
         resetSpaceParams();
-        buf.putInt(numUncompressedRecordsOff, 0);
-        buf.putInt(numPrefixRecordsOff, 0);
+        buf.putInt(uncompressedTupleCountOff, 0);
+        buf.putInt(prefixTupleCountOff, 0);
         buf.put(levelOff, level);
         buf.put(smFlagOff, (byte)0);
         buf.putInt(prevLeafOff, -1);
@@ -293,64 +255,45 @@
     
     @Override
     public void insert(ITupleReference tuple, MultiComparator cmp) throws Exception {    	
-    	int slot = slotManager.findSlot(tuple, cmp, false);        
-        slot = slotManager.insertSlot(slot, buf.getInt(freeSpaceOff));
+    	int slot = slotManager.findSlot(tuple, frameTuple, framePrefixTuple, cmp, false);        
         
-        int suffixSize = spaceNeededForTuple(tuple);
-        int suffixStartOff = tuple.getFieldStart(0);
-        int prefixSlotNum = slotManager.decodeFirstSlotField(slot);
+    	slot = slotManager.insertSlot(slot, buf.getInt(freeSpaceOff));                
         
-        if(prefixSlotNum != FieldPrefixSlotManager.RECORD_UNCOMPRESSED) {
-        	
-        	// TODO relies on length being stored in serialized field
-        	
+        int prefixSlotNum = slotManager.decodeFirstSlotField(slot);        
+        int numPrefixFields = 0;
+        if(prefixSlotNum != FieldPrefixSlotManager.TUPLE_UNCOMPRESSED) {        	        	
             int prefixSlotOff = slotManager.getPrefixSlotOff(prefixSlotNum);            
             int prefixSlot = buf.getInt(prefixSlotOff);
-            int numPrefixFields = slotManager.decodeFirstSlotField(prefixSlot);
-                                    
-            // skip prefix fields
-            for(int i = 0; i < numPrefixFields; i++) {
-                suffixStartOff += cmp.getFields()[i].getLength(tuple.getFieldData(i), suffixStartOff);
-            }
-            
-            // compute suffix size
-            suffixSize = suffixStartOff; 
-            for(int i = numPrefixFields; i < cmp.getFields().length; i++) {
-                suffixSize += cmp.getFields()[i].getLength(tuple.getFieldData(i), suffixSize);
-            }
-            suffixSize -= suffixStartOff;                  
+            numPrefixFields = slotManager.decodeFirstSlotField(prefixSlot);                                                                 
         }
         else {
-        	buf.putInt(numUncompressedRecordsOff, buf.getInt(numUncompressedRecordsOff) + 1);
+        	buf.putInt(uncompressedTupleCountOff, buf.getInt(uncompressedTupleCountOff) + 1);
         }
         
-    	// TODO relies on length being stored in serialized field        
         int freeSpace = buf.getInt(freeSpaceOff);
-        System.arraycopy(tuple.getFieldData(0), suffixStartOff, buf.array(), freeSpace, suffixSize);                    
-        buf.putInt(numRecordsOff, buf.getInt(numRecordsOff) + 1);
-        buf.putInt(freeSpaceOff, buf.getInt(freeSpaceOff) + suffixSize);
-        buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) - suffixSize - slotManager.getSlotSize());
-               
-        //System.out.println(buf.getInt(totalFreeSpaceOff) + " " + buf.getInt(freeSpaceOff) + " " + buf.getInt(numRecordsOff));        
-        //System.out.println("COPIED: " + suffixSize + " / " + data.length);      
+        int bytesWritten = tupleWriter.writeTupleFields(tuple, numPrefixFields, tuple.getFieldCount() - numPrefixFields, buf, freeSpace);
+        
+        buf.putInt(tupleCountOff, buf.getInt(tupleCountOff) + 1);
+        buf.putInt(freeSpaceOff, buf.getInt(freeSpaceOff) + bytesWritten);
+        buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) - bytesWritten - slotManager.getSlotSize());       
     }
     
     public void verifyPrefixes(MultiComparator cmp) {
-    	int numPrefixes = buf.getInt(numPrefixRecordsOff);
+    	int numPrefixes = buf.getInt(prefixTupleCountOff);
     	int totalPrefixBytes = 0;
     	for(int i = 0; i < numPrefixes; i++) {
     		int prefixSlotOff = slotManager.getPrefixSlotOff(i);
     		int prefixSlot = buf.getInt(prefixSlotOff);
     		
     		int numPrefixFields = slotManager.decodeFirstSlotField(prefixSlot);
-    		int prefixRecOff = slotManager.decodeSecondSlotField(prefixSlot);
+    		int prefixTupleOff = slotManager.decodeSecondSlotField(prefixSlot);
     		
-    		System.out.print("VERIFYING " + i + " : " + numPrefixFields + " " + prefixRecOff + " ");
-    		int recOffRunner = prefixRecOff;
+    		System.out.print("VERIFYING " + i + " : " + numPrefixFields + " " + prefixTupleOff + " ");
+    		int tupleOffRunner = prefixTupleOff;
     		for(int j = 0; j < numPrefixFields; j++) {
-    			System.out.print(buf.getInt(prefixRecOff+j*4) + " ");
-    			int length = cmp.getFields()[j].getLength(buf.array(), recOffRunner);
-    			recOffRunner += length;
+    			System.out.print(buf.getInt(prefixTupleOff+j*4) + " ");
+    			int length = cmp.getFields()[j].getLength(buf.array(), tupleOffRunner);
+    			tupleOffRunner += length;
     			totalPrefixBytes += length;    			
     		}
     		System.out.println();
@@ -371,8 +314,8 @@
     }       
         
     @Override
-    public int getNumRecords() {
-        return buf.getInt(numRecordsOff);
+    public int getTupleCount() {
+        return buf.getInt(tupleCountOff);
     }
     
     public ISlotManager getSlotManager() {
@@ -380,28 +323,25 @@
     }
     
     @Override
-    public String printKeys(MultiComparator cmp) {      
-        StringBuilder strBuilder = new StringBuilder();
-        FieldPrefixFieldIterator rec = new FieldPrefixFieldIterator(cmp.getFields(), this);      
-        int numRecords = buf.getInt(numRecordsOff);
-        for(int i = 0; i < numRecords; i++) {                               	
-        	rec.openRecSlotNum(i);        	
-        	//strBuilder.append(String.format("RECORD %5d: ", i));
-        	for(int j = 0; j < cmp.size(); j++) {               
-                strBuilder.append(cmp.getFields()[j].print(buf.array(), rec.getFieldOff()) + " ");
-                rec.nextField();
-            }
-            strBuilder.append(" | ");        	        	                           
-        }
-        strBuilder.append("\n");
-        return strBuilder.toString();
+    public String printKeys(MultiComparator cmp, int fieldCount) {      
+    	StringBuilder strBuilder = new StringBuilder();		
+		int tupleCount = buf.getInt(tupleCountOff);
+		for(int i = 0; i < tupleCount; i++) {						
+			frameTuple.resetByTupleIndex(this, i);			
+			for(int j = 0; j < cmp.getKeyLength(); j++) {
+				strBuilder.append(cmp.getFields()[j].print(buf.array(), frameTuple.getFieldStart(j)) + " ");	
+			}
+			strBuilder.append(" | ");				
+		}
+		strBuilder.append("\n");
+		return strBuilder.toString();
     }
     
     @Override
-    public int getRecordOffset(int slotNum) {        
-    	int recSlotOff = slotManager.getRecSlotOff(slotNum);
-    	int recSlot = buf.getInt(recSlotOff);
-    	return slotManager.decodeSecondSlotField(recSlot);    	
+    public int getTupleOffset(int slotNum) {        
+    	int tupleSlotOff = slotManager.getTupleSlotOff(slotNum);
+    	int tupleSlot = buf.getInt(tupleSlotOff);
+    	return slotManager.decodeSecondSlotField(tupleSlot);    	
     }
     
     @Override
@@ -447,58 +387,40 @@
 			buf.put(smFlagOff, (byte)0);			
 	}		
 	
-	public int getNumPrefixRecords() {
-		return buf.getInt(numPrefixRecordsOff);
+	public int getPrefixTupleCount() {
+		return buf.getInt(prefixTupleCountOff);
 	}
 	
-	public void setNumPrefixRecords(int numPrefixRecords) {
-		buf.putInt(numPrefixRecordsOff, numPrefixRecords);
+	public void setPrefixTupleCount(int prefixTupleCount) {
+		buf.putInt(prefixTupleCountOff, prefixTupleCount);
 	}
-
+	
     @Override
     public void insertSorted(ITupleReference tuple, MultiComparator cmp) throws Exception {    	
     	int freeSpace = buf.getInt(freeSpaceOff);				
 		int fieldsToTruncate = 0;
 		
-		pageTuple.setFrame(this);
-		pageTuple.setFields(cmp.getFields());
-		
-		if(buf.getInt(numPrefixRecordsOff) > 0) {
-			// check if record matches last prefix record
-			int prefixSlotOff = slotManager.getPrefixSlotOff(buf.getInt(numPrefixRecordsOff)-1);
-			int prefixSlot = buf.getInt(prefixSlotOff);
-			int numPrefixFields = slotManager.decodeFirstSlotField(prefixSlot);
-			
-			pagePrefixTuple.setFrame(this);
-			pagePrefixTuple.setFields(cmp.getFields());
-			pagePrefixTuple.openPrefixSlotOff(prefixSlotOff);
-			
-			if(cmp.fieldRangeCompare(tuple, pagePrefixTuple, 0, numPrefixFields) == 0) {
-				fieldsToTruncate = numPrefixFields;													
+		// check if tuple matches last prefix tuple
+		if(buf.getInt(prefixTupleCountOff) > 0) {
+			framePrefixTuple.resetByTupleIndex(this, buf.getInt(prefixTupleCountOff)-1);						
+			if(cmp.fieldRangeCompare(tuple, framePrefixTuple, 0, framePrefixTuple.getFieldCount()) == 0) {
+				fieldsToTruncate = framePrefixTuple.getFieldCount();													
 			}			
 		}
-				
-		// copy truncated record
-    	// TODO relies on length being stored in serialized field   
-		int tupleSpace = spaceNeededForTuple(tuple); 
-		int recStart = tuple.getFieldStart(0);
-		for(int i = 0; i < fieldsToTruncate; i++) {
-			recStart += cmp.getFields()[i].getLength(tuple.getFieldData(0), recStart);
-		}
-		int recLen = tupleSpace - recStart - tuple.getFieldStart(0);
-		System.arraycopy(tuple.getFieldData(0), recStart, buf.array(), freeSpace, recLen);
 		
+        int bytesWritten = tupleWriter.writeTupleFields(tuple, fieldsToTruncate, tuple.getFieldCount() - fieldsToTruncate, buf, freeSpace);
+                
 		// insert slot
-		int prefixSlotNum = FieldPrefixSlotManager.RECORD_UNCOMPRESSED;
-		if(fieldsToTruncate > 0) prefixSlotNum = buf.getInt(numPrefixRecordsOff)-1;					
-		else buf.putInt(numUncompressedRecordsOff, buf.getInt(numUncompressedRecordsOff) + 1);					
+		int prefixSlotNum = FieldPrefixSlotManager.TUPLE_UNCOMPRESSED;
+		if(fieldsToTruncate > 0) prefixSlotNum = buf.getInt(prefixTupleCountOff)-1;					
+		else buf.putInt(uncompressedTupleCountOff, buf.getInt(uncompressedTupleCountOff) + 1);					
 		int insSlot = slotManager.encodeSlotFields(prefixSlotNum, FieldPrefixSlotManager.GREATEST_SLOT);				
 		slotManager.insertSlot(insSlot, freeSpace);
 		
 		// update page metadata
-		buf.putInt(numRecordsOff, buf.getInt(numRecordsOff) + 1);
-		buf.putInt(freeSpaceOff, buf.getInt(freeSpaceOff) + recLen);
-		buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) - recLen - slotManager.getSlotSize());			
+		buf.putInt(tupleCountOff, buf.getInt(tupleCountOff) + 1);
+		buf.putInt(freeSpaceOff, buf.getInt(freeSpaceOff) + bytesWritten);
+		buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) - bytesWritten - slotManager.getSlotSize());		
     }
     
     @Override
@@ -506,150 +428,128 @@
     	
     	FieldPrefixNSMLeafFrame rf = (FieldPrefixNSMLeafFrame)rightFrame;
     	
-    	pageTuple.setFrame(this);
-		pageTuple.setFields(cmp.getFields());
+    	frameTuple.setFieldCount(cmp.getFields().length);
     	
     	// before doing anything check if key already exists
-		int slot = slotManager.findSlot(tuple, cmp, true);
-		int recSlotNum = slotManager.decodeSecondSlotField(slot);		
-		if(recSlotNum != FieldPrefixSlotManager.GREATEST_SLOT) {				
-			pageTuple.openRecSlotNum(recSlotNum);
-			if(cmp.compare(tuple, (ITupleReference)pageTuple) == 0) {			
+		int slot = slotManager.findSlot(tuple, frameTuple, framePrefixTuple, cmp, true);
+		int tupleSlotNum = slotManager.decodeSecondSlotField(slot);		
+		if(tupleSlotNum != FieldPrefixSlotManager.GREATEST_SLOT) {				
+			frameTuple.resetByTupleIndex(this, tupleSlotNum);
+			if(cmp.compare(tuple, frameTuple) == 0) {			
 				throw new BTreeException("Inserting duplicate key into unique index");
 			}
 		}
 		
 		ByteBuffer right = rf.getBuffer();
-		int numRecords = getNumRecords();
-		int numPrefixRecords = getNumPrefixRecords();
+		int tupleCount = getTupleCount();
+		int prefixTupleCount = getPrefixTupleCount();
 		
-		int recordsToLeft;
-		int midSlotNum = numRecords / 2;
-		IBTreeFrame targetFrame = null;
-		int midSlotOff = slotManager.getRecSlotOff(midSlotNum);
-		int midSlot = buf.getInt(midSlotOff);
-		int midPrefixSlotNum = slotManager.decodeFirstSlotField(midSlot);
-		int midRecOff = slotManager.decodeSecondSlotField(midSlot);		
-		pageTuple.openRecSlotNum(midSlotNum);
-		int comparison = cmp.compare(tuple, (ITupleReference)pageTuple);		
+		int tuplesToLeft;
+		int midSlotNum = tupleCount / 2;
+		IBTreeFrame targetFrame = null;		
+		frameTuple.resetByTupleIndex(this, midSlotNum);
+		int comparison = cmp.compare(tuple, frameTuple);		
 		if (comparison >= 0) {
-			recordsToLeft = midSlotNum + (numRecords % 2);
+			tuplesToLeft = midSlotNum + (tupleCount % 2);
 			targetFrame = rf;
 		} else {
-			recordsToLeft = midSlotNum;
+			tuplesToLeft = midSlotNum;
 			targetFrame = this;
 		}
-		int recordsToRight = numRecords - recordsToLeft;
+		int tuplesToRight = tupleCount - tuplesToLeft;
 				
 		// copy entire page
 		System.arraycopy(buf.array(), 0, right.array(), 0, buf.capacity());
 				
 		// determine how many slots go on left and right page
-		int prefixesToLeft = numPrefixRecords;		
-		for(int i = recordsToLeft; i < numRecords; i++) {			
-			int recSlotOff = rf.slotManager.getRecSlotOff(i);						
-			int recSlot = right.getInt(recSlotOff);
-			int prefixSlotNum = rf.slotManager.decodeFirstSlotField(recSlot);			
-			if(prefixSlotNum != FieldPrefixSlotManager.RECORD_UNCOMPRESSED) {
+		int prefixesToLeft = prefixTupleCount;		
+		for(int i = tuplesToLeft; i < tupleCount; i++) {			
+			int tupleSlotOff = rf.slotManager.getTupleSlotOff(i);						
+			int tupleSlot = right.getInt(tupleSlotOff);
+			int prefixSlotNum = rf.slotManager.decodeFirstSlotField(tupleSlot);			
+			if(prefixSlotNum != FieldPrefixSlotManager.TUPLE_UNCOMPRESSED) {
 				prefixesToLeft = prefixSlotNum;			
 				break;
 			}
 		}
 		
-		// if we are splitting in the middle of a prefix both pages need to have the prefix slot and record
-		int bounradyRecSlotOff = rf.slotManager.getRecSlotOff(recordsToLeft-1);
-		int boundaryRecSlot = buf.getInt(bounradyRecSlotOff);
-		int boundaryPrefixSlotNum = rf.slotManager.decodeFirstSlotField(boundaryRecSlot);
-		int prefixesToRight = numPrefixRecords - prefixesToLeft;
-		if(boundaryPrefixSlotNum == prefixesToLeft && boundaryPrefixSlotNum != FieldPrefixSlotManager.RECORD_UNCOMPRESSED) {
-			prefixesToLeft++; // records on both pages share one prefix 
+		// if we are splitting in the middle of a prefix both pages need to have the prefix slot and tuple
+		int boundaryTupleSlotOff = rf.slotManager.getTupleSlotOff(tuplesToLeft-1);
+		int boundaryTupleSlot = buf.getInt(boundaryTupleSlotOff);
+		int boundaryPrefixSlotNum = rf.slotManager.decodeFirstSlotField(boundaryTupleSlot);
+		int prefixesToRight = prefixTupleCount - prefixesToLeft;
+		if(boundaryPrefixSlotNum == prefixesToLeft && boundaryPrefixSlotNum != FieldPrefixSlotManager.TUPLE_UNCOMPRESSED) {
+			prefixesToLeft++; // tuples on both pages share one prefix 
 		}			
 								
-		// move prefix records on right page to beginning of page and adjust prefix slots
-		if(prefixesToRight > 0 && prefixesToLeft > 0 && numPrefixRecords > 1) {			
+		// move prefix tuples on right page to beginning of page and adjust prefix slots
+		if(prefixesToRight > 0 && prefixesToLeft > 0 && prefixTupleCount > 1) {			
 			
 			int freeSpace = rf.getOrigFreeSpaceOff();
 			int lastPrefixSlotNum = -1;
 			
-			for(int i = recordsToLeft; i < numRecords; i++) {			
-				int recSlotOff = rf.slotManager.getRecSlotOff(i);						
-				int recSlot = right.getInt(recSlotOff);
-				int prefixSlotNum = rf.slotManager.decodeFirstSlotField(recSlot);			
-				if(prefixSlotNum != FieldPrefixSlotManager.RECORD_UNCOMPRESSED) {								
-					int prefixSlotOff = rf.slotManager.getPrefixSlotOff(prefixSlotNum);
-					int prefixSlot = right.getInt(prefixSlotOff);
-					int numPrefixFields = rf.slotManager.decodeFirstSlotField(prefixSlot);
+			for(int i = tuplesToLeft; i < tupleCount; i++) {			
+				int tupleSlotOff = rf.slotManager.getTupleSlotOff(i);						
+				int tupleSlot = right.getInt(tupleSlotOff);
+				int prefixSlotNum = rf.slotManager.decodeFirstSlotField(tupleSlot);			
+				if(prefixSlotNum != FieldPrefixSlotManager.TUPLE_UNCOMPRESSED) {
+					framePrefixTuple.resetByTupleIndex(this, prefixSlotNum);
 					
-					int prefixRecSize = 0;
+					int bytesWritten = 0;
 					if(lastPrefixSlotNum != prefixSlotNum) {
-						int prefixRecOff = rf.slotManager.decodeSecondSlotField(prefixSlot);
-						for(int j = 0; j < numPrefixFields; j++) {
-							prefixRecSize += cmp.getFields()[j].getLength(buf.array(), prefixRecOff + prefixRecSize);						
-						}
-						// copy from left page to make sure not to overwrite anything in right page that we may need later
-						System.arraycopy(buf.array(), prefixRecOff, right.array(), freeSpace, prefixRecSize);
-						
-						int newPrefixSlot = rf.slotManager.encodeSlotFields(numPrefixFields, freeSpace);
-						right.putInt(prefixSlotOff, newPrefixSlot);
-						
+						bytesWritten = tupleWriter.writeTuple(framePrefixTuple, right, freeSpace);												
+						int newPrefixSlot = rf.slotManager.encodeSlotFields(framePrefixTuple.getFieldCount(), freeSpace);
+						int prefixSlotOff = rf.slotManager.getPrefixSlotOff(prefixSlotNum);
+						right.putInt(prefixSlotOff, newPrefixSlot);						
 						lastPrefixSlotNum = prefixSlotNum;
-					}
+					}					
 					
-					int recOff = rf.slotManager.decodeSecondSlotField(recSlot);
-					int newRecSlot = rf.slotManager.encodeSlotFields(prefixSlotNum - (numPrefixRecords - prefixesToRight), recOff);
-					right.putInt(recSlotOff, newRecSlot);
-															
-					freeSpace += prefixRecSize;
+					int tupleOff = rf.slotManager.decodeSecondSlotField(tupleSlot);
+					int newTupleSlot = rf.slotManager.encodeSlotFields(prefixSlotNum - (prefixTupleCount - prefixesToRight), tupleOff);
+					right.putInt(tupleSlotOff, newTupleSlot);		
+					freeSpace += bytesWritten;
 				}
-			}
+			}			
 		}
 				
 		// move the modified prefix slots on the right page
 		int prefixSrc = rf.slotManager.getPrefixSlotEndOff();
-		int prefixDest = rf.slotManager.getPrefixSlotEndOff() + (numPrefixRecords - prefixesToRight) * rf.slotManager.getSlotSize();
+		int prefixDest = rf.slotManager.getPrefixSlotEndOff() + (prefixTupleCount - prefixesToRight) * rf.slotManager.getSlotSize();
 		int prefixLength = rf.slotManager.getSlotSize() * prefixesToRight;
 		System.arraycopy(right.array(), prefixSrc, right.array(), prefixDest, prefixLength);
 		
-		// on right page we need to copy rightmost record slots to left
-		int src = rf.slotManager.getRecSlotEndOff();
-		int dest = rf.slotManager.getRecSlotEndOff() + recordsToLeft * rf.slotManager.getSlotSize() + (numPrefixRecords - prefixesToRight) * rf.slotManager.getSlotSize();
-		int length = rf.slotManager.getSlotSize() * recordsToRight;				
+		// on right page we need to copy rightmost tuple slots to left
+		int src = rf.slotManager.getTupleSlotEndOff();
+		int dest = rf.slotManager.getTupleSlotEndOff() + tuplesToLeft * rf.slotManager.getSlotSize() + (prefixTupleCount - prefixesToRight) * rf.slotManager.getSlotSize();
+		int length = rf.slotManager.getSlotSize() * tuplesToRight;				
 		System.arraycopy(right.array(), src, right.array(), dest, length);
 		
-		right.putInt(numRecordsOff, recordsToRight);
-		right.putInt(numPrefixRecordsOff, prefixesToRight);
+		right.putInt(tupleCountOff, tuplesToRight);
+		right.putInt(prefixTupleCountOff, prefixesToRight);
 		
 		// on left page move slots to reflect possibly removed prefixes
-		src = slotManager.getRecSlotEndOff() + recordsToRight * slotManager.getSlotSize();
-		dest = slotManager.getRecSlotEndOff() + recordsToRight * slotManager.getSlotSize() + (numPrefixRecords - prefixesToLeft) * slotManager.getSlotSize();
-		length = slotManager.getSlotSize() * recordsToLeft;				
+		src = slotManager.getTupleSlotEndOff() + tuplesToRight * slotManager.getSlotSize();
+		dest = slotManager.getTupleSlotEndOff() + tuplesToRight * slotManager.getSlotSize() + (prefixTupleCount - prefixesToLeft) * slotManager.getSlotSize();
+		length = slotManager.getSlotSize() * tuplesToLeft;				
 		System.arraycopy(buf.array(), src, buf.array(), dest, length);
 		
-		buf.putInt(numRecordsOff, recordsToLeft);
-		buf.putInt(numPrefixRecordsOff, prefixesToLeft);		
+		buf.putInt(tupleCountOff, tuplesToLeft);
+		buf.putInt(prefixTupleCountOff, prefixesToLeft);		
 				
 		// compact both pages		
-		compact(cmp);
+		compact(cmp);		
 		rightFrame.compact(cmp);
 		
 		// insert last key
 		targetFrame.insert(tuple, cmp);
-		
-    	// TODO relies on length being stored in serialized field  
-		
+				
 		// set split key to be highest value in left page		
-		int splitKeyRecSlotOff = slotManager.getRecSlotEndOff();		
+		frameTuple.resetByTupleIndex(this, getTupleCount()-1);
 		
-		int keySize = 0;
-		FieldPrefixFieldIterator fieldIter = new FieldPrefixFieldIterator(cmp.getFields(), this);
-		fieldIter.openRecSlotOff(splitKeyRecSlotOff);		
-		for(int i = 0; i < cmp.getKeyLength(); i++) {
-			keySize += fieldIter.getFieldSize();
-			fieldIter.nextField();
-		}									
-		splitKey.initData(keySize);
-		fieldIter.copyFields(0, cmp.getKeyLength()-1, splitKey.getBuffer().array(), 0);
-		
+		int splitKeySize = tupleWriter.bytesRequired(frameTuple, 0, cmp.getKeyLength());
+		splitKey.initData(splitKeySize);
+		tupleWriter.writeTupleFields(frameTuple, 0, cmp.getKeyLength(), splitKey.getBuffer(), 0);
+				
 		return 0;
     }
     
@@ -687,12 +587,12 @@
 		return buf.getInt(prevLeafOff);
 	}
 	
-	public int getNumUncompressedRecords() {
-		return buf.getInt(numUncompressedRecordsOff);
+	public int getUncompressedTupleCount() {
+		return buf.getInt(uncompressedTupleCountOff);
 	}
 	
-	public void setNumUncompressedRecords(int numUncompressedRecords) {
-		buf.putInt(numUncompressedRecordsOff, numUncompressedRecords);
+	public void setUncompressedTupleCount(int uncompressedTupleCount) {
+		buf.putInt(uncompressedTupleCountOff, uncompressedTupleCount);
 	}
 	
 	@Override
@@ -701,12 +601,12 @@
 	}
 	
 	@Override
-    public IFieldIterator createFieldIterator() {
-    	return new FieldPrefixFieldIterator();
+    public IBTreeTupleReference createTupleReference() {
+    	return new FieldPrefixTupleReference();
     }
 		
 	@Override
-	public void setPageTupleFields(IFieldAccessor[] fields) {
-		pageTuple.setFields(fields);
-	}		
+	public void setPageTupleFieldCount(int fieldCount) {
+		frameTuple.setFieldCount(fieldCount);
+	}	
 }
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/MetaDataFrame.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/MetaDataFrame.java
index 6a479b0..9c8fa3f 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/MetaDataFrame.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/MetaDataFrame.java
@@ -27,8 +27,8 @@
 
 public class MetaDataFrame implements IBTreeMetaDataFrame {
         
-    protected static final int numRecordsOff = 0;             
-    protected static final int freeSpaceOff = numRecordsOff + 4;
+    protected static final int tupleCountOff = 0;             
+    protected static final int freeSpaceOff = tupleCountOff + 4;
     protected static final int maxPageOff = freeSpaceOff + 4;
     protected static final byte levelOff = maxPageOff + 1;
     protected static final byte nextPageOff = maxPageOff + 8;
@@ -45,13 +45,13 @@
     }
         
     public int getFreePage() {                
-        int numRecords = buf.getInt(numRecordsOff); 
-        if(numRecords > 0) {
+        int tupleCount = buf.getInt(tupleCountOff); 
+        if(tupleCount > 0) {
             // return the last page from the linked list of free pages
             // TODO: this is a dumb policy, but good enough for now
             int lastPageOff = buf.getInt(freeSpaceOff) - 4;
             buf.putInt(freeSpaceOff, lastPageOff);
-            buf.putInt(numRecordsOff, numRecords - 1);
+            buf.putInt(tupleCountOff, tupleCount - 1);
             return buf.getInt(lastPageOff);
         }
         else {
@@ -71,7 +71,7 @@
         int freeSpace = buf.getInt(freeSpaceOff);
         buf.putInt(freeSpace, freePage);
         buf.putInt(freeSpaceOff, freeSpace + 4);
-        buf.putInt(numRecordsOff, buf.getInt(numRecordsOff) + 1);
+        buf.putInt(tupleCountOff, buf.getInt(tupleCountOff) + 1);
     }
 
     @Override
@@ -98,7 +98,7 @@
     @Override
     public void initBuffer(int level) {
         buf.putInt(freeSpaceOff, nextPageOff + 4);
-        buf.putInt(numRecordsOff, 0);
+        buf.putInt(tupleCountOff, 0);
         buf.putInt(levelOff, level);
         buf.putInt(nextPageOff, -1);
     }
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMFrame.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMFrame.java
index 5256766..c6a27b3 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMFrame.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMFrame.java
@@ -21,23 +21,22 @@
 
 import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeFrame;
-import edu.uci.ics.hyracks.storage.am.btree.api.IFieldAccessor;
-import edu.uci.ics.hyracks.storage.am.btree.api.IFieldIterator;
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeTupleReference;
 import edu.uci.ics.hyracks.storage.am.btree.api.ISlotManager;
 import edu.uci.ics.hyracks.storage.am.btree.impls.BTreeException;
 import edu.uci.ics.hyracks.storage.am.btree.impls.MultiComparator;
-import edu.uci.ics.hyracks.storage.am.btree.impls.NSMFieldIterator;
 import edu.uci.ics.hyracks.storage.am.btree.impls.OrderedSlotManager;
-import edu.uci.ics.hyracks.storage.am.btree.impls.SelfDescTupleReference;
-import edu.uci.ics.hyracks.storage.am.btree.impls.SlotOffRecOff;
+import edu.uci.ics.hyracks.storage.am.btree.impls.SimpleTupleReference;
+import edu.uci.ics.hyracks.storage.am.btree.impls.SimpleTupleWriter;
+import edu.uci.ics.hyracks.storage.am.btree.impls.SlotOffTupleOff;
 import edu.uci.ics.hyracks.storage.am.btree.impls.SpaceStatus;
 import edu.uci.ics.hyracks.storage.common.buffercache.ICachedPage;
 
 public abstract class NSMFrame implements IBTreeFrame {
 	
 	protected static final int pageLsnOff = 0;								// 0
-	protected static final int numRecordsOff = pageLsnOff + 4;				// 4
-	protected static final int freeSpaceOff = numRecordsOff + 4;			// 8
+	protected static final int tupleCountOff = pageLsnOff + 4;				// 4
+	protected static final int freeSpaceOff = tupleCountOff + 4;			// 8
 	protected static final int totalFreeSpaceOff = freeSpaceOff + 4;		// 16
 	protected static final byte levelOff = totalFreeSpaceOff + 4; 	
 	protected static final byte smFlagOff = levelOff + 1;	
@@ -46,7 +45,8 @@
 	protected ByteBuffer buf = null;
 	protected ISlotManager slotManager;
 	
-	protected SelfDescTupleReference pageTuple = new SelfDescTupleReference();
+	protected SimpleTupleWriter tupleWriter = new SimpleTupleWriter();
+	protected SimpleTupleReference frameTuple = new SimpleTupleReference();
 	
 	public NSMFrame() {
 		this.slotManager = new OrderedSlotManager();
@@ -55,7 +55,7 @@
 	@Override
 	public void initBuffer(byte level) {
 		buf.putInt(pageLsnOff, 0); // TODO: might to set to a different lsn during creation
-		buf.putInt(numRecordsOff, 0);	
+		buf.putInt(tupleCountOff, 0);	
 		resetSpaceParams();
 		buf.put(levelOff, level);
 		buf.put(smFlagOff, (byte)0);
@@ -116,54 +116,59 @@
 	
 	@Override
 	public void compact(MultiComparator cmp) {		
-		resetSpaceParams();		
+		resetSpaceParams();
+		frameTuple.setFieldCount(cmp.getFields().length);	
 		
-		int numRecords = buf.getInt(numRecordsOff);
-		int freeSpace = buf.getInt(freeSpaceOff);
-		byte[] data = buf.array();
+		int tupleCount = buf.getInt(tupleCountOff);
+		int freeSpace = buf.getInt(freeSpaceOff);		
 		
-		ArrayList<SlotOffRecOff> sortedRecOffs = new ArrayList<SlotOffRecOff>();
-		sortedRecOffs.ensureCapacity(numRecords);
-		for(int i = 0; i < numRecords; i++) {			
+		ArrayList<SlotOffTupleOff> sortedTupleOffs = new ArrayList<SlotOffTupleOff>();
+		sortedTupleOffs.ensureCapacity(tupleCount);
+		for(int i = 0; i < tupleCount; i++) {			
 			int slotOff = slotManager.getSlotOff(i);
-			int recOff = slotManager.getRecOff(slotOff);
-			sortedRecOffs.add(new SlotOffRecOff(slotOff, recOff));
+			int tupleOff = slotManager.getTupleOff(slotOff);
+			sortedTupleOffs.add(new SlotOffTupleOff(i, slotOff, tupleOff));
 		}
-		Collections.sort(sortedRecOffs);
+		Collections.sort(sortedTupleOffs);
 		
-		for(int i = 0; i < sortedRecOffs.size(); i++) {
-			int recOff = sortedRecOffs.get(i).recOff;
-			int recSize = cmp.getRecordSize(data, recOff);
-			System.arraycopy(data, recOff, data, freeSpace, recSize);
-			slotManager.setSlot(sortedRecOffs.get(i).slotOff, freeSpace);
-			freeSpace += recSize;
+		for(int i = 0; i < sortedTupleOffs.size(); i++) {
+			int tupleOff = sortedTupleOffs.get(i).tupleOff;
+			frameTuple.resetByOffset(buf, tupleOff);
+			
+            int tupleEndOff = frameTuple.getFieldStart(frameTuple.getFieldCount()-1) + frameTuple.getFieldLength(frameTuple.getFieldCount()-1);
+            int tupleLength = tupleEndOff - tupleOff;
+            System.arraycopy(buf.array(), tupleOff, buf.array(), freeSpace, tupleLength);
+						
+			slotManager.setSlot(sortedTupleOffs.get(i).slotOff, freeSpace);
+			freeSpace += tupleLength;
 		}
 		
 		buf.putInt(freeSpaceOff, freeSpace);
-		buf.putInt(totalFreeSpaceOff, buf.capacity() - freeSpace - numRecords * slotManager.getSlotSize());				
+		buf.putInt(totalFreeSpaceOff, buf.capacity() - freeSpace - tupleCount * slotManager.getSlotSize());				
 	}
 
 	@Override
 	public void delete(ITupleReference tuple, MultiComparator cmp, boolean exactDelete) throws Exception {		
-		int slotOff = slotManager.findSlot(tuple, cmp, true);
+		frameTuple.setFieldCount(cmp.getFields().length);
+		int slotOff = slotManager.findSlot(tuple, frameTuple, cmp, true);
 		if(slotOff < 0) {
 			throw new BTreeException("Key to be deleted does not exist.");
 		}
 		else {
 			if(exactDelete) {
 				// check the non-key columns for equality by byte-by-byte comparison
-				int storedRecOff = slotManager.getRecOff(slotOff);			
-				pageTuple.setFields(cmp.getFields());
-				pageTuple.reset(buf, storedRecOff);
+				int tupleOff = slotManager.getTupleOff(slotOff);
+				frameTuple.resetByOffset(buf, tupleOff);
 				
-				int comparison = cmp.fieldRangeCompare(tuple, pageTuple, cmp.getKeyLength()-1, cmp.getFields().length - cmp.getKeyLength());
+				int comparison = cmp.fieldRangeCompare(tuple, frameTuple, cmp.getKeyLength()-1, cmp.getFields().length - cmp.getKeyLength());
                 if(comparison != 0) {
-                	throw new BTreeException("Cannot delete record. Byte-by-byte comparison failed to prove equality.");
+                	throw new BTreeException("Cannot delete tuple. Byte-by-byte comparison failed to prove equality.");
                 }										
 			}
 			
-			int recOff = slotManager.getRecOff(slotOff);
-			int recSize = cmp.getRecordSize(buf.array(), recOff);
+			int tupleOff = slotManager.getTupleOff(slotOff);
+			frameTuple.resetByOffset(buf, tupleOff);
+			int tupleSize = tupleWriter.bytesRequired(frameTuple);
 			
 			// perform deletion (we just do a memcpy to overwrite the slot)
 			int slotStartOff = slotManager.getSlotEndOff();
@@ -171,16 +176,16 @@
 			System.arraycopy(buf.array(), slotStartOff, buf.array(), slotStartOff + slotManager.getSlotSize(), length);
 						
 			// maintain space information
-			buf.putInt(numRecordsOff, buf.getInt(numRecordsOff) - 1);
-			buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) + recSize + slotManager.getSlotSize());			
+			buf.putInt(tupleCountOff, buf.getInt(tupleCountOff) - 1);
+			buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) + tupleSize + slotManager.getSlotSize());			
 		}
 	}
 	
 	@Override
 	public SpaceStatus hasSpaceInsert(ITupleReference tuple, MultiComparator cmp) {				
-		int tupleSpace = spaceNeededForTuple(tuple);		
-		if(tupleSpace + slotManager.getSlotSize() <= buf.capacity() - buf.getInt(freeSpaceOff) - (buf.getInt(numRecordsOff) * slotManager.getSlotSize()) ) return SpaceStatus.SUFFICIENT_CONTIGUOUS_SPACE;
-		else if(tupleSpace + slotManager.getSlotSize() <= buf.getInt(totalFreeSpaceOff)) return SpaceStatus.SUFFICIENT_SPACE;
+		int bytesRequired = tupleWriter.bytesRequired(tuple);		
+		if(bytesRequired + slotManager.getSlotSize() <= buf.capacity() - buf.getInt(freeSpaceOff) - (buf.getInt(tupleCountOff) * slotManager.getSlotSize()) ) return SpaceStatus.SUFFICIENT_CONTIGUOUS_SPACE;
+		else if(bytesRequired + slotManager.getSlotSize() <= buf.getInt(totalFreeSpaceOff)) return SpaceStatus.SUFFICIENT_SPACE;
 		else return SpaceStatus.INSUFFICIENT_SPACE;
 	}
 	
@@ -197,33 +202,16 @@
 	
 	@Override
 	public void insert(ITupleReference tuple, MultiComparator cmp) throws Exception {
-		int slotOff = slotManager.findSlot(tuple, cmp, false);		
+		frameTuple.setFieldCount(cmp.getFields().length);
+		int slotOff = slotManager.findSlot(tuple, frameTuple, cmp, false);		
 		slotOff = slotManager.insertSlot(slotOff, buf.getInt(freeSpaceOff));				
-		int bytesWritten = writeTupleFields(tuple, tuple.getFieldCount(), buf, buf.getInt(freeSpaceOff));
+		int bytesWritten = tupleWriter.writeTuple(tuple, buf, buf.getInt(freeSpaceOff));
 		
-		buf.putInt(numRecordsOff, buf.getInt(numRecordsOff) + 1);
+		buf.putInt(tupleCountOff, buf.getInt(tupleCountOff) + 1);
 		buf.putInt(freeSpaceOff, buf.getInt(freeSpaceOff) + bytesWritten);
 		buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) - bytesWritten - slotManager.getSlotSize());				
 	}
-	
-	// TODO: need to change these methods
-	protected int writeTupleFields(ITupleReference src, int numFields, ByteBuffer targetBuf, int targetOff) {
-		int runner = targetOff;		
-		for(int i = 0; i < numFields; i++) {
-			System.arraycopy(src.getFieldData(i), src.getFieldStart(i), targetBuf.array(), runner, src.getFieldLength(i));
-			runner += src.getFieldLength(i);
-		}
-		return runner - targetOff;
-	}
-	
-	protected int spaceNeededForTuple(ITupleReference src) {
-		int space = 0;
-		for(int i = 0; i < src.getFieldCount(); i++) {
-			space += src.getFieldLength(i);
-		}
-		return space;
-	}
-	
+		
 	@Override
 	public void update(int rid, ITupleReference tuple) throws Exception {
 		// TODO Auto-generated method stub
@@ -237,8 +225,8 @@
 	}		
 		
 	@Override
-	public int getNumRecords() {
-		return buf.getInt(numRecordsOff);
+	public int getTupleCount() {
+		return buf.getInt(tupleCountOff);
 	}
 
 	public ISlotManager getSlotManager() {
@@ -246,14 +234,15 @@
 	}
 	
 	@Override
-	public String printKeys(MultiComparator cmp) {		
+	public String printKeys(MultiComparator cmp, int fieldCount) {		
 		StringBuilder strBuilder = new StringBuilder();		
-		int numRecords = buf.getInt(numRecordsOff);
-		for(int i = 0; i < numRecords; i++) {			
-			int recOff = slotManager.getRecOff(slotManager.getSlotOff(i));
-			for(int j = 0; j < cmp.size(); j++) {				
-				strBuilder.append(cmp.getFields()[j].print(buf.array(), recOff) + " ");
-				recOff += cmp.getFields()[j].getLength(buf.array(), recOff);
+		frameTuple.setFieldCount(fieldCount);
+		int tupleCount = buf.getInt(tupleCountOff);
+		for(int i = 0; i < tupleCount; i++) {			
+			int tupleOff = slotManager.getTupleOff(slotManager.getSlotOff(i));			
+			frameTuple.resetByOffset(buf, tupleOff);			
+			for(int j = 0; j < cmp.getKeyLength(); j++) {
+				strBuilder.append(cmp.getFields()[j].print(buf.array(), frameTuple.getFieldStart(j)) + " ");				
 			}
 			strBuilder.append(" | ");				
 		}
@@ -262,8 +251,8 @@
 	}
 	
 	@Override
-	public int getRecordOffset(int slotNum) {
-		return slotManager.getRecOff(slotManager.getSlotStartOff() - slotNum * slotManager.getSlotSize());
+	public int getTupleOffset(int slotNum) {
+		return slotManager.getTupleOff(slotManager.getSlotStartOff() - slotNum * slotManager.getSlotSize());
 	}
 
 	@Override
@@ -292,12 +281,12 @@
     }
     
     @Override
-    public IFieldIterator createFieldIterator() {
-    	return new NSMFieldIterator();
+    public IBTreeTupleReference createTupleReference() {
+    	return new SimpleTupleReference();
     }
     
     @Override
-	public void setPageTupleFields(IFieldAccessor[] fields) {
-		pageTuple.setFields(fields);
+	public void setPageTupleFieldCount(int fieldCount) {
+		frameTuple.setFieldCount(fieldCount);
 	}
 }
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMInteriorFrame.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMInteriorFrame.java
index 51dece6..b6bcaee 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMInteriorFrame.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMInteriorFrame.java
@@ -25,7 +25,9 @@
 import edu.uci.ics.hyracks.storage.am.btree.impls.BTreeException;
 import edu.uci.ics.hyracks.storage.am.btree.impls.MultiComparator;
 import edu.uci.ics.hyracks.storage.am.btree.impls.RangePredicate;
-import edu.uci.ics.hyracks.storage.am.btree.impls.SlotOffRecOff;
+import edu.uci.ics.hyracks.storage.am.btree.impls.SimpleTupleReference;
+import edu.uci.ics.hyracks.storage.am.btree.impls.SlotOffTupleOff;
+import edu.uci.ics.hyracks.storage.am.btree.impls.SpaceStatus;
 import edu.uci.ics.hyracks.storage.am.btree.impls.SplitKey;
 
 public class NSMInteriorFrame extends NSMFrame implements IBTreeInteriorFrame {
@@ -34,15 +36,14 @@
 	
 	private static final int childPtrSize = 4;
 	
+	private SimpleTupleReference cmpFrameTuple = new SimpleTupleReference();
+	
 	public NSMInteriorFrame() {
 		super();
 	}
 	
-	private int getLeftChildPageOff(int recOff, MultiComparator cmp) {
-		for(int i = 0; i < cmp.size(); i++) {
-			recOff += cmp.getFields()[i].getLength(buf.array(), recOff);
-		}		
-		return recOff;
+	private int getLeftChildPageOff(ITupleReference tuple, MultiComparator cmp) {		
+		return tuple.getFieldStart(cmp.getKeyLength()-1) + tuple.getFieldLength(cmp.getKeyLength()-1);		
 	}	
 	
 	@Override
@@ -52,162 +53,177 @@
 	}
 	
 	@Override
+	public SpaceStatus hasSpaceInsert(ITupleReference tuple, MultiComparator cmp) {				
+		int bytesRequired = tupleWriter.bytesRequired(tuple) + 8; // for the two childpointers
+		if(bytesRequired + slotManager.getSlotSize() <= buf.capacity() - buf.getInt(freeSpaceOff) - (buf.getInt(tupleCountOff) * slotManager.getSlotSize()) ) return SpaceStatus.SUFFICIENT_CONTIGUOUS_SPACE;
+		else if(bytesRequired + slotManager.getSlotSize() <= buf.getInt(totalFreeSpaceOff)) return SpaceStatus.SUFFICIENT_SPACE;
+		else return SpaceStatus.INSUFFICIENT_SPACE;
+	}
+	
+	@Override
 	public void insert(ITupleReference tuple, MultiComparator cmp) throws Exception {
-		
-		int slotOff = slotManager.findSlot(tuple, cmp, false);
+		frameTuple.setFieldCount(cmp.getKeyLength());
+		int slotOff = slotManager.findSlot(tuple, frameTuple, cmp, false);
 		boolean isDuplicate = true;				
 		
 		if(slotOff < 0) isDuplicate = false; // greater than all existing keys
-		else {			
-			pageTuple.setFields(cmp.getFields());
-			pageTuple.reset(buf, slotManager.getRecOff(slotOff));
-			if(cmp.compare(tuple, pageTuple) != 0) isDuplicate = false;
+		else {
+			frameTuple.resetByOffset(buf, slotManager.getTupleOff(slotOff));
+			if(cmp.compare(tuple, frameTuple) != 0) isDuplicate = false;
 		}
-		
+				
 		if(isDuplicate) {			
 			throw new BTreeException("Trying to insert duplicate value into interior node.");
 		}
 		else {			
 			slotOff = slotManager.insertSlot(slotOff, buf.getInt(freeSpaceOff));			
-						
-			// TODO: will need to be modified for different record format
-			int freeSpace = buf.getInt(freeSpaceOff);
-			int recSize = writeTupleFields(tuple, tuple.getFieldCount()-1, buf, freeSpace);			
 			
-			buf.putInt(numRecordsOff, buf.getInt(numRecordsOff) + 1);
-			buf.putInt(freeSpaceOff, buf.getInt(freeSpaceOff) + recSize);
-			buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) - recSize - slotManager.getSlotSize());
+			int freeSpace = buf.getInt(freeSpaceOff);			
+			int bytesWritten = tupleWriter.writeTupleFields(tuple, 0, cmp.getKeyLength(), buf, freeSpace);
+			System.arraycopy(tuple.getFieldData(cmp.getKeyLength()-1), getLeftChildPageOff(tuple, cmp), buf.array(), freeSpace + bytesWritten, childPtrSize);
+			int tupleSize = bytesWritten + childPtrSize;	
+			
+			buf.putInt(tupleCountOff, buf.getInt(tupleCountOff) + 1);
+			buf.putInt(freeSpaceOff, buf.getInt(freeSpaceOff) + tupleSize);
+			buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) - tupleSize - slotManager.getSlotSize());
 			
 			// did insert into the rightmost slot?
 			if(slotOff == slotManager.getSlotEndOff()) { 
-				System.arraycopy(tuple.getFieldData(tuple.getFieldCount()-1), tuple.getFieldStart(tuple.getFieldCount()-1), buf.array(), rightLeafOff, childPtrSize);
+				System.arraycopy(tuple.getFieldData(cmp.getKeyLength()-1), getLeftChildPageOff(tuple, cmp) + childPtrSize, buf.array(), rightLeafOff, childPtrSize);
 			}
 			else {
 				// if slotOff has a right (slot-)neighbor then update its child pointer
-				// the only time when this is NOT the case, is when this is the first record 
-				// (or when the splitkey goes into the rightmost slot but that case was handled in the if above)
-				
-				if(buf.getInt(numRecordsOff) > 1) {					
+				// the only time when this is NOT the case, is when this is the first tuple 
+				// (or when the splitkey goes into the rightmost slot but that case was handled in the if above)				
+				if(buf.getInt(tupleCountOff) > 1) {					
 					int rightNeighborOff = slotOff - slotManager.getSlotSize();					
-					pageTuple.reset(buf, slotManager.getRecOff(rightNeighborOff));
-					System.arraycopy(tuple.getFieldData(0), tuple.getFieldStart(cmp.getKeyLength()+1), buf.array(), pageTuple.getFieldStart(cmp.getKeyLength()), childPtrSize);
+					frameTuple.resetByOffset(buf, slotManager.getTupleOff(rightNeighborOff));
+					System.arraycopy(tuple.getFieldData(0), getLeftChildPageOff(tuple, cmp) + childPtrSize, buf.array(), getLeftChildPageOff(frameTuple, cmp), childPtrSize);
 				}				
 			}			
-		}
-		
-		//System.out.println("INSSPACEA: " + (buf.capacity() - buf.getInt(freeSpaceOff) - (buf.getInt(numRecordsOff) * slotManager.getSlotSize())));
-		//System.out.println("INSSPACEB: " + (buf.getInt(totalFreeSpaceOff)));
+		}		
 	}
 	
 	@Override
 	public void insertSorted(ITupleReference tuple, MultiComparator cmp) throws Exception {
 		int freeSpace = buf.getInt(freeSpaceOff);
 		slotManager.insertSlot(-1, freeSpace);
-		int recSize = writeTupleFields(tuple, tuple.getFieldCount()-1, buf, freeSpace);				
-		buf.putInt(numRecordsOff, buf.getInt(numRecordsOff) + 1);
-		buf.putInt(freeSpaceOff, buf.getInt(freeSpaceOff) + recSize);
-		buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) - recSize - slotManager.getSlotSize());
-		System.arraycopy(tuple.getFieldData(tuple.getFieldCount()-1), tuple.getFieldStart(tuple.getFieldCount()-1), buf.array(), rightLeafOff, childPtrSize);
+		int bytesWritten = tupleWriter.writeTupleFields(tuple, 0, cmp.getKeyLength(), buf, freeSpace);
+		System.arraycopy(tuple.getFieldData(cmp.getKeyLength()-1), getLeftChildPageOff(tuple, cmp), buf.array(), freeSpace + bytesWritten, childPtrSize);
+		int tupleSize = bytesWritten + childPtrSize;
+		buf.putInt(tupleCountOff, buf.getInt(tupleCountOff) + 1);
+		buf.putInt(freeSpaceOff, buf.getInt(freeSpaceOff) + tupleSize);
+		buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) - tupleSize - slotManager.getSlotSize());
+		System.arraycopy(tuple.getFieldData(0), getLeftChildPageOff(tuple, cmp) + childPtrSize, buf.array(), rightLeafOff, childPtrSize);
 	}
 		
 	@Override
 	public int split(IBTreeFrame rightFrame, ITupleReference tuple, MultiComparator cmp, SplitKey splitKey) throws Exception {		
 		// before doing anything check if key already exists
-		int slotOff = slotManager.findSlot(tuple, cmp, true);
+		frameTuple.setFieldCount(cmp.getKeyLength());
+		int slotOff = slotManager.findSlot(tuple, frameTuple, cmp, true);
 		if(slotOff >= 0) {			
-			pageTuple.reset(buf, slotManager.getRecOff(slotOff));			
-			if(cmp.compare(tuple, pageTuple) == 0) {				
+			frameTuple.resetByOffset(buf, slotManager.getTupleOff(slotOff));			
+			if(cmp.compare(tuple, frameTuple) == 0) {				
 				throw new BTreeException("Inserting duplicate key in interior node during split");				
 			}
 		}
 		
 		ByteBuffer right = rightFrame.getBuffer();
-		int numRecords = buf.getInt(numRecordsOff);
+		int tupleCount = buf.getInt(tupleCountOff);
 		
-		int recordsToLeft = (numRecords / 2) + (numRecords % 2);
+		int tuplesToLeft = (tupleCount / 2) + (tupleCount % 2);
 		IBTreeFrame targetFrame = null;
-		pageTuple.reset(buf, getRecordOffset(recordsToLeft-1));
-		if(cmp.compare(tuple, pageTuple) <= 0) {
+		frameTuple.resetByOffset(buf, getTupleOffset(tuplesToLeft-1));
+		if(cmp.compare(tuple, frameTuple) <= 0) {
 			targetFrame = this;
 		}
 		else {
 			targetFrame = rightFrame;
 		}			
-		int recordsToRight = numRecords - recordsToLeft;		
+		int tuplesToRight = tupleCount - tuplesToLeft;		
 				
 		// copy entire page
 		System.arraycopy(buf.array(), 0, right.array(), 0, buf.capacity());
 		
 		// on right page we need to copy rightmost slots to left		
 		int src = rightFrame.getSlotManager().getSlotEndOff();
-		int dest = rightFrame.getSlotManager().getSlotEndOff() + recordsToLeft * rightFrame.getSlotManager().getSlotSize();
-		int length = rightFrame.getSlotManager().getSlotSize() * recordsToRight;
+		int dest = rightFrame.getSlotManager().getSlotEndOff() + tuplesToLeft * rightFrame.getSlotManager().getSlotSize();
+		int length = rightFrame.getSlotManager().getSlotSize() * tuplesToRight;
 		System.arraycopy(right.array(), src, right.array(), dest, length);				
-		right.putInt(numRecordsOff, recordsToRight);
+		right.putInt(tupleCountOff, tuplesToRight);
 				
 		// on left page, remove highest key and make its childpointer the rightmost childpointer
-		buf.putInt(numRecordsOff, recordsToLeft);
+		buf.putInt(tupleCountOff, tuplesToLeft);
 				
 		// copy data to be inserted, we need this because creating the splitkey will overwrite the data param (data points to same memory as splitKey.getData())
 		SplitKey savedSplitKey = splitKey.duplicate();		
 		
 		// set split key to be highest value in left page	
-		int recOff = slotManager.getRecOff(slotManager.getSlotEndOff());
-		int splitKeySize = cmp.getKeySize(buf.array(), recOff);
+		int tupleOff = slotManager.getTupleOff(slotManager.getSlotEndOff());
+		frameTuple.resetByOffset(buf, tupleOff);
+		int splitKeySize = tupleWriter.bytesRequired(frameTuple, 0, cmp.getKeyLength());
 		splitKey.initData(splitKeySize);
-		pageTuple.reset(buf, recOff);
-		writeTupleFields((ITupleReference)pageTuple, cmp.getKeyLength(), splitKey.getBuffer(), 0);		
+		tupleWriter.writeTupleFields(frameTuple, 0, cmp.getKeyLength(), splitKey.getBuffer(), 0);			
 		
-		int deleteRecOff = slotManager.getRecOff(slotManager.getSlotEndOff());
-		int deleteKeySize = cmp.getKeySize(buf.array(), deleteRecOff); 		
-		buf.putInt(rightLeafOff, buf.getInt(deleteRecOff + deleteKeySize));
-		buf.putInt(numRecordsOff, recordsToLeft - 1);
-				
+		int deleteTupleOff = slotManager.getTupleOff(slotManager.getSlotEndOff());
+		frameTuple.resetByOffset(buf, deleteTupleOff);
+		buf.putInt(rightLeafOff, buf.getInt(getLeftChildPageOff(frameTuple, cmp)));
+		buf.putInt(tupleCountOff, tuplesToLeft - 1);
+		
 		// compact both pages
 		rightFrame.compact(cmp);
 		compact(cmp);
 		
 		// insert key
-		targetFrame.insert(savedSplitKey.getTuple(), cmp);
-			
+		targetFrame.insert(savedSplitKey.getTuple(), cmp);					
+		
 		return 0;
 	}	
 	
 	@Override
 	public void compact(MultiComparator cmp) {		
-		resetSpaceParams();		
+		resetSpaceParams();
 		
-		int numRecords = buf.getInt(numRecordsOff);
+		frameTuple.setFieldCount(cmp.getKeyLength());
+		
+		int tupleCount = buf.getInt(tupleCountOff);
 		int freeSpace = buf.getInt(freeSpaceOff);
-		byte[] data = buf.array();
 		
-		ArrayList<SlotOffRecOff> sortedRecOffs = new ArrayList<SlotOffRecOff>();
-		sortedRecOffs.ensureCapacity(numRecords);
-		for(int i = 0; i < numRecords; i++) {			
+		ArrayList<SlotOffTupleOff> sortedTupleOffs = new ArrayList<SlotOffTupleOff>();
+		sortedTupleOffs.ensureCapacity(tupleCount);
+		for(int i = 0; i < tupleCount; i++) {			
 			int slotOff = slotManager.getSlotOff(i);
-			int recOff = slotManager.getRecOff(slotOff);
-			sortedRecOffs.add(new SlotOffRecOff(slotOff, recOff));
+			int tupleOff = slotManager.getTupleOff(slotOff);
+			sortedTupleOffs.add(new SlotOffTupleOff(i, slotOff, tupleOff));
 		}
-		Collections.sort(sortedRecOffs);
+		Collections.sort(sortedTupleOffs);
 		
-		for(int i = 0; i < sortedRecOffs.size(); i++) {
-			int recOff = sortedRecOffs.get(i).recOff;
-			int recSize = cmp.getKeySize(data, recOff) + childPtrSize; // only difference from compact in NSMFrame
-			System.arraycopy(data, recOff, data, freeSpace, recSize);
-			slotManager.setSlot(sortedRecOffs.get(i).slotOff, freeSpace);		
-			freeSpace += recSize;
+		for(int i = 0; i < sortedTupleOffs.size(); i++) {
+			int tupleOff = sortedTupleOffs.get(i).tupleOff;
+			frameTuple.resetByOffset(buf, tupleOff);
+			
+			int tupleEndOff = frameTuple.getFieldStart(frameTuple.getFieldCount()-1) + frameTuple.getFieldLength(frameTuple.getFieldCount()-1);
+			int tupleLength = tupleEndOff - tupleOff + childPtrSize;
+			System.arraycopy(buf.array(), tupleOff, buf.array(), freeSpace, tupleLength);
+						
+			slotManager.setSlot(sortedTupleOffs.get(i).slotOff, freeSpace);
+			freeSpace += tupleLength;						
 		}
 		
 		buf.putInt(freeSpaceOff, freeSpace);
-		buf.putInt(totalFreeSpaceOff, buf.capacity() - freeSpace - numRecords * slotManager.getSlotSize());
+		buf.putInt(totalFreeSpaceOff, buf.capacity() - freeSpace - tupleCount * slotManager.getSlotSize());
 	}
 		
 	@Override
 	public int getChildPageId(RangePredicate pred, MultiComparator srcCmp) {				
 		// check for trivial case where there is only a child pointer (and no key)
-		if(buf.getInt(numRecordsOff) == 0) {			
+		if(buf.getInt(tupleCountOff) == 0) {			
 			return buf.getInt(rightLeafOff);
 		}
+				
+		cmpFrameTuple.setFieldCount(srcCmp.getKeyLength());		
+		frameTuple.setFieldCount(srcCmp.getKeyLength());
 		
 		// check for trivial cases where no low key or high key exists (e.g. during an index scan)
 		ITupleReference tuple = null;
@@ -224,22 +240,23 @@
 			}								
 		}
 		
-		// TODO: need to modify this procedure to use TupleReference
-		MultiComparator targetCmp = pred.getComparator();
-		int slotOff = slotManager.findSlot(tuple, targetCmp, false);
+		MultiComparator targetCmp = pred.getComparator();		
+		int slotOff = slotManager.findSlot(tuple, frameTuple, targetCmp, false);
 		if(slotOff < 0) {
 			return buf.getInt(rightLeafOff);
 		}
 		else {
-			int origRecOff = slotManager.getRecOff(slotOff);
-			int cmpRecOff = origRecOff;
+			int origTupleOff = slotManager.getTupleOff(slotOff);
+			cmpFrameTuple.resetByOffset(buf, origTupleOff);
+			int cmpTupleOff = origTupleOff;
 			if(pred.isForward()) {
 				int maxSlotOff = buf.capacity();
 				slotOff += slotManager.getSlotSize();		
 				while(slotOff < maxSlotOff) {
-					cmpRecOff = slotManager.getRecOff(slotOff);
-					if(targetCmp.compare(buf.array(), origRecOff, buf.array(), cmpRecOff) != 0) break;					
-					slotOff += slotManager.getSlotSize();			
+					cmpTupleOff = slotManager.getTupleOff(slotOff);
+					frameTuple.resetByOffset(buf, cmpTupleOff);
+					if(targetCmp.compare(cmpFrameTuple, frameTuple) != 0) break;					
+					slotOff += slotManager.getSlotSize();	
 				}
 				slotOff -= slotManager.getSlotSize();
 			}
@@ -247,33 +264,39 @@
 				int minSlotOff = slotManager.getSlotEndOff() - slotManager.getSlotSize();
 				slotOff -= slotManager.getSlotSize();
 				while(slotOff > minSlotOff) {
-					cmpRecOff = slotManager.getRecOff(slotOff);
-					if(targetCmp.compare(buf.array(), origRecOff, buf.array(), cmpRecOff) != 0) break;
+					cmpTupleOff = slotManager.getTupleOff(slotOff);
+					frameTuple.resetByOffset(buf, cmpTupleOff);
+					if(targetCmp.compare(cmpFrameTuple, frameTuple) != 0) break;
 					slotOff -= slotManager.getSlotSize();										
 				}
 				slotOff += slotManager.getSlotSize();
 			}
 			
-			int childPageOff = getLeftChildPageOff(slotManager.getRecOff(slotOff), srcCmp);			
+			frameTuple.resetByOffset(buf, slotManager.getTupleOff(slotOff));
+			int childPageOff = getLeftChildPageOff(frameTuple, srcCmp);			
 			return buf.getInt(childPageOff);
 		}				
 	}	
 	
 	@Override
 	public void delete(ITupleReference tuple, MultiComparator cmp, boolean exactDelete) throws Exception {
-		int slotOff = slotManager.findSlot(tuple, cmp, false);
-		int recOff;
+		frameTuple.setFieldCount(cmp.getKeyLength());
+		int slotOff = slotManager.findSlot(tuple, frameTuple, cmp, false);
+		int tupleOff;
 		int keySize;
 		
 		if(slotOff < 0) {						
-			recOff = slotManager.getRecOff(slotManager.getSlotEndOff());
-			keySize = cmp.getKeySize(buf.array(), recOff);
+			tupleOff = slotManager.getTupleOff(slotManager.getSlotEndOff());
+			frameTuple.resetByOffset(buf, tupleOff);			
+			keySize = tupleWriter.bytesRequired(frameTuple, 0, cmp.getKeyLength());
+			
 			// copy new rightmost pointer
-			System.arraycopy(buf.array(), recOff + keySize, buf.array(), rightLeafOff, childPtrSize);						
+			System.arraycopy(buf.array(), tupleOff + keySize, buf.array(), rightLeafOff, childPtrSize);						
 		}
 		else {						
-			recOff = slotManager.getRecOff(slotOff);
-			keySize = cmp.getKeySize(buf.array(), recOff);	
+			tupleOff = slotManager.getTupleOff(slotOff);
+			frameTuple.resetByOffset(buf, tupleOff);
+			keySize = tupleWriter.bytesRequired(frameTuple, 0, cmp.getKeyLength());	
 			// perform deletion (we just do a memcpy to overwrite the slot)
 			int slotStartOff = slotManager.getSlotEndOff();
 			int length = slotOff - slotStartOff;
@@ -281,7 +304,7 @@
 		}
 		
 		// maintain space information
-		buf.putInt(numRecordsOff, buf.getInt(numRecordsOff) - 1);
+		buf.putInt(tupleCountOff, buf.getInt(tupleCountOff) - 1);
 		buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) + keySize + childPtrSize + slotManager.getSlotSize());
 	}
 	
@@ -293,8 +316,10 @@
 		
 	@Override
 	public int getLeftmostChildPageId(MultiComparator cmp) {						
-		int recOff = slotManager.getRecOff(slotManager.getSlotStartOff());
-		int childPageOff = getLeftChildPageOff(recOff, cmp);		
+		int tupleOff = slotManager.getTupleOff(slotManager.getSlotStartOff());		
+		frameTuple.setFieldCount(cmp.getKeyLength());
+		frameTuple.resetByOffset(buf, tupleOff);
+		int childPageOff = getLeftChildPageOff(frameTuple, cmp);
 		return buf.getInt(childPageOff);
 	}
 
@@ -310,16 +335,19 @@
 	
 	// for debugging
 	public ArrayList<Integer> getChildren(MultiComparator cmp) {		
+		System.out.println(page);
+		
 		ArrayList<Integer> ret = new ArrayList<Integer>();		
-		int numRecords = buf.getInt(numRecordsOff);
-		for(int i = 0; i < numRecords; i++) {
-			int recOff = slotManager.getRecOff(slotManager.getSlotOff(i));
-			int keySize = cmp.getKeySize(buf.array(), recOff);			
-			int intVal = getInt(buf.array(), recOff + keySize);			
+		frameTuple.setFieldCount(cmp.getKeyLength());
+		int tupleCount = buf.getInt(tupleCountOff);
+		for(int i = 0; i < tupleCount; i++) {
+			int tupleOff = slotManager.getTupleOff(slotManager.getSlotOff(i));
+			frameTuple.resetByOffset(buf, tupleOff);
+			int intVal = getInt(buf.array(), frameTuple.getFieldStart(frameTuple.getFieldCount()-1) + frameTuple.getFieldLength(frameTuple.getFieldCount()-1));
 			ret.add(intVal);
 		}
-		if(!isLeaf()) {
-			int rightLeaf = buf.getInt(rightLeafOff);
+		if(!isLeaf()) {			
+			int rightLeaf = buf.getInt(rightLeafOff);			
 			if(rightLeaf > 0) ret.add(buf.getInt(rightLeafOff));
 		}
 		return ret;
@@ -328,16 +356,18 @@
 	@Override
 	public void deleteGreatest(MultiComparator cmp) {
 		int slotOff = slotManager.getSlotEndOff();
-		int recOff = slotManager.getRecOff(slotOff);
-		int keySize = cmp.getKeySize(buf.array(), recOff); 
-		System.arraycopy(buf.array(), recOff + keySize, buf.array(), rightLeafOff, childPtrSize);
-				
+		int tupleOff = slotManager.getTupleOff(slotOff);
+		frameTuple.setFieldCount(cmp.getKeyLength());
+		frameTuple.resetByOffset(buf, tupleOff);
+		int keySize = tupleWriter.bytesRequired(frameTuple, 0, cmp.getKeyLength());		 
+		System.arraycopy(buf.array(), tupleOff + keySize, buf.array(), rightLeafOff, childPtrSize);
+		
 		// maintain space information
-		buf.putInt(numRecordsOff, buf.getInt(numRecordsOff) - 1);
+		buf.putInt(tupleCountOff, buf.getInt(tupleCountOff) - 1);
 		buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) + keySize + childPtrSize + slotManager.getSlotSize());
 		
 		int freeSpace = buf.getInt(freeSpaceOff);
-		if(freeSpace == recOff + keySize + childPtrSize) {
+		if(freeSpace == tupleOff + keySize + childPtrSize) {
 			buf.putInt(freeSpace, freeSpace - (keySize + childPtrSize));
 		}		
 	}		
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMLeafFrame.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMLeafFrame.java
index f548704..8f88bc6 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMLeafFrame.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMLeafFrame.java
@@ -61,14 +61,14 @@
 
 	@Override
 	public void insert(ITupleReference tuple, MultiComparator cmp) throws Exception {		
-		int slotOff = slotManager.findSlot(tuple, cmp, false);
+		frameTuple.setFieldCount(cmp.getFields().length);
+		int slotOff = slotManager.findSlot(tuple, frameTuple, cmp, false);
 		boolean isDuplicate = true;
 				
 		if (slotOff < 0) isDuplicate = false; // greater than all existing keys
 		else {
-			pageTuple.setFields(cmp.getFields());
-			pageTuple.reset(buf, slotManager.getRecOff(slotOff));					
-			if (cmp.compare(tuple, pageTuple) != 0) isDuplicate = false;
+			frameTuple.resetByOffset(buf, slotManager.getTupleOff(slotOff));				
+			if (cmp.compare(tuple, frameTuple) != 0) isDuplicate = false;
 		}
 				
 		if (isDuplicate) {
@@ -77,10 +77,10 @@
 		else {
 			slotOff = slotManager.insertSlot(slotOff, buf.getInt(freeSpaceOff));
 			
-			int freeSpace = buf.getInt(freeSpaceOff);
-			int bytesWritten = writeTupleFields(tuple, tuple.getFieldCount(), buf, freeSpace);			
+			int freeSpace = buf.getInt(freeSpaceOff);			
+			int bytesWritten = tupleWriter.writeTuple(tuple, buf, freeSpace);			
 			
-			buf.putInt(numRecordsOff, buf.getInt(numRecordsOff) + 1);
+			buf.putInt(tupleCountOff, buf.getInt(tupleCountOff) + 1);
 			buf.putInt(freeSpaceOff, buf.getInt(freeSpaceOff) + bytesWritten);
 			buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) - bytesWritten - slotManager.getSlotSize());
 		}	
@@ -89,70 +89,71 @@
 	@Override
 	public void insertSorted(ITupleReference tuple, MultiComparator cmp) throws Exception {		
 		int freeSpace = buf.getInt(freeSpaceOff);
-		slotManager.insertSlot(-1, freeSpace);
-		int bytesWritten = writeTupleFields(tuple, tuple.getFieldCount(), buf, freeSpace);		
-		buf.putInt(numRecordsOff, buf.getInt(numRecordsOff) + 1);
+		slotManager.insertSlot(-1, freeSpace);		
+		int bytesWritten = tupleWriter.writeTuple(tuple, buf, freeSpace);			
+		buf.putInt(tupleCountOff, buf.getInt(tupleCountOff) + 1);
 		buf.putInt(freeSpaceOff, buf.getInt(freeSpaceOff) + bytesWritten);
 		buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) - bytesWritten - slotManager.getSlotSize());
 	}
 	
 	@Override
 	public int split(IBTreeFrame rightFrame, ITupleReference tuple, MultiComparator cmp, SplitKey splitKey) throws Exception {
-		
-		pageTuple.setFields(cmp.getFields());
+				
+		frameTuple.setFieldCount(cmp.getFields().length);
 		
 		// before doing anything check if key already exists
-		int slotOff = slotManager.findSlot(tuple, cmp, true);
+		int slotOff = slotManager.findSlot(tuple, frameTuple, cmp, true);
 		if (slotOff >= 0) {						
-			pageTuple.reset(buf, slotManager.getRecOff(slotOff));		
-			if (cmp.compare(tuple, pageTuple) == 0) {
+			frameTuple.resetByOffset(buf, slotManager.getTupleOff(slotOff));		
+			if (cmp.compare(tuple, frameTuple) == 0) {
 				throw new BTreeException("Inserting duplicate key into unique index");
 			}
 		}
 		
 		ByteBuffer right = rightFrame.getBuffer();
-		int numRecords = getNumRecords();
+		int tupleCount = getTupleCount();
 
-		int recordsToLeft;
-		int mid = numRecords / 2;
+		int tuplesToLeft;
+		int mid = tupleCount / 2;
 		IBTreeFrame targetFrame = null;
-		int recOff = slotManager.getRecOff(slotManager.getSlotEndOff() + slotManager.getSlotSize() * mid);
-		pageTuple.reset(buf, recOff);		
-		if (cmp.compare(tuple, pageTuple) >= 0) {
-			recordsToLeft = mid + (numRecords % 2);
+		int tupleOff = slotManager.getTupleOff(slotManager.getSlotEndOff() + slotManager.getSlotSize() * mid);
+		frameTuple.resetByOffset(buf, tupleOff);	
+		if (cmp.compare(tuple, frameTuple) >= 0) {
+			tuplesToLeft = mid + (tupleCount % 2);
 			targetFrame = rightFrame;
 		} else {
-			recordsToLeft = mid;
+			tuplesToLeft = mid;
 			targetFrame = this;
 		}
-		int recordsToRight = numRecords - recordsToLeft;
+		int tuplesToRight = tupleCount - tuplesToLeft;
 				
 		// copy entire page
 		System.arraycopy(buf.array(), 0, right.array(), 0, buf.capacity());
 		
 		// on right page we need to copy rightmost slots to left
 		int src = rightFrame.getSlotManager().getSlotEndOff();
-		int dest = rightFrame.getSlotManager().getSlotEndOff() + recordsToLeft * rightFrame.getSlotManager().getSlotSize();
-		int length = rightFrame.getSlotManager().getSlotSize() * recordsToRight; 
+		int dest = rightFrame.getSlotManager().getSlotEndOff() + tuplesToLeft * rightFrame.getSlotManager().getSlotSize();
+		int length = rightFrame.getSlotManager().getSlotSize() * tuplesToRight; 
 		System.arraycopy(right.array(), src, right.array(), dest, length);
-		right.putInt(numRecordsOff, recordsToRight);
+		right.putInt(tupleCountOff, tuplesToRight);
 		
-		// on left page only change the numRecords indicator
-		buf.putInt(numRecordsOff, recordsToLeft);
+		// on left page only change the tupleCount indicator
+		buf.putInt(tupleCountOff, tuplesToLeft);				
 			
 		// compact both pages
-		rightFrame.compact(cmp);
+		rightFrame.compact(cmp);		
 		compact(cmp);
 		
 		// insert last key
 		targetFrame.insert(tuple, cmp);	
 		
 		// set split key to be highest value in left page
-		recOff = slotManager.getRecOff(slotManager.getSlotEndOff());
-		int keySize = cmp.getKeySize(buf.array(), recOff);		
-		splitKey.initData(keySize);
-		pageTuple.reset(buf, recOff);
-		writeTupleFields((ITupleReference)pageTuple, cmp.getKeyLength(), splitKey.getBuffer(), 0);
+		tupleOff = slotManager.getTupleOff(slotManager.getSlotEndOff());
+		frameTuple.resetByOffset(buf, tupleOff);
+		
+		int splitKeySize = tupleWriter.bytesRequired(frameTuple, 0, cmp.getKeyLength());
+		splitKey.initData(splitKeySize);				
+		tupleWriter.writeTupleFields(frameTuple, 0, cmp.getKeyLength(), splitKey.getBuffer(), 0);
 		
 		return 0;
 	}
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/BTree.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/BTree.java
index 6386734..58e044c 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/BTree.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/BTree.java
@@ -15,7 +15,6 @@
 
 package edu.uci.ics.hyracks.storage.am.btree.impls;
 
-import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Stack;
 import java.util.concurrent.locks.ReadWriteLock;
@@ -32,7 +31,6 @@
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeMetaDataFrame;
 import edu.uci.ics.hyracks.storage.am.btree.api.IFieldAccessor;
 import edu.uci.ics.hyracks.storage.am.btree.frames.NSMInteriorFrame;
-import edu.uci.ics.hyracks.storage.am.btree.types.Int32Accessor;
 import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
 import edu.uci.ics.hyracks.storage.common.buffercache.ICachedPage;
 import edu.uci.ics.hyracks.storage.common.file.FileInfo;
@@ -102,12 +100,10 @@
         this.treeLatch = new ReentrantReadWriteLock(true);
         this.diskOrderScanPredicate = new RangePredicate(true, null, null, cmp);
         
-        IFieldAccessor[] interiorFields = new IFieldAccessor[cmp.getKeyLength() + 2];
+        IFieldAccessor[] interiorFields = new IFieldAccessor[cmp.getKeyLength()];
         for(int i = 0; i < cmp.getKeyLength(); i++) {
         	interiorFields[i] = cmp.getFields()[i];
-        }
-        interiorFields[interiorFields.length-1] = new Int32Accessor();
-        interiorFields[interiorFields.length-2] = new Int32Accessor();        
+        }               
         this.interiorCmp = new MultiComparator(cmp.getComparators(), interiorFields);        
     }
     
@@ -373,16 +369,23 @@
             String keyString;
             if(interiorFrame.isLeaf()) {
             	leafFrame.setPage(node);
-            	keyString = leafFrame.printKeys(cmp);
+            	keyString = leafFrame.printKeys(cmp, cmp.getFields().length);
             }
             else {
-            	keyString = interiorFrame.printKeys(cmp);
+            	keyString = interiorFrame.printKeys(interiorCmp, cmp.getKeyLength());
             }
 
             System.out.format(keyString);
-            
             if (!interiorFrame.isLeaf()) {
-                ArrayList<Integer> children = ((NSMInteriorFrame) (interiorFrame)).getChildren(cmp);
+            	ArrayList<Integer> children = ((NSMInteriorFrame) (interiorFrame)).getChildren(cmp);
+            	
+            	// debug
+            	System.out.println("CHILDREN OF PAGE: " + interiorFrame.isLeaf() + " " + pageId);
+            	for (int i = 0; i < children.size(); i++) {
+                    System.out.print(children.get(i) + " ");
+                }
+            	System.out.println();
+            	            	
                 for (int i = 0; i < children.size(); i++) {
                     printTree(children.get(i), node, i == children.size() - 1, leafFrame, interiorFrame);
                 }
@@ -560,7 +563,7 @@
         ctx.metaFrame = metaFrame;
         ctx.pred = new RangePredicate(true, tuple, tuple, cmp);
         ctx.splitKey = new SplitKey();
-        ctx.splitKey.getTuple().setFields(interiorCmp.getFields());
+        ctx.splitKey.getTuple().setFieldCount(cmp.getKeyLength());
         ctx.smPages = new ArrayList<Integer>();
         ctx.pageLsns = new Stack<Integer>();
 
@@ -592,7 +595,7 @@
     
     private void insertLeaf(ICachedPage node, int pageId, ITupleReference tuple, BTreeOpContext ctx) throws Exception {
     	ctx.leafFrame.setPage(node);
-    	ctx.leafFrame.setPageTupleFields(cmp.getFields());
+    	ctx.leafFrame.setPageTupleFieldCount(cmp.getFields().length);
     	SpaceStatus spaceStatus = ctx.leafFrame.hasSpaceInsert(tuple, cmp);
     	switch (spaceStatus) {
     	
@@ -652,7 +655,8 @@
     					IBTreeLeafFrame rightFrame = leafFrameFactory.getFrame();
     					rightFrame.setPage(rightNode);
     					rightFrame.initBuffer((byte) 0);
-
+    					rightFrame.setPageTupleFieldCount(cmp.getFields().length);
+    					
     					int ret = ctx.leafFrame.split(rightFrame, tuple, cmp, ctx.splitKey);
 
     					ctx.smPages.add(pageId);
@@ -722,11 +726,10 @@
 
     private void insertInterior(ICachedPage node, int pageId, ITupleReference tuple, BTreeOpContext ctx) throws Exception {
         ctx.interiorFrame.setPage(node);
-        ctx.interiorFrame.setPageTupleFields(interiorCmp.getFields());
+        ctx.interiorFrame.setPageTupleFieldCount(cmp.getKeyLength());
         SpaceStatus spaceStatus = ctx.interiorFrame.hasSpaceInsert(tuple, cmp);
         switch (spaceStatus) {
             case INSUFFICIENT_SPACE: {
-                // System.out.println("SPLIT INTERIOR! " + pageId);
                 splitsByLevel[ctx.interiorFrame.getLevel()]++; // debug
                 int rightPageId = getFreePage(ctx.metaFrame);
                 ICachedPage rightNode = bufferCache.pin(FileInfo.getDiskPageId(fileId, rightPageId), true);
@@ -737,6 +740,7 @@
                     IBTreeFrame rightFrame = interiorFrameFactory.getFrame();
                     rightFrame.setPage(rightNode);
                     rightFrame.initBuffer((byte) ctx.interiorFrame.getLevel());
+                    rightFrame.setPageTupleFieldCount(cmp.getKeyLength());
                     // instead of creating a new split key, use the existing
                     // splitKey
                     int ret = ctx.interiorFrame.split(rightFrame, ctx.splitKey.getTuple(), cmp, ctx.splitKey);
@@ -796,7 +800,7 @@
         ctx.metaFrame = metaFrame;
         ctx.pred = new RangePredicate(true, tuple, tuple, cmp);
         ctx.splitKey = new SplitKey();
-        ctx.splitKey.getTuple().setFields(interiorCmp.getFields());
+        ctx.splitKey.getTuple().setFieldCount(cmp.getKeyLength());
         ctx.smPages = new ArrayList<Integer>();
         ctx.pageLsns = new Stack<Integer>();
         ctx.freePages = new ArrayList<Integer>();
@@ -843,11 +847,11 @@
     // TODO: to avoid latch deadlock, must modify cursor to detect empty leaves
     private void deleteLeaf(ICachedPage node, int pageId, ITupleReference tuple, BTreeOpContext ctx) throws Exception {
         ctx.leafFrame.setPage(node);
-
+        
         // will this leaf become empty?
-        if (ctx.leafFrame.getNumRecords() == 1) {
+        if (ctx.leafFrame.getTupleCount() == 1) {
             IBTreeLeafFrame siblingFrame = leafFrameFactory.getFrame();
-
+            
             ICachedPage leftNode = null;
             ICachedPage rightNode = null;
             int nextLeaf = ctx.leafFrame.getNextLeaf();
@@ -866,7 +870,7 @@
                     treeLatchesAcquired++;
 
                     try {
-                        ctx.leafFrame.delete(tuple, cmp, true);
+                        ctx.leafFrame.delete(tuple, cmp, true);                        
                         // to propagate the deletion we only need to make the
                         // splitKey != null
                         // we can reuse data to identify which key to delete in
@@ -954,7 +958,7 @@
 
         // this means there is only a child pointer but no key, this case
         // propagates the split
-        if (ctx.interiorFrame.getNumRecords() == 0) {
+        if (ctx.interiorFrame.getTupleCount() == 0) {
             ctx.interiorFrame.setPageLsn(ctx.interiorFrame.getPageLsn() + 1); // TODO:
             // tie
             // together
@@ -1047,7 +1051,7 @@
                     // the descent
                     boolean repeatOp = true;
                     while (repeatOp && ctx.opRestarts < MAX_RESTARTS) {
-                        int childPageId = ctx.interiorFrame.getChildPageId(ctx.pred, cmp);
+                        int childPageId = ctx.interiorFrame.getChildPageId(ctx.pred, cmp);                                                
                         performOp(childPageId, node, ctx);
 
                         if (!ctx.pageLsns.isEmpty() && ctx.pageLsns.peek().equals(RESTART_OP)) {
@@ -1130,7 +1134,7 @@
                     releaseLatch(node, ctx.op, isLeaf);
                     bufferCache.unpin(node);
                     unpins++;
-
+                    
                     // TODO: this should be an instant duration lock, how to do
                     // this in java?
                     // instead we just immediately release the lock. this is
@@ -1209,6 +1213,8 @@
         IBTreeInteriorFrame interiorFrame;
         IBTreeMetaDataFrame metaFrame;
 
+        SimpleTupleWriter tupleWriter = new SimpleTupleWriter();
+        
         public BulkLoadContext(float fillFactor, IBTreeLeafFrame leafFrame, IBTreeInteriorFrame interiorFrame,
                 IBTreeMetaDataFrame metaFrame) throws Exception {
             NodeFrontier leafFrontier = new NodeFrontier();
@@ -1238,7 +1244,7 @@
             frontier.pageId = getFreePage(metaFrame);
             frontier.page = bufferCache.pin(FileInfo.getDiskPageId(fileId, frontier.pageId), bulkNewPage);
             frontier.page.acquireWriteLatch();
-            frontier.lastTuple.setFields(interiorCmp.getFields());
+            frontier.lastTuple.setFieldCount(cmp.getKeyLength());
             interiorFrame.setPage(frontier.page);
             interiorFrame.initBuffer((byte) nodeFrontiers.size());
             nodeFrontiers.add(frontier);
@@ -1255,28 +1261,28 @@
 
         NodeFrontier frontier = ctx.nodeFrontiers.get(level);
         ctx.interiorFrame.setPage(frontier.page);
-        ITupleReference tuple = ctx.splitKey.getTuple();
-        int keySize = 0; // TODO: somehow incorporate this into interface
-        for(int i = 0; i < cmp.getKeyLength(); i++) {
-        	keySize += tuple.getFieldLength(i);
-        }
-        int spaceNeeded = keySize + ctx.slotSize + 4; // TODO: this is a dirty constant for the size of a child pointer
+        
+        ITupleReference tuple = ctx.splitKey.getTuple();        
+        int spaceNeeded = ctx.tupleWriter.bytesRequired(tuple, 0, cmp.getKeyLength()) + ctx.slotSize + 4;                
         int spaceUsed = ctx.interiorFrame.getBuffer().capacity() - ctx.interiorFrame.getTotalFreeSpace();
         if (spaceUsed + spaceNeeded > ctx.interiorMaxBytes) {
             //ctx.interiorFrame.deleteGreatest(cmp);
-            
-            frontier.lastTuple.reset(frontier.page.getBuffer(), ctx.interiorFrame.getRecordOffset(ctx.interiorFrame.getNumRecords()-1));        	
-        	int splitKeySize = cmp.getKeySize(frontier.page.getBuffer().array(), frontier.lastTuple.getFieldStart(0));
+        	
+        	SplitKey copyKey = ctx.splitKey.duplicate();
+        	tuple = copyKey.getTuple();
+        	
+            frontier.lastTuple.resetByOffset(frontier.page.getBuffer(), ctx.interiorFrame.getTupleOffset(ctx.interiorFrame.getTupleCount()-1));            
+            int splitKeySize = ctx.tupleWriter.bytesRequired(frontier.lastTuple, 0, cmp.getKeyLength());
             ctx.splitKey.initData(splitKeySize);
-            writeTupleFields((ITupleReference)frontier.lastTuple, cmp.getKeyLength(), ctx.splitKey.getBuffer(), 0);                       
+            ctx.tupleWriter.writeTupleFields(frontier.lastTuple, 0, cmp.getKeyLength(), ctx.splitKey.getBuffer(), 0);                  
             ctx.splitKey.setLeftPage(frontier.pageId);
-
+            
             ctx.interiorFrame.deleteGreatest(cmp);
             
             frontier.page.releaseWriteLatch();
             bufferCache.unpin(frontier.page);
             frontier.pageId = getFreePage(ctx.metaFrame);
-
+            
             ctx.splitKey.setRightPage(frontier.pageId);
             propagateBulk(ctx, level + 1);
 
@@ -1291,16 +1297,16 @@
     // assumes btree has been created and opened
     public BulkLoadContext beginBulkLoad(float fillFactor, IBTreeLeafFrame leafFrame, IBTreeInteriorFrame interiorFrame, IBTreeMetaDataFrame metaFrame) throws Exception {
         BulkLoadContext ctx = new BulkLoadContext(fillFactor, leafFrame, interiorFrame, metaFrame);
-        ctx.nodeFrontiers.get(0).lastTuple.setFields(cmp.getFields());
-        ctx.splitKey.tuple.setFields(interiorCmp.getFields());
+        ctx.nodeFrontiers.get(0).lastTuple.setFieldCount(cmp.getFields().length);
+        ctx.splitKey.getTuple().setFieldCount(cmp.getKeyLength());
         return ctx;
     }
     
-    public void bulkLoadAddRecord(BulkLoadContext ctx, ITupleReference tuple) throws Exception {                        
+    public void bulkLoadAddTuple(BulkLoadContext ctx, ITupleReference tuple) throws Exception {                        
         NodeFrontier leafFrontier = ctx.nodeFrontiers.get(0);
         IBTreeLeafFrame leafFrame = ctx.leafFrame;
         
-        int spaceNeeded = spaceNeededForTuple(tuple) + ctx.slotSize;
+        int spaceNeeded =  ctx.tupleWriter.bytesRequired(tuple) + ctx.slotSize;
         int spaceUsed = leafFrame.getBuffer().capacity() - leafFrame.getTotalFreeSpace();
         
         // try to free space by compression
@@ -1310,20 +1316,21 @@
         }
         
         if (spaceUsed + spaceNeeded > ctx.leafMaxBytes) {        	
-        	leafFrontier.lastTuple.reset(leafFrontier.page.getBuffer(), leafFrame.getRecordOffset(leafFrame.getNumRecords()-1));        	
-        	int splitKeySize = cmp.getKeySize(leafFrontier.page.getBuffer().array(), leafFrontier.lastTuple.getFieldStart(0));
+        	leafFrontier.lastTuple.resetByOffset(leafFrontier.page.getBuffer(), leafFrame.getTupleOffset(leafFrame.getTupleCount()-1));        	
+        	int splitKeySize = ctx.tupleWriter.bytesRequired(leafFrontier.lastTuple, 0, cmp.getKeyLength());        	
             ctx.splitKey.initData(splitKeySize);
-            writeTupleFields((ITupleReference)leafFrontier.lastTuple, cmp.getKeyLength(), ctx.splitKey.getBuffer(), 0);            
+            ctx.tupleWriter.writeTupleFields(leafFrontier.lastTuple, 0, cmp.getKeyLength(), ctx.splitKey.getBuffer(), 0);
             ctx.splitKey.setLeftPage(leafFrontier.pageId);
             int prevPageId = leafFrontier.pageId;
             leafFrontier.pageId = getFreePage(ctx.metaFrame);
+                        
             leafFrame.setNextLeaf(leafFrontier.pageId);
             leafFrontier.page.releaseWriteLatch();
             bufferCache.unpin(leafFrontier.page);
-
+                       
             ctx.splitKey.setRightPage(leafFrontier.pageId);
             propagateBulk(ctx, 1);
-
+                       
             leafFrontier.page = bufferCache.pin(FileInfo.getDiskPageId(fileId, leafFrontier.pageId), bulkNewPage);
             leafFrontier.page.acquireWriteLatch();
             leafFrame.setPage(leafFrontier.page);
@@ -1355,25 +1362,7 @@
         // debug
         currentLevel = (byte) ctx.nodeFrontiers.size();
     }
-    
-    // TODO: need to change these methods
-	protected int writeTupleFields(ITupleReference src, int numFields, ByteBuffer targetBuf, int targetOff) {
-		int runner = targetOff;		
-		for(int i = 0; i < numFields; i++) {
-			System.arraycopy(src.getFieldData(i), src.getFieldStart(i), targetBuf.array(), runner, src.getFieldLength(i));
-			runner += src.getFieldLength(i);
-		}
-		return runner - targetOff;
-	}
-    
-	protected int spaceNeededForTuple(ITupleReference src) {
-		int space = 0;
-		for(int i = 0; i < src.getFieldCount(); i++) {
-			space += src.getFieldLength(i);
-		}
-		return space;
-	}
-	
+        
     public IBTreeInteriorFrameFactory getInteriorFrameFactory() {
         return interiorFrameFactory;
     }
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/DiskOrderScanCursor.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/DiskOrderScanCursor.java
index 443cf2a..a9d8560 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/DiskOrderScanCursor.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/DiskOrderScanCursor.java
@@ -17,7 +17,7 @@
 
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeCursor;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeLeafFrame;
-import edu.uci.ics.hyracks.storage.am.btree.api.IFieldIterator;
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeTupleReference;
 import edu.uci.ics.hyracks.storage.am.btree.api.ISearchPredicate;
 import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
 import edu.uci.ics.hyracks.storage.common.buffercache.ICachedPage;
@@ -25,9 +25,9 @@
 
 public class DiskOrderScanCursor implements IBTreeCursor {
     
-    // TODO: might want to return records in physical order, not logical order to speed up access
+    // TODO: might want to return tuples in physical order, not logical order to speed up access
     
-    private int recordNum = 0;
+    private int tupleIndex = 0;
     private int fileId = -1;    
     int currentPageId = -1;
     int maxPageId = -1; // TODO: figure out how to scan to the end of file, this is dirty and may not with concurrent updates
@@ -35,12 +35,11 @@
     private IBTreeLeafFrame frame = null;
     private IBufferCache bufferCache = null;
     
-    private IFieldIterator fieldIter;
+    private IBTreeTupleReference frameTuple;
     
     public DiskOrderScanCursor(IBTreeLeafFrame frame) {
-        this.frame = frame;
-        this.fieldIter = frame.createFieldIterator();
-		this.fieldIter.setFrame(frame);
+    	this.frame = frame;
+		this.frameTuple = frame.createTupleReference();
     }
     
     @Override
@@ -51,9 +50,8 @@
     }
     
     @Override
-    public IFieldIterator getFieldIterator() {
-		fieldIter.reset();
-		return fieldIter;
+    public IBTreeTupleReference getTuple() {		
+		return frameTuple;
 	}
     
     @Override
@@ -62,7 +60,7 @@
     }
     
     private boolean positionToNextLeaf(boolean skipCurrent) throws Exception {
-        while( (frame.getLevel() != 0 || skipCurrent) && currentPageId <= maxPageId) {            
+        while( (frame.getLevel() != 0 || skipCurrent) && (currentPageId <= maxPageId) || (frame.getTupleCount() == 0) ) {            
             currentPageId++;
             
             ICachedPage nextPage = bufferCache.pin(FileInfo.getDiskPageId(fileId, currentPageId), false);
@@ -73,7 +71,7 @@
             
             page = nextPage;
             frame.setPage(page);
-            recordNum = 0;
+            tupleIndex = 0;
             skipCurrent = false;
         }   
         if(currentPageId <= maxPageId) return true;
@@ -81,11 +79,11 @@
     }
     
     @Override
-    public boolean hasNext() throws Exception {        
-        if(recordNum >= frame.getNumRecords()) {
+    public boolean hasNext() throws Exception {            	
+    	if(tupleIndex >= frame.getTupleCount()) {
             boolean nextLeafExists = positionToNextLeaf(true);
             if(nextLeafExists) {
-            	fieldIter.openRecSlotNum(recordNum);
+            	frameTuple.resetByTupleIndex(frame, tupleIndex);
                 return true;
             }
             else {
@@ -93,13 +91,13 @@
             }
         }        
         
-        fieldIter.openRecSlotNum(recordNum);        
+        frameTuple.resetByTupleIndex(frame, tupleIndex); 
         return true;
     }
 
     @Override
-    public void next() throws Exception {        
-        recordNum++;
+    public void next() throws Exception {       
+    	tupleIndex++;
     }
     
     @Override
@@ -111,11 +109,11 @@
         }
         
         this.page = page;
-        recordNum = 0;
+        tupleIndex = 0;
         frame.setPage(page);
         RangePredicate pred = (RangePredicate)searchPred;
 		MultiComparator cmp = pred.getComparator();
-		this.fieldIter.setFields(cmp.getFields());
+		frameTuple.setFieldCount(cmp.getFields().length);
         boolean leafExists = positionToNextLeaf(false);
         if(!leafExists) {
             throw new Exception("Failed to open disk-order scan cursor for B-tree. Traget B-tree has no leaves.");
@@ -124,7 +122,7 @@
     
     @Override
     public void reset() {
-        recordNum = 0;
+        tupleIndex = 0;
         currentPageId = -1;
         maxPageId = -1;
         page = null;     
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixFieldIterator.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixFieldIterator.java
deleted file mode 100644
index cba3929..0000000
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixFieldIterator.java
+++ /dev/null
@@ -1,136 +0,0 @@
-/*
- * Copyright 2009-2010 by The Regents of the University of California
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * you may obtain a copy of the License from
- * 
- *     http://www.apache.org/licenses/LICENSE-2.0
- * 
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package edu.uci.ics.hyracks.storage.am.btree.impls;
-
-import java.nio.ByteBuffer;
-
-import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeFrame;
-import edu.uci.ics.hyracks.storage.am.btree.api.IFieldAccessor;
-import edu.uci.ics.hyracks.storage.am.btree.api.IFieldIterator;
-import edu.uci.ics.hyracks.storage.am.btree.frames.FieldPrefixNSMLeafFrame;
-
-// TODO: make members private, only for debugging now
-public class FieldPrefixFieldIterator implements IFieldIterator {
-	
-	public int recSlotOff = -1;
-    public int recOff = -1;
-    public int prefixSlotNum = FieldPrefixSlotManager.RECORD_UNCOMPRESSED;
-    public int numPrefixFields = 0;
-    public IFieldAccessor[] fields;
-    public FieldPrefixNSMLeafFrame frame;
-    
-    public int currentField = -1;
-    public int recOffRunner = -1;
-    
-    public FieldPrefixFieldIterator() {    	
-    }
-    
-    public FieldPrefixFieldIterator(IFieldAccessor[] fields, FieldPrefixNSMLeafFrame frame) {
-        this.fields = fields;
-        this.frame = frame;
-    }
-    
-    public void setFrame(IBTreeFrame frame) {
-        this.frame = (FieldPrefixNSMLeafFrame)frame;        
-    }
-    
-    public void setFields(IFieldAccessor[] fields) {
-        this.fields = fields;
-    }
-    
-    public void openRecSlotNum(int recSlotNum) {          
-        openRecSlotOff(frame.slotManager.getRecSlotOff(recSlotNum));         
-    }
-    
-    public void openRecSlotOff(int recSlotOff) {
-        this.recSlotOff = recSlotOff;
-        reset();
-    }
-       
-    // re-position to first field
-    public void reset() {
-        currentField = 0;
-        numPrefixFields = 0;
-        int recSlot = frame.getBuffer().getInt(recSlotOff);
-        prefixSlotNum = frame.slotManager.decodeFirstSlotField(recSlot);
-        recOff = frame.slotManager.decodeSecondSlotField(recSlot);
-                
-        // position to prefix records first (if record is compressed)
-        if(prefixSlotNum != FieldPrefixSlotManager.RECORD_UNCOMPRESSED) {
-            int prefixSlotOff = frame.slotManager.getPrefixSlotOff(prefixSlotNum);
-            int prefixSlot = frame.getBuffer().getInt(prefixSlotOff);
-            numPrefixFields = frame.slotManager.decodeFirstSlotField(prefixSlot);
-            recOffRunner = frame.slotManager.decodeSecondSlotField(prefixSlot);               
-        }
-        else {
-            recOffRunner = recOff;
-        }        
-    }
-        
-    public void nextField() {                   
-            	
-    	// if we have passed the prefix fields of any of the two records, position them to the suffix record
-        if(currentField+1 == numPrefixFields) recOffRunner = recOff;
-        else recOffRunner += fields[currentField].getLength(frame.getBuffer().array(), recOffRunner);
-        currentField++;        
-    }
-    
-    public int getFieldOff() {
-        return recOffRunner;
-    }        
-    
-    public int getFieldSize() {
-        return fields[currentField].getLength(frame.getBuffer().array(), recOffRunner);                
-    }
-    
-    // TODO: this is the simplest implementation, could be improved to minimize the number calls to System.arrayCopy()
-    // copies the fields [startField, endField] into dest at position destOff
-    // returns the total number of bytes written
-    // this operation does not change the instances state
-    public int copyFields(int startField, int endField, byte[] dest, int destOff) {
-        
-        // remember state
-        int oldCurrentField = currentField;
-        int oldRecOffRunner = recOffRunner;
-        
-        // do we need to reposition from start?
-        if(currentField != startField) {
-            reset();
-            while(currentField != startField) nextField();
-        }
-        
-        // perform copy
-        int bytesWritten = 0;
-        for(int i = startField; i <= endField; i++) {
-            int fieldSize = getFieldSize();                        
-            System.arraycopy(frame.getBuffer().array(), recOffRunner, dest, destOff, fieldSize);
-            bytesWritten += fieldSize;             
-            destOff += fieldSize;
-            nextField();
-        }
-                    
-        // restore original state
-        currentField = oldCurrentField;
-        recOffRunner = oldRecOffRunner;
-                    
-        return bytesWritten;
-    }
-    
-    @Override
-    public ByteBuffer getBuffer() {
-    	return frame.getBuffer();
-    }
-}
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixPrefixTuple.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixPrefixTuple.java
deleted file mode 100644
index bd9de19..0000000
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixPrefixTuple.java
+++ /dev/null
@@ -1,92 +0,0 @@
-package edu.uci.ics.hyracks.storage.am.btree.impls;
-
-import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
-import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeFrame;
-import edu.uci.ics.hyracks.storage.am.btree.api.IFieldAccessor;
-import edu.uci.ics.hyracks.storage.am.btree.frames.FieldPrefixNSMLeafFrame;
-
-public class FieldPrefixPrefixTuple implements ITupleReference {
-	
-	public int recSlotOff = -1;
-    public int prefixSlotOff = -1;
-    public int prefixSlotNum = FieldPrefixSlotManager.RECORD_UNCOMPRESSED;
-    public int numPrefixFields = 0;
-    public IFieldAccessor[] fields;
-    public FieldPrefixNSMLeafFrame frame;
-    
-    public int currentField = -1;
-    public int recOffRunner = -1;    
-    
-    public FieldPrefixPrefixTuple() {    	
-    }
-    
-    public FieldPrefixPrefixTuple(IFieldAccessor[] fields, FieldPrefixNSMLeafFrame frame) {
-        this.fields = fields;
-        this.frame = frame;
-    }
-    
-    public void setFrame(IBTreeFrame frame) {
-        this.frame = (FieldPrefixNSMLeafFrame)frame;        
-    }
-    
-    public void setFields(IFieldAccessor[] fields) {
-        this.fields = fields;
-    }
-    
-    public void openPrefixSlotNum(int prefixSlotNum) {          
-        openPrefixSlotOff(frame.slotManager.getPrefixSlotOff(prefixSlotNum));        
-    }
-    
-    public void openPrefixSlotOff(int prefixSlotOff) {
-        this.prefixSlotOff = prefixSlotOff;
-        reset();
-    }
-       
-    // re-position to first field
-    public void reset() {
-        currentField = 0;        
-        int prefixSlot = frame.getBuffer().getInt(prefixSlotOff);
-        numPrefixFields = frame.slotManager.decodeFirstSlotField(prefixSlot);
-        recOffRunner = frame.slotManager.decodeSecondSlotField(prefixSlot);               
-    }
-        
-    public void nextField() {           
-    	recOffRunner += fields[currentField].getLength(frame.getBuffer().array(), recOffRunner);    	
-    }
-    
-    public int getFieldOff() {
-        return recOffRunner;
-    }        
-    
-    public int getFieldSize() {
-        return fields[currentField].getLength(frame.getBuffer().array(), recOffRunner);                
-    }
-    
-	@Override
-	public int getFieldCount() {
-		return numPrefixFields;
-	}
-
-	@Override
-	public byte[] getFieldData(int fIdx) {
-		return frame.getBuffer().array();
-	}
-
-	@Override
-	public int getFieldLength(int fIdx) {
-		reset();
-		for(int i = 0; i < fIdx; i++) {
-			nextField();
-		}						
-		return getFieldSize();
-	}
-	
-	@Override
-	public int getFieldStart(int fIdx) {
-		reset();
-		for(int i = 0; i < fIdx; i++) {
-			nextField();
-		}						
-		return recOffRunner;
-	}
-}
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixPrefixTupleReference.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixPrefixTupleReference.java
new file mode 100644
index 0000000..0727cea
--- /dev/null
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixPrefixTupleReference.java
@@ -0,0 +1,18 @@
+package edu.uci.ics.hyracks.storage.am.btree.impls;
+
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeFrame;
+import edu.uci.ics.hyracks.storage.am.btree.frames.FieldPrefixNSMLeafFrame;
+
+public class FieldPrefixPrefixTupleReference extends SimpleTupleReference {
+	
+	// assumes tuple index refers to prefix tuples
+	@Override
+	public void resetByTupleIndex(IBTreeFrame frame, int tupleIndex) {
+		FieldPrefixNSMLeafFrame concreteFrame = (FieldPrefixNSMLeafFrame)frame;		
+		int prefixSlotOff = concreteFrame.slotManager.getPrefixSlotOff(tupleIndex);
+		int prefixSlot = concreteFrame.getBuffer().getInt(prefixSlotOff);
+		setFieldCount(concreteFrame.slotManager.decodeFirstSlotField(prefixSlot));
+		tupleStartOff = concreteFrame.slotManager.decodeSecondSlotField(prefixSlot);
+		buf = concreteFrame.getBuffer();
+	}	
+}
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixSlotManager.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixSlotManager.java
index 84fbde2..f510dad 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixSlotManager.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixSlotManager.java
@@ -18,21 +18,19 @@
 import java.nio.ByteBuffer;
 
 import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeTupleReference;
 import edu.uci.ics.hyracks.storage.am.btree.api.IPrefixSlotManager;
 import edu.uci.ics.hyracks.storage.am.btree.frames.FieldPrefixNSMLeafFrame;
 
 public class FieldPrefixSlotManager implements IPrefixSlotManager {
 	
 	private static final int slotSize = 4;		
-	public static final int RECORD_UNCOMPRESSED = 0xFF;
+	public static final int TUPLE_UNCOMPRESSED = 0xFF;
 	public static final int MAX_PREFIX_SLOTS = 0xFE;
 	public static final int GREATEST_SLOT = 0x00FFFFFF;
 	
 	private ByteBuffer buf;
-	private FieldPrefixNSMLeafFrame frame;
-	//private FieldPrefixFieldIterator fieldIter = new FieldPrefixFieldIterator(null, null);
-	private FieldPrefixTuple pageTuple = new FieldPrefixTuple();
-	private FieldPrefixPrefixTuple pagePrefixTuple = new FieldPrefixPrefixTuple();
+	private FieldPrefixNSMLeafFrame frame;	
 	
 	public int decodeFirstSlotField(int slot) {
 		return (slot & 0xFF000000) >>> 24;
@@ -43,119 +41,107 @@
 	}
 	
 	public int encodeSlotFields(int firstField, int secondField) {
-		return ((firstField & 0x000000FF) << 24) | (secondField & 0x00FFFFFF);		
+		return ((firstField & 0x000000FF) << 24) | (secondField & 0x00FFFFFF);
 	}
 	
-	// returns prefix slot number, or RECORD_UNCOMPRESSED of no match was found
-	public int findPrefix(ITupleReference tuple, MultiComparator multiCmp) {
+	// returns prefix slot number, or TUPLE_UNCOMPRESSED of no match was found
+	public int findPrefix(ITupleReference tuple, IBTreeTupleReference framePrefixTuple, MultiComparator multiCmp) {
 		int prefixMid;
 	    int prefixBegin = 0;
-	    int prefixEnd = frame.getNumPrefixRecords() - 1;
-	     
-	    pagePrefixTuple.setFrame(frame);
-	    pagePrefixTuple.setFields(multiCmp.getFields());
+	    int prefixEnd = frame.getPrefixTupleCount() - 1;
 	    
-	    if(frame.getNumPrefixRecords() > 0) {
+	    if(frame.getPrefixTupleCount() > 0) {
 	    	while(prefixBegin <= prefixEnd) {
 	    		prefixMid = (prefixBegin + prefixEnd) / 2;	    			    		
-	    		int prefixSlotOff = getPrefixSlotOff(prefixMid);
-	    		int prefixSlot = buf.getInt(prefixSlotOff);    		
-	    		int numPrefixFields = decodeFirstSlotField(prefixSlot);
-	    		pagePrefixTuple.openPrefixSlotNum(prefixMid);
-	    		int cmp = multiCmp.fieldRangeCompare(tuple, pagePrefixTuple, 0, numPrefixFields);
+	    		framePrefixTuple.resetByTupleIndex(frame, prefixMid);
+	    		int cmp = multiCmp.fieldRangeCompare(tuple, framePrefixTuple, 0, framePrefixTuple.getFieldCount());
 	    		if(cmp < 0) prefixEnd = prefixMid - 1;				    		
 	    		else if(cmp > 0) prefixBegin = prefixMid + 1;	    		
 	    		else return prefixMid;
 	    	}
 	    }
 	    
-	    return FieldPrefixSlotManager.RECORD_UNCOMPRESSED;
+	    return FieldPrefixSlotManager.TUPLE_UNCOMPRESSED;
 	}
 	
-	public int findSlot(ITupleReference tuple, MultiComparator multiCmp, boolean exact) {				
-		if(frame.getNumRecords() <= 0) encodeSlotFields(RECORD_UNCOMPRESSED, GREATEST_SLOT);
+	public int findSlot(ITupleReference tuple, IBTreeTupleReference frameTuple, IBTreeTupleReference framePrefixTuple, MultiComparator multiCmp, boolean exact) {				
+		if(frame.getTupleCount() <= 0) encodeSlotFields(TUPLE_UNCOMPRESSED, GREATEST_SLOT);
 								
-		pageTuple.setFrame(frame);
-		pageTuple.setFields(multiCmp.getFields());		
-		pagePrefixTuple.setFrame(frame);
-		pagePrefixTuple.setFields(multiCmp.getFields());
+		frameTuple.setFieldCount(multiCmp.getFields().length);
 		
 	    int prefixMid;
 	    int prefixBegin = 0;
-	    int prefixEnd = frame.getNumPrefixRecords() - 1;
-	    int prefixMatch = RECORD_UNCOMPRESSED;
+	    int prefixEnd = frame.getPrefixTupleCount() - 1;
+	    int prefixMatch = TUPLE_UNCOMPRESSED;
 	    
 	    // bounds are inclusive on both ends
-	    int recPrefixSlotNumLbound = prefixBegin;
-	    int recPrefixSlotNumUbound = prefixEnd;
+	    int tuplePrefixSlotNumLbound = prefixBegin;
+	    int tuplePrefixSlotNumUbound = prefixEnd;
 	    
-	    // binary search on the prefix slots to determine upper and lower bounds for the prefixSlotNums in record slots 
+	    // binary search on the prefix slots to determine upper and lower bounds for the prefixSlotNums in tuple slots 
 	    while(prefixBegin <= prefixEnd) {
 	    	prefixMid = (prefixBegin + prefixEnd) / 2;	    			    		
-	    	int prefixSlotOff = getPrefixSlotOff(prefixMid);
-	    	int prefixSlot = buf.getInt(prefixSlotOff);    		
-	    	int numPrefixFields = decodeFirstSlotField(prefixSlot);
-	    	pagePrefixTuple.openPrefixSlotNum(prefixMid);
-	    	int cmp = multiCmp.fieldRangeCompare(tuple, pagePrefixTuple, 0, numPrefixFields);
+	    	framePrefixTuple.resetByTupleIndex(frame, prefixMid);	    		    	
+	    	int cmp = multiCmp.fieldRangeCompare(tuple, framePrefixTuple, 0, framePrefixTuple.getFieldCount());
 	    	if(cmp < 0) {	    			
 	    		prefixEnd = prefixMid - 1;
-	    		recPrefixSlotNumLbound = prefixMid - 1;	    				    		
+	    		tuplePrefixSlotNumLbound = prefixMid - 1;	    				    		
 	    	}
 	    	else if(cmp > 0) {	    			
 	    		prefixBegin = prefixMid + 1;
-	    		recPrefixSlotNumUbound = prefixMid + 1;
+	    		tuplePrefixSlotNumUbound = prefixMid + 1;
 	    	}
 	    	else {
-	    		recPrefixSlotNumLbound = prefixMid;
-	    		recPrefixSlotNumUbound = prefixMid;
+	    		tuplePrefixSlotNumLbound = prefixMid;
+	    		tuplePrefixSlotNumUbound = prefixMid;
 	    		prefixMatch = prefixMid;	    			
 	    		break;
 	    	}
 	    }
 	    
-	    //System.out.println("SLOTLBOUND: " + recPrefixSlotNumLbound);
-	    //System.out.println("SLOTUBOUND: " + recPrefixSlotNumUbound);
+	    //System.out.println("SLOTLBOUND: " + tuplePrefixSlotNumLbound);
+	    //System.out.println("SLOTUBOUND: " + tuplePrefixSlotNumUbound);
 	    
-	    int recMid = -1;
-	    int recBegin = 0;
-	    int recEnd = frame.getNumRecords() - 1;
+	    int tupleMid = -1;
+	    int tupleBegin = 0;
+	    int tupleEnd = frame.getTupleCount() - 1;
 	    
-	    // binary search on records, guided by the lower and upper bounds on prefixSlotNum
-	    while(recBegin <= recEnd) {
-            recMid = (recBegin + recEnd) / 2;      
-            int recSlotOff = getRecSlotOff(recMid);
-            int recSlot = buf.getInt(recSlotOff);              
-            int prefixSlotNum = decodeFirstSlotField(recSlot);
+	    // binary search on tuples, guided by the lower and upper bounds on prefixSlotNum
+	    while(tupleBegin <= tupleEnd) {
+            tupleMid = (tupleBegin + tupleEnd) / 2;      
+            int tupleSlotOff = getTupleSlotOff(tupleMid);
+            int tupleSlot = buf.getInt(tupleSlotOff);              
+            int prefixSlotNum = decodeFirstSlotField(tupleSlot);
             
             //System.out.println("RECS: " + recBegin + " " + recMid + " " + recEnd);
             int cmp = 0;            
-            if(prefixSlotNum == RECORD_UNCOMPRESSED) {            	            	
-            	pageTuple.openRecSlotNum(recMid);
-            	cmp = multiCmp.compare(tuple, (ITupleReference)pageTuple);      
+            if(prefixSlotNum == TUPLE_UNCOMPRESSED) {            	            	
+            	frameTuple.resetByTupleIndex(frame, tupleMid);
+            	cmp = multiCmp.compare(tuple, frameTuple);      
             }
             else {             	
-            	if(prefixSlotNum < recPrefixSlotNumLbound) cmp = 1;
-            	else if(prefixSlotNum > recPrefixSlotNumUbound) cmp = -1;
+            	if(prefixSlotNum < tuplePrefixSlotNumLbound) cmp = 1;
+            	else if(prefixSlotNum > tuplePrefixSlotNumUbound) cmp = -1;
             	else {
-            		pageTuple.openRecSlotNum(recMid);
-            		cmp = multiCmp.compare(tuple, (ITupleReference)pageTuple);            	            	            	
+            		frameTuple.resetByTupleIndex(frame, tupleMid);
+            		cmp = multiCmp.compare(tuple, frameTuple);            	            	            	
             	}
             }    
             
-            if(cmp < 0) recEnd = recMid - 1;
-            else if(cmp > 0) recBegin = recMid + 1;
-            else return encodeSlotFields(prefixMatch, recMid);
+            if(cmp < 0) tupleEnd = tupleMid - 1;
+            else if(cmp > 0) tupleBegin = tupleMid + 1;
+            else return encodeSlotFields(prefixMatch, tupleMid);
         }
 	    
 	    //System.out.println("RECS: " + recBegin + " " + recMid + " " + recEnd);
 	    
         if(exact) return encodeSlotFields(prefixMatch, GREATEST_SLOT);                       
-        if(recBegin > (frame.getNumRecords() - 1)) return encodeSlotFields(prefixMatch, GREATEST_SLOT);
+        if(tupleBegin > (frame.getTupleCount() - 1)) return encodeSlotFields(prefixMatch, GREATEST_SLOT);
         
-        // do final comparison to determine whether the search key is greater than all keys or in between some existing keys        
-        pageTuple.openRecSlotNum(recBegin);
-        int cmp = multiCmp.compare(tuple, (ITupleReference)pageTuple);        
-        if(cmp < 0) return encodeSlotFields(prefixMatch, recBegin);
+        // do final comparison to determine whether the search key is greater than all keys or in between some existing keys
+        frameTuple.resetByTupleIndex(frame, tupleBegin);
+        int cmp = multiCmp.compare(tuple, (ITupleReference)frameTuple);        
+        if(cmp < 0) return encodeSlotFields(prefixMatch, tupleBegin);
         else return encodeSlotFields(prefixMatch, GREATEST_SLOT);
 	}	
 	
@@ -164,15 +150,15 @@
 	}
 	
 	public int getPrefixSlotEndOff() {
-	    return buf.capacity() - slotSize * frame.getNumPrefixRecords();
+	    return buf.capacity() - slotSize * frame.getPrefixTupleCount();
 	}
 	
-	public int getRecSlotStartOff() {
+	public int getTupleSlotStartOff() {
 	    return getPrefixSlotEndOff() - slotSize;
 	}
 	
-	public int getRecSlotEndOff() {		
-		return buf.capacity() - slotSize * (frame.getNumPrefixRecords() + frame.getNumRecords());	    
+	public int getTupleSlotEndOff() {		
+		return buf.capacity() - slotSize * (frame.getPrefixTupleCount() + frame.getTupleCount());	    
 	}	
 	
 	public int getSlotSize() {
@@ -183,25 +169,22 @@
 		frame.getBuffer().putInt(offset, value);
 	}
 	
-	public int insertSlot(int slot, int recOff) {
+	public int insertSlot(int slot, int tupleOff) {
 		int slotNum = decodeSecondSlotField(slot);
 		if(slotNum == GREATEST_SLOT) {			
-			int slotOff = getRecSlotEndOff() - slotSize;
-			int newSlot = encodeSlotFields(decodeFirstSlotField(slot), recOff);
-			setSlot(slotOff, newSlot);
-			//System.out.println("SETTING A: " + slotOff + " " + recOff);
+			int slotOff = getTupleSlotEndOff() - slotSize;
+			int newSlot = encodeSlotFields(decodeFirstSlotField(slot), tupleOff);
+			setSlot(slotOff, newSlot);			
 			return newSlot;
 		}
 		else {
-			int slotEndOff = getRecSlotEndOff();
-			int slotOff = getRecSlotOff(slotNum);
+			int slotEndOff = getTupleSlotEndOff();
+			int slotOff = getTupleSlotOff(slotNum);
 			int length = (slotOff - slotEndOff) + slotSize;			
-			System.arraycopy(frame.getBuffer().array(), slotEndOff, frame.getBuffer().array(), slotEndOff - slotSize, length);
-			//System.out.println("MOVING SLOTS: " + length + " " + (frame.getNumRecords()*4));
+			System.arraycopy(frame.getBuffer().array(), slotEndOff, frame.getBuffer().array(), slotEndOff - slotSize, length);			
 			
-			int newSlot = encodeSlotFields(decodeFirstSlotField(slot), recOff);
-			setSlot(slotOff, newSlot);			
-			//System.out.println("SETTING B: " + slotOff + " " + recOff);
+			int newSlot = encodeSlotFields(decodeFirstSlotField(slot), tupleOff);
+			setSlot(slotOff, newSlot);
 			return newSlot;
 		}
 	}
@@ -215,8 +198,8 @@
 		return getPrefixSlotStartOff() - slotNum * slotSize;
 	}
 	
-	public int getRecSlotOff(int slotNum) {
-		return getRecSlotStartOff() - slotNum * slotSize;
+	public int getTupleSlotOff(int slotNum) {
+		return getTupleSlotStartOff() - slotNum * slotSize;
 	}
 	
 	public void setPrefixSlot(int slotNum, int slot) {
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixTuple.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixTuple.java
deleted file mode 100644
index 684edb3..0000000
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixTuple.java
+++ /dev/null
@@ -1,34 +0,0 @@
-package edu.uci.ics.hyracks.storage.am.btree.impls;
-
-import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
-
-public class FieldPrefixTuple extends FieldPrefixFieldIterator implements ITupleReference {
-
-	@Override
-	public int getFieldCount() {
-		return fields.length;
-	}
-
-	@Override
-	public byte[] getFieldData(int fIdx) {
-		return frame.getBuffer().array();
-	}
-
-	@Override
-	public int getFieldLength(int fIdx) {
-		reset();
-		for(int i = 0; i < fIdx; i++) {
-			nextField();
-		}						
-		return getFieldSize();
-	}
-	
-	@Override
-	public int getFieldStart(int fIdx) {
-		reset();
-		for(int i = 0; i < fIdx; i++) {
-			nextField();
-		}						
-		return recOffRunner;
-	}
-}
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixTupleReference.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixTupleReference.java
new file mode 100644
index 0000000..c417b86
--- /dev/null
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixTupleReference.java
@@ -0,0 +1,87 @@
+package edu.uci.ics.hyracks.storage.am.btree.impls;
+
+import java.nio.ByteBuffer;
+
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeFrame;
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeTupleReference;
+import edu.uci.ics.hyracks.storage.am.btree.frames.FieldPrefixNSMLeafFrame;
+
+public class FieldPrefixTupleReference implements IBTreeTupleReference {
+	
+	private FieldPrefixNSMLeafFrame frame;
+	private int prefixTupleStartOff;
+	private int suffixTupleStartOff;
+	private int numPrefixFields;
+	private int fieldCount;
+	private SimpleTupleReference helperTuple = new SimpleTupleReference();	
+	
+	@Override
+	public void resetByTupleIndex(IBTreeFrame frame, int tupleIndex) {
+		this.frame = (FieldPrefixNSMLeafFrame)frame;
+		
+		int tupleSlotOff = this.frame.slotManager.getTupleSlotOff(tupleIndex);
+		int tupleSlot = this.frame.getBuffer().getInt(tupleSlotOff);
+        int prefixSlotNum = this.frame.slotManager.decodeFirstSlotField(tupleSlot);
+        suffixTupleStartOff = this.frame.slotManager.decodeSecondSlotField(tupleSlot);
+        
+        if(prefixSlotNum != FieldPrefixSlotManager.TUPLE_UNCOMPRESSED) {
+            int prefixSlotOff = this.frame.slotManager.getPrefixSlotOff(prefixSlotNum);
+            int prefixSlot = this.frame.getBuffer().getInt(prefixSlotOff);
+            numPrefixFields = this.frame.slotManager.decodeFirstSlotField(prefixSlot);
+            prefixTupleStartOff = this.frame.slotManager.decodeSecondSlotField(prefixSlot); 
+        }
+        else {
+        	numPrefixFields = 0;
+        	prefixTupleStartOff = -1;
+        }
+	}
+	
+	@Override
+	public void setFieldCount(int fieldCount) {
+		this.fieldCount = fieldCount;
+	}
+
+	@Override
+	public int getFieldCount() {		
+		return fieldCount;
+	}
+
+	@Override
+	public byte[] getFieldData(int fIdx) {
+		return frame.getBuffer().array();
+	}
+
+	@Override
+	public int getFieldLength(int fIdx) {		
+		if(fIdx < numPrefixFields) {
+			helperTuple.setFieldCount(numPrefixFields);			
+			helperTuple.resetByOffset(frame.getBuffer(), prefixTupleStartOff);
+			return helperTuple.getFieldLength(fIdx);
+		}
+		else {
+			helperTuple.setFieldCount(fieldCount - numPrefixFields);
+			helperTuple.resetByOffset(frame.getBuffer(), suffixTupleStartOff);
+			return helperTuple.getFieldLength(fIdx - numPrefixFields);
+		}				
+	}
+
+	@Override
+	public int getFieldStart(int fIdx) {
+		if(fIdx < numPrefixFields) {
+			helperTuple.setFieldCount(numPrefixFields);			
+			helperTuple.resetByOffset(frame.getBuffer(), prefixTupleStartOff);
+			return helperTuple.getFieldStart(fIdx);
+		}
+		else {
+			helperTuple.setFieldCount(fieldCount - numPrefixFields);
+			helperTuple.resetByOffset(frame.getBuffer(), suffixTupleStartOff);
+			return helperTuple.getFieldStart(fIdx - numPrefixFields);
+		}
+	}
+	
+	// unsupported operation
+	@Override
+	public void resetByOffset(ByteBuffer buf, int tupleStartOffset) {
+		frame = null;		
+	}	
+}
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/MultiComparator.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/MultiComparator.java
index 0945d12..fe8c25a 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/MultiComparator.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/MultiComparator.java
@@ -15,13 +15,19 @@
 
 package edu.uci.ics.hyracks.storage.am.btree.impls;
 
+import java.io.ByteArrayInputStream;
+import java.io.DataInput;
+import java.io.DataInputStream;
+
 import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparator;
+import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
+import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
 import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
 import edu.uci.ics.hyracks.storage.am.btree.api.IFieldAccessor;
-import edu.uci.ics.hyracks.storage.am.btree.api.IFieldIterator;
 
+@SuppressWarnings("unchecked")
 public class MultiComparator {
-    	
+	
 	private static final long serialVersionUID = 1L;
 	
 	private IBinaryComparator[] cmps = null;
@@ -55,21 +61,6 @@
 	public IFieldAccessor[] getFields() {
 		return fields;
 	}
-		
-	public int compare(byte[] dataA, int recOffA, byte[] dataB, int recOffB) {
-		int lenA;
-		int lenB;
-		for(int i = 0; i < cmps.length; i++) {
-			lenA = fields[i].getLength(dataA, recOffA);
-			lenB = fields[i].getLength(dataB, recOffB);			
-			int cmp = cmps[i].compare(dataA, recOffA, lenA, dataB, recOffB, lenB);	
-    		if(cmp < 0) return -1;
-    		else if(cmp > 0) return 1;
-    		recOffA += lenA; 
-   			recOffB += lenB;    			
-    	}
-    	return 0;
-	}
 	
 	public int compare(ITupleReference tupleA, ITupleReference tupleB) {			
 		for(int i = 0; i < cmps.length; i++) {						
@@ -84,40 +75,7 @@
     	}
     	return 0;
 	}
-	
-	public int compare(ITupleReference tuple, IFieldIterator fieldIter) {
-		fieldIter.reset();
-        
-        int cmp = 0;
-        for(int i = 0; i < cmps.length; i++) {       	
-       	 cmp = cmps[i].compare(tuple.getFieldData(i), tuple.getFieldStart(i), tuple.getFieldLength(i), fieldIter.getBuffer().array(), fieldIter.getFieldOff(), fieldIter.getFieldSize());          
-            if(cmp < 0) return -1;
-            else if(cmp > 0) return 1;             
-            fieldIter.nextField();
-        }
-        
-        fieldIter.reset();
-        return 0;
-	}
-	
-	public int compare(byte[] data, int recOff, IFieldIterator fieldIter) {
-		 fieldIter.reset();
-         
-         int recRunner = 0;
-         int cmp = 0;
-         for(int i = 0; i < cmps.length; i++) {
-        	 int recFieldLen = fields[i].getLength(data, recRunner);
-        	 cmp = cmps[i].compare(data, recRunner, recFieldLen, fieldIter.getBuffer().array(), fieldIter.getFieldOff(), fieldIter.getFieldSize());          
-             if(cmp < 0) return -1;
-             else if(cmp > 0) return 1;             
-             fieldIter.nextField();
-             recRunner += recFieldLen;
-         }
-         
-         fieldIter.reset();
-         return 0;
-	}
-	
+		
 	public int fieldRangeCompare(ITupleReference tupleA, ITupleReference tupleB, int startFieldIndex, int numFields) {
 		for(int i = startFieldIndex; i < startFieldIndex + numFields; i++) {						
 			int cmp = cmps[i].compare(
@@ -133,49 +91,14 @@
     	return 0;
 	}
 	
-	public int getRecordSize(byte[] data, int recOff) {
-		int runner = recOff;
-		for(int i = 0; i < fields.length; i++) {
-			runner += fields[i].getLength(data, runner);
-		}
-		return runner - recOff;
-	}
-	
-	public int getKeySize(byte[] data, int recOff) {
-		int runner = recOff;
-		for(int i = 0; i < cmps.length; i++) {
-			runner += fields[i].getLength(data, runner);
-		}
-		return runner - recOff;
-	}		
-	
-	public String printRecord(IFieldIterator fieldIter) {
-		StringBuilder strBuilder = new StringBuilder();	    
-		fieldIter.reset();
-		for(int i = 0; i < fields.length; i++) {
-			strBuilder.append(fields[i].print(fieldIter.getBuffer().array(), fieldIter.getFieldOff()) + " ");						
-			fieldIter.nextField();
-		}
-		return strBuilder.toString();
-	}
-		
-	public String printRecord(byte[] data, int recOff) {
+	public String printTuple(ITupleReference tuple, ISerializerDeserializer[] fields) throws HyracksDataException {										
 		StringBuilder strBuilder = new StringBuilder();
-	    int runner = recOff;
-		for(int i = 0; i < fields.length; i++) {
-			strBuilder.append(fields[i].print(data, runner) + " ");
-			runner += fields[i].getLength(data, runner);
+		for(int i = 0; i < tuple.getFieldCount(); i++) {
+			ByteArrayInputStream inStream = new ByteArrayInputStream(tuple.getFieldData(i), tuple.getFieldStart(i), tuple.getFieldLength(i));
+			DataInput dataIn = new DataInputStream(inStream);
+			Object o = fields[i].deserialize(dataIn);
+			strBuilder.append(o.toString() + " ");
 		}
 		return strBuilder.toString();
-	}
-		
-	public String printKey(byte[] data, int recOff) {
-		StringBuilder strBuilder = new StringBuilder();		
-		int runner = recOff;
-		for(int i = 0; i < cmps.length; i++) {
-			strBuilder.append(fields[i].print(data, runner) + " ");			
-			runner += fields[i].getLength(data, runner);
-		}
-		return strBuilder.toString();
-	}		
+	}			
 }
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/NSMFieldIterator.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/NSMFieldIterator.java
deleted file mode 100644
index 760fa8a..0000000
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/NSMFieldIterator.java
+++ /dev/null
@@ -1,91 +0,0 @@
-/*
- * Copyright 2009-2010 by The Regents of the University of California
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * you may obtain a copy of the License from
- * 
- *     http://www.apache.org/licenses/LICENSE-2.0
- * 
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package edu.uci.ics.hyracks.storage.am.btree.impls;
-
-import java.nio.ByteBuffer;
-
-import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeFrame;
-import edu.uci.ics.hyracks.storage.am.btree.api.IFieldAccessor;
-import edu.uci.ics.hyracks.storage.am.btree.api.IFieldIterator;
-import edu.uci.ics.hyracks.storage.am.btree.frames.NSMFrame;
-
-public class NSMFieldIterator implements IFieldIterator {
-    public int recOff = -1;
-    public IFieldAccessor[] fields;
-    public NSMFrame frame;
-    
-    public int currentField = -1;
-    public int recOffRunner = -1;
-		
-	public NSMFieldIterator() {    	
-    }
-    
-    public NSMFieldIterator(IFieldAccessor[] fields, NSMFrame frame) {
-        this.fields = fields;
-        this.frame = frame;
-    }
-    
-    public void setFrame(NSMFrame frame) {
-        this.frame = frame;        
-    }
-    
-    public void setFields(IFieldAccessor[] fields) {
-        this.fields = fields;
-    }
-    
-    public void openRecSlotNum(int recSlotNum) {                 
-    	int recSlotOff = frame.getSlotManager().getSlotOff(recSlotNum);
-    	openRecSlotOff(recSlotOff);
-    }
-    
-    public void openRecSlotOff(int recSlotOff) {
-    	recOff = frame.getSlotManager().getRecOff(recSlotOff);
-        reset();
-    }
-    
-    // re-position to first field
-    public void reset() {
-        currentField = 0;
-        recOffRunner = recOff;               
-    }
-    
-    public void nextField() {                   
-    	recOffRunner += fields[currentField].getLength(frame.getBuffer().array(), recOffRunner);    
-    }
-    
-    public int getFieldOff() {
-        return recOffRunner;
-    }        
-    
-    public int getFieldSize() {
-        return fields[currentField].getLength(frame.getBuffer().array(), recOffRunner);                
-    }
-    
-    public int copyFields(int startField, int endField, byte[] dest, int destOff) {
-    
-    	return 0;
-    }
-
-	@Override
-	public void setFrame(IBTreeFrame frame) {
-		this.frame = (NSMFrame)frame;
-	}	
-	
-	@Override
-    public ByteBuffer getBuffer() {
-    	return frame.getBuffer();
-    }
-}
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/NodeFrontier.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/NodeFrontier.java
index a6efd7a..c578735 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/NodeFrontier.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/NodeFrontier.java
@@ -20,6 +20,6 @@
 public class NodeFrontier {
 	public ICachedPage page;
 	public int pageId;
-	public SelfDescTupleReference lastTuple = new SelfDescTupleReference();
+	public SimpleTupleReference lastTuple = new SimpleTupleReference();
 }
 
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/OrderedSlotManager.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/OrderedSlotManager.java
index e2dafc0..db9ee0e 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/OrderedSlotManager.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/OrderedSlotManager.java
@@ -17,31 +17,29 @@
 
 import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeFrame;
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeTupleReference;
 import edu.uci.ics.hyracks.storage.am.btree.api.ISlotManager;
 
 public class OrderedSlotManager implements ISlotManager {
 	
 	private static final int slotSize = 4;
 	private IBTreeFrame frame;
-		
-	private SelfDescTupleReference pageTuple = new SelfDescTupleReference();
-		
+	
 	// TODO: mix in interpolation search
 	@Override
-	public int findSlot(ITupleReference tuple, MultiComparator multiCmp, boolean exact) {
-		if(frame.getNumRecords() <= 0) return -1;
-		
-		pageTuple.setFields(multiCmp.getFields());
-		
+	public int findSlot(ITupleReference tuple, IBTreeTupleReference pageTuple, MultiComparator multiCmp, boolean exact) {
+		if(frame.getTupleCount() <= 0) return -1;
+				
 		int mid;
 		int begin = 0;
-		int end = frame.getNumRecords() - 1;
-		
+		int end = frame.getTupleCount() - 1;
+				
         while(begin <= end) {
             mid = (begin + end) / 2;
         	int slotOff = getSlotOff(mid);        	
-        	int recOff = getRecOff(slotOff);
-        	pageTuple.reset(frame.getBuffer(), recOff);
+        	int tupleOff = getTupleOff(slotOff);
+        	pageTuple.resetByOffset(frame.getBuffer(), tupleOff);
+        	
         	int cmp = multiCmp.compare(tuple, pageTuple);
         	if(cmp < 0)
         		end = mid - 1;
@@ -52,11 +50,11 @@
         }
                         
         if(exact) return -1;             
-        if(begin > frame.getNumRecords() - 1) return -1;   
+        if(begin > frame.getTupleCount() - 1) return -1;   
         
         int slotOff = getSlotOff(begin);
-        int recOff = getRecOff(slotOff);
-        pageTuple.reset(frame.getBuffer(), recOff);
+        int tupleOff = getTupleOff(slotOff);
+        pageTuple.resetByOffset(frame.getBuffer(), tupleOff);
         if(multiCmp.compare(tuple, pageTuple)  < 0)
         	return slotOff;
         else
@@ -64,7 +62,7 @@
 	}
 	
 	@Override
-	public int getRecOff(int offset) {		
+	public int getTupleOff(int offset) {		
 		return frame.getBuffer().getInt(offset);
 	}
 	
@@ -75,7 +73,7 @@
 	
 	@Override
 	public int getSlotEndOff() {
-		return frame.getBuffer().capacity() - (frame.getNumRecords() * slotSize);
+		return frame.getBuffer().capacity() - (frame.getTupleCount() * slotSize);
 	}
 	
 	@Override
@@ -89,17 +87,17 @@
 	}
 	
 	@Override
-	public int insertSlot(int slotOff, int recOff) {
+	public int insertSlot(int slotOff, int tupleOff) {
 		if(slotOff < 0) {
 			slotOff = getSlotEndOff() - slotSize;
-			setSlot(slotOff, recOff);
+			setSlot(slotOff, tupleOff);
 			return slotOff;
 		}
 		else {
 			int slotEndOff = getSlotEndOff();
 			int length = (slotOff - slotEndOff) + slotSize;			
 			System.arraycopy(frame.getBuffer().array(), slotEndOff, frame.getBuffer().array(), slotEndOff - slotSize, length);
-			setSlot(slotOff, recOff);
+			setSlot(slotOff, tupleOff);
 			return slotOff;
 		}		
 	}
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/RangeSearchCursor.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/RangeSearchCursor.java
index 293ad6b..22687d4 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/RangeSearchCursor.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/RangeSearchCursor.java
@@ -18,7 +18,7 @@
 import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeCursor;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeLeafFrame;
-import edu.uci.ics.hyracks.storage.am.btree.api.IFieldIterator;
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeTupleReference;
 import edu.uci.ics.hyracks.storage.am.btree.api.ISearchPredicate;
 import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
 import edu.uci.ics.hyracks.storage.common.buffercache.ICachedPage;
@@ -27,20 +27,17 @@
 public class RangeSearchCursor implements IBTreeCursor {
 
 	private ISearchPredicate searchPred = null;	
-	private int recordNum = 0;
-	private int recordOffset = -1;
+	private int tupleIndex = 0;
 	private int fileId = -1;
 	private ICachedPage page = null;
 	private IBTreeLeafFrame frame = null;
 	private IBufferCache bufferCache = null;
 	
-	private IFieldIterator fieldIter;
-	
-	
+	private IBTreeTupleReference frameTuple;
+			
 	public RangeSearchCursor(IBTreeLeafFrame frame) {
 		this.frame = frame;
-		this.fieldIter = frame.createFieldIterator();
-		this.fieldIter.setFrame(frame);
+		this.frameTuple = frame.createTupleReference();
 	}
 	
 	@Override
@@ -50,9 +47,8 @@
 		page = null;
 	}
 	
-	public IFieldIterator getFieldIterator() {
-		fieldIter.reset();
-		return fieldIter;
+	public ITupleReference getTuple() {
+		return frameTuple;
 	}
 	
 	@Override
@@ -62,7 +58,7 @@
 		
 	@Override
 	public boolean hasNext() throws Exception {
-		if(recordNum >= frame.getNumRecords()) {
+		if(tupleIndex >= frame.getTupleCount()) {
 			int nextLeafPage = -1;
 			if(searchPred.isForward()) {
 				nextLeafPage = frame.getNextLeaf();
@@ -81,7 +77,7 @@
 				page = nextLeaf;
 				frame.setPage(page);
 				
-				recordNum = 0;
+				tupleIndex = 0;
 			}
 			else {
 				return false;
@@ -93,12 +89,10 @@
 		MultiComparator cmp = pred.getComparator();
 		if(searchPred.isForward()) {
 			ITupleReference highKey = pred.getHighKey();		
-			fieldIter.openRecSlotNum(recordNum);
-			//recordOffset = frame.getRecordOffset(recordNum);
+			frameTuple.resetByTupleIndex(frame, tupleIndex);
 									
-			if(highKey == null) return true;									
-			//if(cmp.compare(highKeys, 0, page.getBuffer().array(), recordOffset) < 0) {
-			if(cmp.compare(highKey, fieldIter) < 0) {
+			if(highKey == null) return true;
+			if(cmp.compare(highKey, frameTuple) < 0) {
 				return false;
 			}
 			else {
@@ -107,11 +101,10 @@
 		}
 		else {
 			ITupleReference lowKey = pred.getLowKey();						
-			fieldIter.openRecSlotNum(frame.getNumRecords() - recordNum - 1);
-			//recordOffset = frame.getRecordOffset(frame.getNumRecords() - recordNum - 1);
+			frameTuple.resetByTupleIndex(frame, frame.getTupleCount() - tupleIndex - 1);
 			if(lowKey == null) return true;
 			
-			if(cmp.compare(lowKey, fieldIter) > 0) {
+			if(cmp.compare(lowKey, frameTuple) > 0) {
 				return false;
 			}
 			else {
@@ -121,8 +114,8 @@
 	}
 
 	@Override
-	public void next() throws Exception {				
-		recordNum++;
+	public void next() throws Exception {		
+		tupleIndex++;
 	}
 	
 	@Override
@@ -137,43 +130,38 @@
 		this.page = page;
 		frame.setPage(page);					
 		
-		// position recordNum to the first appropriate key
+		// position tupleIndex to the first appropriate key
 		// TODO: can be done more efficiently with binary search but this needs some thinking/refactoring
 		RangePredicate pred = (RangePredicate)searchPred;
 		MultiComparator cmp = pred.getComparator();
-		this.fieldIter.setFields(cmp.getFields());
+		frameTuple.setFieldCount(cmp.getFields().length);
 		if(searchPred.isForward()) {
-			ITupleReference lowKey = pred.getLowKey();
-						
-			//recordOffset = frame.getRecordOffset(recordNum);			
-			fieldIter.openRecSlotNum(recordNum);
+			ITupleReference lowKey = pred.getLowKey();			
+			
+			frameTuple.resetByTupleIndex(frame, tupleIndex);
 			if(lowKey == null) return; // null means -infinity
 						
-			while(cmp.compare(lowKey, fieldIter) > 0 && recordNum < frame.getNumRecords()) {
-			//while(cmp.compare(lowKeys, 0, page.getBuffer().array(), recordOffset) > 0 && recordNum < frame.getNumRecords()) {
-			    recordNum++;
-			    fieldIter.openRecSlotNum(recordNum);
+			while(cmp.compare(lowKey, frameTuple) > 0 && tupleIndex < frame.getTupleCount()) {
+			    tupleIndex++;
+			    frameTuple.resetByTupleIndex(frame, tupleIndex);
 			}						
 		}
 		else {
 			ITupleReference highKey = pred.getHighKey();
 			
-			//recordOffset = frame.getRecordOffset(frame.getNumRecords() - recordNum - 1);
-			fieldIter.openRecSlotNum(frame.getNumRecords() - recordNum - 1);
-			if(highKey != null) return; // null means +infinity
+			frameTuple.resetByTupleIndex(frame, frame.getTupleCount() - tupleIndex - 1);
+			if(highKey == null) return; // null means +infinity
 			    
-			while(cmp.compare(highKey, fieldIter) < 0 && recordNum < frame.getNumRecords()) {				
-			    recordNum++;
-			    fieldIter.openRecSlotNum(recordNum);
-			    //recordOffset = frame.getRecordOffset(frame.getNumRecords() - recordNum - 1);			
+			while(cmp.compare(highKey, frameTuple) < 0 && tupleIndex < frame.getTupleCount()) {				
+			    tupleIndex++;
+			    frameTuple.resetByTupleIndex(frame, frame.getTupleCount() - tupleIndex - 1);
 			}						
 		}
 	}
 	
 	@Override
 	public void reset() {
-		recordNum = 0;
-		recordOffset = 0;
+		tupleIndex = 0;
 		page = null;	
 		searchPred = null;
 	}
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/SimpleTupleReference.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/SimpleTupleReference.java
new file mode 100644
index 0000000..d793e43
--- /dev/null
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/SimpleTupleReference.java
@@ -0,0 +1,69 @@
+package edu.uci.ics.hyracks.storage.am.btree.impls;
+
+import java.nio.ByteBuffer;
+
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeFrame;
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeTupleReference;
+
+public class SimpleTupleReference implements IBTreeTupleReference {
+	
+	protected ByteBuffer buf;
+	protected int fieldCount;	
+	protected int tupleStartOff;
+	protected int nullFlagsBytes;
+	protected int fieldSlotsBytes;
+	
+	public void resetByOffset(ByteBuffer buf, int tupleStartOff) {
+		this.buf = buf;
+		this.tupleStartOff = tupleStartOff;
+	}
+	
+	public void setFieldCount(int fieldCount) {
+		this.fieldCount = fieldCount;
+		nullFlagsBytes = getNullFlagsBytes();
+		fieldSlotsBytes = getFieldSlotsBytes();
+	}
+	
+	@Override
+	public int getFieldCount() {
+		return fieldCount;
+	}
+
+	@Override
+	public byte[] getFieldData(int fIdx) {
+		return buf.array();
+	}
+
+	@Override
+	public int getFieldLength(int fIdx) {
+		if(fIdx == 0) {
+			return buf.getShort(tupleStartOff + nullFlagsBytes);			
+		}
+		else {
+			return buf.getShort(tupleStartOff + nullFlagsBytes + fIdx * 2) - buf.getShort(tupleStartOff + nullFlagsBytes + ((fIdx-1) * 2));
+		}
+	}
+
+	@Override
+	public int getFieldStart(int fIdx) {				
+		if(fIdx == 0) {			
+			return tupleStartOff + nullFlagsBytes + fieldSlotsBytes;
+		}
+		else {			
+			return tupleStartOff + nullFlagsBytes + fieldSlotsBytes + buf.getShort(tupleStartOff + nullFlagsBytes + ((fIdx-1) * 2));
+		}		
+	}
+	
+	protected int getNullFlagsBytes() {
+		return (int)Math.ceil(fieldCount / 8.0);
+	}
+	
+	protected int getFieldSlotsBytes() {
+		return fieldCount * 2;
+	}
+
+	@Override
+	public void resetByTupleIndex(IBTreeFrame frame, int tupleIndex) {
+		resetByOffset(frame.getBuffer(), frame.getTupleOffset(tupleIndex));		
+	}	
+}
\ No newline at end of file
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/SimpleTupleWriter.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/SimpleTupleWriter.java
new file mode 100644
index 0000000..8fa9677
--- /dev/null
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/SimpleTupleWriter.java
@@ -0,0 +1,92 @@
+package edu.uci.ics.hyracks.storage.am.btree.impls;
+
+import java.nio.ByteBuffer;
+
+import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
+import edu.uci.ics.hyracks.storage.am.btree.api.ITupleWriter;
+
+public class SimpleTupleWriter implements ITupleWriter {
+
+	@Override
+	public int bytesRequired(ITupleReference tuple) {
+		int bytes = getNullFlagsBytes(tuple) + getFieldSlotsBytes(tuple);
+		for(int i = 0; i < tuple.getFieldCount(); i++) {
+			bytes += tuple.getFieldLength(i);
+		}
+		return bytes;
+	}
+	
+	@Override
+	public int bytesRequired(ITupleReference tuple, int startField, int numFields) {		
+		int bytes = getNullFlagsBytes(tuple, startField, numFields) + getFieldSlotsBytes(tuple, startField, numFields);
+		for(int i = startField; i < startField + numFields; i++) {
+			bytes += tuple.getFieldLength(i);
+		}
+		return bytes;	
+	}
+
+	@Override
+	public ITupleReference createTupleReference() {
+		// TODO Auto-generated method stub
+		return null;
+	}
+	
+	@Override
+	public int writeTuple(ITupleReference tuple, ByteBuffer targetBuf, int targetOff) {
+		int runner = targetOff;
+		int nullFlagsBytes = getNullFlagsBytes(tuple);
+		int fieldSlotsBytes = getFieldSlotsBytes(tuple);
+		for(int i = 0; i < nullFlagsBytes; i++) {
+			targetBuf.put(runner++, (byte)0);			
+		}
+		runner += fieldSlotsBytes;
+		
+		int fieldEndOff = 0;
+		for(int i = 0; i < tuple.getFieldCount(); i++) {			
+			System.arraycopy(tuple.getFieldData(i), tuple.getFieldStart(i), targetBuf.array(), runner, tuple.getFieldLength(i));
+			fieldEndOff += tuple.getFieldLength(i);
+			runner += tuple.getFieldLength(i);
+			targetBuf.putShort(targetOff + nullFlagsBytes + i * 2, (short)fieldEndOff);	
+		}
+		
+		return runner - targetOff;
+	}
+
+	@Override
+	public int writeTupleFields(ITupleReference tuple, int startField, int numFields, ByteBuffer targetBuf, int targetOff) {	
+		int runner = targetOff;
+		int nullFlagsBytes = getNullFlagsBytes(tuple, startField, numFields);
+		for(int i = 0; i < nullFlagsBytes; i++) {
+			targetBuf.put(runner++, (byte)0);
+		}
+		runner += getFieldSlotsBytes(tuple, startField, numFields);
+		
+		int fieldEndOff = 0;
+		int fieldCounter = 0;
+		for(int i = startField; i < startField + numFields; i++) {			
+			System.arraycopy(tuple.getFieldData(i), tuple.getFieldStart(i), targetBuf.array(), runner, tuple.getFieldLength(i));			
+			fieldEndOff += tuple.getFieldLength(i);			
+			runner += tuple.getFieldLength(i);		
+			targetBuf.putShort(targetOff + nullFlagsBytes + fieldCounter * 2, (short)fieldEndOff);
+			fieldCounter++;
+		}
+		
+		return runner - targetOff;				
+	}
+	
+	private int getNullFlagsBytes(ITupleReference tuple) {
+		return (int)Math.ceil((double)tuple.getFieldCount() / 8.0);
+	}
+	
+	private int getFieldSlotsBytes(ITupleReference tuple) {
+		return tuple.getFieldCount() * 2;
+	}
+	
+	private int getNullFlagsBytes(ITupleReference tuple, int startField, int numFields) {
+		return (int)Math.ceil((double)numFields / 8.0);
+	}
+	
+	private int getFieldSlotsBytes(ITupleReference tuple, int startField, int numFields) {
+		return numFields * 2;
+	}
+}
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/SlotOffRecOff.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/SlotOffTupleOff.java
similarity index 70%
rename from hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/SlotOffRecOff.java
rename to hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/SlotOffTupleOff.java
index 0c49fb5..24001ff 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/SlotOffRecOff.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/SlotOffTupleOff.java
@@ -15,18 +15,20 @@
 
 package edu.uci.ics.hyracks.storage.am.btree.impls;
 
-public class SlotOffRecOff implements Comparable<SlotOffRecOff> {
+public class SlotOffTupleOff implements Comparable<SlotOffTupleOff> {
+	public int tupleIndex;
 	public int slotOff;
-	public int recOff;
+	public int tupleOff;	
 	
-	public SlotOffRecOff(int slotOff, int recOff) {
+	public SlotOffTupleOff(int tupleIndex, int slotOff, int recOff) {
+		this.tupleIndex = tupleIndex;
 		this.slotOff = slotOff;
-		this.recOff = recOff;
+		this.tupleOff = recOff;
 	}
 	
 	@Override
-	public int compareTo(SlotOffRecOff o) {			
-		return recOff - o.recOff;
+	public int compareTo(SlotOffTupleOff o) {			
+		return tupleOff - o.tupleOff;
 	}
 }
 
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/SplitKey.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/SplitKey.java
index 4fb52b4..809e3da 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/SplitKey.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/SplitKey.java
@@ -20,12 +20,12 @@
 public class SplitKey {		
 	public byte[] data = null;	
 	public ByteBuffer buf = null;
-	public SelfDescTupleReference tuple = new SelfDescTupleReference();
+	public SimpleTupleReference tuple = new SimpleTupleReference();
+	public int keySize = 0;
 	
 	public void initData(int keySize) {
-		// try to reuse existing memory from a lower-level split if possible
-		// in general this is always possible for fixed-sized keys		
-		/*
+		// try to reuse existing memory from a lower-level split if possible				
+		this.keySize = keySize;
 		if(data != null) {
 			if(data.length < keySize + 8) {
 				data = new byte[keySize + 8]; // add 8 for left and right page
@@ -35,11 +35,9 @@
 		else { 
 			data = new byte[keySize + 8]; // add 8 for left and right page
 			buf = ByteBuffer.wrap(data);
-		}
-		*/		
-		data = new byte[keySize + 8]; // add 8 for left and right page
-		buf = ByteBuffer.wrap(data);
-		tuple.reset(buf, 0);
+		}		
+				
+		tuple.resetByOffset(buf, 0);
 	}
 	
 	public void reset() {
@@ -47,46 +45,42 @@
 		buf = null;
 	}
 	
-	public void setData(byte[] data) {
-		this.data = data;
-	}
-	
 	public ByteBuffer getBuffer() {
 		return buf;
 	}
 	
-	public SelfDescTupleReference getTuple() {
+	public SimpleTupleReference getTuple() {
 		return tuple;
 	}
 	
 	public int getLeftPage() {
-		return buf.getInt(data.length - 8);
+		return buf.getInt(keySize);
 	}
 	
 	public int getRightPage() {
-		return buf.getInt(data.length - 4);
+		return buf.getInt(keySize + 4);
 	}
 		
 	public void setLeftPage(int leftPage) {
-		buf.putInt(data.length - 8, leftPage);
+		buf.putInt(keySize, leftPage);
 	}
 	
 	public void setRightPage(int rightPage) {
-		buf.putInt(data.length - 4, rightPage);
+		buf.putInt(keySize + 4, rightPage);
 	}
 	
-	public void setPages(int leftPage, int rightPage) {
-		buf.putInt(data.length - 8, leftPage);
-		buf.putInt(data.length - 4, rightPage);
+	public void setPages(int leftPage, int rightPage) {		
+		buf.putInt(keySize, leftPage);
+		buf.putInt(keySize + 4, rightPage);
 	}
 	
 	public SplitKey duplicate() {
 		SplitKey copy = new SplitKey();
 		copy.data = data.clone();		
 		copy.buf = ByteBuffer.wrap(copy.data);
-		copy.tuple = new SelfDescTupleReference();
-		copy.tuple.setFields(tuple.getFields());
-		copy.tuple.reset(copy.buf, 0);	
+		copy.tuple = new SimpleTupleReference();
+		copy.tuple.setFieldCount(tuple.getFieldCount());
+		copy.tuple.resetByOffset(copy.buf, 0);	
 		return copy;
 	}
 }
\ No newline at end of file
diff --git a/hyracks/hyracks-storage-am-btree/src/test/java/edu/uci/ics/hyracks/storage/am/btree/BTreeFieldPrefixNSMTest.java b/hyracks/hyracks-storage-am-btree/src/test/java/edu/uci/ics/hyracks/storage/am/btree/BTreeFieldPrefixNSMTest.java
index 77c114d..207ffa3 100644
--- a/hyracks/hyracks-storage-am-btree/src/test/java/edu/uci/ics/hyracks/storage/am/btree/BTreeFieldPrefixNSMTest.java
+++ b/hyracks/hyracks-storage-am-btree/src/test/java/edu/uci/ics/hyracks/storage/am/btree/BTreeFieldPrefixNSMTest.java
@@ -153,7 +153,7 @@
             frame.setPage(page);            
             frame.initBuffer((byte)0);
             slotManager.setFrame(frame);          
-            frame.setNumPrefixRecords(0);
+            frame.setPrefixTupleCount(0);
             
             String before = new String();
             String after = new String();
@@ -190,16 +190,16 @@
         		savedFields[i][2] = c;
         		            
         		if(rnd.nextInt() % compactFreq == 0) {
-        			before = frame.printKeys(cmp);
+        			before = frame.printKeys(cmp, fields.length);
         			frame.compact(cmp);
-        			after = frame.printKeys(cmp);
+        			after = frame.printKeys(cmp, fields.length);
         			Assert.assertEquals(before, after);
         		}
         		
         		if(rnd.nextInt() % compressFreq == 0) {
-        			before = frame.printKeys(cmp);
+        			before = frame.printKeys(cmp, fields.length);
         			frame.compress(cmp);
-        			after = frame.printKeys(cmp);
+        			after = frame.printKeys(cmp, fields.length);
         			Assert.assertEquals(before, after);
         		}
         	}
@@ -217,16 +217,16 @@
         		}
         		
         		if(rnd.nextInt() % compactFreq == 0) {
-        			before = frame.printKeys(cmp);
+        			before = frame.printKeys(cmp, fields.length);
         			frame.compact(cmp);
-        			after = frame.printKeys(cmp);
+        			after = frame.printKeys(cmp, fields.length);
         			Assert.assertEquals(before, after);
         		}
         		
         		if(rnd.nextInt() % compressFreq == 0) {
-        			before = frame.printKeys(cmp);
+        			before = frame.printKeys(cmp, fields.length);
         			frame.compress(cmp);
-        			after = frame.printKeys(cmp);
+        			after = frame.printKeys(cmp, fields.length);
         			Assert.assertEquals(before, after);
         		}  
         	}
diff --git a/hyracks/hyracks-storage-am-btree/src/test/java/edu/uci/ics/hyracks/storage/am/btree/BTreeTest.java b/hyracks/hyracks-storage-am-btree/src/test/java/edu/uci/ics/hyracks/storage/am/btree/BTreeTest.java
index f616f56..be4d4c2 100644
--- a/hyracks/hyracks-storage-am-btree/src/test/java/edu/uci/ics/hyracks/storage/am/btree/BTreeTest.java
+++ b/hyracks/hyracks-storage-am-btree/src/test/java/edu/uci/ics/hyracks/storage/am/btree/BTreeTest.java
@@ -35,6 +35,7 @@
 import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAccessor;
 import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAppender;
 import edu.uci.ics.hyracks.dataflow.common.data.accessors.FrameTupleReference;
+import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
 import edu.uci.ics.hyracks.dataflow.common.data.comparators.IntegerBinaryComparatorFactory;
 import edu.uci.ics.hyracks.dataflow.common.data.comparators.UTF8StringBinaryComparatorFactory;
 import edu.uci.ics.hyracks.dataflow.common.data.marshalling.IntegerSerializerDeserializer;
@@ -47,7 +48,6 @@
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeMetaDataFrame;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeMetaDataFrameFactory;
 import edu.uci.ics.hyracks.storage.am.btree.api.IFieldAccessor;
-import edu.uci.ics.hyracks.storage.am.btree.api.IFieldIterator;
 import edu.uci.ics.hyracks.storage.am.btree.frames.MetaDataFrameFactory;
 import edu.uci.ics.hyracks.storage.am.btree.frames.NSMInteriorFrameFactory;
 import edu.uci.ics.hyracks.storage.am.btree.frames.NSMLeafFrameFactory;
@@ -73,7 +73,8 @@
 public class BTreeTest {    
 	
     //private static final int PAGE_SIZE = 128;
-    private static final int PAGE_SIZE = 8192;
+    //private static final int PAGE_SIZE = 8192;
+	private static final int PAGE_SIZE = 256;
     private static final int NUM_PAGES = 10;
     private static final int HYRACKS_FRAME_SIZE = 128;
     
@@ -96,7 +97,7 @@
             return buffers;
         }
     }
-    
+        
     // FIXED-LENGTH KEY TEST
     // create a B-tree with one fixed-length "key" field and one fixed-length "value" field
     // fill B-tree with random values using insertions (not bulk load)
@@ -128,7 +129,7 @@
         IFieldAccessor[] fields = new IFieldAccessor[2];
         fields[0] = new Int32Accessor(); // key field
         fields[1] = new Int32Accessor(); // value field
-
+        
         int keyLen = 1;
         IBinaryComparator[] cmps = new IBinaryComparator[keyLen];
         cmps[0] = IntegerBinaryComparatorFactory.INSTANCE.createBinaryComparator();
@@ -158,10 +159,11 @@
 		FrameTupleReference tuple = new FrameTupleReference();
 		
 		// 10000
-        for (int i = 0; i < 10000; i++) {        	
+        for (int i = 0; i < 10000; i++) {        	        	
+        	
         	int f0 = rnd.nextInt() % 10000;
         	int f1 = 5;
-        	
+        	        	
         	tb.reset();
         	IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
         	tb.addFieldEndOffset();
@@ -191,6 +193,7 @@
             //System.out.println();
         }
         //btree.printTree(leafFrame, interiorFrame);
+        //System.out.println();
         
         print("TOTALSPACE: " + f.length() + "\n");
         
@@ -200,24 +203,26 @@
         long end = System.currentTimeMillis();
         long duration = end - start;
         print("DURATION: " + duration + "\n");
-        
+                
         // ordered scan
+        
         print("ORDERED SCAN:\n");
         IBTreeCursor scanCursor = new RangeSearchCursor(leafFrame);
         RangePredicate nullPred = new RangePredicate(true, null, null, null);
         btree.search(scanCursor, nullPred, leafFrame, interiorFrame);
         try {
             while (scanCursor.hasNext()) {
-                scanCursor.next();
-                IFieldIterator fieldIter = scanCursor.getFieldIterator();                                
-                String rec = cmp.printRecord(fieldIter);
+                scanCursor.next();                
+                ITupleReference frameTuple = scanCursor.getTuple();                                
+                String rec = cmp.printTuple(frameTuple, recDescSers);
                 print(rec + "\n");
             }
         } catch (Exception e) {
             e.printStackTrace();
         } finally {
             scanCursor.close();
-        }                
+        }
+                       
         
         // disk-order scan
         print("DISK-ORDER SCAN:\n");        
@@ -226,8 +231,8 @@
         try {
             while (diskOrderCursor.hasNext()) {
                 diskOrderCursor.next();
-                IFieldIterator fieldIter = diskOrderCursor.getFieldIterator();                                
-                String rec = cmp.printRecord(fieldIter);
+                ITupleReference frameTuple = diskOrderCursor.getTuple();                
+                String rec = cmp.printTuple(frameTuple, recDescSers);
                 print(rec + "\n");
             }
         } catch (Exception e) {
@@ -235,7 +240,8 @@
         } finally {
             diskOrderCursor.close();
         }
-               
+         
+        
         // range search in [-1000, 1000]
         print("RANGE SEARCH:\n");        
  
@@ -269,22 +275,23 @@
         try {
             while (rangeCursor.hasNext()) {
                 rangeCursor.next();
-                IFieldIterator fieldIter = rangeCursor.getFieldIterator();                                
-                String rec = cmp.printRecord(fieldIter);
-                print(rec + "\n"); 
+                ITupleReference frameTuple = rangeCursor.getTuple();                
+                String rec = cmp.printTuple(frameTuple, recDescSers);
+                print(rec + "\n");
             }
         } catch (Exception e) {
             e.printStackTrace();
         } finally {
             rangeCursor.close();
-        }
+        }  
                 
         btree.close();
 
         bufferCache.close();
         fileManager.close();
         print("\n");
-    }    
+    }
+    
     
     // COMPOSITE KEY TEST (NON-UNIQUE B-TREE)
     // create a B-tree with one two fixed-length "key" fields and one fixed-length "value" field
@@ -317,8 +324,8 @@
         IFieldAccessor[] fields = new IFieldAccessor[3];
         fields[0] = new Int32Accessor(); // key field 1
         fields[1] = new Int32Accessor(); // key field 2
-        fields[2] = new Int32Accessor(); // value field
-
+        fields[2] = new Int32Accessor(); // value field        
+        
         int keyLen = 2;
         IBinaryComparator[] cmps = new IBinaryComparator[keyLen];
         cmps[0] = IntegerBinaryComparatorFactory.INSTANCE.createBinaryComparator();
@@ -391,9 +398,9 @@
         try {
             while (scanCursor.hasNext()) {
                 scanCursor.next();
-                IFieldIterator fieldIter = scanCursor.getFieldIterator();                                
-                String rec = cmp.printRecord(fieldIter);
-                print(rec + "\n");         
+                ITupleReference frameTuple = scanCursor.getTuple();                                
+                String rec = cmp.printTuple(frameTuple, recDescSers);
+                print(rec + "\n");
             }
         } catch (Exception e) {
             e.printStackTrace();
@@ -433,9 +440,9 @@
         try {
             while (rangeCursor.hasNext()) {
                 rangeCursor.next();
-                IFieldIterator fieldIter = rangeCursor.getFieldIterator();                                
-                String rec = cmp.printRecord(fieldIter);
-                print(rec + "\n");         
+                ITupleReference frameTuple = rangeCursor.getTuple();                                
+                String rec = cmp.printTuple(frameTuple, recDescSers);
+                print(rec + "\n"); 
             }
         } catch (Exception e) {
             e.printStackTrace();
@@ -449,7 +456,8 @@
         fileManager.close();
         
         print("\n");
-    }
+    }    
+    
     
     // VARIABLE-LENGTH TEST
     // create a B-tree with one variable-length "key" field and one variable-length "value" field
@@ -482,7 +490,7 @@
     	IFieldAccessor[] fields = new IFieldAccessor[2];
     	fields[0] = new UTF8StringAccessor(); // key        
     	fields[1] = new UTF8StringAccessor(); // value
-
+    	
     	int keyLen = 1;
     	IBinaryComparator[] cmps = new IBinaryComparator[keyLen];
     	cmps[0] = UTF8StringBinaryComparatorFactory.INSTANCE.createBinaryComparator();
@@ -531,7 +539,9 @@
     		try {
     			btree.insert(tuple, leafFrame, interiorFrame, metaFrame);
     		} catch (Exception e) {
+    			//e.printStackTrace();
     		}
+    		    		
     	}     
     	// btree.printTree();
 
@@ -544,9 +554,9 @@
         try {
             while (scanCursor.hasNext()) {
                 scanCursor.next();
-                IFieldIterator fieldIter = scanCursor.getFieldIterator();                                
-                String rec = cmp.printRecord(fieldIter);
-                print(rec + "\n");          
+                ITupleReference frameTuple = scanCursor.getTuple();                                
+                String rec = cmp.printTuple(frameTuple, recDescSers);
+                print(rec + "\n");  
             }
         } catch (Exception e) {
             e.printStackTrace();
@@ -587,9 +597,9 @@
         try {
             while (rangeCursor.hasNext()) {
                 rangeCursor.next();
-                IFieldIterator fieldIter = rangeCursor.getFieldIterator();                                
-                String rec = cmp.printRecord(fieldIter);
-                print(rec + "\n");    
+                ITupleReference frameTuple = rangeCursor.getTuple();                                
+                String rec = cmp.printTuple(frameTuple, recDescSers);
+                print(rec + "\n");  
             }
         } catch (Exception e) {
             e.printStackTrace();
@@ -604,6 +614,7 @@
         
         print("\n");
     }
+        
     
     // DELETION TEST
     // create a B-tree with one variable-length "key" field and one variable-length "value" field
@@ -760,6 +771,7 @@
         
         print("\n");
     }
+        
     
     // BULK LOAD TEST
     // insert 100,000 records in bulk
@@ -788,14 +800,13 @@
         IBTreeLeafFrame leafFrame = leafFrameFactory.getFrame();
         IBTreeInteriorFrame interiorFrame = interiorFrameFactory.getFrame();
         IBTreeMetaDataFrame metaFrame = metaFrameFactory.getFrame();          
-
-        int keyLen = 2;
-
+        
         IFieldAccessor[] fields = new IFieldAccessor[3];
         fields[0] = new Int32Accessor();
         fields[1] = new Int32Accessor();
         fields[2] = new Int32Accessor();
-
+        
+        int keyLen = 2;
         IBinaryComparator[] cmps = new IBinaryComparator[keyLen];
         cmps[0] = IntegerBinaryComparatorFactory.INSTANCE.createBinaryComparator();
         cmps[1] = IntegerBinaryComparatorFactory.INSTANCE.createBinaryComparator();
@@ -836,18 +847,19 @@
         	tb.addFieldEndOffset();        	
         	IntegerSerializerDeserializer.INSTANCE.serialize(5, dos);
         	tb.addFieldEndOffset();
-        	
+        	        	
         	appender.reset(frame, true);
         	appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
         	
         	tuple.reset(accessor, 0);
         	
-        	btree.bulkLoadAddRecord(bulkLoadCtx, tuple);
-        	
+        	btree.bulkLoadAddTuple(bulkLoadCtx, tuple);        	
         }
 
         btree.endBulkLoad(bulkLoadCtx);
         
+        //btree.printTree(leafFrame, interiorFrame);
+        
         long end = System.currentTimeMillis();
         long duration = end - start;
         print("DURATION: " + duration + "\n");
@@ -881,12 +893,12 @@
         // TODO: check when searching backwards
         RangePredicate rangePred = new RangePredicate(true, lowKey, highKey, searchCmp);
         btree.search(rangeCursor, rangePred, leafFrame, interiorFrame);
-
+        
         try {
             while (rangeCursor.hasNext()) {
-                rangeCursor.next();
-                IFieldIterator fieldIter = rangeCursor.getFieldIterator();                                
-                String rec = cmp.printRecord(fieldIter);
+            	rangeCursor.next();
+                ITupleReference frameTuple = rangeCursor.getTuple();                                
+                String rec = cmp.printTuple(frameTuple, recDescSers);
                 print(rec + "\n");
             }
         } catch (Exception e) {
@@ -902,7 +914,7 @@
         
         print("\n");
     }    
-    
+        
     // TIME-INTERVAL INTERSECTION DEMO FOR EVENT PEOPLE
     // demo for Arjun to show easy support of intersection queries on time-intervals
     @Test
@@ -1040,8 +1052,8 @@
         try {
             while (scanCursor.hasNext()) {
                 scanCursor.next();
-                IFieldIterator fieldIter = scanCursor.getFieldIterator();                                
-                String rec = cmp.printRecord(fieldIter);
+                ITupleReference frameTuple = scanCursor.getTuple();                                
+                String rec = cmp.printTuple(frameTuple, recDescSers);
                 print(rec + "\n");
             }
         } catch (Exception e) {
@@ -1087,8 +1099,8 @@
         try {
             while (rangeCursor.hasNext()) {
                 rangeCursor.next();
-                IFieldIterator fieldIter = rangeCursor.getFieldIterator();                                
-                String rec = cmp.printRecord(fieldIter);
+                ITupleReference frameTuple = rangeCursor.getTuple();                                
+                String rec = cmp.printTuple(frameTuple, recDescSers);
                 print(rec + "\n");
             }
         } catch (Exception e) {
@@ -1112,5 +1124,5 @@
             strBuilder.append(s.charAt(Math.abs(random.nextInt()) % s.length()));
         }
         return strBuilder.toString();
-    }
+    }    
 }
\ No newline at end of file