diff --git a/hyracks-dataflow-common/src/main/java/edu/uci/ics/hyracks/dataflow/common/data/accessors/FrameTupleReference.java b/hyracks-dataflow-common/src/main/java/edu/uci/ics/hyracks/dataflow/common/data/accessors/FrameTupleReference.java
index aa5e447..d9c9fb2 100644
--- a/hyracks-dataflow-common/src/main/java/edu/uci/ics/hyracks/dataflow/common/data/accessors/FrameTupleReference.java
+++ b/hyracks-dataflow-common/src/main/java/edu/uci/ics/hyracks/dataflow/common/data/accessors/FrameTupleReference.java
@@ -33,7 +33,7 @@
 
     @Override
     public int getFieldStart(int fIdx) {
-        return fta.getFieldStartOffset(tIndex, fIdx);
+    	return fta.getTupleStartOffset(tIndex) + fta.getFieldSlotsLength() + fta.getFieldStartOffset(tIndex, fIdx);
     }
 
     @Override
diff --git a/hyracks-examples/hyracks-integration-tests/src/test/java/edu/uci/ics/hyracks/tests/btree/BTreeOperatorsTest.java b/hyracks-examples/hyracks-integration-tests/src/test/java/edu/uci/ics/hyracks/tests/btree/BTreeOperatorsTest.java
index 43b97d1..49ed678 100644
--- a/hyracks-examples/hyracks-integration-tests/src/test/java/edu/uci/ics/hyracks/tests/btree/BTreeOperatorsTest.java
+++ b/hyracks-examples/hyracks-integration-tests/src/test/java/edu/uci/ics/hyracks/tests/btree/BTreeOperatorsTest.java
@@ -15,7 +15,6 @@
 
 package edu.uci.ics.hyracks.tests.btree;
 
-import java.io.DataOutputStream;
 import java.io.File;
 import java.io.RandomAccessFile;
 
@@ -25,16 +24,12 @@
 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.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;
@@ -45,7 +40,6 @@
 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;
@@ -57,7 +51,6 @@
 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;
@@ -69,7 +62,6 @@
 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.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;
@@ -123,11 +115,10 @@
 		IBinaryComparatorFactory[] comparatorFactories = new IBinaryComparatorFactory[keyLen];
 		comparatorFactories[0] = UTF8StringBinaryComparatorFactory.INSTANCE;		
 				
-        int[] keyFields = { 0 };
-        int[] payloadFields = { 4, 5 };
+		int[] fieldPermutation = { 0, 4, 5 };  
         int btreeFileId = 0;
         
-		BTreeBulkLoadOperatorDescriptor btreeBulkLoad = new BTreeBulkLoadOperatorDescriptor(spec, ordersSplitProvider, ordersDesc, bufferCacheProvider, btreeRegistryProvider, btreeFileId, "/tmp/btreetest.bin", interiorFrameFactory, leafFrameFactory, fieldAccessorFactories, comparatorFactories, keyFields, payloadFields, 0.7f);		
+		BTreeBulkLoadOperatorDescriptor btreeBulkLoad = new BTreeBulkLoadOperatorDescriptor(spec, ordersSplitProvider, ordersDesc, bufferCacheProvider, btreeRegistryProvider, btreeFileId, "/tmp/btreetest.bin", interiorFrameFactory, leafFrameFactory, fieldAccessorFactories, comparatorFactories, fieldPermutation, 0.7f);		
 		PartitionConstraint btreePartitionConstraintA = new ExplicitPartitionConstraint(new LocationConstraint[] { new AbsoluteLocationConstraint(NC1_ID) });
 		btreeBulkLoad.setPartitionConstraint(btreePartitionConstraintA);
 				
@@ -170,6 +161,7 @@
         }                        
 	}
 		
+	/*
 	@Test
 	public void btreeSearchTest() throws Exception {
 		JobSpecification spec = new JobSpecification();
@@ -222,7 +214,8 @@
         spec.addRoot(printer);
         runTest(spec);
     }
-		
+    */
+	
 	@Test
 	public void insertTest() throws Exception {
 		JobSpecification spec = new JobSpecification();
@@ -346,23 +339,20 @@
         // create insert operators
         
         // primary index
-        int[] keyFieldsA = { 0 };
-        int[] payloadFieldsA = { 1,2,3,4,5 };                
-        BTreeInsertUpdateDeleteOperatorDescriptor insertOpA = new BTreeInsertUpdateDeleteOperatorDescriptor(spec, ordersSplitProvider, ordersDesc, bufferCacheProvider, btreeRegistryProvider, fileIdA, "/tmp/btreetestA.ix", interiorFrameFactory, leafFrameFactory, primaryFieldAccessorFactories, primaryComparatorFactories, keyFieldsA, payloadFieldsA, BTreeOp.BTO_INSERT);
+        int[] fieldPermutationA = { 0,1,2,3,4,5 };                       
+        BTreeInsertUpdateDeleteOperatorDescriptor insertOpA = new BTreeInsertUpdateDeleteOperatorDescriptor(spec, ordersSplitProvider, ordersDesc, bufferCacheProvider, btreeRegistryProvider, fileIdA, "/tmp/btreetestA.ix", interiorFrameFactory, leafFrameFactory, primaryFieldAccessorFactories, primaryComparatorFactories, fieldPermutationA, BTreeOp.BTO_INSERT);
         PartitionConstraint insertPartitionConstraintA = new ExplicitPartitionConstraint(new LocationConstraint[] { new AbsoluteLocationConstraint(NC1_ID) });
         insertOpA.setPartitionConstraint(insertPartitionConstraintA);
         
         // first secondary index
-        int[] keyFieldsB = { 3, 0 };
-        int[] payloadFieldsB = { };                
-        BTreeInsertUpdateDeleteOperatorDescriptor insertOpB = new BTreeInsertUpdateDeleteOperatorDescriptor(spec, ordersSplitProvider, ordersDesc, bufferCacheProvider, btreeRegistryProvider, fileIdB, "/tmp/btreetestB.ix", interiorFrameFactory, leafFrameFactory, secondaryFieldAccessorFactories, secondaryComparatorFactories, keyFieldsB, payloadFieldsB, BTreeOp.BTO_INSERT);
+        int[] fieldPermutationB = { 3, 0 };                    
+        BTreeInsertUpdateDeleteOperatorDescriptor insertOpB = new BTreeInsertUpdateDeleteOperatorDescriptor(spec, ordersSplitProvider, ordersDesc, bufferCacheProvider, btreeRegistryProvider, fileIdB, "/tmp/btreetestB.ix", interiorFrameFactory, leafFrameFactory, secondaryFieldAccessorFactories, secondaryComparatorFactories, fieldPermutationB, BTreeOp.BTO_INSERT);
         PartitionConstraint insertPartitionConstraintB = new ExplicitPartitionConstraint(new LocationConstraint[] { new AbsoluteLocationConstraint(NC1_ID) });
         insertOpB.setPartitionConstraint(insertPartitionConstraintB);
 		
         // second secondary index
-        int[] keyFieldsC = { 4, 0 };
-        int[] payloadFieldsC = { };                
-        BTreeInsertUpdateDeleteOperatorDescriptor insertOpC = new BTreeInsertUpdateDeleteOperatorDescriptor(spec, ordersSplitProvider, ordersDesc, bufferCacheProvider, btreeRegistryProvider, fileIdC, "/tmp/btreetestC.ix", interiorFrameFactory, leafFrameFactory, secondaryFieldAccessorFactories, secondaryComparatorFactories, keyFieldsC, payloadFieldsC, BTreeOp.BTO_INSERT);
+        int[] fieldPermutationC = { 4, 0 };                       
+        BTreeInsertUpdateDeleteOperatorDescriptor insertOpC = new BTreeInsertUpdateDeleteOperatorDescriptor(spec, ordersSplitProvider, ordersDesc, bufferCacheProvider, btreeRegistryProvider, fileIdC, "/tmp/btreetestC.ix", interiorFrameFactory, leafFrameFactory, secondaryFieldAccessorFactories, secondaryComparatorFactories, fieldPermutationC, BTreeOp.BTO_INSERT);
         PartitionConstraint insertPartitionConstraintC = new ExplicitPartitionConstraint(new LocationConstraint[] { new AbsoluteLocationConstraint(NC1_ID) });
         insertOpC.setPartitionConstraint(insertPartitionConstraintC);
                 
diff --git a/hyracks-storage-am-btree/pom.xml b/hyracks-storage-am-btree/pom.xml
index 5d91b5a..90b5872 100644
--- a/hyracks-storage-am-btree/pom.xml
+++ b/hyracks-storage-am-btree/pom.xml
@@ -46,6 +46,13 @@
   		<scope>compile</scope>
   	</dependency>  	
   	<dependency>
+  		<groupId>edu.uci.ics.hyracks</groupId>
+  		<artifactId>hyracks-control-nc</artifactId>
+  		<version>0.1.2-SNAPSHOT</version>
+  		<type>jar</type>
+  		<scope>compile</scope>
+  	</dependency>  	
+  	<dependency>
   		<groupId>junit</groupId>
   		<artifactId>junit</artifactId>
   		<version>4.8.1</version>
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IBTreeFrame.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IBTreeFrame.java
index aabd005..511d808 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IBTreeFrame.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IBTreeFrame.java
@@ -17,6 +17,7 @@
 
 import java.nio.ByteBuffer;
 
+import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
 import edu.uci.ics.hyracks.storage.am.btree.impls.MultiComparator;
 import edu.uci.ics.hyracks.storage.am.btree.impls.SpaceStatus;
 import edu.uci.ics.hyracks.storage.am.btree.impls.SplitKey;
@@ -27,9 +28,9 @@
 	public ICachedPage getPage();
 	public ByteBuffer getBuffer();
 	
-	public void insert(byte[] data, MultiComparator cmp) throws Exception;
-	public void update(int rid, byte[] data) throws Exception;
-	public void delete(byte[] data, MultiComparator cmp, boolean exactDelete) throws Exception;
+	public void insert(ITupleReference tuple, MultiComparator cmp) throws Exception;
+	public void update(int rid, ITupleReference tuple) throws Exception;
+	public void delete(ITupleReference tuple, MultiComparator cmp, boolean exactDelete) throws Exception;
 	
 	public void compact(MultiComparator cmp);
 	public boolean compress(MultiComparator cmp) throws Exception;
@@ -39,8 +40,8 @@
 	public int getNumRecords();
 		
 	// assumption: page must be write-latched at this point
-	public SpaceStatus hasSpaceInsert(byte[] data, MultiComparator cmp);
-	public SpaceStatus hasSpaceUpdate(int rid, byte[] data, MultiComparator cmp);
+	public SpaceStatus hasSpaceInsert(ITupleReference tuple, MultiComparator cmp);
+	public SpaceStatus hasSpaceUpdate(int rid, ITupleReference tuple, MultiComparator cmp);
 	
 	public int getRecordOffset(int slotNum);
 	
@@ -55,7 +56,7 @@
 	
 	
 	// TODO; what if records more than half-page size?
-	public int split(IBTreeFrame rightFrame, byte[] data, MultiComparator cmp, SplitKey splitKey) throws Exception;		
+	public int split(IBTreeFrame rightFrame, ITupleReference tuple, MultiComparator cmp, SplitKey splitKey) throws Exception;		
 	
 	// TODO: check if we do something nicer than returning object
 	public ISlotManager getSlotManager();
@@ -68,12 +69,15 @@
 	public boolean getSmFlag(); // structure modification flag
 	public void setSmFlag(boolean smFlag);	
 	
-	public void insertSorted(byte[] data, MultiComparator cmp) throws Exception;
+	public void insertSorted(ITupleReference tuple, MultiComparator cmp) throws Exception;
 	
 	public int getSlotSize();
 	
 	public IFieldIterator createFieldIterator();
 	
+	// TODO: should be removed after new record format		
+	public void setPageTupleFields(IFieldAccessor[] fields);
+	
 	// for debugging
 	public int getFreeSpaceOff();
 	public void setFreeSpaceOff(int freeSpace);
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IPrefixSlotManager.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IPrefixSlotManager.java
index 7792fa0..1b72b0f 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IPrefixSlotManager.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IPrefixSlotManager.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.am.btree.frames.FieldPrefixNSMLeafFrame;
 import edu.uci.ics.hyracks.storage.am.btree.impls.MultiComparator;
 
@@ -45,11 +46,11 @@
 	public int decodeSecondSlotField(int slot);		
 	public int encodeSlotFields(int firstField, int secondField);
 	
-	public int findSlot(byte[] data, MultiComparator multiCmp, boolean exact);
+	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(byte[] data, MultiComparator multiCmp);
+	public int findPrefix(ITupleReference tuple, MultiComparator multiCmp);
 	
 	public int getRecSlotStartOff();
 	public int getRecSlotEndOff();
@@ -63,9 +64,7 @@
 	public int getSlotSize();		
 	
 	public void setSlot(int offset, int value);
-	
-	public int compareCompressed(byte[] record, byte[] page, int prefixSlotNum, int recSlotNum, MultiComparator multiCmp);
-	
+		
 	// functions for testing
 	public void setPrefixSlot(int slotNum, int slot);
 }
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/ISlotManager.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/ISlotManager.java
index b92f9e1..04fb2c1 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/ISlotManager.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/ISlotManager.java
@@ -15,12 +15,13 @@
 
 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.am.btree.impls.MultiComparator;
 
 public interface ISlotManager {
 	public void setFrame(IBTreeFrame frame);
 	
-	public int findSlot(byte[] data, MultiComparator multiCmp, boolean exact);
+	public int findSlot(ITupleReference tuple, MultiComparator multiCmp, boolean exact);
 	public int insertSlot(int slotOff, int recOff);
 	
 	public int getSlotStartOff();
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/compressors/FieldPrefixCompressor.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/compressors/FieldPrefixCompressor.java
index a0f0258..08dca87 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/compressors/FieldPrefixCompressor.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/compressors/FieldPrefixCompressor.java
@@ -30,7 +30,7 @@
 import edu.uci.ics.hyracks.storage.am.btree.impls.MultiComparator;
 
 public class FieldPrefixCompressor implements IFrameCompressor {
-		
+	
 	// minimum ratio of uncompressed records to total records to consider re-compression
 	private float ratioThreshold;
 	
@@ -428,5 +428,5 @@
         public int compare(KeyPartition a, KeyPartition b) {
             return a.firstRecSlotNum - b.firstRecSlotNum;
         }
-    }	
+    }
 }
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/AbstractBTreeOperatorNodePushable.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/AbstractBTreeOperatorNodePushable.java
index 1949651..9dd430c 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/AbstractBTreeOperatorNodePushable.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/AbstractBTreeOperatorNodePushable.java
@@ -15,15 +15,21 @@
 
 package edu.uci.ics.hyracks.storage.am.btree.dataflow;
 
-import java.io.DataOutputStream;
+import java.io.DataOutput;
 import java.io.File;
 import java.io.RandomAccessFile;
+import java.nio.ByteBuffer;
 import java.util.Random;
 
 import edu.uci.ics.hyracks.api.comm.IFrameTupleAccessor;
 import edu.uci.ics.hyracks.api.context.IHyracksContext;
 import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparator;
-import edu.uci.ics.hyracks.dataflow.common.comm.io.ByteArrayAccessibleOutputStream;
+import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
+import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
+import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
+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.marshalling.IntegerSerializerDeserializer;
 import edu.uci.ics.hyracks.dataflow.std.base.AbstractUnaryOutputOperatorNodePushable;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeInteriorFrame;
@@ -37,7 +43,7 @@
 import edu.uci.ics.hyracks.storage.common.file.FileManager;
 
 public abstract class AbstractBTreeOperatorNodePushable extends AbstractUnaryOutputOperatorNodePushable {
-		
+	
 	protected IBTreeInteriorFrame interiorFrame;
 	protected IBTreeLeafFrame leafFrame;
 	
@@ -118,34 +124,50 @@
 	// debug
 	protected void fill() throws Exception {
 		
+		
+		// TODO: uncomment and fix
 		MetaDataFrame metaFrame = new MetaDataFrame();                		
 		btree.create(opDesc.getBtreeFileId(), leafFrame, metaFrame);
 		
 		Random rnd = new Random();
 		rnd.setSeed(50);				
 		
-		for (int i = 0; i < 10000; i++) {
-			ByteArrayAccessibleOutputStream baaos = new ByteArrayAccessibleOutputStream();
-			DataOutputStream dos = new DataOutputStream(baaos);        	        	
+		ByteBuffer frame = ctx.getResourceManager().allocateFrame();
+		FrameTupleAppender appender = new FrameTupleAppender(ctx);				
+		ArrayTupleBuilder tb = new ArrayTupleBuilder(2);
+		DataOutput dos = tb.getDataOutput();
 
+		ISerializerDeserializer[] recDescSers = { IntegerSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE};
+		RecordDescriptor recDesc = new RecordDescriptor(recDescSers);
+		IFrameTupleAccessor accessor = new FrameTupleAccessor(ctx, recDesc);
+		accessor.reset(frame);
+		FrameTupleReference tuple = new FrameTupleReference();
+		
+		for (int i = 0; i < 10000; i++) {			
 			int f0 = rnd.nextInt() % 10000;
 			int f1 = 5;
 
-			IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
-			IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
-
-			byte[] record = baaos.toByteArray();
-
+			tb.reset();
+        	IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+        	tb.addFieldEndOffset();
+        	IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+        	tb.addFieldEndOffset();        	
+        	        	
+        	appender.reset(frame, true);
+        	appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+        	
+        	tuple.reset(accessor, 0);
+						
 			if (i % 1000 == 0) {
 				System.out.println("INSERTING " + i + " : " + f0 + " " + f1);            	
 			}
 
 			try {                                
-				btree.insert(record, leafFrame, interiorFrame, metaFrame);
+				btree.insert(tuple, leafFrame, interiorFrame, metaFrame);
 			} catch (Exception e) {
 			}
 		}
-
+		
 		/*
         IFieldAccessor[] fields = new IFieldAccessor[2];
         fields[0] = new Int32Accessor(); // key field
@@ -217,5 +239,5 @@
 		}						
 		
 		return btreeRecord;
-	}
+	}	
 }
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeBulkLoadOperatorDescriptor.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeBulkLoadOperatorDescriptor.java
index 0df2581..446538b 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeBulkLoadOperatorDescriptor.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeBulkLoadOperatorDescriptor.java
@@ -31,9 +31,7 @@
 	
 	private static final long serialVersionUID = 1L;
 	
-	private final int[] keyFields;
-	private final int[] payloadFields;
-	
+	private int[] fieldPermutation;	
     private float fillFactor;
 	
 	public BTreeBulkLoadOperatorDescriptor(JobSpecification spec,
@@ -43,21 +41,19 @@
 			String btreeFileName, IBTreeInteriorFrameFactory interiorFactory,
 			IBTreeLeafFrameFactory leafFactory, IFieldAccessorFactory[] fieldAccessorFactories, 
 			IBinaryComparatorFactory[] comparatorFactories,			
-			int[] keyFields, int[] payloadFields, float fillFactor) {
+			int[] fieldPermutation, float fillFactor) {
 		super(spec, 1, 0, fileSplitProvider, recDesc, bufferCacheProvider,
 				btreeRegistryProvider, btreeFileId, btreeFileName, interiorFactory,
 				leafFactory, fieldAccessorFactories, comparatorFactories);
-		this.keyFields = keyFields;
-		this.payloadFields = payloadFields;
+		this.fieldPermutation = fieldPermutation;
 		this.fillFactor = fillFactor;
 	}
-	
+		
 	@Override
 	public IOperatorNodePushable createPushRuntime(IHyracksContext ctx,
 			IOperatorEnvironment env,
 			IRecordDescriptorProvider recordDescProvider, int partition,
 			int nPartitions) {
-		return new BTreeBulkLoadOperatorNodePushable(this, ctx, keyFields, payloadFields, fillFactor, recordDescProvider);
-	}
-	
+		return new BTreeBulkLoadOperatorNodePushable(this, ctx, fieldPermutation, fillFactor, recordDescProvider);
+	}	
 }
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeBulkLoadOperatorNodePushable.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeBulkLoadOperatorNodePushable.java
index 8376260..1b9db48 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeBulkLoadOperatorNodePushable.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeBulkLoadOperatorNodePushable.java
@@ -27,10 +27,7 @@
 import edu.uci.ics.hyracks.storage.am.btree.impls.BTree;
 
 public class BTreeBulkLoadOperatorNodePushable extends AbstractBTreeOperatorNodePushable {
-	
-	private final int[] keyFields;
-	private final int[] payloadFields;
-	
+		
     private float fillFactor;
     
     private FrameTupleAccessor accessor;
@@ -38,12 +35,13 @@
     
     private IRecordDescriptorProvider recordDescProvider;
     
-	public BTreeBulkLoadOperatorNodePushable(AbstractBTreeOperatorDescriptor opDesc, IHyracksContext ctx, int[] keyFields, int[] payloadFields, float fillFactor, IRecordDescriptorProvider recordDescProvider) {
+    private PermutingFrameTupleReference tuple = new PermutingFrameTupleReference();
+    
+	public BTreeBulkLoadOperatorNodePushable(AbstractBTreeOperatorDescriptor opDesc, IHyracksContext ctx, int[] fieldPermutation, float fillFactor, IRecordDescriptorProvider recordDescProvider) {
 		super(opDesc, ctx, true);
-		this.keyFields = keyFields;
-		this.payloadFields = payloadFields;
 		this.fillFactor = fillFactor;
 		this.recordDescProvider = recordDescProvider;
+		tuple.setFieldPermutation(fieldPermutation);
 	}
 	
 	@Override
@@ -61,9 +59,9 @@
 		                      
 		int tupleCount = accessor.getTupleCount();
 		for(int i = 0; i < tupleCount; i++) {
-			byte[] btreeRecord = buildBTreeRecordFromHyraxRecord(accessor, i, keyFields, payloadFields);
+			tuple.reset(accessor, i);			
 			try {
-				btree.bulkLoadAddRecord(bulkLoadCtx, btreeRecord);
+				btree.bulkLoadAddRecord(bulkLoadCtx, tuple);
 			} catch (Exception e) {
 				e.printStackTrace();
 			}
@@ -86,5 +84,5 @@
 
     @Override
     public void flush() throws HyracksDataException {
-    }	
+    }    
 }
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeInsertUpdateDeleteOperatorDescriptor.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeInsertUpdateDeleteOperatorDescriptor.java
index e88b633..e3da2c2 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeInsertUpdateDeleteOperatorDescriptor.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeInsertUpdateDeleteOperatorDescriptor.java
@@ -32,8 +32,7 @@
 	
 	private static final long serialVersionUID = 1L;
 	
-	private final int[] keyFields;
-	private final int[] payloadFields;
+	private final int[] fieldPermutation;
 	
 	private BTreeOp op;
 	
@@ -44,12 +43,11 @@
 			String btreeFileName, IBTreeInteriorFrameFactory interiorFactory,
 			IBTreeLeafFrameFactory leafFactory, IFieldAccessorFactory[] fieldAccessorFactories, 
 			IBinaryComparatorFactory[] comparatorFactories,			
-			int[] keyFields, int[] payloadFields, BTreeOp op) {
+			int[] fieldPermutation, BTreeOp op) {
 		super(spec, 1, 1, fileSplitProvider, recDesc, bufferCacheProvider,
 				btreeRegistryProvider, btreeFileId, btreeFileName, interiorFactory,
 				leafFactory, fieldAccessorFactories, comparatorFactories);
-		this.keyFields = keyFields;
-		this.payloadFields = payloadFields;
+		this.fieldPermutation = fieldPermutation;		
 		this.op = op;
 	}
 	
@@ -58,7 +56,6 @@
 			IOperatorEnvironment env,
 			IRecordDescriptorProvider recordDescProvider, int partition,
 			int nPartitions) {
-		return new BTreeInsertUpdateDeleteOperatorNodePushable(this, ctx, keyFields, payloadFields, recordDescProvider, op);
-	}
-	
+		return new BTreeInsertUpdateDeleteOperatorNodePushable(this, ctx, fieldPermutation, recordDescProvider, op);
+	}	
 }
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeInsertUpdateDeleteOperatorNodePushable.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeInsertUpdateDeleteOperatorNodePushable.java
index f578538..09668fe 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeInsertUpdateDeleteOperatorNodePushable.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeInsertUpdateDeleteOperatorNodePushable.java
@@ -28,10 +28,7 @@
 import edu.uci.ics.hyracks.storage.am.btree.impls.BTreeOp;
 
 public class BTreeInsertUpdateDeleteOperatorNodePushable extends AbstractBTreeOperatorNodePushable {
-	
-	private final int[] keyFields;
-	private final int[] payloadFields;
-	    
+		
     private FrameTupleAccessor accessor;
     
     private IRecordDescriptorProvider recordDescProvider;
@@ -40,12 +37,13 @@
     
     private BTreeOp op;
     
-	public BTreeInsertUpdateDeleteOperatorNodePushable(AbstractBTreeOperatorDescriptor opDesc, IHyracksContext ctx, int[] keyFields, int[] payloadFields, IRecordDescriptorProvider recordDescProvider, BTreeOp op) {
-		super(opDesc, ctx, false);
-		this.keyFields = keyFields;
-		this.payloadFields = payloadFields;
+    private PermutingFrameTupleReference tuple = new PermutingFrameTupleReference();
+    
+	public BTreeInsertUpdateDeleteOperatorNodePushable(AbstractBTreeOperatorDescriptor opDesc, IHyracksContext ctx, int[] fieldPermutation, IRecordDescriptorProvider recordDescProvider, BTreeOp op) {
+		super(opDesc, ctx, false);		
 		this.recordDescProvider = recordDescProvider;
 		this.op = op;
+		tuple.setFieldPermutation(fieldPermutation);
 	}
 	
 	@Override
@@ -59,17 +57,17 @@
 		
 		int tupleCount = accessor.getTupleCount();
 		for(int i = 0; i < tupleCount; i++) {
-			byte[] btreeRecord = buildBTreeRecordFromHyraxRecord(accessor, i, keyFields, payloadFields);
+			tuple.reset(accessor, i);
 			try {
 				
 				switch(op) {
 				
 				case BTO_INSERT: {
-					btree.insert(btreeRecord, leafFrame, interiorFrame, metaFrame);				
+					btree.insert(tuple, leafFrame, interiorFrame, metaFrame);				
 				} break;
 				
 				case BTO_DELETE: {
-					btree.delete(btreeRecord, leafFrame, interiorFrame, metaFrame);				
+					btree.delete(tuple, leafFrame, interiorFrame, metaFrame);				
 				} break;
 				
 				default: {
@@ -102,5 +100,5 @@
 
     @Override
     public void flush() throws HyracksDataException {
-    }	
+    }    
 }
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeSearchOperatorDescriptor.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeSearchOperatorDescriptor.java
index ce5795b..c0b246f 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeSearchOperatorDescriptor.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeSearchOperatorDescriptor.java
@@ -22,21 +22,22 @@
 import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
 import edu.uci.ics.hyracks.api.job.IOperatorEnvironment;
 import edu.uci.ics.hyracks.api.job.JobSpecification;
+import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
 import edu.uci.ics.hyracks.dataflow.std.file.IFileSplitProvider;
 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.IFieldAccessorFactory;
 
 public class BTreeSearchOperatorDescriptor extends AbstractBTreeOperatorDescriptor {
-
+	
 	private static final long serialVersionUID = 1L;
 
 	private boolean isForward;
-	private byte[] lowKey;
-	private byte[] highKey;
+	private ITupleReference lowKey;
+	private ITupleReference highKey;
 	private int searchKeyFields;
 	
-	public BTreeSearchOperatorDescriptor(JobSpecification spec, IFileSplitProvider fileSplitProvider, RecordDescriptor recDesc, IBufferCacheProvider bufferCacheProvider, IBTreeRegistryProvider btreeRegistryProvider,  int btreeFileId, String btreeFileName, IBTreeInteriorFrameFactory interiorFactory, IBTreeLeafFrameFactory leafFactory, IFieldAccessorFactory[] fieldAccessorFactories, IBinaryComparatorFactory[] comparatorFactories, boolean isForward, byte[] lowKey, byte[] highKey, int searchKeyFields) {
+	public BTreeSearchOperatorDescriptor(JobSpecification spec, IFileSplitProvider fileSplitProvider, RecordDescriptor recDesc, IBufferCacheProvider bufferCacheProvider, IBTreeRegistryProvider btreeRegistryProvider,  int btreeFileId, String btreeFileName, IBTreeInteriorFrameFactory interiorFactory, IBTreeLeafFrameFactory leafFactory, IFieldAccessorFactory[] fieldAccessorFactories, IBinaryComparatorFactory[] comparatorFactories, boolean isForward, ITupleReference lowKey, ITupleReference highKey, int searchKeyFields) {
 		super(spec, 0, 1, fileSplitProvider, recDesc, bufferCacheProvider, btreeRegistryProvider, btreeFileId, btreeFileName, interiorFactory, leafFactory, fieldAccessorFactories, comparatorFactories);
 		this.isForward = isForward;
 		this.lowKey = lowKey;
@@ -48,5 +49,5 @@
 	public IOperatorNodePushable createPushRuntime(final IHyracksContext ctx, final IOperatorEnvironment env,
 			IRecordDescriptorProvider recordDescProvider, int partition, int nPartitions) {
 		return new BTreeSearchOperatorNodePushable(this, ctx, isForward, lowKey, highKey, searchKeyFields);
-	}	
+	}
 }
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeSearchOperatorNodePushable.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeSearchOperatorNodePushable.java
index 67a4ff2..3bd3e36 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeSearchOperatorNodePushable.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeSearchOperatorNodePushable.java
@@ -25,6 +25,7 @@
 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.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;
@@ -33,13 +34,13 @@
 import edu.uci.ics.hyracks.storage.am.btree.impls.RangeSearchCursor;
 
 public class BTreeSearchOperatorNodePushable extends AbstractBTreeOperatorNodePushable {
-	
+
 	private boolean isForward;
-	private byte[] lowKey;
-	private byte[] highKey;
+	private ITupleReference lowKey;
+	private ITupleReference highKey;
 	private int searchKeyFields;
 	
-	public BTreeSearchOperatorNodePushable(AbstractBTreeOperatorDescriptor opDesc, IHyracksContext ctx, boolean isForward, byte[] lowKey, byte[] highKey, int searchKeyFields) {
+	public BTreeSearchOperatorNodePushable(AbstractBTreeOperatorDescriptor opDesc, IHyracksContext ctx, boolean isForward, ITupleReference lowKey, ITupleReference highKey, int searchKeyFields) {
 		super(opDesc, ctx, false);
 		this.isForward = isForward;
 		this.lowKey = lowKey;
@@ -126,5 +127,5 @@
 
     @Override
     public void flush() throws HyracksDataException {
-    }			
+    }
 }
\ No newline at end of file
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/PermutingFrameTupleReference.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/PermutingFrameTupleReference.java
new file mode 100644
index 0000000..db74a9c
--- /dev/null
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/PermutingFrameTupleReference.java
@@ -0,0 +1,49 @@
+package edu.uci.ics.hyracks.storage.am.btree.dataflow;
+
+import edu.uci.ics.hyracks.api.comm.IFrameTupleAccessor;
+import edu.uci.ics.hyracks.dataflow.common.data.accessors.IFrameTupleReference;
+
+public class PermutingFrameTupleReference implements IFrameTupleReference {
+	private IFrameTupleAccessor fta;
+    private int tIndex;
+    private int[] fieldPermutation;
+    
+    public void setFieldPermutation(int[] fieldPermutation) {
+    	this.fieldPermutation = fieldPermutation;
+    }
+    
+    public void reset(IFrameTupleAccessor fta, int tIndex) {
+        this.fta = fta;
+        this.tIndex = tIndex;
+    }
+
+    @Override
+    public IFrameTupleAccessor getFrameTupleAccessor() {
+        return fta;
+    }
+
+    @Override
+    public int getTupleIndex() {
+        return tIndex;
+    }
+
+    @Override
+    public int getFieldCount() {
+        return fieldPermutation.length;
+    }
+
+    @Override
+    public byte[] getFieldData(int fIdx) {
+        return fta.getBuffer().array();
+    }
+
+    @Override
+    public int getFieldStart(int fIdx) {    	    	
+    	return fta.getTupleStartOffset(tIndex) + fta.getFieldSlotsLength() + fta.getFieldStartOffset(tIndex, fieldPermutation[fIdx]);
+    }
+
+    @Override
+    public int getFieldLength(int fIdx) {
+        return fta.getFieldLength(tIndex, fieldPermutation[fIdx]);
+    }
+}
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/FieldPrefixNSMLeafFrame.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/FieldPrefixNSMLeafFrame.java
index 96884e4..6c05eb6 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/FieldPrefixNSMLeafFrame.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/FieldPrefixNSMLeafFrame.java
@@ -19,7 +19,7 @@
 import java.util.ArrayList;
 import java.util.Collections;
 
-import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparator;
+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;
@@ -30,7 +30,9 @@
 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.FieldPrefixSlotManager;
+import edu.uci.ics.hyracks.storage.am.btree.impls.FieldPrefixTuple;
 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.SpaceStatus;
@@ -56,6 +58,9 @@
     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(); 
+    
     public FieldPrefixNSMLeafFrame() {
         this.slotManager = new FieldPrefixSlotManager();
         this.compressor = new FieldPrefixCompressor(0.001f, 2);        
@@ -152,8 +157,8 @@
     }
     
     @Override
-    public void delete(byte[] data, MultiComparator cmp, boolean exactDelete) throws Exception {        
-        int slot = slotManager.findSlot(data, cmp, true);
+    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) {
             throw new BTreeException("Key to be deleted does not exist.");   
@@ -162,25 +167,15 @@
             int prefixSlotNum = slotManager.decodeFirstSlotField(slot);            
             int recSlotOff = slotManager.getRecSlotOff(recSlotNum);
             
-            if(exactDelete) {                    
-                IBinaryComparator[] cmps = cmp.getComparators();
-                IFieldAccessor[] fields = cmp.getFields();
-                FieldPrefixFieldIterator fieldIter = new FieldPrefixFieldIterator(fields, this);                    
-                fieldIter.openRecSlotOff(recSlotOff);
+            if(exactDelete) {                                    
+                pageTuple.setFields(cmp.getFields());
+                pageTuple.setFrame(this);
+                pageTuple.openRecSlotNum(recSlotNum);
                 
-                int dataRunner = 0;
-                for(int i = 0; i < cmp.getKeyLength(); i++) {
-                    dataRunner += fields[i].getLength(data, dataRunner);
-                    fieldIter.nextField();                   
-                }                
-                for(int i = cmp.getKeyLength(); i < fields.length; i++) {
-                    int dataFieldLen = fields[i].getLength(data, dataRunner);
-                	if(cmps[i].compare(data, dataRunner, dataFieldLen, buf.array(), fieldIter.getFieldOff(), fieldIter.getFieldSize()) != 0) {
-                        throw new BTreeException("Cannot delete record. Byte-by-byte comparison failed to prove equality.");
-                    }
-                    dataRunner += dataFieldLen;
-                    fieldIter.nextField();    
-                }                                
+                int comparison = cmp.fieldRangeCompare(tuple, pageTuple, 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.");
+                }                                  
             }
             
             // perform deletion (we just do a memcpy to overwrite the slot)
@@ -216,35 +211,56 @@
     }
     
     @Override
-    public SpaceStatus hasSpaceInsert(byte[] data, MultiComparator cmp) {                    	
+    public SpaceStatus hasSpaceInsert(ITupleReference tuple, MultiComparator cmp) {                    	
     	int freeContiguous = buf.capacity() - buf.getInt(freeSpaceOff) - ((buf.getInt(numRecordsOff) + buf.getInt(numPrefixRecordsOff)) * slotManager.getSlotSize());     	    	                        
         
+    	int tupleSpace = spaceNeededForTuple(tuple);
+    	
         // see if the record would fit uncompressed
-        if(data.length + slotManager.getSlotSize() <= freeContiguous) return SpaceStatus.SUFFICIENT_CONTIGUOUS_SPACE;
+        if(tupleSpace + slotManager.getSlotSize() <= freeContiguous) return SpaceStatus.SUFFICIENT_CONTIGUOUS_SPACE;
         
         // see if record would fit into remaining space after compaction
-        if(data.length + slotManager.getSlotSize() <= buf.getInt(totalFreeSpaceOff)) return SpaceStatus.SUFFICIENT_SPACE;
+        if(tupleSpace + 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(data, cmp);
+        int prefixSlotNum = slotManager.findPrefix(tuple, cmp);
         if(prefixSlotNum != FieldPrefixSlotManager.RECORD_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(data, recRunner);
+        		recRunner += cmp.getFields()[i].getLength(tuple.getFieldData(i), recRunner);
         	}
-        	int compressedSize = data.length - recRunner;
+        	int compressedSize = tupleSpace - recRunner;
         	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, byte[] data, MultiComparator cmp) {
+    public SpaceStatus hasSpaceUpdate(int rid, ITupleReference tuple, MultiComparator cmp) {
         // TODO Auto-generated method stub
         return SpaceStatus.INSUFFICIENT_SPACE;
     }
@@ -276,29 +292,31 @@
     }
     
     @Override
-    public void insert(byte[] data, MultiComparator cmp) throws Exception {    	
-    	int slot = slotManager.findSlot(data, cmp, false);        
+    public void insert(ITupleReference tuple, MultiComparator cmp) throws Exception {    	
+    	int slot = slotManager.findSlot(tuple, cmp, false);        
         slot = slotManager.insertSlot(slot, buf.getInt(freeSpaceOff));
         
-        int suffixSize = data.length;
-        int suffixStartOff = 0;
+        int suffixSize = spaceNeededForTuple(tuple);
+        int suffixStartOff = tuple.getFieldStart(0);
         int prefixSlotNum = slotManager.decodeFirstSlotField(slot);
         
         if(prefixSlotNum != FieldPrefixSlotManager.RECORD_UNCOMPRESSED) {
         	
+        	// TODO relies on length being stored in serialized field
+        	
             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(data, suffixStartOff);
+                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(data, suffixSize);
+                suffixSize += cmp.getFields()[i].getLength(tuple.getFieldData(i), suffixSize);
             }
             suffixSize -= suffixStartOff;                  
         }
@@ -306,8 +324,9 @@
         	buf.putInt(numUncompressedRecordsOff, buf.getInt(numUncompressedRecordsOff) + 1);
         }
         
+    	// TODO relies on length being stored in serialized field        
         int freeSpace = buf.getInt(freeSpaceOff);
-        System.arraycopy(data, suffixStartOff, buf.array(), freeSpace, suffixSize);                    
+        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());
@@ -340,7 +359,7 @@
     }
     
     @Override
-    public void update(int rid, byte[] data) throws Exception {
+    public void update(int rid, ITupleReference tuple) throws Exception {
         // TODO Auto-generated method stub
         
     }   
@@ -437,28 +456,37 @@
 	}
 
     @Override
-    public void insertSorted(byte[] data, MultiComparator cmp) throws Exception {    	
+    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);
-			int prefixRecOff = slotManager.decodeSecondSlotField(prefixSlot);
 			
-			if(cmp.fieldRangeCompare(data, 0, buf.array(), prefixRecOff, 0, numPrefixFields) == 0) {
+			pagePrefixTuple.setFrame(this);
+			pagePrefixTuple.setFields(cmp.getFields());
+			pagePrefixTuple.openPrefixSlotOff(prefixSlotOff);
+			
+			if(cmp.fieldRangeCompare(tuple, pagePrefixTuple, 0, numPrefixFields) == 0) {
 				fieldsToTruncate = numPrefixFields;													
 			}			
 		}
 				
 		// copy truncated record
-		int recStart = 0;		
+    	// 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(data, recStart);
+			recStart += cmp.getFields()[i].getLength(tuple.getFieldData(0), recStart);
 		}
-		int recLen = data.length - recStart;
-		System.arraycopy(data, recStart, buf.array(), freeSpace, recLen);
+		int recLen = tupleSpace - recStart - tuple.getFieldStart(0);
+		System.arraycopy(tuple.getFieldData(0), recStart, buf.array(), freeSpace, recLen);
 		
 		// insert slot
 		int prefixSlotNum = FieldPrefixSlotManager.RECORD_UNCOMPRESSED;
@@ -474,16 +502,19 @@
     }
     
     @Override
-    public int split(IBTreeFrame rightFrame, byte[] data, MultiComparator cmp, SplitKey splitKey) throws Exception {
-    	    	
+    public int split(IBTreeFrame rightFrame, ITupleReference tuple, MultiComparator cmp, SplitKey splitKey) throws Exception {
+    	
     	FieldPrefixNSMLeafFrame rf = (FieldPrefixNSMLeafFrame)rightFrame;
     	
+    	pageTuple.setFrame(this);
+		pageTuple.setFields(cmp.getFields());
+    	
     	// before doing anything check if key already exists
-		int slot = slotManager.findSlot(data, cmp, true);
-		int recSlotNum = slotManager.decodeSecondSlotField(slot);	
-		if(recSlotNum != FieldPrefixSlotManager.GREATEST_SLOT) {
-			int prefixSlotNum = slotManager.decodeFirstSlotField(slot);				
-			if(slotManager.compareCompressed(data, buf.array(), prefixSlotNum, recSlotNum, cmp) == 0) {			
+		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) {			
 				throw new BTreeException("Inserting duplicate key into unique index");
 			}
 		}
@@ -498,15 +529,9 @@
 		int midSlotOff = slotManager.getRecSlotOff(midSlotNum);
 		int midSlot = buf.getInt(midSlotOff);
 		int midPrefixSlotNum = slotManager.decodeFirstSlotField(midSlot);
-		int midRecOff = slotManager.decodeSecondSlotField(midSlot);
-		int comparison = 0;
-		if(midPrefixSlotNum == FieldPrefixSlotManager.RECORD_UNCOMPRESSED) {
-			comparison = cmp.compare(data, 0, buf.array(), midRecOff);
-		}		
-		else {			
-			comparison = slotManager.compareCompressed(data, buf.array(), midPrefixSlotNum, midSlotNum, cmp);
-		}
-				
+		int midRecOff = slotManager.decodeSecondSlotField(midSlot);		
+		pageTuple.openRecSlotNum(midSlotNum);
+		int comparison = cmp.compare(tuple, (ITupleReference)pageTuple);		
 		if (comparison >= 0) {
 			recordsToLeft = midSlotNum + (numRecords % 2);
 			targetFrame = rf;
@@ -608,7 +633,9 @@
 		rightFrame.compact(cmp);
 		
 		// insert last key
-		targetFrame.insert(data, cmp);			
+		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();		
@@ -621,7 +648,7 @@
 			fieldIter.nextField();
 		}									
 		splitKey.initData(keySize);
-		fieldIter.copyFields(0, cmp.getKeyLength()-1, splitKey.getData(), 0);
+		fieldIter.copyFields(0, cmp.getKeyLength()-1, splitKey.getBuffer().array(), 0);
 		
 		return 0;
     }
@@ -677,4 +704,9 @@
     public IFieldIterator createFieldIterator() {
     	return new FieldPrefixFieldIterator();
     }
+		
+	@Override
+	public void setPageTupleFields(IFieldAccessor[] fields) {
+		pageTuple.setFields(fields);
+	}		
 }
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMFrame.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMFrame.java
index cca5268..5256766 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMFrame.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMFrame.java
@@ -19,13 +19,16 @@
 import java.util.ArrayList;
 import java.util.Collections;
 
+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.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.SpaceStatus;
 import edu.uci.ics.hyracks.storage.common.buffercache.ICachedPage;
@@ -43,6 +46,8 @@
 	protected ByteBuffer buf = null;
 	protected ISlotManager slotManager;
 	
+	protected SelfDescTupleReference pageTuple = new SelfDescTupleReference();
+	
 	public NSMFrame() {
 		this.slotManager = new OrderedSlotManager();
 	}
@@ -96,7 +101,7 @@
 	public void setPage(ICachedPage page) {
 		this.page = page;
 		this.buf = page.getBuffer();
-		slotManager.setFrame(this);
+		slotManager.setFrame(this);		
 	}
 	
 	@Override
@@ -139,24 +144,22 @@
 	}
 
 	@Override
-	public void delete(byte[] data, MultiComparator cmp, boolean exactDelete) throws Exception {		
-		int slotOff = slotManager.findSlot(data, cmp, true);
+	public void delete(ITupleReference tuple, MultiComparator cmp, boolean exactDelete) throws Exception {		
+		int slotOff = slotManager.findSlot(tuple, 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);
-				int storedRecSize = cmp.getRecordSize(buf.array(), storedRecOff);						
-				if(data.length != storedRecSize) {
-					throw new BTreeException("Cannot delete record. Given record size does not match candidate record.");
-				}
-				for(int i = 0; i < storedRecSize; i++) {
-					if(data[i] != buf.array()[storedRecOff+i]) {
-						throw new BTreeException("Cannot delete record. Byte-by-byte comparison failed to prove equality.");
-					}				
-				}
+				int storedRecOff = slotManager.getRecOff(slotOff);			
+				pageTuple.setFields(cmp.getFields());
+				pageTuple.reset(buf, storedRecOff);
+				
+				int comparison = cmp.fieldRangeCompare(tuple, pageTuple, 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.");
+                }										
 			}
 			
 			int recOff = slotManager.getRecOff(slotOff);
@@ -174,14 +177,15 @@
 	}
 	
 	@Override
-	public SpaceStatus hasSpaceInsert(byte[] data, MultiComparator cmp) {				
-		if(data.length + slotManager.getSlotSize() <= buf.capacity() - buf.getInt(freeSpaceOff) - (buf.getInt(numRecordsOff) * slotManager.getSlotSize()) ) return SpaceStatus.SUFFICIENT_CONTIGUOUS_SPACE;
-		else if(data.length + slotManager.getSlotSize() <= buf.getInt(totalFreeSpaceOff)) return SpaceStatus.SUFFICIENT_SPACE;
+	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;
 		else return SpaceStatus.INSUFFICIENT_SPACE;
 	}
 	
 	@Override
-	public SpaceStatus hasSpaceUpdate(int rid, byte[] data, MultiComparator cmp) {
+	public SpaceStatus hasSpaceUpdate(int rid, ITupleReference tuple, MultiComparator cmp) {
 		// TODO Auto-generated method stub
 		return SpaceStatus.INSUFFICIENT_SPACE;
 	}
@@ -192,20 +196,36 @@
 	}
 	
 	@Override
-	public void insert(byte[] data, MultiComparator cmp) throws Exception {
-		int slotOff = slotManager.findSlot(data, cmp, false);
-		slotOff = slotManager.insertSlot(slotOff, buf.getInt(freeSpaceOff));
+	public void insert(ITupleReference tuple, MultiComparator cmp) throws Exception {
+		int slotOff = slotManager.findSlot(tuple, cmp, false);		
+		slotOff = slotManager.insertSlot(slotOff, buf.getInt(freeSpaceOff));				
+		int bytesWritten = writeTupleFields(tuple, tuple.getFieldCount(), buf, buf.getInt(freeSpaceOff));
 		
-		int recOff = buf.getInt(freeSpaceOff);
-		System.arraycopy(data, 0, buf.array(), recOff, data.length);		
-					
 		buf.putInt(numRecordsOff, buf.getInt(numRecordsOff) + 1);
-		buf.putInt(freeSpaceOff, buf.getInt(freeSpaceOff) + data.length);
-		buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) - data.length - slotManager.getSlotSize());				
+		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, byte[] data) throws Exception {
+	public void update(int rid, ITupleReference tuple) throws Exception {
 		// TODO Auto-generated method stub
 		
 	}	
@@ -275,4 +295,9 @@
     public IFieldIterator createFieldIterator() {
     	return new NSMFieldIterator();
     }
+    
+    @Override
+	public void setPageTupleFields(IFieldAccessor[] fields) {
+		pageTuple.setFields(fields);
+	}
 }
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMInteriorFrame.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMInteriorFrame.java
index a3af5b2..51dece6 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMInteriorFrame.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMInteriorFrame.java
@@ -19,6 +19,7 @@
 import java.util.ArrayList;
 import java.util.Collections;
 
+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.IBTreeInteriorFrame;
 import edu.uci.ics.hyracks.storage.am.btree.impls.BTreeException;
@@ -51,13 +52,17 @@
 	}
 	
 	@Override
-	public void insert(byte[] data, MultiComparator cmp) throws Exception {
+	public void insert(ITupleReference tuple, MultiComparator cmp) throws Exception {
 		
-		int slotOff = slotManager.findSlot(data, cmp, false);
-		boolean isDuplicate = true;
+		int slotOff = slotManager.findSlot(tuple, cmp, false);
+		boolean isDuplicate = true;				
 		
 		if(slotOff < 0) isDuplicate = false; // greater than all existing keys
-		else if(cmp.compare(data, 0, buf.array(), slotManager.getRecOff(slotOff)) != 0) isDuplicate = false;
+		else {			
+			pageTuple.setFields(cmp.getFields());
+			pageTuple.reset(buf, slotManager.getRecOff(slotOff));
+			if(cmp.compare(tuple, pageTuple) != 0) isDuplicate = false;
+		}
 		
 		if(isDuplicate) {			
 			throw new BTreeException("Trying to insert duplicate value into interior node.");
@@ -65,9 +70,9 @@
 		else {			
 			slotOff = slotManager.insertSlot(slotOff, buf.getInt(freeSpaceOff));			
 						
-			int recOff = buf.getInt(freeSpaceOff);
-			int recSize = cmp.getKeySize(data, 0) + childPtrSize;
-			System.arraycopy(data, 0, buf.array(), recOff, recSize);		
+			// 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);
@@ -75,7 +80,7 @@
 			
 			// did insert into the rightmost slot?
 			if(slotOff == slotManager.getSlotEndOff()) { 
-				System.arraycopy(data, recSize, buf.array(), rightLeafOff, childPtrSize);
+				System.arraycopy(tuple.getFieldData(tuple.getFieldCount()-1), tuple.getFieldStart(tuple.getFieldCount()-1), buf.array(), rightLeafOff, childPtrSize);
 			}
 			else {
 				// if slotOff has a right (slot-)neighbor then update its child pointer
@@ -83,9 +88,9 @@
 				// (or when the splitkey goes into the rightmost slot but that case was handled in the if above)
 				
 				if(buf.getInt(numRecordsOff) > 1) {					
-					int rightNeighborOff = slotOff - slotManager.getSlotSize();
-					int keySize = cmp.getKeySize(buf.array(), slotManager.getRecOff(rightNeighborOff));
-					System.arraycopy(data, recSize, buf.array(), slotManager.getRecOff(rightNeighborOff) + keySize, childPtrSize);
+					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);
 				}				
 			}			
 		}
@@ -95,23 +100,23 @@
 	}
 	
 	@Override
-	public void insertSorted(byte[] data, MultiComparator cmp) throws Exception {
-		int recOff = buf.getInt(freeSpaceOff);
-		slotManager.insertSlot(-1, buf.getInt(freeSpaceOff));
-		int recSize = cmp.getKeySize(data, 0) + childPtrSize;
-		System.arraycopy(data, 0, buf.array(), recOff, recSize);
+	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(data, recSize, buf.array(), rightLeafOff, childPtrSize);
+		System.arraycopy(tuple.getFieldData(tuple.getFieldCount()-1), tuple.getFieldStart(tuple.getFieldCount()-1), buf.array(), rightLeafOff, childPtrSize);
 	}
 		
 	@Override
-	public int split(IBTreeFrame rightFrame, byte[] data, MultiComparator cmp, SplitKey splitKey) throws Exception {		
+	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(data, cmp, true);
-		if(slotOff >= 0) {
-			if(cmp.compare(data, 0, buf.array(), slotManager.getRecOff(slotOff)) == 0) {				
+		int slotOff = slotManager.findSlot(tuple, cmp, true);
+		if(slotOff >= 0) {			
+			pageTuple.reset(buf, slotManager.getRecOff(slotOff));			
+			if(cmp.compare(tuple, pageTuple) == 0) {				
 				throw new BTreeException("Inserting duplicate key in interior node during split");				
 			}
 		}
@@ -121,7 +126,8 @@
 		
 		int recordsToLeft = (numRecords / 2) + (numRecords % 2);
 		IBTreeFrame targetFrame = null;
-		if(cmp.compare(data, 0, buf.array(), getRecordOffset(recordsToLeft-1)) <= 0) {
+		pageTuple.reset(buf, getRecordOffset(recordsToLeft-1));
+		if(cmp.compare(tuple, pageTuple) <= 0) {
 			targetFrame = this;
 		}
 		else {
@@ -143,14 +149,14 @@
 		buf.putInt(numRecordsOff, recordsToLeft);
 				
 		// 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())
-		byte[] savedData = new byte[data.length];
-		System.arraycopy(data, 0, savedData, 0, data.length);
+		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);
 		splitKey.initData(splitKeySize);
-		System.arraycopy(buf.array(), recOff, splitKey.getData(), 0, splitKeySize);
+		pageTuple.reset(buf, recOff);
+		writeTupleFields((ITupleReference)pageTuple, cmp.getKeyLength(), splitKey.getBuffer(), 0);		
 		
 		int deleteRecOff = slotManager.getRecOff(slotManager.getSlotEndOff());
 		int deleteKeySize = cmp.getKeySize(buf.array(), deleteRecOff); 		
@@ -160,9 +166,9 @@
 		// compact both pages
 		rightFrame.compact(cmp);
 		compact(cmp);
-			
+		
 		// insert key
-		targetFrame.insert(savedData, cmp);
+		targetFrame.insert(savedSplitKey.getTuple(), cmp);
 			
 		return 0;
 	}	
@@ -204,22 +210,23 @@
 		}
 		
 		// check for trivial cases where no low key or high key exists (e.g. during an index scan)
-		byte[] data = null;
+		ITupleReference tuple = null;
 		if(pred.isForward()) {						
-			data = pred.getLowKeys();
-			if(data == null) {
+			tuple = pred.getLowKey();
+			if(tuple == null) {
 				return getLeftmostChildPageId(srcCmp);
 			}			
 		}
 		else {
-			data = pred.getHighKeys();
-			if(data == null) {
+			tuple = pred.getHighKey();
+			if(tuple == null) {
 				return getRightmostChildPageId(srcCmp);				
 			}								
 		}
 		
+		// TODO: need to modify this procedure to use TupleReference
 		MultiComparator targetCmp = pred.getComparator();
-		int slotOff = slotManager.findSlot(data, targetCmp, false);
+		int slotOff = slotManager.findSlot(tuple, targetCmp, false);
 		if(slotOff < 0) {
 			return buf.getInt(rightLeafOff);
 		}
@@ -253,8 +260,8 @@
 	}	
 	
 	@Override
-	public void delete(byte[] data, MultiComparator cmp, boolean exactDelete) throws Exception {
-		int slotOff = slotManager.findSlot(data, cmp, false);
+	public void delete(ITupleReference tuple, MultiComparator cmp, boolean exactDelete) throws Exception {
+		int slotOff = slotManager.findSlot(tuple, cmp, false);
 		int recOff;
 		int keySize;
 		
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMLeafFrame.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMLeafFrame.java
index c41106d..f548704 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMLeafFrame.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMLeafFrame.java
@@ -17,6 +17,7 @@
 
 import java.nio.ByteBuffer;
 
+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.impls.BTreeException;
@@ -59,45 +60,52 @@
 	}
 
 	@Override
-	public void insert(byte[] data, MultiComparator cmp) throws Exception {		
-		int slotOff = slotManager.findSlot(data, cmp, false);
+	public void insert(ITupleReference tuple, MultiComparator cmp) throws Exception {		
+		int slotOff = slotManager.findSlot(tuple, cmp, false);
 		boolean isDuplicate = true;
-		
+				
 		if (slotOff < 0) isDuplicate = false; // greater than all existing keys
-		else if (cmp.compare(data, 0, buf.array(), slotManager.getRecOff(slotOff)) != 0) isDuplicate = false;
-
+		else {
+			pageTuple.setFields(cmp.getFields());
+			pageTuple.reset(buf, slotManager.getRecOff(slotOff));					
+			if (cmp.compare(tuple, pageTuple) != 0) isDuplicate = false;
+		}
+				
 		if (isDuplicate) {
 			throw new BTreeException("Trying to insert duplicate value into leaf of unique index");
 		} 
 		else {
 			slotOff = slotManager.insertSlot(slotOff, buf.getInt(freeSpaceOff));
 			
-			int recOff = buf.getInt(freeSpaceOff);
-			System.arraycopy(data, 0, buf.array(), recOff, data.length);
-
+			int freeSpace = buf.getInt(freeSpaceOff);
+			int bytesWritten = writeTupleFields(tuple, tuple.getFieldCount(), buf, freeSpace);			
+			
 			buf.putInt(numRecordsOff, buf.getInt(numRecordsOff) + 1);
-			buf.putInt(freeSpaceOff, buf.getInt(freeSpaceOff) + data.length);
-			buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) - data.length - slotManager.getSlotSize());
+			buf.putInt(freeSpaceOff, buf.getInt(freeSpaceOff) + bytesWritten);
+			buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) - bytesWritten - slotManager.getSlotSize());
 		}	
 	}
 	
 	@Override
-	public void insertSorted(byte[] data, MultiComparator cmp) throws Exception {		
-		int recOff = buf.getInt(freeSpaceOff);
-		slotManager.insertSlot(-1, recOff);
-		System.arraycopy(data, 0, buf.array(), recOff, data.length);
+	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);
-		buf.putInt(freeSpaceOff, buf.getInt(freeSpaceOff) + data.length);
-		buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) - data.length - slotManager.getSlotSize());
+		buf.putInt(freeSpaceOff, buf.getInt(freeSpaceOff) + bytesWritten);
+		buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) - bytesWritten - slotManager.getSlotSize());
 	}
 	
 	@Override
-	public int split(IBTreeFrame rightFrame, byte[] data, MultiComparator cmp, SplitKey splitKey) throws Exception {
+	public int split(IBTreeFrame rightFrame, ITupleReference tuple, MultiComparator cmp, SplitKey splitKey) throws Exception {
+		
+		pageTuple.setFields(cmp.getFields());
 		
 		// before doing anything check if key already exists
-		int slotOff = slotManager.findSlot(data, cmp, true);
-		if (slotOff >= 0) {			
-			if (cmp.compare(data, 0, buf.array(), slotManager.getRecOff(slotOff)) == 0) {
+		int slotOff = slotManager.findSlot(tuple, cmp, true);
+		if (slotOff >= 0) {						
+			pageTuple.reset(buf, slotManager.getRecOff(slotOff));		
+			if (cmp.compare(tuple, pageTuple) == 0) {
 				throw new BTreeException("Inserting duplicate key into unique index");
 			}
 		}
@@ -108,7 +116,9 @@
 		int recordsToLeft;
 		int mid = numRecords / 2;
 		IBTreeFrame targetFrame = null;
-		if ((cmp.compare(data, 0, buf.array(), slotManager.getRecOff(slotManager.getSlotEndOff() + slotManager.getSlotSize() * mid))) >= 0) {
+		int recOff = slotManager.getRecOff(slotManager.getSlotEndOff() + slotManager.getSlotSize() * mid);
+		pageTuple.reset(buf, recOff);		
+		if (cmp.compare(tuple, pageTuple) >= 0) {
 			recordsToLeft = mid + (numRecords % 2);
 			targetFrame = rightFrame;
 		} else {
@@ -135,14 +145,15 @@
 		compact(cmp);
 		
 		// insert last key
-		targetFrame.insert(data, cmp);			
+		targetFrame.insert(tuple, cmp);	
 		
 		// set split key to be highest value in left page
-		int recOff = slotManager.getRecOff(slotManager.getSlotEndOff());
-		int keySize = cmp.getKeySize(buf.array(), recOff);				
+		recOff = slotManager.getRecOff(slotManager.getSlotEndOff());
+		int keySize = cmp.getKeySize(buf.array(), recOff);		
 		splitKey.initData(keySize);
-		System.arraycopy(buf.array(), recOff, splitKey.getData(), 0, keySize);
-			
+		pageTuple.reset(buf, recOff);
+		writeTupleFields((ITupleReference)pageTuple, cmp.getKeyLength(), splitKey.getBuffer(), 0);
+		
 		return 0;
 	}
 
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/BTree.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/BTree.java
index 42d375c..6386734 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/BTree.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/BTree.java
@@ -15,12 +15,14 @@
 
 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;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 
 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.IBTreeCursor;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeFrame;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeInteriorFrame;
@@ -28,7 +30,9 @@
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeLeafFrame;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeLeafFrameFactory;
 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;
@@ -51,6 +55,7 @@
     private final MultiComparator cmp;
     private final ReadWriteLock treeLatch;    
     private final RangePredicate diskOrderScanPredicate;
+    private final MultiComparator interiorCmp;
     
     public int rootSplits = 0;
     public int[] splitsByLevel = new int[500];
@@ -96,6 +101,14 @@
         this.cmp = cmp;
         this.treeLatch = new ReentrantReadWriteLock(true);
         this.diskOrderScanPredicate = new RangePredicate(true, null, null, cmp);
+        
+        IFieldAccessor[] interiorFields = new IFieldAccessor[cmp.getKeyLength() + 2];
+        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);        
     }
     
     public void create(int fileId, IBTreeLeafFrame leafFrame, IBTreeMetaDataFrame metaFrame) throws Exception {
@@ -517,7 +530,7 @@
                     ctx.interiorFrame.setSmFlag(true); // will be cleared later
                     // in unsetSmPages
                     ctx.splitKey.setLeftPage(newLeftId);
-                    ctx.interiorFrame.insert(ctx.splitKey.getData(), cmp);
+                    ctx.interiorFrame.insert(ctx.splitKey.getTuple(), cmp);
                 } finally {
                     newLeftNode.releaseWriteLatch();
                     writeLatchesReleased++;
@@ -538,15 +551,16 @@
         }
     }
 
-    public void insert(byte[] data, IBTreeLeafFrame leafFrame, IBTreeInteriorFrame interiorFrame,
+    public void insert(ITupleReference tuple, IBTreeLeafFrame leafFrame, IBTreeInteriorFrame interiorFrame,
             IBTreeMetaDataFrame metaFrame) throws Exception {
         BTreeOpContext ctx = new BTreeOpContext();
         ctx.op = BTreeOp.BTO_INSERT;
         ctx.leafFrame = leafFrame;
         ctx.interiorFrame = interiorFrame;
         ctx.metaFrame = metaFrame;
-        ctx.pred = new RangePredicate(true, data, data, cmp);
+        ctx.pred = new RangePredicate(true, tuple, tuple, cmp);
         ctx.splitKey = new SplitKey();
+        ctx.splitKey.getTuple().setFields(interiorCmp.getFields());
         ctx.smPages = new ArrayList<Integer>();
         ctx.pageLsns = new Stack<Integer>();
 
@@ -564,7 +578,7 @@
             }
 
             // we split the root, here is the key for a new root
-            if (ctx.splitKey.getData() != null) {
+            if (ctx.splitKey.getBuffer() != null) {
                 createNewRoot(ctx);
             }
 
@@ -576,32 +590,36 @@
     
     public long uselessCompressionTime = 0;
     
-    private void insertLeaf(ICachedPage node, int pageId, byte[] data, BTreeOpContext ctx) throws Exception {
+    private void insertLeaf(ICachedPage node, int pageId, ITupleReference tuple, BTreeOpContext ctx) throws Exception {
     	ctx.leafFrame.setPage(node);
-    	SpaceStatus spaceStatus = ctx.leafFrame.hasSpaceInsert(data, cmp);
+    	ctx.leafFrame.setPageTupleFields(cmp.getFields());
+    	SpaceStatus spaceStatus = ctx.leafFrame.hasSpaceInsert(tuple, cmp);
     	switch (spaceStatus) {
     	
     	case SUFFICIENT_CONTIGUOUS_SPACE: {
-    		// System.out.println("INSERT LEAF");
-    		ctx.leafFrame.insert(data, cmp);
+    		//System.out.println("SUFFICIENT_CONTIGUOUS_SPACE");
+    		ctx.leafFrame.insert(tuple, cmp);
     		ctx.splitKey.reset();
     	} break;
 
     	case SUFFICIENT_SPACE: {
+    		//System.out.println("SUFFICIENT_SPACE");
     		ctx.leafFrame.compact(cmp);
-    		ctx.leafFrame.insert(data, cmp);
+    		ctx.leafFrame.insert(tuple, cmp);
     		ctx.splitKey.reset();
     	} break;
 
     	case INSUFFICIENT_SPACE: {
+    		//System.out.println("INSUFFICIENT_SPACE");
+    		
     		// try compressing the page first and see if there is space available    		
     		long start = System.currentTimeMillis();
     		boolean reCompressed = ctx.leafFrame.compress(cmp);
     		long end = System.currentTimeMillis();
-    		if(reCompressed) spaceStatus = ctx.leafFrame.hasSpaceInsert(data, cmp);
+    		if(reCompressed) spaceStatus = ctx.leafFrame.hasSpaceInsert(tuple, cmp);
     		
     		if(spaceStatus == SpaceStatus.SUFFICIENT_CONTIGUOUS_SPACE) {    		
-    			ctx.leafFrame.insert(data, cmp);
+    			ctx.leafFrame.insert(tuple, cmp);
     			ctx.splitKey.reset();    			    			
     			
     			usefulCompression++;    			
@@ -635,7 +653,7 @@
     					rightFrame.setPage(rightNode);
     					rightFrame.initBuffer((byte) 0);
 
-    					int ret = ctx.leafFrame.split(rightFrame, data, cmp, ctx.splitKey);
+    					int ret = ctx.leafFrame.split(rightFrame, tuple, cmp, ctx.splitKey);
 
     					ctx.smPages.add(pageId);
     					ctx.smPages.add(rightPageId);
@@ -702,9 +720,10 @@
     	unpins++;
     }
 
-    private void insertInterior(ICachedPage node, int pageId, byte[] data, BTreeOpContext ctx) throws Exception {
+    private void insertInterior(ICachedPage node, int pageId, ITupleReference tuple, BTreeOpContext ctx) throws Exception {
         ctx.interiorFrame.setPage(node);
-        SpaceStatus spaceStatus = ctx.interiorFrame.hasSpaceInsert(data, cmp);
+        ctx.interiorFrame.setPageTupleFields(interiorCmp.getFields());
+        SpaceStatus spaceStatus = ctx.interiorFrame.hasSpaceInsert(tuple, cmp);
         switch (spaceStatus) {
             case INSUFFICIENT_SPACE: {
                 // System.out.println("SPLIT INTERIOR! " + pageId);
@@ -720,7 +739,7 @@
                     rightFrame.initBuffer((byte) ctx.interiorFrame.getLevel());
                     // instead of creating a new split key, use the existing
                     // splitKey
-                    int ret = ctx.interiorFrame.split(rightFrame, ctx.splitKey.getData(), cmp, ctx.splitKey);
+                    int ret = ctx.interiorFrame.split(rightFrame, ctx.splitKey.getTuple(), cmp, ctx.splitKey);
 
                     ctx.smPages.add(pageId);
                     ctx.smPages.add(rightPageId);
@@ -753,30 +772,31 @@
 
             case SUFFICIENT_CONTIGUOUS_SPACE: {
                 // System.out.println("INSERT INTERIOR: " + pageId);
-                ctx.interiorFrame.insert(data, cmp);
+                ctx.interiorFrame.insert(tuple, cmp);
                 ctx.splitKey.reset();
             }
                 break;
 
             case SUFFICIENT_SPACE: {
                 ctx.interiorFrame.compact(cmp);
-                ctx.interiorFrame.insert(data, cmp);
+                ctx.interiorFrame.insert(tuple, cmp);
                 ctx.splitKey.reset();
             }
                 break;
 
         }
     }
-
-    public void delete(byte[] data, IBTreeLeafFrame leafFrame, IBTreeInteriorFrame interiorFrame,
+    
+    public void delete(ITupleReference tuple, IBTreeLeafFrame leafFrame, IBTreeInteriorFrame interiorFrame,
             IBTreeMetaDataFrame metaFrame) throws Exception {
         BTreeOpContext ctx = new BTreeOpContext();
         ctx.op = BTreeOp.BTO_DELETE;
         ctx.leafFrame = leafFrame;
         ctx.interiorFrame = interiorFrame;
         ctx.metaFrame = metaFrame;
-        ctx.pred = new RangePredicate(true, data, data, cmp);
+        ctx.pred = new RangePredicate(true, tuple, tuple, cmp);
         ctx.splitKey = new SplitKey();
+        ctx.splitKey.getTuple().setFields(interiorCmp.getFields());
         ctx.smPages = new ArrayList<Integer>();
         ctx.pageLsns = new Stack<Integer>();
         ctx.freePages = new ArrayList<Integer>();
@@ -795,7 +815,7 @@
             }
 
             // tree is empty, reset level to zero
-            if (ctx.splitKey.getData() != null) {
+            if (ctx.splitKey.getBuffer() != null) {
                 ICachedPage rootNode = bufferCache.pin(FileInfo.getDiskPageId(fileId, rootPage), false);
                 pins++;
                 rootNode.acquireWriteLatch();
@@ -821,7 +841,7 @@
     }
 
     // TODO: to avoid latch deadlock, must modify cursor to detect empty leaves
-    private void deleteLeaf(ICachedPage node, int pageId, byte[] data, BTreeOpContext ctx) throws Exception {
+    private void deleteLeaf(ICachedPage node, int pageId, ITupleReference tuple, BTreeOpContext ctx) throws Exception {
         ctx.leafFrame.setPage(node);
 
         // will this leaf become empty?
@@ -846,7 +866,7 @@
                     treeLatchesAcquired++;
 
                     try {
-                        ctx.leafFrame.delete(data, 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
@@ -921,7 +941,7 @@
                 }
             }
         } else { // leaf will not become empty
-            ctx.leafFrame.delete(data, cmp, true);
+            ctx.leafFrame.delete(tuple, cmp, true);
             node.releaseWriteLatch();
             writeLatchesReleased++;
             bufferCache.unpin(node);
@@ -929,7 +949,7 @@
         }
     }
 
-    private void deleteInterior(ICachedPage node, int pageId, byte[] data, BTreeOpContext ctx) throws Exception {
+    private void deleteInterior(ICachedPage node, int pageId, ITupleReference tuple, BTreeOpContext ctx) throws Exception {
         ctx.interiorFrame.setPage(node);
 
         // this means there is only a child pointer but no key, this case
@@ -948,7 +968,7 @@
             ctx.freePages.add(pageId);
 
         } else {
-            ctx.interiorFrame.delete(data, cmp, false);
+            ctx.interiorFrame.delete(tuple, cmp, false);
             ctx.interiorFrame.setPageLsn(ctx.interiorFrame.getPageLsn() + 1); // TODO:
             // tie
             // together
@@ -1055,13 +1075,13 @@
                         switch (ctx.op) {
 
                             case BTO_INSERT: {
-                                if (ctx.splitKey.getData() != null) {
+                                if (ctx.splitKey.getBuffer() != null) {
                                     node = bufferCache.pin(FileInfo.getDiskPageId(fileId, pageId), false);
                                     pins++;
                                     node.acquireWriteLatch();
                                     writeLatchesAcquired++;
                                     try {
-                                        insertInterior(node, pageId, ctx.splitKey.getData(), ctx);
+                                        insertInterior(node, pageId, ctx.splitKey.getTuple(), ctx);
                                     } finally {
                                         node.releaseWriteLatch();
                                         writeLatchesReleased++;
@@ -1075,13 +1095,13 @@
                                 break;
 
                             case BTO_DELETE: {
-                                if (ctx.splitKey.getData() != null) {
+                            	if (ctx.splitKey.getBuffer() != null) {
                                     node = bufferCache.pin(FileInfo.getDiskPageId(fileId, pageId), false);
                                     pins++;
                                     node.acquireWriteLatch();
                                     writeLatchesAcquired++;
                                     try {
-                                        deleteInterior(node, pageId, ctx.pred.getLowKeys(), ctx);
+                                        deleteInterior(node, pageId, ctx.pred.getLowKey(), ctx);
                                     } finally {
                                         node.releaseWriteLatch();
                                         writeLatchesReleased++;
@@ -1128,27 +1148,23 @@
                 }
             } else { // isLeaf and !smFlag
                 switch (ctx.op) {
-                    case BTO_INSERT: {
-                        insertLeaf(node, pageId, ctx.pred.getLowKeys(), ctx);
-                    }
-                        break;
+                    case BTO_INSERT: {                    	
+                    	insertLeaf(node, pageId, ctx.pred.getLowKey(), ctx);
+                    } break;
 
                     case BTO_DELETE: {
-                        deleteLeaf(node, pageId, ctx.pred.getLowKeys(), ctx);
-                    }
-                        break;
+                        deleteLeaf(node, pageId, ctx.pred.getLowKey(), ctx);
+                    } break;
 
                     case BTO_SEARCH: {
                         ctx.cursor.open(node, ctx.pred);
-                    }
-                        break;
-
+                    } break;
                 }
             }
         } catch (BTreeException e) {
-            // System.out.println("BTREE EXCEPTION");
-            // System.out.println(e.getMessage());
-            // e.printStackTrace();
+            //System.out.println("BTREE EXCEPTION");
+            //System.out.println(e.getMessage());
+            //e.printStackTrace();
             if (!e.getHandled()) {
                 releaseLatch(node, ctx.op, isLeaf);
                 bufferCache.unpin(node);
@@ -1172,7 +1188,7 @@
             throw propException;
         }
     }
-
+    
     private boolean bulkNewPage = false;
 
     public final class BulkLoadContext {               
@@ -1222,6 +1238,7 @@
             frontier.pageId = getFreePage(metaFrame);
             frontier.page = bufferCache.pin(FileInfo.getDiskPageId(fileId, frontier.pageId), bulkNewPage);
             frontier.page.acquireWriteLatch();
+            frontier.lastTuple.setFields(interiorCmp.getFields());
             interiorFrame.setPage(frontier.page);
             interiorFrame.initBuffer((byte) nodeFrontiers.size());
             nodeFrontiers.add(frontier);
@@ -1230,25 +1247,32 @@
 
     private void propagateBulk(BulkLoadContext ctx, int level) throws Exception {
 
-        if (ctx.splitKey.getData() == null)
+        if (ctx.splitKey.getBuffer() == null)
             return;
-
+        
         if (level >= ctx.nodeFrontiers.size())
             ctx.addLevel();
 
         NodeFrontier frontier = ctx.nodeFrontiers.get(level);
         ctx.interiorFrame.setPage(frontier.page);
-        byte[] insBytes = ctx.splitKey.getData();
-        int keySize = cmp.getKeySize(insBytes, 0);
+        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
         int spaceUsed = ctx.interiorFrame.getBuffer().capacity() - ctx.interiorFrame.getTotalFreeSpace();
         if (spaceUsed + spaceNeeded > ctx.interiorMaxBytes) {
-            ctx.interiorFrame.deleteGreatest(cmp);
-            int splitKeySize = cmp.getKeySize(frontier.lastRecord, 0);
+            //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));
             ctx.splitKey.initData(splitKeySize);
-            System.arraycopy(frontier.lastRecord, 0, ctx.splitKey.getData(), 0, splitKeySize);
+            writeTupleFields((ITupleReference)frontier.lastTuple, 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);
@@ -1261,21 +1285,22 @@
             ctx.interiorFrame.setPage(frontier.page);
             ctx.interiorFrame.initBuffer((byte) level);
         }
-        ctx.interiorFrame.insertSorted(insBytes, cmp);
-        frontier.lastRecord = insBytes;
+        ctx.interiorFrame.insertSorted(tuple, cmp);
     }
     
     // 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);        
+        BulkLoadContext ctx = new BulkLoadContext(fillFactor, leafFrame, interiorFrame, metaFrame);
+        ctx.nodeFrontiers.get(0).lastTuple.setFields(cmp.getFields());
+        ctx.splitKey.tuple.setFields(interiorCmp.getFields());
         return ctx;
     }
     
-    public void bulkLoadAddRecord(BulkLoadContext ctx, byte[] record) throws Exception {                        
+    public void bulkLoadAddRecord(BulkLoadContext ctx, ITupleReference tuple) throws Exception {                        
         NodeFrontier leafFrontier = ctx.nodeFrontiers.get(0);
         IBTreeLeafFrame leafFrame = ctx.leafFrame;
         
-        int spaceNeeded = record.length + ctx.slotSize;
+        int spaceNeeded = spaceNeededForTuple(tuple) + ctx.slotSize;
         int spaceUsed = leafFrame.getBuffer().capacity() - leafFrame.getTotalFreeSpace();
         
         // try to free space by compression
@@ -1284,10 +1309,11 @@
         	spaceUsed = leafFrame.getBuffer().capacity() - leafFrame.getTotalFreeSpace();
         }
         
-        if (spaceUsed + spaceNeeded > ctx.leafMaxBytes) {
-            int splitKeySize = cmp.getKeySize(leafFrontier.lastRecord, 0);
-            ctx.splitKey.initData(splitKeySize);            
-            System.arraycopy(leafFrontier.lastRecord, 0, ctx.splitKey.getData(), 0, splitKeySize);
+        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));
+            ctx.splitKey.initData(splitKeySize);
+            writeTupleFields((ITupleReference)leafFrontier.lastTuple, cmp.getKeyLength(), ctx.splitKey.getBuffer(), 0);            
             ctx.splitKey.setLeftPage(leafFrontier.pageId);
             int prevPageId = leafFrontier.pageId;
             leafFrontier.pageId = getFreePage(ctx.metaFrame);
@@ -1306,8 +1332,7 @@
         }
         
         leafFrame.setPage(leafFrontier.page);
-        leafFrame.insertSorted(record, cmp);
-        leafFrontier.lastRecord = record;
+        leafFrame.insertSorted(tuple, cmp);
     }
     
     public void endBulkLoad(BulkLoadContext ctx) throws Exception {                
@@ -1331,6 +1356,24 @@
         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-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixFieldIterator.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixFieldIterator.java
index f0fdf1d..cba3929 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixFieldIterator.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixFieldIterator.java
@@ -24,7 +24,8 @@
 
 // TODO: make members private, only for debugging now
 public class FieldPrefixFieldIterator implements IFieldIterator {
-    public int recSlotOff = -1;
+	
+	public int recSlotOff = -1;
     public int recOff = -1;
     public int prefixSlotNum = FieldPrefixSlotManager.RECORD_UNCOMPRESSED;
     public int numPrefixFields = 0;
@@ -58,7 +59,7 @@
         this.recSlotOff = recSlotOff;
         reset();
     }
-    
+       
     // re-position to first field
     public void reset() {
         currentField = 0;
@@ -78,7 +79,7 @@
             recOffRunner = recOff;
         }        
     }
-    
+        
     public void nextField() {                   
             	
     	// if we have passed the prefix fields of any of the two records, position them to the suffix record
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixPrefixTuple.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixPrefixTuple.java
new file mode 100644
index 0000000..bd9de19
--- /dev/null
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixPrefixTuple.java
@@ -0,0 +1,92 @@
+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-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixSlotManager.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixSlotManager.java
index e6d28bd..84fbde2 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixSlotManager.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixSlotManager.java
@@ -17,7 +17,7 @@
 
 import java.nio.ByteBuffer;
 
-import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparator;
+import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
 import edu.uci.ics.hyracks.storage.am.btree.api.IPrefixSlotManager;
 import edu.uci.ics.hyracks.storage.am.btree.frames.FieldPrefixNSMLeafFrame;
 
@@ -30,7 +30,9 @@
 	
 	private ByteBuffer buf;
 	private FieldPrefixNSMLeafFrame frame;
-	FieldPrefixFieldIterator fieldIter = new FieldPrefixFieldIterator(null, null);
+	//private FieldPrefixFieldIterator fieldIter = new FieldPrefixFieldIterator(null, null);
+	private FieldPrefixTuple pageTuple = new FieldPrefixTuple();
+	private FieldPrefixPrefixTuple pagePrefixTuple = new FieldPrefixPrefixTuple();
 	
 	public int decodeFirstSlotField(int slot) {
 		return (slot & 0xFF000000) >>> 24;
@@ -45,19 +47,22 @@
 	}
 	
 	// returns prefix slot number, or RECORD_UNCOMPRESSED of no match was found
-	public int findPrefix(byte[] data, MultiComparator multiCmp) {
+	public int findPrefix(ITupleReference tuple, MultiComparator multiCmp) {
 		int prefixMid;
 	    int prefixBegin = 0;
 	    int prefixEnd = frame.getNumPrefixRecords() - 1;
 	     
+	    pagePrefixTuple.setFrame(frame);
+	    pagePrefixTuple.setFields(multiCmp.getFields());
+	    
 	    if(frame.getNumPrefixRecords() > 0) {
 	    	while(prefixBegin <= prefixEnd) {
 	    		prefixMid = (prefixBegin + prefixEnd) / 2;	    			    		
 	    		int prefixSlotOff = getPrefixSlotOff(prefixMid);
 	    		int prefixSlot = buf.getInt(prefixSlotOff);    		
 	    		int numPrefixFields = decodeFirstSlotField(prefixSlot);
-	    		int prefixRecOff = decodeSecondSlotField(prefixSlot);
-	    		int cmp = multiCmp.fieldRangeCompare(data, 0, buf.array(), prefixRecOff, 0, numPrefixFields);
+	    		pagePrefixTuple.openPrefixSlotNum(prefixMid);
+	    		int cmp = multiCmp.fieldRangeCompare(tuple, pagePrefixTuple, 0, numPrefixFields);
 	    		if(cmp < 0) prefixEnd = prefixMid - 1;				    		
 	    		else if(cmp > 0) prefixBegin = prefixMid + 1;	    		
 	    		else return prefixMid;
@@ -67,9 +72,14 @@
 	    return FieldPrefixSlotManager.RECORD_UNCOMPRESSED;
 	}
 	
-	public int findSlot(byte[] data, MultiComparator multiCmp, boolean exact) {				
+	public int findSlot(ITupleReference tuple, MultiComparator multiCmp, boolean exact) {				
 		if(frame.getNumRecords() <= 0) encodeSlotFields(RECORD_UNCOMPRESSED, GREATEST_SLOT);
 								
+		pageTuple.setFrame(frame);
+		pageTuple.setFields(multiCmp.getFields());		
+		pagePrefixTuple.setFrame(frame);
+		pagePrefixTuple.setFields(multiCmp.getFields());
+		
 	    int prefixMid;
 	    int prefixBegin = 0;
 	    int prefixEnd = frame.getNumPrefixRecords() - 1;
@@ -85,9 +95,8 @@
 	    	int prefixSlotOff = getPrefixSlotOff(prefixMid);
 	    	int prefixSlot = buf.getInt(prefixSlotOff);    		
 	    	int numPrefixFields = decodeFirstSlotField(prefixSlot);
-	    	int prefixRecOff = decodeSecondSlotField(prefixSlot);
-	    	//System.out.println("PREFIX: " + prefixRecOff + " " + buf.getInt(prefixRecOff) + " " + buf.getInt(prefixRecOff+4));
-	    	int cmp = multiCmp.fieldRangeCompare(data, 0, buf.array(), prefixRecOff, 0, numPrefixFields);
+	    	pagePrefixTuple.openPrefixSlotNum(prefixMid);
+	    	int cmp = multiCmp.fieldRangeCompare(tuple, pagePrefixTuple, 0, numPrefixFields);
 	    	if(cmp < 0) {	    			
 	    		prefixEnd = prefixMid - 1;
 	    		recPrefixSlotNumLbound = prefixMid - 1;	    				    		
@@ -117,17 +126,20 @@
             int recSlotOff = getRecSlotOff(recMid);
             int recSlot = buf.getInt(recSlotOff);              
             int prefixSlotNum = decodeFirstSlotField(recSlot);
-            int recOff = decodeSecondSlotField(recSlot);
             
             //System.out.println("RECS: " + recBegin + " " + recMid + " " + recEnd);
-            int cmp = 0;
-            if(prefixSlotNum == RECORD_UNCOMPRESSED) {                
-                cmp = multiCmp.compare(data, 0, buf.array(), recOff);                
+            int cmp = 0;            
+            if(prefixSlotNum == RECORD_UNCOMPRESSED) {            	            	
+            	pageTuple.openRecSlotNum(recMid);
+            	cmp = multiCmp.compare(tuple, (ITupleReference)pageTuple);      
             }
             else {             	
             	if(prefixSlotNum < recPrefixSlotNumLbound) cmp = 1;
             	else if(prefixSlotNum > recPrefixSlotNumUbound) cmp = -1;
-            	else cmp = compareCompressed(data, buf.array(), prefixSlotNum, recMid, multiCmp);            	            	            	
+            	else {
+            		pageTuple.openRecSlotNum(recMid);
+            		cmp = multiCmp.compare(tuple, (ITupleReference)pageTuple);            	            	            	
+            	}
             }    
             
             if(cmp < 0) recEnd = recMid - 1;
@@ -140,39 +152,13 @@
         if(exact) return encodeSlotFields(prefixMatch, GREATEST_SLOT);                       
         if(recBegin > (frame.getNumRecords() - 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
-        int recSlotOff = getRecSlotOff(recBegin);
-        int recSlot = buf.getInt(recSlotOff);
-        int prefixSlotNum = decodeFirstSlotField(recSlot);
-        int recOff = decodeSecondSlotField(recSlot);
-        
-        int cmp = 0;
-        if(prefixSlotNum == RECORD_UNCOMPRESSED) cmp = multiCmp.compare(data, 0, buf.array(), recOff);        	
-        else cmp = compareCompressed(data, buf.array(), prefixSlotNum, recBegin, multiCmp);
-        
+        // 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);
         else return encodeSlotFields(prefixMatch, GREATEST_SLOT);
 	}	
 	
-	public int compareCompressed(byte[] record, byte[] page, int prefixSlotNum, int recSlotNum, MultiComparator multiCmp) {                          
-         IBinaryComparator[] cmps = multiCmp.getComparators();
-         fieldIter.setFields(multiCmp.getFields());
-         fieldIter.setFrame(frame);         
-         fieldIter.openRecSlotNum(recSlotNum);
-         
-         int recRunner = 0;
-         int cmp = 0;
-         for(int i = 0; i < multiCmp.getKeyLength(); i++) {
-        	 int recFieldLen = multiCmp.getFields()[i].getLength(record, recRunner);
-        	 cmp = cmps[i].compare(record, recRunner, recFieldLen, buf.array(), fieldIter.getFieldOff(), fieldIter.getFieldSize());             
-             if(cmp < 0) return -1;                 
-             else if(cmp > 0) return 1;             
-             fieldIter.nextField();
-             recRunner += recFieldLen;
-         }
-         return 0;
-	}
-	
 	public int getPrefixSlotStartOff() {		
 	    return buf.capacity() - slotSize;	    
 	}
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixTuple.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixTuple.java
new file mode 100644
index 0000000..684edb3
--- /dev/null
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixTuple.java
@@ -0,0 +1,34 @@
+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-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/MultiComparator.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/MultiComparator.java
index 6dbd9c5..0945d12 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/MultiComparator.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/MultiComparator.java
@@ -16,6 +16,7 @@
 package edu.uci.ics.hyracks.storage.am.btree.impls;
 
 import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparator;
+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;
 
@@ -70,6 +71,35 @@
     	return 0;
 	}
 	
+	public int compare(ITupleReference tupleA, ITupleReference tupleB) {			
+		for(int i = 0; i < cmps.length; i++) {						
+			int cmp = cmps[i].compare(tupleA.getFieldData(i), 
+					tupleA.getFieldStart(i), 
+					tupleA.getFieldLength(i),
+					tupleB.getFieldData(i), 
+					tupleB.getFieldStart(i), 
+					tupleB.getFieldLength(i));
+    		if(cmp < 0) return -1;
+    		else if(cmp > 0) return 1;    			
+    	}
+    	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();
          
@@ -88,17 +118,17 @@
          return 0;
 	}
 	
-	public int fieldRangeCompare(byte[] dataA, int recOffA, byte[] dataB, int recOffB, int startFieldIndex, int numFields) {
-		int lenA;
-		int lenB;
-		for(int i = startFieldIndex; i < startFieldIndex + numFields; i++) {
-			lenA = fields[i].getLength(dataA, recOffA);
-			lenB = fields[i].getLength(dataB, recOffB);			
-			int cmp = cmps[i].compare(dataA, recOffA, lenA, dataB, recOffB, lenB);
+	public int fieldRangeCompare(ITupleReference tupleA, ITupleReference tupleB, int startFieldIndex, int numFields) {
+		for(int i = startFieldIndex; i < startFieldIndex + numFields; i++) {						
+			int cmp = cmps[i].compare(
+					tupleA.getFieldData(i), 
+					tupleA.getFieldStart(i), 
+					tupleA.getFieldLength(i), 
+					tupleB.getFieldData(i), 
+					tupleB.getFieldStart(i), 
+					tupleB.getFieldLength(i));
     		if(cmp < 0) return -1;
     		else if(cmp > 0) return 1;
-    		recOffA += lenA; 
-   			recOffB += lenB;
     	}
     	return 0;
 	}
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/NodeFrontier.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/NodeFrontier.java
index 6dcafb4..a6efd7a 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/NodeFrontier.java
+++ b/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 byte[] lastRecord;
+	public SelfDescTupleReference lastTuple = new SelfDescTupleReference();
 }
 
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/OrderedSlotManager.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/OrderedSlotManager.java
index bd4acba..e2dafc0 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/OrderedSlotManager.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/OrderedSlotManager.java
@@ -15,6 +15,7 @@
 
 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.ISlotManager;
 
@@ -23,11 +24,15 @@
 	private static final int slotSize = 4;
 	private IBTreeFrame frame;
 		
+	private SelfDescTupleReference pageTuple = new SelfDescTupleReference();
+		
 	// TODO: mix in interpolation search
 	@Override
-	public int findSlot(byte[] data, MultiComparator multiCmp, boolean exact) {
+	public int findSlot(ITupleReference tuple, MultiComparator multiCmp, boolean exact) {
 		if(frame.getNumRecords() <= 0) return -1;
 		
+		pageTuple.setFields(multiCmp.getFields());
+		
 		int mid;
 		int begin = 0;
 		int end = frame.getNumRecords() - 1;
@@ -36,7 +41,8 @@
             mid = (begin + end) / 2;
         	int slotOff = getSlotOff(mid);        	
         	int recOff = getRecOff(slotOff);
-        	int cmp = multiCmp.compare(data, 0, frame.getBuffer().array(), recOff);
+        	pageTuple.reset(frame.getBuffer(), recOff);
+        	int cmp = multiCmp.compare(tuple, pageTuple);
         	if(cmp < 0)
         		end = mid - 1;
         	else if(cmp > 0)
@@ -50,7 +56,8 @@
         
         int slotOff = getSlotOff(begin);
         int recOff = getRecOff(slotOff);
-        if(multiCmp.compare(data, 0, frame.getBuffer().array(), recOff)  < 0)
+        pageTuple.reset(frame.getBuffer(), recOff);
+        if(multiCmp.compare(tuple, pageTuple)  < 0)
         	return slotOff;
         else
         	return -1;		
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/RangePredicate.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/RangePredicate.java
index d71019f..74f385c 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/RangePredicate.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/RangePredicate.java
@@ -15,6 +15,7 @@
 
 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.ISearchPredicate;
 
 public class RangePredicate implements ISearchPredicate {
@@ -22,20 +23,20 @@
 	private static final long serialVersionUID = 1L;
 	
 	protected boolean isForward = true;
-	protected byte[] lowKeys = null;
-	protected byte[] highKeys = null;
+	protected ITupleReference lowKey = null;
+	protected ITupleReference highKey = null;
 	protected MultiComparator cmp;
-			
+	
 	public RangePredicate() {
 	}
 	
 	// TODO: for now range is [lowKey, highKey] but depending on user predicate the range could be exclusive on any end
 	// need to model this somehow	
 	// for point queries just use same value for low and high key
-	public RangePredicate(boolean isForward, byte[] lowKeys, byte[] highKeys, MultiComparator cmp) {
+	public RangePredicate(boolean isForward, ITupleReference lowKey, ITupleReference highKey, MultiComparator cmp) {
 		this.isForward = isForward;
-		this.lowKeys = lowKeys;
-		this.highKeys = highKeys;
+		this.lowKey = lowKey;
+		this.highKey = highKey;
 		this.cmp = cmp;
 	}
 	
@@ -51,11 +52,11 @@
 		return isForward;
 	}	
 	
-	public byte[] getLowKeys() {
-		return lowKeys;
+	public ITupleReference getLowKey() {
+		return lowKey;
 	}
 	
-	public byte[] getHighKeys() {
-		return highKeys;
+	public ITupleReference getHighKey() {
+		return highKey;
 	}
 }
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/RangeSearchCursor.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/RangeSearchCursor.java
index b6234fb..293ad6b 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/RangeSearchCursor.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/RangeSearchCursor.java
@@ -15,6 +15,7 @@
 
 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.IBTreeCursor;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeLeafFrame;
 import edu.uci.ics.hyracks.storage.am.btree.api.IFieldIterator;
@@ -35,6 +36,7 @@
 	
 	private IFieldIterator fieldIter;
 	
+	
 	public RangeSearchCursor(IBTreeLeafFrame frame) {
 		this.frame = frame;
 		this.fieldIter = frame.createFieldIterator();
@@ -90,13 +92,13 @@
 		RangePredicate pred = (RangePredicate)searchPred;
 		MultiComparator cmp = pred.getComparator();
 		if(searchPred.isForward()) {
-			byte[] highKeys = pred.getHighKeys();			
+			ITupleReference highKey = pred.getHighKey();		
 			fieldIter.openRecSlotNum(recordNum);
 			//recordOffset = frame.getRecordOffset(recordNum);
-			
-			if(highKeys == null) return true;									
+									
+			if(highKey == null) return true;									
 			//if(cmp.compare(highKeys, 0, page.getBuffer().array(), recordOffset) < 0) {
-			if(cmp.compare(highKeys, 0, fieldIter) < 0) {
+			if(cmp.compare(highKey, fieldIter) < 0) {
 				return false;
 			}
 			else {
@@ -104,11 +106,12 @@
 			}
 		}
 		else {
-			byte[] lowKeys = pred.getLowKeys();			
-			recordOffset = frame.getRecordOffset(frame.getNumRecords() - recordNum - 1);
-			if(lowKeys == null) return true;
+			ITupleReference lowKey = pred.getLowKey();						
+			fieldIter.openRecSlotNum(frame.getNumRecords() - recordNum - 1);
+			//recordOffset = frame.getRecordOffset(frame.getNumRecords() - recordNum - 1);
+			if(lowKey == null) return true;
 			
-			if(cmp.compare(lowKeys, 0, page.getBuffer().array(), recordOffset) > 0) {
+			if(cmp.compare(lowKey, fieldIter) > 0) {
 				return false;
 			}
 			else {
@@ -140,26 +143,26 @@
 		MultiComparator cmp = pred.getComparator();
 		this.fieldIter.setFields(cmp.getFields());
 		if(searchPred.isForward()) {
-			byte[] lowKeys = pred.getLowKeys();
+			ITupleReference lowKey = pred.getLowKey();
 						
 			//recordOffset = frame.getRecordOffset(recordNum);			
 			fieldIter.openRecSlotNum(recordNum);
-			if(lowKeys == null) return; // null means -infinity
+			if(lowKey == null) return; // null means -infinity
 						
-			while(cmp.compare(lowKeys, 0, fieldIter) > 0 && recordNum < frame.getNumRecords()) {
+			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);
 			}						
 		}
 		else {
-			byte[] highKeys = pred.getHighKeys();
-						
+			ITupleReference highKey = pred.getHighKey();
+			
 			//recordOffset = frame.getRecordOffset(frame.getNumRecords() - recordNum - 1);
-			fieldIter.openRecSlotNum(recordNum);
-			if(highKeys != null) return; // null means +infinity
+			fieldIter.openRecSlotNum(frame.getNumRecords() - recordNum - 1);
+			if(highKey != null) return; // null means +infinity
 			    
-			while(cmp.compare(highKeys, 0, page.getBuffer().array(), recordOffset) < 0 && recordNum < frame.getNumRecords()) {				
+			while(cmp.compare(highKey, fieldIter) < 0 && recordNum < frame.getNumRecords()) {				
 			    recordNum++;
 			    fieldIter.openRecSlotNum(recordNum);
 			    //recordOffset = frame.getRecordOffset(frame.getNumRecords() - recordNum - 1);			
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/SelfDescTupleReference.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/SelfDescTupleReference.java
new file mode 100644
index 0000000..b264c66
--- /dev/null
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/SelfDescTupleReference.java
@@ -0,0 +1,59 @@
+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.IFieldAccessor;
+
+public class SelfDescTupleReference implements ITupleReference {
+
+	private ByteBuffer buf;
+	private int startOff;
+	private IFieldAccessor[] fields;
+	
+	public SelfDescTupleReference(IFieldAccessor[] fields) {
+		this.fields = fields;
+	}
+	
+	public SelfDescTupleReference() {
+	}
+	
+	public IFieldAccessor[] getFields() {
+		return fields;
+	}
+	
+	public void reset(ByteBuffer buf, int startOff) {
+		this.buf = buf;
+		this.startOff = startOff;
+	}
+	
+	public void setFields(IFieldAccessor[] fields) {
+		this.fields = fields;
+	}
+	
+	@Override
+	public int getFieldCount() {
+		return fields.length;
+	}
+
+	@Override
+	public byte[] getFieldData(int fIdx) {
+		return buf.array();
+	}
+
+	@Override
+	public int getFieldLength(int fIdx) {				
+		int fieldStart = getFieldStart(fIdx);
+		return fields[fIdx].getLength(buf.array(), fieldStart);
+	}
+
+	@Override
+	public int getFieldStart(int fIdx) {
+		int runner = startOff;
+		for(int i = 0; i < fIdx; i++) {
+			runner += fields[i].getLength(buf.array(), runner);
+		}
+		return runner;
+	}
+	
+}
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/SplitKey.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/SplitKey.java
index aaaa8ab..4fb52b4 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/SplitKey.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/SplitKey.java
@@ -18,8 +18,9 @@
 import java.nio.ByteBuffer;
 
 public class SplitKey {		
-	private byte[] data = null;	
-	private ByteBuffer buf = null;
+	public byte[] data = null;	
+	public ByteBuffer buf = null;
+	public SelfDescTupleReference tuple = new SelfDescTupleReference();
 	
 	public void initData(int keySize) {
 		// try to reuse existing memory from a lower-level split if possible
@@ -38,6 +39,7 @@
 		*/		
 		data = new byte[keySize + 8]; // add 8 for left and right page
 		buf = ByteBuffer.wrap(data);
+		tuple.reset(buf, 0);
 	}
 	
 	public void reset() {
@@ -49,8 +51,12 @@
 		this.data = data;
 	}
 	
-	public byte[] getData() {
-		return data;
+	public ByteBuffer getBuffer() {
+		return buf;
+	}
+	
+	public SelfDescTupleReference getTuple() {
+		return tuple;
 	}
 	
 	public int getLeftPage() {
@@ -73,4 +79,14 @@
 		buf.putInt(data.length - 8, leftPage);
 		buf.putInt(data.length - 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);	
+		return copy;
+	}
 }
\ No newline at end of file
diff --git a/hyracks-storage-am-btree/src/test/java/edu/uci/ics/hyracks/storage/am/btree/BTreeFieldPrefixNSMTest.java b/hyracks-storage-am-btree/src/test/java/edu/uci/ics/hyracks/storage/am/btree/BTreeFieldPrefixNSMTest.java
index a4d5095..77c114d 100644
--- a/hyracks-storage-am-btree/src/test/java/edu/uci/ics/hyracks/storage/am/btree/BTreeFieldPrefixNSMTest.java
+++ b/hyracks-storage-am-btree/src/test/java/edu/uci/ics/hyracks/storage/am/btree/BTreeFieldPrefixNSMTest.java
@@ -15,23 +15,33 @@
 
 package edu.uci.ics.hyracks.storage.am.btree;
 
-import java.io.DataOutputStream;
+import java.io.DataOutput;
 import java.io.File;
 import java.io.RandomAccessFile;
 import java.nio.ByteBuffer;
-import java.util.ArrayList;
 import java.util.Random;
 
 import org.junit.Assert;
 import org.junit.Test;
 
+import edu.uci.ics.hyracks.api.comm.IFrameTupleAccessor;
+import edu.uci.ics.hyracks.api.context.IHyracksContext;
 import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparator;
-import edu.uci.ics.hyracks.dataflow.common.comm.io.ByteArrayAccessibleOutputStream;
+import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
+import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
+import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
+import edu.uci.ics.hyracks.control.nc.runtime.HyracksContext;
+import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
+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.marshalling.IntegerSerializerDeserializer;
 import edu.uci.ics.hyracks.storage.am.btree.api.IFieldAccessor;
 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.BTreeException;
 import edu.uci.ics.hyracks.storage.am.btree.impls.FieldPrefixSlotManager;
 import edu.uci.ics.hyracks.storage.am.btree.impls.MultiComparator;
 import edu.uci.ics.hyracks.storage.am.btree.types.Int32Accessor;
@@ -52,6 +62,7 @@
 	//private static final int PAGE_SIZE = 65536; // 64K
 	private static final int PAGE_SIZE = 131072; // 128K
     private static final int NUM_PAGES = 40;
+    private static final int HYRACKS_FRAME_SIZE = 128;
     
     // to help with the logger madness
     private void print(String str) {
@@ -73,22 +84,37 @@
         }
     }
     
-    private void tupleInsert(FieldPrefixNSMLeafFrame frame, MultiComparator cmp, int f0, int f1, int f2, boolean print, ArrayList<byte[]> records) throws Exception {
-    	if(print) System.out.println("INSERTING: " + f0 + " " + f1 + " " + f2);
+    private ITupleReference createTuple(int f0, int f1, int f2, boolean print) throws HyracksDataException {
+    	if(print) System.out.println("CREATING: " + f0 + " " + f1 + " " + f2);
     	
-    	ByteArrayAccessibleOutputStream baaos = new ByteArrayAccessibleOutputStream();
-    	DataOutputStream dos = new DataOutputStream(baaos);        	        	
-    	    	
+    	IHyracksContext ctx = new HyracksContext(HYRACKS_FRAME_SIZE);        
+        ByteBuffer buf = ctx.getResourceManager().allocateFrame();
+		FrameTupleAppender appender = new FrameTupleAppender(ctx);				
+		ArrayTupleBuilder tb = new ArrayTupleBuilder(3);
+		DataOutput dos = tb.getDataOutput();
+		
+		ISerializerDeserializer[] recDescSers = { IntegerSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE };
+		RecordDescriptor recDesc = new RecordDescriptor(recDescSers);
+		IFrameTupleAccessor accessor = new FrameTupleAccessor(ctx, recDesc);
+		accessor.reset(buf);
+		FrameTupleReference tuple = new FrameTupleReference();
+		
+		tb.reset();
     	IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+    	tb.addFieldEndOffset();
     	IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
-    	IntegerSerializerDeserializer.INSTANCE.serialize(f2, dos);
+    	tb.addFieldEndOffset();
+    	IntegerSerializerDeserializer.INSTANCE.serialize(f2, dos);    	        	
+    	tb.addFieldEndOffset();
     	
-    	byte[] record = baaos.toByteArray();        
-        frame.insert(record, cmp);
-        
-        if(records != null) records.add(record);
+    	appender.reset(buf, true);
+    	appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+    	
+    	tuple.reset(accessor, 0);
+    	
+    	return tuple;
     }
-    
+        
     @Test
     public void test01() throws Exception {
     	
@@ -136,9 +162,9 @@
         	int compressFreq = 5;
         	int smallMax = 10;        	
         	int numRecords = 5000;
-        	ArrayList<byte[]> records = new ArrayList<byte[]>();
-        	records.ensureCapacity(numRecords);
-        	        	
+        	
+        	int[][] savedFields = new int[numRecords][3];   	
+        	
         	// insert records with random calls to compact and compress
         	for(int i = 0; i < numRecords; i++) {
         		
@@ -147,8 +173,22 @@
         		int a = rnd.nextInt() % smallMax;
         		int b = rnd.nextInt() % smallMax;
         		int c = i;
-        		tupleInsert(frame, cmp, a, b, c, false, records);
         		
+        		ITupleReference tuple = createTuple(a, b, c, false);
+        		try {
+        			frame.insert(tuple, cmp);
+        		} 
+        		catch (BTreeException e) {    
+        			e.printStackTrace();
+        		}
+        		catch (Exception e) {
+        			e.printStackTrace();
+        		}        		        	
+        		
+        		savedFields[i][0] = a;
+        		savedFields[i][1] = b;
+        		savedFields[i][2] = c;
+        		            
         		if(rnd.nextInt() % compactFreq == 0) {
         			before = frame.printKeys(cmp);
         			frame.compact(cmp);
@@ -165,11 +205,16 @@
         	}
         	
         	// delete records with random calls to compact and compress        	
-        	for(int i = 0; i < records.size(); i++) {        	            	    
+        	for(int i = 0; i < numRecords; i++) {        	            	    
         		
         		if((i+1) % 100 == 0) print("DELETING " + (i+1) + " / " + numRecords + "\n");
         		
-        		frame.delete(records.get(i), cmp, true);    
+        		ITupleReference tuple = createTuple(savedFields[i][0], savedFields[i][1], savedFields[i][2], false);
+        		try {
+        			frame.delete(tuple, cmp, true);
+        		}
+        		catch (Exception e) {        			
+        		}
         		
         		if(rnd.nextInt() % compactFreq == 0) {
         			before = frame.printKeys(cmp);
diff --git a/hyracks-storage-am-btree/src/test/java/edu/uci/ics/hyracks/storage/am/btree/BTreeTest.java b/hyracks-storage-am-btree/src/test/java/edu/uci/ics/hyracks/storage/am/btree/BTreeTest.java
index 3caba23..f616f56 100644
--- a/hyracks-storage-am-btree/src/test/java/edu/uci/ics/hyracks/storage/am/btree/BTreeTest.java
+++ b/hyracks-storage-am-btree/src/test/java/edu/uci/ics/hyracks/storage/am/btree/BTreeTest.java
@@ -15,6 +15,7 @@
 
 package edu.uci.ics.hyracks.storage.am.btree;
 
+import java.io.DataOutput;
 import java.io.DataOutputStream;
 import java.io.File;
 import java.io.RandomAccessFile;
@@ -23,8 +24,17 @@
 
 import org.junit.Test;
 
+import edu.uci.ics.hyracks.api.comm.IFrameTupleAccessor;
+import edu.uci.ics.hyracks.api.context.IHyracksContext;
 import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparator;
+import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
+import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
+import edu.uci.ics.hyracks.control.nc.runtime.HyracksContext;
+import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
 import edu.uci.ics.hyracks.dataflow.common.comm.io.ByteArrayAccessibleOutputStream;
+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.comparators.IntegerBinaryComparatorFactory;
 import edu.uci.ics.hyracks.dataflow.common.data.comparators.UTF8StringBinaryComparatorFactory;
 import edu.uci.ics.hyracks.dataflow.common.data.marshalling.IntegerSerializerDeserializer;
@@ -42,10 +52,12 @@
 import edu.uci.ics.hyracks.storage.am.btree.frames.NSMInteriorFrameFactory;
 import edu.uci.ics.hyracks.storage.am.btree.frames.NSMLeafFrameFactory;
 import edu.uci.ics.hyracks.storage.am.btree.impls.BTree;
+import edu.uci.ics.hyracks.storage.am.btree.impls.BTreeException;
 import edu.uci.ics.hyracks.storage.am.btree.impls.DiskOrderScanCursor;
-import edu.uci.ics.hyracks.storage.am.btree.impls.RangeSearchCursor;
 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.Int32Accessor;
 import edu.uci.ics.hyracks.storage.am.btree.types.UTF8StringAccessor;
 import edu.uci.ics.hyracks.storage.common.buffercache.BufferCache;
@@ -56,11 +68,14 @@
 import edu.uci.ics.hyracks.storage.common.file.FileInfo;
 import edu.uci.ics.hyracks.storage.common.file.FileManager;
 
+@SuppressWarnings("unchecked")
+
 public class BTreeTest {    
 	
-    private static final int PAGE_SIZE = 128;
-    // private static final int PAGE_SIZE = 8192;
+    //private static final int PAGE_SIZE = 128;
+    private static final int PAGE_SIZE = 8192;
     private static final int NUM_PAGES = 10;
+    private static final int HYRACKS_FRAME_SIZE = 128;
     
     // to help with the logger madness
     private void print(String str) {
@@ -88,7 +103,7 @@
     // perform ordered scan and range search
     @Test
     public void test01() throws Exception {
-
+    	
     	print("FIXED-LENGTH KEY TEST\n");
     	
         FileManager fileManager = new FileManager();
@@ -130,27 +145,50 @@
         
         print("INSERTING INTO TREE\n");
         
-        for (int i = 0; i < 10000; i++) {
-        	ByteArrayAccessibleOutputStream baaos = new ByteArrayAccessibleOutputStream();
-        	DataOutputStream dos = new DataOutputStream(baaos);        	        	
-        	
+        IHyracksContext ctx = new HyracksContext(HYRACKS_FRAME_SIZE);        
+        ByteBuffer frame = ctx.getResourceManager().allocateFrame();
+		FrameTupleAppender appender = new FrameTupleAppender(ctx);				
+		ArrayTupleBuilder tb = new ArrayTupleBuilder(cmp.getFields().length);
+		DataOutput dos = tb.getDataOutput();
+		
+		ISerializerDeserializer[] recDescSers = { IntegerSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE};
+		RecordDescriptor recDesc = new RecordDescriptor(recDescSers);
+		IFrameTupleAccessor accessor = new FrameTupleAccessor(ctx, recDesc);
+		accessor.reset(frame);
+		FrameTupleReference tuple = new FrameTupleReference();
+		
+		// 10000
+        for (int i = 0; i < 10000; i++) {        	
         	int f0 = rnd.nextInt() % 10000;
         	int f1 = 5;
         	
+        	tb.reset();
         	IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+        	tb.addFieldEndOffset();
         	IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+        	tb.addFieldEndOffset();        	
+        	        	
+        	appender.reset(frame, true);
+        	appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
         	
-        	byte[] record = baaos.toByteArray();
-        	        	           
+        	tuple.reset(accessor, 0);
+        	
+        	//System.out.println(tuple.getFieldCount() + " " + tuple.getFieldLength(0) + " " + tuple.getFieldLength(1));
+        	
             if (i % 1000 == 0) {
                 long end = System.currentTimeMillis();
                 print("INSERTING " + i + " : " + f0 + " " + f1 + " " + (end - start) + "\n");            	
             }
             
             try {                                
-                btree.insert(record, leafFrame, interiorFrame, metaFrame);
+                btree.insert(tuple, leafFrame, interiorFrame, metaFrame);
+            } catch (BTreeException e) {            	
             } catch (Exception e) {
+            	e.printStackTrace();
             }
+            
+            //btree.printTree(leafFrame, interiorFrame);
+            //System.out.println();
         }
         //btree.printTree(leafFrame, interiorFrame);
         
@@ -211,8 +249,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);
         
         IBinaryComparator[] searchCmps = new IBinaryComparator[1];
         searchCmps[0] = IntegerBinaryComparatorFactory.INSTANCE.createBinaryComparator();
@@ -239,7 +284,7 @@
         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
@@ -290,26 +335,44 @@
         long start = System.currentTimeMillis();
         
         print("INSERTING INTO TREE\n");
-        for (int i = 0; i < 10000; i++) {
-        	ByteArrayAccessibleOutputStream baaos = new ByteArrayAccessibleOutputStream();
-        	DataOutputStream dos = new DataOutputStream(baaos);        	        	
-        	
+        
+        IHyracksContext ctx = new HyracksContext(HYRACKS_FRAME_SIZE);        
+        ByteBuffer frame = ctx.getResourceManager().allocateFrame();
+		FrameTupleAppender appender = new FrameTupleAppender(ctx);				
+		ArrayTupleBuilder tb = new ArrayTupleBuilder(cmp.getFields().length);
+		DataOutput dos = tb.getDataOutput();
+		
+		ISerializerDeserializer[] recDescSers = { IntegerSerializerDeserializer.INSTANCE, 
+				IntegerSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE};
+		RecordDescriptor recDesc = new RecordDescriptor(recDescSers);
+		IFrameTupleAccessor accessor = new FrameTupleAccessor(ctx, recDesc);
+		accessor.reset(frame);
+		FrameTupleReference tuple = new FrameTupleReference();
+        
+        for (int i = 0; i < 10000; i++) {        	
         	int f0 = rnd.nextInt() % 2000;
         	int f1 = rnd.nextInt() % 1000;
         	int f2 = 5;
         	        	
+        	tb.reset();
         	IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+        	tb.addFieldEndOffset();
         	IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+        	tb.addFieldEndOffset();
         	IntegerSerializerDeserializer.INSTANCE.serialize(f2, dos);
+        	tb.addFieldEndOffset();
         	
-        	byte[] record = baaos.toByteArray();
+        	appender.reset(frame, true);
+        	appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
         	
+        	tuple.reset(accessor, 0);
+        	        	
             if (i % 1000 == 0) {
             	print("INSERTING " + i + " : " + f0 + " " + f1 + "\n");
             }
             
             try {
-                btree.insert(record, leafFrame, interiorFrame, metaFrame);
+                btree.insert(tuple, leafFrame, interiorFrame, metaFrame);
             } catch (Exception e) {
             }
         }
@@ -349,9 +412,16 @@
     	ByteArrayAccessibleOutputStream hkbaaos = new ByteArrayAccessibleOutputStream();
     	DataOutputStream hkdos = new DataOutputStream(hkbaaos);    	    	    	
     	IntegerSerializerDeserializer.INSTANCE.serialize(3, 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);
         
         IBinaryComparator[] searchCmps = new IBinaryComparator[1];
         searchCmps[0] = IntegerBinaryComparatorFactory.INSTANCE.createBinaryComparator();       
@@ -425,25 +495,41 @@
     	Random rnd = new Random();
     	rnd.setSeed(50);
 
+    	IHyracksContext ctx = new HyracksContext(HYRACKS_FRAME_SIZE);        
+        ByteBuffer frame = ctx.getResourceManager().allocateFrame();
+		FrameTupleAppender appender = new FrameTupleAppender(ctx);				
+		ArrayTupleBuilder tb = new ArrayTupleBuilder(cmp.getFields().length);
+		DataOutput dos = tb.getDataOutput();
+		
+		ISerializerDeserializer[] recDescSers = { UTF8StringSerializerDeserializer.INSTANCE, UTF8StringSerializerDeserializer.INSTANCE };
+		RecordDescriptor recDesc = new RecordDescriptor(recDescSers);
+		IFrameTupleAccessor accessor = new FrameTupleAccessor(ctx, recDesc);
+		accessor.reset(frame);
+		FrameTupleReference tuple = new FrameTupleReference();
+    	
     	int maxLength = 10; // max string length to be generated
     	for (int i = 0; i < 10000; i++) {
     		String f0 = randomString(Math.abs(rnd.nextInt()) % maxLength + 1, rnd);
     		String f1 = randomString(Math.abs(rnd.nextInt()) % maxLength + 1, rnd);
     		
-    		ByteArrayAccessibleOutputStream baaos = new ByteArrayAccessibleOutputStream();
-        	DataOutputStream dos = new DataOutputStream(baaos);        	        	
-        	
-        	UTF8StringSerializerDeserializer.INSTANCE.serialize(f0, dos);
+    		tb.reset();
+    		UTF8StringSerializerDeserializer.INSTANCE.serialize(f0, dos);
+        	tb.addFieldEndOffset();
         	UTF8StringSerializerDeserializer.INSTANCE.serialize(f1, dos);
+        	tb.addFieldEndOffset();        	
         	
-    		byte[] record = baaos.toByteArray();
-    		
+        	appender.reset(frame, true);
+        	appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+        	
+        	tuple.reset(accessor, 0);
+    		    		
     		if (i % 1000 == 0) {
-    			print("INSERTING " + i + ": " + cmp.printRecord(record, 0) + "\n");
+    			//print("INSERTING " + i + ": " + cmp.printRecord(record, 0) + "\n");
+    			print("INSERTING " + i + "\n");
     		}
 
     		try {
-    			btree.insert(record, leafFrame, interiorFrame, metaFrame);
+    			btree.insert(tuple, leafFrame, interiorFrame, metaFrame);
     		} catch (Exception e) {
     		}
     	}     
@@ -481,8 +567,15 @@
     	DataOutputStream hkdos = new DataOutputStream(hkbaaos);    	    	    	
     	UTF8StringSerializerDeserializer.INSTANCE.serialize("cc7", 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);
         
         IBinaryComparator[] searchCmps = new IBinaryComparator[1];
         searchCmps[0] = UTF8StringBinaryComparatorFactory.INSTANCE.createBinaryComparator();
@@ -557,6 +650,18 @@
         Random rnd = new Random();
         rnd.setSeed(50);
 
+        IHyracksContext ctx = new HyracksContext(HYRACKS_FRAME_SIZE);        
+        ByteBuffer frame = ctx.getResourceManager().allocateFrame();
+		FrameTupleAppender appender = new FrameTupleAppender(ctx);				
+		ArrayTupleBuilder tb = new ArrayTupleBuilder(cmp.getFields().length);
+		DataOutput dos = tb.getDataOutput();
+		
+		ISerializerDeserializer[] recDescSers = { UTF8StringSerializerDeserializer.INSTANCE, UTF8StringSerializerDeserializer.INSTANCE };
+		RecordDescriptor recDesc = new RecordDescriptor(recDescSers);
+		IFrameTupleAccessor accessor = new FrameTupleAccessor(ctx, recDesc);
+		accessor.reset(frame);
+		FrameTupleReference tuple = new FrameTupleReference();
+        
         int runs = 3;
         for (int run = 0; run < runs; run++) {
         	
@@ -565,30 +670,39 @@
             print("INSERTING INTO BTREE\n");
             int maxLength = 10;
             int ins = 10000;
-            byte[][] records = new byte[ins][];
+            String[] f0s = new String[ins];
+            String[] f1s = new String[ins];                                     
             int insDone = 0;
             int[] insDoneCmp = new int[ins];
             for (int i = 0; i < ins; i++) {
                 String f0 = randomString(Math.abs(rnd.nextInt()) % maxLength + 1, rnd);
                 String f1 = randomString(Math.abs(rnd.nextInt()) % maxLength + 1, rnd);
                 
-        		ByteArrayAccessibleOutputStream baaos = new ByteArrayAccessibleOutputStream();
-            	DataOutputStream dos = new DataOutputStream(baaos);        	        	
-            	
-            	UTF8StringSerializerDeserializer.INSTANCE.serialize(f0, dos);
-            	UTF8StringSerializerDeserializer.INSTANCE.serialize(f1, dos);
+                f0s[i] = f0;
+                f1s[i] = f1;
                 
-            	byte[] record = baaos.toByteArray();            	
-                records[i] = record;
+        		tb.reset();
+        		UTF8StringSerializerDeserializer.INSTANCE.serialize(f0, dos);
+            	tb.addFieldEndOffset();
+            	UTF8StringSerializerDeserializer.INSTANCE.serialize(f1, dos);
+            	tb.addFieldEndOffset();        	
+            	
+            	appender.reset(frame, true);
+            	appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+            	
+            	tuple.reset(accessor, 0);                        		                            	            	
 
                 if (i % 1000 == 0) {
-                    print("INSERTING " + i + ": " + cmp.printRecord(record, 0) + "\n");                                       
+                	print("INSERTING " + i + "\n");
+                	//print("INSERTING " + i + ": " + cmp.printRecord(record, 0) + "\n");                                       
                 }
 
                 try {
-                    btree.insert(record, leafFrame, interiorFrame, metaFrame);
+                    btree.insert(tuple, leafFrame, interiorFrame, metaFrame);
                     insDone++;
+                } catch (BTreeException e) {                	
                 } catch (Exception e) {
+                	e.printStackTrace();
                 }
 
                 insDoneCmp[i] = insDone;
@@ -599,17 +713,31 @@
             print("DELETING FROM BTREE\n");
             int delDone = 0;
             for (int i = 0; i < ins; i++) {
-
+            	
+            	tb.reset();
+        		UTF8StringSerializerDeserializer.INSTANCE.serialize(f0s[i], dos);
+            	tb.addFieldEndOffset();
+            	UTF8StringSerializerDeserializer.INSTANCE.serialize(f1s[i], dos);
+            	tb.addFieldEndOffset();        	
+            	
+            	appender.reset(frame, true);
+            	appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+            	
+            	tuple.reset(accessor, 0);    
+            	
                 if (i % 1000 == 0) {
-                    print("DELETING " + i + ": " + cmp.printRecord(records[i], 0) + "\n");                                    
+                    //print("DELETING " + i + ": " + cmp.printRecord(records[i], 0) + "\n");
+                	print("DELETING " + i + "\n");
                 }
 
                 try {
-                    btree.delete(records[i], leafFrame, interiorFrame, metaFrame);
+                    btree.delete(tuple, leafFrame, interiorFrame, metaFrame);
                     delDone++;
+                } catch (BTreeException e) {                	
                 } catch (Exception e) {
+                	e.printStackTrace();
                 }
-                
+
                 if (insDoneCmp[i] != delDone) {
                     print("INCONSISTENT STATE, ERROR IN DELETION TEST\n");
                     print("INSDONECMP: " + insDoneCmp[i] + " " + delDone + "\n"); 
@@ -632,7 +760,7 @@
         
         print("\n");
     }
-
+    
     // BULK LOAD TEST
     // insert 100,000 records in bulk
     // B-tree has a composite key to "simulate" non-unique index creation
@@ -681,32 +809,44 @@
         Random rnd = new Random();
         rnd.setSeed(50);
 
+        IHyracksContext ctx = new HyracksContext(HYRACKS_FRAME_SIZE);        
+        ByteBuffer frame = ctx.getResourceManager().allocateFrame();
+		FrameTupleAppender appender = new FrameTupleAppender(ctx);				
+		ArrayTupleBuilder tb = new ArrayTupleBuilder(cmp.getFields().length);
+		DataOutput dos = tb.getDataOutput();
+		
+		ISerializerDeserializer[] recDescSers = { IntegerSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE };
+		RecordDescriptor recDesc = new RecordDescriptor(recDescSers);
+		IFrameTupleAccessor accessor = new FrameTupleAccessor(ctx, recDesc);
+		accessor.reset(frame);
+		FrameTupleReference tuple = new FrameTupleReference();
+        
+		BTree.BulkLoadContext bulkLoadCtx = btree.beginBulkLoad(0.7f, leafFrame, interiorFrame, metaFrame);
+		
         // generate sorted records
         int ins = 100000;
-        byte[][] records = new byte[ins][];
-        for (int i = 0; i < ins; i++) {
-        	ByteArrayAccessibleOutputStream baaos = new ByteArrayAccessibleOutputStream();
-        	DataOutputStream dos = new DataOutputStream(baaos);        	        	
-        	
-        	int f0 = i;
-        	int f1 = i;
-        	int f2 = 5;
-        	
-        	IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
-        	IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
-        	IntegerSerializerDeserializer.INSTANCE.serialize(f2, dos);
-        	
-            records[i] = baaos.toByteArray();
-        }
-
         print("BULK LOADING " + ins + " RECORDS\n");
         long start = System.currentTimeMillis();
-        
-        BTree.BulkLoadContext ctx = btree.beginBulkLoad(0.7f, leafFrame, interiorFrame, metaFrame);                
         for (int i = 0; i < ins; i++) {
-            btree.bulkLoadAddRecord(ctx, records[i]);
+        	
+        	tb.reset();
+        	IntegerSerializerDeserializer.INSTANCE.serialize(i, dos);
+        	tb.addFieldEndOffset();
+        	IntegerSerializerDeserializer.INSTANCE.serialize(i, dos);
+        	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.endBulkLoad(ctx);
+
+        btree.endBulkLoad(bulkLoadCtx);
         
         long end = System.currentTimeMillis();
         long duration = end - start;
@@ -724,9 +864,16 @@
     	DataOutputStream hkdos = new DataOutputStream(hkbaaos);    	    	    	
     	IntegerSerializerDeserializer.INSTANCE.serialize(44500, 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);
+    	        
         IBinaryComparator[] searchCmps = new IBinaryComparator[1];
         searchCmps[0] = IntegerBinaryComparatorFactory.INSTANCE.createBinaryComparator();
         MultiComparator searchCmp = new MultiComparator(searchCmps, fields);
@@ -754,7 +901,7 @@
         fileManager.close();
         
         print("\n");
-    }
+    }    
     
     // TIME-INTERVAL INTERSECTION DEMO FOR EVENT PEOPLE
     // demo for Arjun to show easy support of intersection queries on time-intervals
@@ -801,6 +948,18 @@
         Random rnd = new Random();
         rnd.setSeed(50);
         
+        IHyracksContext ctx = new HyracksContext(HYRACKS_FRAME_SIZE);        
+        ByteBuffer frame = ctx.getResourceManager().allocateFrame();
+		FrameTupleAppender appender = new FrameTupleAppender(ctx);				
+		ArrayTupleBuilder tb = new ArrayTupleBuilder(cmp.getFields().length);
+		DataOutput dos = tb.getDataOutput();
+		
+		ISerializerDeserializer[] recDescSers = { IntegerSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE };
+		RecordDescriptor recDesc = new RecordDescriptor(recDescSers);
+		IFrameTupleAccessor accessor = new FrameTupleAccessor(ctx, recDesc);
+		accessor.reset(frame);
+		FrameTupleReference tuple = new FrameTupleReference();
+        
         long start = System.currentTimeMillis();
         
         int intervalCount = 10;
@@ -837,24 +996,29 @@
         intervals[9][1] = 35;
 
         // int exceptionCount = 0;
-        for (int i = 0; i < intervalCount; i++) {
-        	ByteArrayAccessibleOutputStream baaos = new ByteArrayAccessibleOutputStream();
-        	DataOutputStream dos = new DataOutputStream(baaos);        	        	
-        	
+        for (int i = 0; i < intervalCount; i++) {        	
         	int f0 = intervals[i][0];
         	int f1 = intervals[i][1];
         	int f2 = rnd.nextInt() % 100;
         	
+        	tb.reset();
         	IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+        	tb.addFieldEndOffset();
         	IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+        	tb.addFieldEndOffset();        	
         	IntegerSerializerDeserializer.INSTANCE.serialize(f2, dos);
+        	tb.addFieldEndOffset();
         	
-        	byte[] record = baaos.toByteArray();
+        	appender.reset(frame, true);
+        	appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
         	
-            print("INSERTING " + i + " : " + f0 + " " + f1 + "\n");
+        	tuple.reset(accessor, 0);
+        	        	
+            //print("INSERTING " + i + " : " + f0 + " " + f1 + "\n");
+        	print("INSERTING " + i + "\n");
 
             try {
-                btree.insert(record, leafFrame, interiorFrame, metaFrame);
+                btree.insert(tuple, leafFrame, interiorFrame, metaFrame);
             } catch (Exception e) {
                 // e.printStackTrace();
             }
@@ -899,16 +1063,23 @@
     	DataOutputStream hkdos = new DataOutputStream(hkbaaos);    	    	    	
     	IntegerSerializerDeserializer.INSTANCE.serialize(19, hkdos);
     	IntegerSerializerDeserializer.INSTANCE.serialize(19, 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);
         
         IBinaryComparator[] searchCmps = new IBinaryComparator[2];
         searchCmps[0] = IntegerBinaryComparatorFactory.INSTANCE.createBinaryComparator();
         searchCmps[1] = IntegerBinaryComparatorFactory.INSTANCE.createBinaryComparator();
         MultiComparator searchCmp = new MultiComparator(searchCmps, fields);
-        
-        print("INDEX RANGE SEARCH ON: " + cmp.printKey(lowKey, 0) + " " + cmp.printKey(highKey, 0) + "\n");                
+                
+        //print("INDEX RANGE SEARCH ON: " + cmp.printKey(lowKey, 0) + " " + cmp.printKey(highKey, 0) + "\n");                
         
         RangePredicate rangePred = new RangePredicate(true, lowKey, highKey, searchCmp);
         btree.search(rangeCursor, rangePred, leafFrame, interiorFrame);
@@ -933,7 +1104,7 @@
         
         print("\n");
     }
-
+    
     public static String randomString(int length, Random random) {
         String s = Long.toHexString(Double.doubleToLongBits(random.nextDouble()));
         StringBuilder strBuilder = new StringBuilder();