diff --git a/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/edu/uci/ics/hyracks/tests/btree/BTreeOperatorsTest.java b/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/edu/uci/ics/hyracks/tests/btree/BTreeOperatorsTest.java
index daa2425..00fe7e7 100644
--- a/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/edu/uci/ics/hyracks/tests/btree/BTreeOperatorsTest.java
+++ b/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/edu/uci/ics/hyracks/tests/btree/BTreeOperatorsTest.java
@@ -15,7 +15,7 @@
 
 package edu.uci.ics.hyracks.tests.btree;
 
-import java.io.DataOutputStream;
+import java.io.DataOutput;
 import java.io.File;
 import java.io.RandomAccessFile;
 import java.nio.ByteBuffer;
@@ -26,13 +26,16 @@
 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.context.IHyracksContext;
 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.control.nc.runtime.HyracksContext;
+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.data.accessors.ITupleReference;
 import edu.uci.ics.hyracks.dataflow.common.data.comparators.IntegerBinaryComparatorFactory;
 import edu.uci.ics.hyracks.dataflow.common.data.comparators.UTF8StringBinaryComparatorFactory;
@@ -52,16 +55,16 @@
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeCursor;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeInteriorFrameFactory;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeLeafFrameFactory;
-import edu.uci.ics.hyracks.storage.am.btree.api.IFieldAccessor;
-import edu.uci.ics.hyracks.storage.am.btree.api.IFieldAccessorFactory;
 import edu.uci.ics.hyracks.storage.am.btree.dataflow.BTreeBulkLoadOperatorDescriptor;
 import edu.uci.ics.hyracks.storage.am.btree.dataflow.BTreeInsertUpdateDeleteOperatorDescriptor;
 import edu.uci.ics.hyracks.storage.am.btree.dataflow.BTreeRegistry;
 import edu.uci.ics.hyracks.storage.am.btree.dataflow.BTreeRegistryProvider;
 import edu.uci.ics.hyracks.storage.am.btree.dataflow.BTreeSearchOperatorDescriptor;
 import edu.uci.ics.hyracks.storage.am.btree.dataflow.BufferCacheProvider;
+import edu.uci.ics.hyracks.storage.am.btree.dataflow.FrameTupleReferenceFactory;
 import edu.uci.ics.hyracks.storage.am.btree.dataflow.IBTreeRegistryProvider;
 import edu.uci.ics.hyracks.storage.am.btree.dataflow.IBufferCacheProvider;
+import edu.uci.ics.hyracks.storage.am.btree.dataflow.ITupleReferenceFactory;
 import edu.uci.ics.hyracks.storage.am.btree.frames.MetaDataFrame;
 import edu.uci.ics.hyracks.storage.am.btree.frames.NSMInteriorFrameFactory;
 import edu.uci.ics.hyracks.storage.am.btree.frames.NSMLeafFrameFactory;
@@ -70,16 +73,13 @@
 import edu.uci.ics.hyracks.storage.am.btree.impls.MultiComparator;
 import edu.uci.ics.hyracks.storage.am.btree.impls.RangePredicate;
 import edu.uci.ics.hyracks.storage.am.btree.impls.RangeSearchCursor;
-import edu.uci.ics.hyracks.storage.am.btree.impls.SelfDescTupleReference;
-import edu.uci.ics.hyracks.storage.am.btree.types.Int32AccessorFactory;
-import edu.uci.ics.hyracks.storage.am.btree.types.UTF8StringAccessorFactory;
 import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
 import edu.uci.ics.hyracks.storage.common.file.FileInfo;
 import edu.uci.ics.hyracks.storage.common.file.FileManager;
 import edu.uci.ics.hyracks.tests.integration.AbstractIntegrationTest;
 
 public class BTreeOperatorsTest extends AbstractIntegrationTest {
-	
+		
 	@Test
 	public void bulkLoadTest() throws Exception {
 		JobSpecification spec = new JobSpecification();
@@ -115,20 +115,16 @@
 		
 		IBufferCacheProvider bufferCacheProvider = new BufferCacheProvider();		
 		IBTreeRegistryProvider btreeRegistryProvider = new BTreeRegistryProvider();
-        
-		IFieldAccessorFactory[] fieldAccessorFactories = new IFieldAccessorFactory[3];
-		fieldAccessorFactories[0] = new UTF8StringAccessorFactory(); // key
-		fieldAccessorFactories[1] = new UTF8StringAccessorFactory(); // payload
-		fieldAccessorFactories[2] = new UTF8StringAccessorFactory(); // payload
-
-		int keyLen = 1;
-		IBinaryComparatorFactory[] comparatorFactories = new IBinaryComparatorFactory[keyLen];
+        		
+		int fieldCount = 3;
+		int keyFieldCount = 1;
+		IBinaryComparatorFactory[] comparatorFactories = new IBinaryComparatorFactory[keyFieldCount];
 		comparatorFactories[0] = UTF8StringBinaryComparatorFactory.INSTANCE;		
 				
 		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, fieldPermutation, 0.7f);		
+		BTreeBulkLoadOperatorDescriptor btreeBulkLoad = new BTreeBulkLoadOperatorDescriptor(spec, ordersSplitProvider, ordersDesc, bufferCacheProvider, btreeRegistryProvider, btreeFileId, "/tmp/btreetest.bin", interiorFrameFactory, leafFrameFactory, fieldCount, comparatorFactories, fieldPermutation, 0.7f);		
 		PartitionConstraint btreePartitionConstraintA = new ExplicitPartitionConstraint(new LocationConstraint[] { new AbsoluteLocationConstraint(NC1_ID) });
 		btreeBulkLoad.setPartitionConstraint(btreePartitionConstraintA);
 				
@@ -139,18 +135,13 @@
         spec.addRoot(btreeBulkLoad);
         runTest(spec);
         
-        // construct a multicomparator from the factories (only for printing purposes)                
-        IFieldAccessor[] fields = new IFieldAccessor[fieldAccessorFactories.length];
-    	for(int i = 0; i < fieldAccessorFactories.length; i++) {
-    		fields[i] = fieldAccessorFactories[i].getFieldAccessor();
-    	}
-    	
+        // construct a multicomparator from the factories (only for printing purposes)        
     	IBinaryComparator[] comparators = new IBinaryComparator[comparatorFactories.length];
     	for(int i = 0; i < comparatorFactories.length; i++) {
     		comparators[i] = comparatorFactories[i].createBinaryComparator();
     	}
     	
-        MultiComparator cmp = new MultiComparator(comparators, fields);
+        MultiComparator cmp = new MultiComparator(fieldCount, comparators);
 
         // try an ordered scan on the bulk-loaded btree
         BTree btreeA = btreeRegistryProvider.getBTreeRegistry().get(btreeFileId);
@@ -169,9 +160,8 @@
         } finally {
         	scanCursor.close();
         }                        
-	}
+	}	
 	
-	/*
 	@Test
 	public void btreeSearchTest() throws Exception {
 		JobSpecification spec = new JobSpecification();
@@ -181,46 +171,49 @@
 		
 		IBTreeInteriorFrameFactory interiorFrameFactory = new NSMInteriorFrameFactory();
 		IBTreeLeafFrameFactory leafFrameFactory = new NSMLeafFrameFactory();        
-		
-		IFieldAccessorFactory[] fieldAccessorFactories = new IFieldAccessorFactory[2];
-		fieldAccessorFactories[0] = new Int32AccessorFactory(); // key
-		fieldAccessorFactories[1] = new Int32AccessorFactory(); // value
-						
-		int keyLen = 1;
-		IBinaryComparatorFactory[] comparatorFactories = new IBinaryComparatorFactory[keyLen];
+				
+		int fieldCount = 2;
+		int keyFieldCount = 1;
+		IBinaryComparatorFactory[] comparatorFactories = new IBinaryComparatorFactory[keyFieldCount];
 		comparatorFactories[0] = IntegerBinaryComparatorFactory.INSTANCE;	
 		
-		// construct a multicomparator from the factories (only for printing purposes)                
-        IFieldAccessor[] fields = new IFieldAccessor[fieldAccessorFactories.length];
-    	for(int i = 0; i < fieldAccessorFactories.length; i++) {
-    		fields[i] = fieldAccessorFactories[i].getFieldAccessor();
-    	}
-    	
+		// construct a multicomparator from the factories (only for printing purposes)        
     	IBinaryComparator[] comparators = new IBinaryComparator[comparatorFactories.length];
     	for(int i = 0; i < comparatorFactories.length; i++) {
     		comparators[i] = comparatorFactories[i].createBinaryComparator();
     	}
-    	
-        MultiComparator cmp = new MultiComparator(comparators, fields);
+    	    	    	    	
+        MultiComparator cmp = new MultiComparator(fieldCount, comparators);
 		
-		ByteArrayAccessibleOutputStream lkbaaos = new ByteArrayAccessibleOutputStream();
-		DataOutputStream lkdos = new DataOutputStream(lkbaaos);    	    	    	
-		IntegerSerializerDeserializer.INSTANCE.serialize(-1000, lkdos);
-
-		ByteArrayAccessibleOutputStream hkbaaos = new ByteArrayAccessibleOutputStream();
-		DataOutputStream hkdos = new DataOutputStream(hkbaaos);    	    	    	
-		IntegerSerializerDeserializer.INSTANCE.serialize(1000, hkdos);
-				
-		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);
+        // put search keys into frame and create tuplereference factories
+        IHyracksContext ctx = new HyracksContext(32768); // WARNING: make sure frame size is same as on NCs
+        ByteBuffer keyFrame = ctx.getResourceManager().allocateFrame();
+		FrameTupleAppender appender = new FrameTupleAppender(ctx);				
+		appender.reset(keyFrame, true);
+		ArrayTupleBuilder tb = new ArrayTupleBuilder(cmp.getKeyFieldCount());
+		DataOutput dos = tb.getDataOutput();
 		
+		ISerializerDeserializer[] keyRecDescSers = { IntegerSerializerDeserializer.INSTANCE};
+		RecordDescriptor keyRecDesc = new RecordDescriptor(keyRecDescSers);
+		
+		// build low key
+		tb.reset();
+    	IntegerSerializerDeserializer.INSTANCE.serialize(-1000, dos);
+    	tb.addFieldEndOffset();    	  
+    	        	    	
+    	appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+    	    	
+    	// build high key
+    	tb.reset();
+    	IntegerSerializerDeserializer.INSTANCE.serialize(1000, dos);
+    	tb.addFieldEndOffset();
+    	
+    	// build search key factories
+    	ITupleReferenceFactory[] searchKeys = new ITupleReferenceFactory[2]; 
+    	searchKeys[0] = new FrameTupleReferenceFactory(keyFrame.array(), 0, keyRecDesc);
+    	searchKeys[1] = new FrameTupleReferenceFactory(keyFrame.array(), 1, keyRecDesc);
+    	    	    	
 		IBufferCacheProvider bufferCacheProvider = new BufferCacheProvider();
 		IBTreeRegistryProvider btreeRegistryProvider = new BTreeRegistryProvider();
 		
@@ -228,7 +221,7 @@
                 new ISerializerDeserializer[] { IntegerSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE });
 		
 		int btreeFileId = 2;		
-		BTreeSearchOperatorDescriptor btreeSearchOp = new BTreeSearchOperatorDescriptor(spec, splitProvider, recDesc, bufferCacheProvider, btreeRegistryProvider, btreeFileId, "/tmp/btreetest.bin", interiorFrameFactory, leafFrameFactory, fieldAccessorFactories, comparatorFactories, true, lowKey, highKey, comparatorFactories.length);
+		BTreeSearchOperatorDescriptor btreeSearchOp = new BTreeSearchOperatorDescriptor(spec, splitProvider, recDesc, bufferCacheProvider, btreeRegistryProvider, btreeFileId, "/tmp/btreetest.bin", interiorFrameFactory, leafFrameFactory, fieldCount, comparatorFactories, true, searchKeys, comparatorFactories.length);
 		//BTreeDiskOrderScanOperatorDescriptor btreeSearchOp = new BTreeDiskOrderScanOperatorDescriptor(spec, splitProvider, recDesc, bufferCacheProvider, btreeRegistryProvider, 0, "/tmp/btreetest.bin", interiorFrameFactory, leafFrameFactory, cmp);
 		
 		PartitionConstraint btreePartitionConstraint = new ExplicitPartitionConstraint(new LocationConstraint[] { new AbsoluteLocationConstraint(NC1_ID) });
@@ -244,7 +237,7 @@
         spec.addRoot(printer);
         runTest(spec);
     }
-	*/
+	
 	
 	@Test
 	public void insertTest() throws Exception {
@@ -277,55 +270,35 @@
 		IBTreeRegistryProvider btreeRegistryProvider = new BTreeRegistryProvider();
 		
 		// we will create a primary index and 2 secondary indexes
-		// first create fieldaccessors and comparators for primary index
-		IFieldAccessorFactory[] primaryFieldAccessorFactories = new IFieldAccessorFactory[6];
-		primaryFieldAccessorFactories[0] = new UTF8StringAccessorFactory(); // key
-		primaryFieldAccessorFactories[1] = new UTF8StringAccessorFactory(); // payload
-		primaryFieldAccessorFactories[2] = new UTF8StringAccessorFactory(); // payload
-		primaryFieldAccessorFactories[3] = new UTF8StringAccessorFactory(); // payload
-		primaryFieldAccessorFactories[4] = new UTF8StringAccessorFactory(); // payload
-		primaryFieldAccessorFactories[5] = new UTF8StringAccessorFactory(); // payload		
-		
-		int primaryKeyLen = 1;
-		IBinaryComparatorFactory[] primaryComparatorFactories = new IBinaryComparatorFactory[primaryKeyLen];
+		// first create comparators for primary index		
+		int primaryFieldCount = 6;
+		int primaryKeyFieldCount = 1;
+		IBinaryComparatorFactory[] primaryComparatorFactories = new IBinaryComparatorFactory[primaryKeyFieldCount];
 		primaryComparatorFactories[0] = UTF8StringBinaryComparatorFactory.INSTANCE;		
 		
-		// construct a multicomparator for the primary index                
-        IFieldAccessor[] primaryFields = new IFieldAccessor[primaryFieldAccessorFactories.length];
-    	for(int i = 0; i < primaryFieldAccessorFactories.length; i++) {
-    		primaryFields[i] = primaryFieldAccessorFactories[i].getFieldAccessor();
-    	}
-    	
+		// construct a multicomparator for the primary index       
     	IBinaryComparator[] primaryComparators = new IBinaryComparator[primaryComparatorFactories.length];
     	for(int i = 0; i < primaryComparatorFactories.length; i++) {
     		primaryComparators[i] = primaryComparatorFactories[i].createBinaryComparator();
     	}
     	
-        MultiComparator primaryCmp = new MultiComparator(primaryComparators, primaryFields);
+        MultiComparator primaryCmp = new MultiComparator(primaryFieldCount, primaryComparators);
         
         
-        // now create fieldaccessors and comparators for secondary indexes
-		IFieldAccessorFactory[] secondaryFieldAccessorFactories = new IFieldAccessorFactory[2];
-		secondaryFieldAccessorFactories[0] = new UTF8StringAccessorFactory(); // key
-		secondaryFieldAccessorFactories[1] = new UTF8StringAccessorFactory(); // key (key in primary index)
-		
-		int secondaryKeyLen = 2;
-		IBinaryComparatorFactory[] secondaryComparatorFactories = new IBinaryComparatorFactory[secondaryKeyLen];
+        // now create comparators for secondary indexes		
+		int secondaryFieldCount = 2;
+		int secondaryKeyFieldCount = 2;
+		IBinaryComparatorFactory[] secondaryComparatorFactories = new IBinaryComparatorFactory[secondaryKeyFieldCount];
 		secondaryComparatorFactories[0] = UTF8StringBinaryComparatorFactory.INSTANCE;
 		secondaryComparatorFactories[1] = UTF8StringBinaryComparatorFactory.INSTANCE;		
 		
-		// construct a multicomparator for the secondary indexes                
-        IFieldAccessor[] secondaryFields = new IFieldAccessor[secondaryFieldAccessorFactories.length];
-    	for(int i = 0; i < secondaryFieldAccessorFactories.length; i++) {
-    		secondaryFields[i] = secondaryFieldAccessorFactories[i].getFieldAccessor();
-    	}
-    	
+		// construct a multicomparator for the secondary indexes        
     	IBinaryComparator[] secondaryComparators = new IBinaryComparator[secondaryComparatorFactories.length];
     	for(int i = 0; i < secondaryComparatorFactories.length; i++) {
     		secondaryComparators[i] = secondaryComparatorFactories[i].createBinaryComparator();
     	}
     	
-        MultiComparator secondaryCmp = new MultiComparator(secondaryComparators, secondaryFields);
+        MultiComparator secondaryCmp = new MultiComparator(secondaryFieldCount, secondaryComparators);
         
         // we create and register 3 btrees for in an insert pipeline being fed from a filescan op        
         IBufferCache bufferCache = bufferCacheProvider.getBufferCache();
@@ -370,19 +343,19 @@
         
         // primary index
         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);
+        BTreeInsertUpdateDeleteOperatorDescriptor insertOpA = new BTreeInsertUpdateDeleteOperatorDescriptor(spec, ordersSplitProvider, ordersDesc, bufferCacheProvider, btreeRegistryProvider, fileIdA, "/tmp/btreetestA.ix", interiorFrameFactory, leafFrameFactory, primaryFieldCount, primaryComparatorFactories, fieldPermutationA, BTreeOp.BTO_INSERT);
         PartitionConstraint insertPartitionConstraintA = new ExplicitPartitionConstraint(new LocationConstraint[] { new AbsoluteLocationConstraint(NC1_ID) });
         insertOpA.setPartitionConstraint(insertPartitionConstraintA);
         
         // first secondary index
         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);
+        BTreeInsertUpdateDeleteOperatorDescriptor insertOpB = new BTreeInsertUpdateDeleteOperatorDescriptor(spec, ordersSplitProvider, ordersDesc, bufferCacheProvider, btreeRegistryProvider, fileIdB, "/tmp/btreetestB.ix", interiorFrameFactory, leafFrameFactory, secondaryFieldCount, secondaryComparatorFactories, fieldPermutationB, BTreeOp.BTO_INSERT);
         PartitionConstraint insertPartitionConstraintB = new ExplicitPartitionConstraint(new LocationConstraint[] { new AbsoluteLocationConstraint(NC1_ID) });
         insertOpB.setPartitionConstraint(insertPartitionConstraintB);
 		
         // second secondary index
         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);
+        BTreeInsertUpdateDeleteOperatorDescriptor insertOpC = new BTreeInsertUpdateDeleteOperatorDescriptor(spec, ordersSplitProvider, ordersDesc, bufferCacheProvider, btreeRegistryProvider, fileIdC, "/tmp/btreetestC.ix", interiorFrameFactory, leafFrameFactory, secondaryFieldCount, secondaryComparatorFactories, fieldPermutationC, BTreeOp.BTO_INSERT);
         PartitionConstraint insertPartitionConstraintC = new ExplicitPartitionConstraint(new LocationConstraint[] { new AbsoluteLocationConstraint(NC1_ID) });
         insertOpC.setPartitionConstraint(insertPartitionConstraintC);
                 
@@ -457,5 +430,5 @@
         	scanCursorC.close();
         }        
         System.out.println();
-	}
+	}	
 }
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IBTreeFrame.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IBTreeFrame.java
index b0ad734..018c42f 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IBTreeFrame.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IBTreeFrame.java
@@ -17,6 +17,8 @@
 
 import java.nio.ByteBuffer;
 
+import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
+import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
 import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
 import edu.uci.ics.hyracks.storage.am.btree.impls.MultiComparator;
 import edu.uci.ics.hyracks.storage.am.btree.impls.SpaceStatus;
@@ -52,7 +54,7 @@
 	
 	// for debugging
 	public void printHeader();
-	public String printKeys(MultiComparator cmp, int fieldCount);
+	public String printKeys(MultiComparator cmp, ISerializerDeserializer[] fields) throws HyracksDataException;
 	
 	
 	// TODO; what if tuples more than half-page size?
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IFieldAccessor.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IFieldAccessor.java
deleted file mode 100644
index a076f0f..0000000
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IFieldAccessor.java
+++ /dev/null
@@ -1,21 +0,0 @@
-/*
- * Copyright 2009-2010 by The Regents of the University of California
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * you may obtain a copy of the License from
- * 
- *     http://www.apache.org/licenses/LICENSE-2.0
- * 
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package edu.uci.ics.hyracks.storage.am.btree.api;
-
-public interface IFieldAccessor {	    
-    public int getLength(byte[] data, int offset); // skip to next field (equivalent to adding length of field to offset)        
-	public String print(byte[] data, int offset); // debug
-}
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IFieldAccessorFactory.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IFieldAccessorFactory.java
deleted file mode 100644
index 4fbcceb..0000000
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IFieldAccessorFactory.java
+++ /dev/null
@@ -1,22 +0,0 @@
-/*
- * Copyright 2009-2010 by The Regents of the University of California
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * you may obtain a copy of the License from
- * 
- *     http://www.apache.org/licenses/LICENSE-2.0
- * 
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package edu.uci.ics.hyracks.storage.am.btree.api;
-
-import java.io.Serializable;
-
-public interface IFieldAccessorFactory extends Serializable {
-	public IFieldAccessor getFieldAccessor();
-}
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/compressors/FieldPrefixCompressor.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/compressors/FieldPrefixCompressor.java
index cd637ea..d342e6f 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/compressors/FieldPrefixCompressor.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/compressors/FieldPrefixCompressor.java
@@ -21,7 +21,6 @@
 import java.util.Comparator;
 
 import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparator;
-import edu.uci.ics.hyracks.storage.am.btree.api.IFieldAccessor;
 import edu.uci.ics.hyracks.storage.am.btree.api.IFrameCompressor;
 import edu.uci.ics.hyracks.storage.am.btree.api.IPrefixSlotManager;
 import edu.uci.ics.hyracks.storage.am.btree.frames.FieldPrefixNSMLeafFrame;
@@ -58,7 +57,7 @@
     	if(ratio < ratioThreshold) return false;    	
     	
         IBinaryComparator[] cmps = cmp.getComparators();
-        IFieldAccessor[] fields = cmp.getFields();
+        int fieldCount = cmp.getKeyFieldCount();
         
         ByteBuffer buf = frame.getBuffer();
         byte[] pageArray = buf.array();
@@ -147,7 +146,7 @@
         uncompressedTupleCount = 0;
         
         FieldPrefixTupleReference tupleToWrite = new FieldPrefixTupleReference();
-        tupleToWrite.setFieldCount(fields.length);
+        tupleToWrite.setFieldCount(fieldCount);
         
         SimpleTupleWriter tupleWriter = new SimpleTupleWriter();
         
@@ -165,10 +164,10 @@
             		//System.out.println("PROCESSING KEYPARTITION: " + kpIndex + " RANGE: " + keyPartitions.get(kpIndex).firstRecSlotNum + " " + keyPartitions.get(kpIndex).lastRecSlotNum + " FIELDSTOCOMPRESS: " + numFieldsToCompress);
             		
             		FieldPrefixTupleReference prevTuple = new FieldPrefixTupleReference();
-            		prevTuple.setFieldCount(fields.length);
+            		prevTuple.setFieldCount(fieldCount);
             		
             		FieldPrefixTupleReference tuple = new FieldPrefixTupleReference();
-            		tuple.setFieldCount(fields.length);
+            		tuple.setFieldCount(fieldCount);
             		            	
                     for(int i = tupleIndex + 1; i <= keyPartitions.get(kpIndex).lastTupleIndex; i++) {
                     	prevTuple.resetByTupleIndex(frame, i - 1);
@@ -218,7 +217,7 @@
                         	        int currTupleIndex = segmentStart + j;
                                     tupleToWrite.resetByTupleIndex(frame, currTupleIndex);
                                     newTupleSlots[tupleCount - 1 - currTupleIndex] = slotManager.encodeSlotFields(prefixTupleIndex, tupleFreeSpace);                                    
-                                    tupleFreeSpace += tupleWriter.writeTupleFields(tupleToWrite, numFieldsToCompress, fields.length - numFieldsToCompress, byteBuffer, tupleFreeSpace);
+                                    tupleFreeSpace += tupleWriter.writeTupleFields(tupleToWrite, numFieldsToCompress, fieldCount - numFieldsToCompress, byteBuffer, tupleFreeSpace);
                                 }
                         	    
                         	    prefixTupleIndex++;
@@ -309,7 +308,7 @@
     // the occurrenceThreshold determines the minimum number of tuples that must share a common prefix in order for us to consider compressing them        
     private ArrayList<KeyPartition> getKeyPartitions(FieldPrefixNSMLeafFrame frame, MultiComparator cmp, int occurrenceThreshold) {        
     	IBinaryComparator[] cmps = cmp.getComparators();
-        IFieldAccessor[] fields = cmp.getFields();
+        int fieldCount = cmp.getKeyFieldCount();
         
         int maxCmps = cmps.length - 1;
         ByteBuffer buf = frame.getBuffer();
@@ -321,10 +320,10 @@
         keyPartitions.add(kp);
         
         FieldPrefixTupleReference prevTuple = new FieldPrefixTupleReference();
-		prevTuple.setFieldCount(fields.length);
+		prevTuple.setFieldCount(fieldCount);
 		
 		FieldPrefixTupleReference tuple = new FieldPrefixTupleReference();
-		tuple.setFieldCount(fields.length);
+		tuple.setFieldCount(fieldCount);
 		
 		SimpleTupleWriter tupleWriter = new SimpleTupleWriter();
 		
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/AbstractBTreeOperatorDescriptor.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/AbstractBTreeOperatorDescriptor.java
index d1ade0c..54a86a4 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/AbstractBTreeOperatorDescriptor.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/AbstractBTreeOperatorDescriptor.java
@@ -22,7 +22,6 @@
 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 abstract class AbstractBTreeOperatorDescriptor extends AbstractSingleActivityOperatorDescriptor {
 	
@@ -31,7 +30,7 @@
 	protected String btreeFileName;
 	protected int btreeFileId;
 		
-	protected IFieldAccessorFactory[] fieldAccessorFactories;
+	protected int fieldCount;
 	protected IBinaryComparatorFactory[] comparatorFactories;	
 	
 	protected IBTreeInteriorFrameFactory interiorFrameFactory;
@@ -40,7 +39,7 @@
 	protected IBufferCacheProvider bufferCacheProvider;
 	protected IBTreeRegistryProvider btreeRegistryProvider;
 	
-	public AbstractBTreeOperatorDescriptor(JobSpecification spec, int inputArity, int outputArity, IFileSplitProvider fileSplitProvider, RecordDescriptor recDesc, IBufferCacheProvider bufferCacheProvider, IBTreeRegistryProvider btreeRegistryProvider,  int btreeFileId, String btreeFileName, IBTreeInteriorFrameFactory interiorFactory, IBTreeLeafFrameFactory leafFactory, IFieldAccessorFactory[] fieldAccessorFactories, IBinaryComparatorFactory[] comparatorFactories) {
+	public AbstractBTreeOperatorDescriptor(JobSpecification spec, int inputArity, int outputArity, IFileSplitProvider fileSplitProvider, RecordDescriptor recDesc, IBufferCacheProvider bufferCacheProvider, IBTreeRegistryProvider btreeRegistryProvider,  int btreeFileId, String btreeFileName, IBTreeInteriorFrameFactory interiorFactory, IBTreeLeafFrameFactory leafFactory, int fieldCount, IBinaryComparatorFactory[] comparatorFactories) {
         super(spec, inputArity, outputArity);
         this.btreeFileId = btreeFileId;
         this.btreeFileName = btreeFileName;
@@ -48,7 +47,7 @@
         this.btreeRegistryProvider = btreeRegistryProvider;        
         this.interiorFrameFactory = interiorFactory;
         this.leafFrameFactory = leafFactory;
-        this.fieldAccessorFactories = fieldAccessorFactories;
+        this.fieldCount = fieldCount;
         this.comparatorFactories = comparatorFactories;        
         if(outputArity > 0) recordDescriptors[0] = recDesc;   
     }
@@ -65,8 +64,8 @@
 		return comparatorFactories;
 	}
 	
-	public IFieldAccessorFactory[] getFieldAccessorFactories() {
-		return fieldAccessorFactories;
+	public int getFieldCount() {
+		return fieldCount;
 	}
 		
 	public IBTreeInteriorFrameFactory getInteriorFactory() {
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeBulkLoadOperatorDescriptor.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeBulkLoadOperatorDescriptor.java
index 446538b..403da8f 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeBulkLoadOperatorDescriptor.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeBulkLoadOperatorDescriptor.java
@@ -25,7 +25,6 @@
 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 BTreeBulkLoadOperatorDescriptor extends AbstractBTreeOperatorDescriptor {
 	
@@ -39,12 +38,12 @@
 			IBufferCacheProvider bufferCacheProvider,
 			IBTreeRegistryProvider btreeRegistryProvider, int btreeFileId,
 			String btreeFileName, IBTreeInteriorFrameFactory interiorFactory,
-			IBTreeLeafFrameFactory leafFactory, IFieldAccessorFactory[] fieldAccessorFactories, 
+			IBTreeLeafFrameFactory leafFactory, int fieldCount, 
 			IBinaryComparatorFactory[] comparatorFactories,			
 			int[] fieldPermutation, float fillFactor) {
 		super(spec, 1, 0, fileSplitProvider, recDesc, bufferCacheProvider,
 				btreeRegistryProvider, btreeFileId, btreeFileName, interiorFactory,
-				leafFactory, fieldAccessorFactories, comparatorFactories);
+				leafFactory, fieldCount, comparatorFactories);
 		this.fieldPermutation = fieldPermutation;
 		this.fillFactor = fillFactor;
 	}
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeDiskOrderScanOperatorDescriptor.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeDiskOrderScanOperatorDescriptor.java
index 8b1da5a..28f5648 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeDiskOrderScanOperatorDescriptor.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeDiskOrderScanOperatorDescriptor.java
@@ -25,7 +25,6 @@
 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 BTreeDiskOrderScanOperatorDescriptor extends AbstractBTreeOperatorDescriptor {
 	
@@ -37,10 +36,10 @@
 			IBTreeRegistryProvider btreeRegistryProvider, int btreeFileId,
 			String btreeFileName, IBTreeInteriorFrameFactory interiorFactory,
 			IBTreeLeafFrameFactory leafFactory, 
-			IFieldAccessorFactory[] fieldAccessorFactories, IBinaryComparatorFactory[] comparatorFactories) {
+			int fieldCount, IBinaryComparatorFactory[] comparatorFactories) {
 		super(spec, 0, 1, fileSplitProvider, recDesc, bufferCacheProvider,
 				btreeRegistryProvider, btreeFileId, btreeFileName, interiorFactory,
-				leafFactory, fieldAccessorFactories, comparatorFactories);
+				leafFactory, fieldCount, comparatorFactories);
 	}
 	
 	@Override
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeDiskOrderScanOperatorNodePushable.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeDiskOrderScanOperatorNodePushable.java
index bcfcd08..13b27b6 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeDiskOrderScanOperatorNodePushable.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeDiskOrderScanOperatorNodePushable.java
@@ -59,7 +59,7 @@
         ByteBuffer frame = btreeOpHelper.getHyracksContext().getResourceManager().allocateFrame();
         FrameTupleAppender appender = new FrameTupleAppender(btreeOpHelper.getHyracksContext());
         appender.reset(frame, true);
-        ArrayTupleBuilder tb = new ArrayTupleBuilder(cmp.getFields().length);
+        ArrayTupleBuilder tb = new ArrayTupleBuilder(cmp.getFieldCount());
         DataOutput dos = tb.getDataOutput();
 
         try {
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeInsertUpdateDeleteOperatorDescriptor.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeInsertUpdateDeleteOperatorDescriptor.java
index e3da2c2..8352b2a 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeInsertUpdateDeleteOperatorDescriptor.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeInsertUpdateDeleteOperatorDescriptor.java
@@ -25,7 +25,6 @@
 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;
 import edu.uci.ics.hyracks.storage.am.btree.impls.BTreeOp;
 
 public class BTreeInsertUpdateDeleteOperatorDescriptor extends AbstractBTreeOperatorDescriptor {
@@ -41,12 +40,12 @@
 			IBufferCacheProvider bufferCacheProvider,
 			IBTreeRegistryProvider btreeRegistryProvider, int btreeFileId,
 			String btreeFileName, IBTreeInteriorFrameFactory interiorFactory,
-			IBTreeLeafFrameFactory leafFactory, IFieldAccessorFactory[] fieldAccessorFactories, 
+			IBTreeLeafFrameFactory leafFactory, int fieldCount, 
 			IBinaryComparatorFactory[] comparatorFactories,			
 			int[] fieldPermutation, BTreeOp op) {
 		super(spec, 1, 1, fileSplitProvider, recDesc, bufferCacheProvider,
 				btreeRegistryProvider, btreeFileId, btreeFileName, interiorFactory,
-				leafFactory, fieldAccessorFactories, comparatorFactories);
+				leafFactory, fieldCount, comparatorFactories);
 		this.fieldPermutation = fieldPermutation;		
 		this.op = op;
 	}
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeOpHelper.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeOpHelper.java
index 68edce1..8ddcf08 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeOpHelper.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeOpHelper.java
@@ -32,7 +32,6 @@
 import edu.uci.ics.hyracks.dataflow.common.data.marshalling.IntegerSerializerDeserializer;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeInteriorFrame;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeLeafFrame;
-import edu.uci.ics.hyracks.storage.am.btree.api.IFieldAccessor;
 import edu.uci.ics.hyracks.storage.am.btree.frames.MetaDataFrame;
 import edu.uci.ics.hyracks.storage.am.btree.impls.BTree;
 import edu.uci.ics.hyracks.storage.am.btree.impls.MultiComparator;
@@ -58,30 +57,30 @@
     }
 
     void init() throws Exception {
-        IBufferCache bufferCache = opDesc.getBufferCacheProvider().getBufferCache();
+    	
+    	IBufferCache bufferCache = opDesc.getBufferCacheProvider().getBufferCache();
         FileManager fileManager = opDesc.getBufferCacheProvider().getFileManager();
 
         File f = new File(opDesc.getBtreeFileName());
         RandomAccessFile raf = new RandomAccessFile(f, "rw");
-
+        
         if (!f.exists() && !createBTree) {
-            throw new Exception("Trying to open btree from file " + opDesc.getBtreeFileName()
-                    + " but file doesn't exist.");
+            throw new Exception("Trying to open btree from file " + opDesc.getBtreeFileName() + " but file doesn't exist.");
         }
-
+        
         try {
             FileInfo fi = new FileInfo(opDesc.getBtreeFileId(), raf);
             fileManager.registerFile(fi);
         } catch (Exception e) {
         }
-
+        
         interiorFrame = opDesc.getInteriorFactory().getFrame();
         leafFrame = opDesc.getLeafFactory().getFrame();
 
         BTreeRegistry btreeRegistry = opDesc.getBtreeRegistryProvider().getBTreeRegistry();
         btree = btreeRegistry.get(opDesc.getBtreeFileId());
         if (btree == null) {
-
+        	
             // create new btree and register it            
             btreeRegistry.lock();
             try {
@@ -89,20 +88,14 @@
                 btree = btreeRegistry.get(opDesc.getBtreeFileId());
                 if (btree == null) {
                     // this thread should create and register the btee
-
-                    // start by building the multicomparator from the factories
-                    IFieldAccessor[] fields = new IFieldAccessor[opDesc.getFieldAccessorFactories().length];
-                    for (int i = 0; i < opDesc.getFieldAccessorFactories().length; i++) {
-                        fields[i] = opDesc.getFieldAccessorFactories()[i].getFieldAccessor();
-                    }
-
+                	                   
                     IBinaryComparator[] comparators = new IBinaryComparator[opDesc.getComparatorFactories().length];
                     for (int i = 0; i < opDesc.getComparatorFactories().length; i++) {
                         comparators[i] = opDesc.getComparatorFactories()[i].createBinaryComparator();
                     }
 
-                    MultiComparator cmp = new MultiComparator(comparators, fields);
-
+                    MultiComparator cmp = new MultiComparator(opDesc.getFieldCount(), comparators);
+                    
                     btree = new BTree(bufferCache, opDesc.getInteriorFactory(), opDesc.getLeafFactory(), cmp);
                     if (createBTree) {
                         MetaDataFrame metaFrame = new MetaDataFrame();
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeSearchOperatorDescriptor.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeSearchOperatorDescriptor.java
index c0b246f..7f2f6d1 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeSearchOperatorDescriptor.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeSearchOperatorDescriptor.java
@@ -22,32 +22,28 @@
 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 ITupleReference lowKey;
-	private ITupleReference highKey;
-	private int searchKeyFields;
+	private ITupleReferenceFactory[] searchKeys; // create tuples for low and high keys	
+	private int searchKeyFieldCount;
 	
-	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);
+	public BTreeSearchOperatorDescriptor(JobSpecification spec, IFileSplitProvider fileSplitProvider, RecordDescriptor recDesc, IBufferCacheProvider bufferCacheProvider, IBTreeRegistryProvider btreeRegistryProvider,  int btreeFileId, String btreeFileName, IBTreeInteriorFrameFactory interiorFactory, IBTreeLeafFrameFactory leafFactory, int fieldCount, IBinaryComparatorFactory[] comparatorFactories, boolean isForward, ITupleReferenceFactory[] searchKeys, int searchKeyFields) {
+		super(spec, 0, 1, fileSplitProvider, recDesc, bufferCacheProvider, btreeRegistryProvider, btreeFileId, btreeFileName, interiorFactory, leafFactory, fieldCount, comparatorFactories);
 		this.isForward = isForward;
-		this.lowKey = lowKey;
-		this.highKey = highKey;
-		this.searchKeyFields = searchKeyFields;
+		this.searchKeys = searchKeys;
+		this.searchKeyFieldCount = searchKeyFields;
 	}
 	
 	@Override
 	public IOperatorNodePushable createPushRuntime(final IHyracksContext ctx, final IOperatorEnvironment env,
 			IRecordDescriptorProvider recordDescProvider, int partition, int nPartitions) {
-		return new BTreeSearchOperatorNodePushable(this, ctx, isForward, lowKey, highKey, searchKeyFields);
+		return new BTreeSearchOperatorNodePushable(this, ctx, isForward, searchKeys, searchKeyFieldCount);
 	}
 }
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeSearchOperatorNodePushable.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeSearchOperatorNodePushable.java
index 84b8113..826c7ec 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeSearchOperatorNodePushable.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeSearchOperatorNodePushable.java
@@ -37,41 +37,45 @@
 public class BTreeSearchOperatorNodePushable extends AbstractUnaryOutputSourceOperatorNodePushable {
     private BTreeOpHelper btreeOpHelper;
     private boolean isForward;
-    private ITupleReference lowKey;
-    private ITupleReference highKey;
-    private int searchKeyFields;
+    private ITupleReferenceFactory[] searchKeys;    
+    private int searchKeyFieldCount;
 
     public BTreeSearchOperatorNodePushable(AbstractBTreeOperatorDescriptor opDesc, IHyracksContext ctx,
-            boolean isForward, ITupleReference lowKey, ITupleReference highKey, int searchKeyFields) {
+            boolean isForward, ITupleReferenceFactory[] searchKeys, int searchKeyFields) {
         btreeOpHelper = new BTreeOpHelper(opDesc, ctx, false);
         this.isForward = isForward;
-        this.lowKey = lowKey;
-        this.highKey = highKey;
-        this.searchKeyFields = searchKeyFields;
+        this.searchKeys = searchKeys;
+        this.searchKeyFieldCount = searchKeyFields;
     }
-
+    
     @Override
     public void initialize() throws HyracksDataException {
-        AbstractBTreeOperatorDescriptor opDesc = btreeOpHelper.getOperatorDescriptor();
-        BTree btree = btreeOpHelper.getBTree();
-        IBTreeLeafFrame leafFrame = btreeOpHelper.getLeafFrame();
-        IBTreeInteriorFrame interiorFrame = btreeOpHelper.getInteriorFrame();
+        AbstractBTreeOperatorDescriptor opDesc = btreeOpHelper.getOperatorDescriptor();        
         IHyracksContext ctx = btreeOpHelper.getHyracksContext();
 
         IBTreeLeafFrame cursorFrame = opDesc.getLeafFactory().getFrame();
         IBTreeCursor cursor = new RangeSearchCursor(cursorFrame);
 
+        BTree btree = null;        
         try {
-            btreeOpHelper.init();
+        	btreeOpHelper.init();
             btreeOpHelper.fill();
+            btree = btreeOpHelper.getBTree();
 
+            IBTreeLeafFrame leafFrame = btreeOpHelper.getLeafFrame();
+            IBTreeInteriorFrame interiorFrame = btreeOpHelper.getInteriorFrame();
+            
             // construct range predicate
-            assert (searchKeyFields <= btree.getMultiComparator().getKeyLength());
-            IBinaryComparator[] searchComparators = new IBinaryComparator[searchKeyFields];
-            for (int i = 0; i < searchKeyFields; i++) {
-                searchComparators[i] = btree.getMultiComparator().getComparators()[i];
+            assert (searchKeyFieldCount <= btree.getMultiComparator().getKeyFieldCount());
+            IBinaryComparator[] searchComparators = new IBinaryComparator[searchKeyFieldCount];
+            for (int i = 0; i < searchKeyFieldCount; i++) {                
+            	searchComparators[i] = btree.getMultiComparator().getComparators()[i];
             }
-            MultiComparator searchCmp = new MultiComparator(searchComparators, btree.getMultiComparator().getFields());
+            MultiComparator searchCmp = new MultiComparator(btree.getMultiComparator().getFieldCount(), searchComparators);
+            
+            ITupleReference lowKey = searchKeys[0].createTuple(ctx);
+            ITupleReference highKey = searchKeys[1].createTuple(ctx);
+            
             RangePredicate rangePred = new RangePredicate(isForward, lowKey, highKey, searchCmp);
 
             btree.search(cursor, rangePred, leafFrame, interiorFrame);
@@ -85,7 +89,7 @@
         ByteBuffer frame = ctx.getResourceManager().allocateFrame();
         FrameTupleAppender appender = new FrameTupleAppender(ctx);
         appender.reset(frame, true);
-        ArrayTupleBuilder tb = new ArrayTupleBuilder(cmp.getFields().length);
+        ArrayTupleBuilder tb = new ArrayTupleBuilder(cmp.getFieldCount());
         DataOutput dos = tb.getDataOutput();
 
         try {
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/FrameTupleReferenceFactory.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/FrameTupleReferenceFactory.java
new file mode 100644
index 0000000..0702c29
--- /dev/null
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/FrameTupleReferenceFactory.java
@@ -0,0 +1,35 @@
+package edu.uci.ics.hyracks.storage.am.btree.dataflow;
+
+import java.nio.ByteBuffer;
+
+import edu.uci.ics.hyracks.api.comm.IFrameTupleAccessor;
+import edu.uci.ics.hyracks.api.context.IHyracksContext;
+import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
+import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAccessor;
+import edu.uci.ics.hyracks.dataflow.common.data.accessors.FrameTupleReference;
+import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
+
+public class FrameTupleReferenceFactory implements ITupleReferenceFactory {
+	
+	private static final long serialVersionUID = 1L;
+
+	private byte[] frame;
+	private int tupleIndex;
+	private RecordDescriptor recDesc;
+	
+	public FrameTupleReferenceFactory(byte[] frame, int tupleIndex, RecordDescriptor recDesc) {
+		this.frame = frame;
+		this.tupleIndex = tupleIndex;		
+		this.recDesc = recDesc;
+	}
+	
+	// TODO: lots of object creation, fix later
+	@Override
+	public ITupleReference createTuple(IHyracksContext ctx) {				
+		IFrameTupleAccessor accessor = new FrameTupleAccessor(ctx, recDesc);
+		accessor.reset(ByteBuffer.wrap(frame));
+		FrameTupleReference tuple = new FrameTupleReference();
+		tuple.reset(accessor, tupleIndex);
+		return tuple;
+	}
+}
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/ITupleReferenceFactory.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/ITupleReferenceFactory.java
new file mode 100644
index 0000000..a370f84
--- /dev/null
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/ITupleReferenceFactory.java
@@ -0,0 +1,10 @@
+package edu.uci.ics.hyracks.storage.am.btree.dataflow;
+
+import java.io.Serializable;
+
+import edu.uci.ics.hyracks.api.context.IHyracksContext;
+import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
+
+public interface ITupleReferenceFactory extends Serializable {
+	ITupleReference createTuple(IHyracksContext ctx);
+}
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/FieldPrefixNSMLeafFrame.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/FieldPrefixNSMLeafFrame.java
index 98dbf4c..eb8263c 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/FieldPrefixNSMLeafFrame.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/FieldPrefixNSMLeafFrame.java
@@ -15,10 +15,15 @@
 
 package edu.uci.ics.hyracks.storage.am.btree.frames;
 
+import java.io.ByteArrayInputStream;
+import java.io.DataInput;
+import java.io.DataInputStream;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Collections;
 
+import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
+import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
 import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeFrame;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeLeafFrame;
@@ -98,7 +103,7 @@
     public void compact(MultiComparator cmp) {
         resetSpaceParams();
         
-        frameTuple.setFieldCount(cmp.getFields().length);
+        frameTuple.setFieldCount(cmp.getFieldCount());
         
         int tupleCount = buf.getInt(tupleCountOff);
         
@@ -161,10 +166,10 @@
             int tupleSlotOff = slotManager.getTupleSlotOff(tupleIndex);
             
             if(exactDelete) {                
-                frameTuple.setFieldCount(cmp.getFields().length);
+                frameTuple.setFieldCount(cmp.getFieldCount());
                 frameTuple.resetByTupleIndex(this, tupleIndex);
                 
-                int comparison = cmp.fieldRangeCompare(tuple, frameTuple, cmp.getKeyLength()-1, cmp.getFields().length - cmp.getKeyLength());
+                int comparison = cmp.fieldRangeCompare(tuple, frameTuple, cmp.getKeyFieldCount()-1, cmp.getFieldCount() - cmp.getKeyFieldCount());
                 if(comparison != 0) {
                 	throw new BTreeException("Cannot delete tuple. Byte-by-byte comparison failed to prove equality.");
                 }                                  
@@ -278,29 +283,6 @@
         buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) - bytesWritten - slotManager.getSlotSize());       
     }
     
-    public void verifyPrefixes(MultiComparator cmp) {
-    	int numPrefixes = buf.getInt(prefixTupleCountOff);
-    	int totalPrefixBytes = 0;
-    	for(int i = 0; i < numPrefixes; i++) {
-    		int prefixSlotOff = slotManager.getPrefixSlotOff(i);
-    		int prefixSlot = buf.getInt(prefixSlotOff);
-    		
-    		int numPrefixFields = slotManager.decodeFirstSlotField(prefixSlot);
-    		int prefixTupleOff = slotManager.decodeSecondSlotField(prefixSlot);
-    		
-    		System.out.print("VERIFYING " + i + " : " + numPrefixFields + " " + prefixTupleOff + " ");
-    		int tupleOffRunner = prefixTupleOff;
-    		for(int j = 0; j < numPrefixFields; j++) {
-    			System.out.print(buf.getInt(prefixTupleOff+j*4) + " ");
-    			int length = cmp.getFields()[j].getLength(buf.array(), tupleOffRunner);
-    			tupleOffRunner += length;
-    			totalPrefixBytes += length;    			
-    		}
-    		System.out.println();
-    	}    	        
-    	System.out.println("VER TOTALPREFIXBYTES: " + totalPrefixBytes + " " + numPrefixes);
-    }
-    
     @Override
     public void update(int rid, ITupleReference tuple) throws Exception {
         // TODO Auto-generated method stub
@@ -323,13 +305,17 @@
     }
     
     @Override
-    public String printKeys(MultiComparator cmp, int fieldCount) {      
+    public String printKeys(MultiComparator cmp, ISerializerDeserializer[] fields) throws HyracksDataException {   
     	StringBuilder strBuilder = new StringBuilder();		
 		int tupleCount = buf.getInt(tupleCountOff);
+		frameTuple.setFieldCount(fields.length);
 		for(int i = 0; i < tupleCount; i++) {						
-			frameTuple.resetByTupleIndex(this, i);			
-			for(int j = 0; j < cmp.getKeyLength(); j++) {
-				strBuilder.append(cmp.getFields()[j].print(buf.array(), frameTuple.getFieldStart(j)) + " ");	
+			frameTuple.resetByTupleIndex(this, i);												
+			for(int j = 0; j < cmp.getKeyFieldCount(); j++) {
+				ByteArrayInputStream inStream = new ByteArrayInputStream(frameTuple.getFieldData(j), frameTuple.getFieldStart(j), frameTuple.getFieldLength(j));
+				DataInput dataIn = new DataInputStream(inStream);
+				Object o = fields[j].deserialize(dataIn);
+				strBuilder.append(o.toString() + " ");				
 			}
 			strBuilder.append(" | ");				
 		}
@@ -428,7 +414,7 @@
     	
     	FieldPrefixNSMLeafFrame rf = (FieldPrefixNSMLeafFrame)rightFrame;
     	
-    	frameTuple.setFieldCount(cmp.getFields().length);
+    	frameTuple.setFieldCount(cmp.getFieldCount());
     	
     	// before doing anything check if key already exists
 		int slot = slotManager.findSlot(tuple, frameTuple, framePrefixTuple, cmp, true);
@@ -546,9 +532,9 @@
 		// set split key to be highest value in left page		
 		frameTuple.resetByTupleIndex(this, getTupleCount()-1);
 		
-		int splitKeySize = tupleWriter.bytesRequired(frameTuple, 0, cmp.getKeyLength());
+		int splitKeySize = tupleWriter.bytesRequired(frameTuple, 0, cmp.getKeyFieldCount());
 		splitKey.initData(splitKeySize);
-		tupleWriter.writeTupleFields(frameTuple, 0, cmp.getKeyLength(), splitKey.getBuffer(), 0);
+		tupleWriter.writeTupleFields(frameTuple, 0, cmp.getKeyFieldCount(), splitKey.getBuffer(), 0);
 				
 		return 0;
     }
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMFrame.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMFrame.java
index c6a27b3..a378a26 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMFrame.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMFrame.java
@@ -15,10 +15,15 @@
 
 package edu.uci.ics.hyracks.storage.am.btree.frames;
 
+import java.io.ByteArrayInputStream;
+import java.io.DataInput;
+import java.io.DataInputStream;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Collections;
 
+import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
+import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
 import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeFrame;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeTupleReference;
@@ -117,7 +122,7 @@
 	@Override
 	public void compact(MultiComparator cmp) {		
 		resetSpaceParams();
-		frameTuple.setFieldCount(cmp.getFields().length);	
+		frameTuple.setFieldCount(cmp.getFieldCount());	
 		
 		int tupleCount = buf.getInt(tupleCountOff);
 		int freeSpace = buf.getInt(freeSpaceOff);		
@@ -149,7 +154,7 @@
 
 	@Override
 	public void delete(ITupleReference tuple, MultiComparator cmp, boolean exactDelete) throws Exception {		
-		frameTuple.setFieldCount(cmp.getFields().length);
+		frameTuple.setFieldCount(cmp.getFieldCount());
 		int slotOff = slotManager.findSlot(tuple, frameTuple, cmp, true);
 		if(slotOff < 0) {
 			throw new BTreeException("Key to be deleted does not exist.");
@@ -160,7 +165,7 @@
 				int tupleOff = slotManager.getTupleOff(slotOff);
 				frameTuple.resetByOffset(buf, tupleOff);
 				
-				int comparison = cmp.fieldRangeCompare(tuple, frameTuple, cmp.getKeyLength()-1, cmp.getFields().length - cmp.getKeyLength());
+				int comparison = cmp.fieldRangeCompare(tuple, frameTuple, cmp.getKeyFieldCount()-1, cmp.getFieldCount() - cmp.getKeyFieldCount());
                 if(comparison != 0) {
                 	throw new BTreeException("Cannot delete tuple. Byte-by-byte comparison failed to prove equality.");
                 }										
@@ -202,7 +207,7 @@
 	
 	@Override
 	public void insert(ITupleReference tuple, MultiComparator cmp) throws Exception {
-		frameTuple.setFieldCount(cmp.getFields().length);
+		frameTuple.setFieldCount(cmp.getFieldCount());
 		int slotOff = slotManager.findSlot(tuple, frameTuple, cmp, false);		
 		slotOff = slotManager.insertSlot(slotOff, buf.getInt(freeSpaceOff));				
 		int bytesWritten = tupleWriter.writeTuple(tuple, buf, buf.getInt(freeSpaceOff));
@@ -234,15 +239,17 @@
 	}
 	
 	@Override
-	public String printKeys(MultiComparator cmp, int fieldCount) {		
+	public String printKeys(MultiComparator cmp, ISerializerDeserializer[] fields) throws HyracksDataException {		
 		StringBuilder strBuilder = new StringBuilder();		
-		frameTuple.setFieldCount(fieldCount);
 		int tupleCount = buf.getInt(tupleCountOff);
-		for(int i = 0; i < tupleCount; i++) {			
-			int tupleOff = slotManager.getTupleOff(slotManager.getSlotOff(i));			
-			frameTuple.resetByOffset(buf, tupleOff);			
-			for(int j = 0; j < cmp.getKeyLength(); j++) {
-				strBuilder.append(cmp.getFields()[j].print(buf.array(), frameTuple.getFieldStart(j)) + " ");				
+		frameTuple.setFieldCount(fields.length);
+		for(int i = 0; i < tupleCount; i++) {						
+			frameTuple.resetByTupleIndex(this, i);												
+			for(int j = 0; j < cmp.getKeyFieldCount(); j++) {
+				ByteArrayInputStream inStream = new ByteArrayInputStream(frameTuple.getFieldData(j), frameTuple.getFieldStart(j), frameTuple.getFieldLength(j));
+				DataInput dataIn = new DataInputStream(inStream);
+				Object o = fields[j].deserialize(dataIn);
+				strBuilder.append(o.toString() + " ");				
 			}
 			strBuilder.append(" | ");				
 		}
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMInteriorFrame.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMInteriorFrame.java
index b6bcaee..78424e6 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMInteriorFrame.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMInteriorFrame.java
@@ -43,7 +43,7 @@
 	}
 	
 	private int getLeftChildPageOff(ITupleReference tuple, MultiComparator cmp) {		
-		return tuple.getFieldStart(cmp.getKeyLength()-1) + tuple.getFieldLength(cmp.getKeyLength()-1);		
+		return tuple.getFieldStart(cmp.getKeyFieldCount()-1) + tuple.getFieldLength(cmp.getKeyFieldCount()-1);		
 	}	
 	
 	@Override
@@ -62,7 +62,7 @@
 	
 	@Override
 	public void insert(ITupleReference tuple, MultiComparator cmp) throws Exception {
-		frameTuple.setFieldCount(cmp.getKeyLength());
+		frameTuple.setFieldCount(cmp.getKeyFieldCount());
 		int slotOff = slotManager.findSlot(tuple, frameTuple, cmp, false);
 		boolean isDuplicate = true;				
 		
@@ -79,8 +79,8 @@
 			slotOff = slotManager.insertSlot(slotOff, buf.getInt(freeSpaceOff));			
 			
 			int freeSpace = buf.getInt(freeSpaceOff);			
-			int bytesWritten = tupleWriter.writeTupleFields(tuple, 0, cmp.getKeyLength(), buf, freeSpace);
-			System.arraycopy(tuple.getFieldData(cmp.getKeyLength()-1), getLeftChildPageOff(tuple, cmp), buf.array(), freeSpace + bytesWritten, childPtrSize);
+			int bytesWritten = tupleWriter.writeTupleFields(tuple, 0, cmp.getKeyFieldCount(), buf, freeSpace);
+			System.arraycopy(tuple.getFieldData(cmp.getKeyFieldCount()-1), getLeftChildPageOff(tuple, cmp), buf.array(), freeSpace + bytesWritten, childPtrSize);
 			int tupleSize = bytesWritten + childPtrSize;	
 			
 			buf.putInt(tupleCountOff, buf.getInt(tupleCountOff) + 1);
@@ -89,7 +89,7 @@
 			
 			// did insert into the rightmost slot?
 			if(slotOff == slotManager.getSlotEndOff()) { 
-				System.arraycopy(tuple.getFieldData(cmp.getKeyLength()-1), getLeftChildPageOff(tuple, cmp) + childPtrSize, buf.array(), rightLeafOff, childPtrSize);
+				System.arraycopy(tuple.getFieldData(cmp.getKeyFieldCount()-1), getLeftChildPageOff(tuple, cmp) + childPtrSize, buf.array(), rightLeafOff, childPtrSize);
 			}
 			else {
 				// if slotOff has a right (slot-)neighbor then update its child pointer
@@ -108,8 +108,8 @@
 	public void insertSorted(ITupleReference tuple, MultiComparator cmp) throws Exception {
 		int freeSpace = buf.getInt(freeSpaceOff);
 		slotManager.insertSlot(-1, freeSpace);
-		int bytesWritten = tupleWriter.writeTupleFields(tuple, 0, cmp.getKeyLength(), buf, freeSpace);
-		System.arraycopy(tuple.getFieldData(cmp.getKeyLength()-1), getLeftChildPageOff(tuple, cmp), buf.array(), freeSpace + bytesWritten, childPtrSize);
+		int bytesWritten = tupleWriter.writeTupleFields(tuple, 0, cmp.getKeyFieldCount(), buf, freeSpace);
+		System.arraycopy(tuple.getFieldData(cmp.getKeyFieldCount()-1), getLeftChildPageOff(tuple, cmp), buf.array(), freeSpace + bytesWritten, childPtrSize);
 		int tupleSize = bytesWritten + childPtrSize;
 		buf.putInt(tupleCountOff, buf.getInt(tupleCountOff) + 1);
 		buf.putInt(freeSpaceOff, buf.getInt(freeSpaceOff) + tupleSize);
@@ -120,7 +120,7 @@
 	@Override
 	public int split(IBTreeFrame rightFrame, ITupleReference tuple, MultiComparator cmp, SplitKey splitKey) throws Exception {		
 		// before doing anything check if key already exists
-		frameTuple.setFieldCount(cmp.getKeyLength());
+		frameTuple.setFieldCount(cmp.getKeyFieldCount());
 		int slotOff = slotManager.findSlot(tuple, frameTuple, cmp, true);
 		if(slotOff >= 0) {			
 			frameTuple.resetByOffset(buf, slotManager.getTupleOff(slotOff));			
@@ -162,9 +162,9 @@
 		// set split key to be highest value in left page	
 		int tupleOff = slotManager.getTupleOff(slotManager.getSlotEndOff());
 		frameTuple.resetByOffset(buf, tupleOff);
-		int splitKeySize = tupleWriter.bytesRequired(frameTuple, 0, cmp.getKeyLength());
+		int splitKeySize = tupleWriter.bytesRequired(frameTuple, 0, cmp.getKeyFieldCount());
 		splitKey.initData(splitKeySize);
-		tupleWriter.writeTupleFields(frameTuple, 0, cmp.getKeyLength(), splitKey.getBuffer(), 0);			
+		tupleWriter.writeTupleFields(frameTuple, 0, cmp.getKeyFieldCount(), splitKey.getBuffer(), 0);			
 		
 		int deleteTupleOff = slotManager.getTupleOff(slotManager.getSlotEndOff());
 		frameTuple.resetByOffset(buf, deleteTupleOff);
@@ -185,7 +185,7 @@
 	public void compact(MultiComparator cmp) {		
 		resetSpaceParams();
 		
-		frameTuple.setFieldCount(cmp.getKeyLength());
+		frameTuple.setFieldCount(cmp.getKeyFieldCount());
 		
 		int tupleCount = buf.getInt(tupleCountOff);
 		int freeSpace = buf.getInt(freeSpaceOff);
@@ -222,8 +222,8 @@
 			return buf.getInt(rightLeafOff);
 		}
 				
-		cmpFrameTuple.setFieldCount(srcCmp.getKeyLength());		
-		frameTuple.setFieldCount(srcCmp.getKeyLength());
+		cmpFrameTuple.setFieldCount(srcCmp.getKeyFieldCount());		
+		frameTuple.setFieldCount(srcCmp.getKeyFieldCount());
 		
 		// check for trivial cases where no low key or high key exists (e.g. during an index scan)
 		ITupleReference tuple = null;
@@ -280,7 +280,7 @@
 	
 	@Override
 	public void delete(ITupleReference tuple, MultiComparator cmp, boolean exactDelete) throws Exception {
-		frameTuple.setFieldCount(cmp.getKeyLength());
+		frameTuple.setFieldCount(cmp.getKeyFieldCount());
 		int slotOff = slotManager.findSlot(tuple, frameTuple, cmp, false);
 		int tupleOff;
 		int keySize;
@@ -288,7 +288,7 @@
 		if(slotOff < 0) {						
 			tupleOff = slotManager.getTupleOff(slotManager.getSlotEndOff());
 			frameTuple.resetByOffset(buf, tupleOff);			
-			keySize = tupleWriter.bytesRequired(frameTuple, 0, cmp.getKeyLength());
+			keySize = tupleWriter.bytesRequired(frameTuple, 0, cmp.getKeyFieldCount());
 			
 			// copy new rightmost pointer
 			System.arraycopy(buf.array(), tupleOff + keySize, buf.array(), rightLeafOff, childPtrSize);						
@@ -296,7 +296,7 @@
 		else {						
 			tupleOff = slotManager.getTupleOff(slotOff);
 			frameTuple.resetByOffset(buf, tupleOff);
-			keySize = tupleWriter.bytesRequired(frameTuple, 0, cmp.getKeyLength());	
+			keySize = tupleWriter.bytesRequired(frameTuple, 0, cmp.getKeyFieldCount());	
 			// perform deletion (we just do a memcpy to overwrite the slot)
 			int slotStartOff = slotManager.getSlotEndOff();
 			int length = slotOff - slotStartOff;
@@ -317,7 +317,7 @@
 	@Override
 	public int getLeftmostChildPageId(MultiComparator cmp) {						
 		int tupleOff = slotManager.getTupleOff(slotManager.getSlotStartOff());		
-		frameTuple.setFieldCount(cmp.getKeyLength());
+		frameTuple.setFieldCount(cmp.getKeyFieldCount());
 		frameTuple.resetByOffset(buf, tupleOff);
 		int childPageOff = getLeftChildPageOff(frameTuple, cmp);
 		return buf.getInt(childPageOff);
@@ -338,7 +338,7 @@
 		System.out.println(page);
 		
 		ArrayList<Integer> ret = new ArrayList<Integer>();		
-		frameTuple.setFieldCount(cmp.getKeyLength());
+		frameTuple.setFieldCount(cmp.getKeyFieldCount());
 		int tupleCount = buf.getInt(tupleCountOff);
 		for(int i = 0; i < tupleCount; i++) {
 			int tupleOff = slotManager.getTupleOff(slotManager.getSlotOff(i));
@@ -357,9 +357,9 @@
 	public void deleteGreatest(MultiComparator cmp) {
 		int slotOff = slotManager.getSlotEndOff();
 		int tupleOff = slotManager.getTupleOff(slotOff);
-		frameTuple.setFieldCount(cmp.getKeyLength());
+		frameTuple.setFieldCount(cmp.getKeyFieldCount());
 		frameTuple.resetByOffset(buf, tupleOff);
-		int keySize = tupleWriter.bytesRequired(frameTuple, 0, cmp.getKeyLength());		 
+		int keySize = tupleWriter.bytesRequired(frameTuple, 0, cmp.getKeyFieldCount());		 
 		System.arraycopy(buf.array(), tupleOff + keySize, buf.array(), rightLeafOff, childPtrSize);
 		
 		// maintain space information
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMLeafFrame.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMLeafFrame.java
index 8f88bc6..d82c47f 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMLeafFrame.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMLeafFrame.java
@@ -61,7 +61,7 @@
 
 	@Override
 	public void insert(ITupleReference tuple, MultiComparator cmp) throws Exception {		
-		frameTuple.setFieldCount(cmp.getFields().length);
+		frameTuple.setFieldCount(cmp.getFieldCount());
 		int slotOff = slotManager.findSlot(tuple, frameTuple, cmp, false);
 		boolean isDuplicate = true;
 				
@@ -99,7 +99,7 @@
 	@Override
 	public int split(IBTreeFrame rightFrame, ITupleReference tuple, MultiComparator cmp, SplitKey splitKey) throws Exception {
 				
-		frameTuple.setFieldCount(cmp.getFields().length);
+		frameTuple.setFieldCount(cmp.getFieldCount());
 		
 		// before doing anything check if key already exists
 		int slotOff = slotManager.findSlot(tuple, frameTuple, cmp, true);
@@ -151,9 +151,9 @@
 		tupleOff = slotManager.getTupleOff(slotManager.getSlotEndOff());
 		frameTuple.resetByOffset(buf, tupleOff);
 		
-		int splitKeySize = tupleWriter.bytesRequired(frameTuple, 0, cmp.getKeyLength());
+		int splitKeySize = tupleWriter.bytesRequired(frameTuple, 0, cmp.getKeyFieldCount());
 		splitKey.initData(splitKeySize);				
-		tupleWriter.writeTupleFields(frameTuple, 0, cmp.getKeyLength(), splitKey.getBuffer(), 0);
+		tupleWriter.writeTupleFields(frameTuple, 0, cmp.getKeyFieldCount(), splitKey.getBuffer(), 0);
 		
 		return 0;
 	}
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/BTree.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/BTree.java
index 58e044c..b170d080b 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/BTree.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/BTree.java
@@ -20,6 +20,7 @@
 import java.util.concurrent.locks.ReadWriteLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 
+import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
 import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
 import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeCursor;
@@ -29,7 +30,6 @@
 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.common.buffercache.IBufferCache;
 import edu.uci.ics.hyracks.storage.common.buffercache.ICachedPage;
@@ -53,7 +53,6 @@
     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];
@@ -98,13 +97,7 @@
         this.leafFrameFactory = leafFrameFactory;       
         this.cmp = cmp;
         this.treeLatch = new ReentrantReadWriteLock(true);
-        this.diskOrderScanPredicate = new RangePredicate(true, null, null, cmp);
-        
-        IFieldAccessor[] interiorFields = new IFieldAccessor[cmp.getKeyLength()];
-        for(int i = 0; i < cmp.getKeyLength(); i++) {
-        	interiorFields[i] = cmp.getFields()[i];
-        }               
-        this.interiorCmp = new MultiComparator(cmp.getComparators(), interiorFields);        
+        this.diskOrderScanPredicate = new RangePredicate(true, null, null, cmp);                
     }
     
     public void create(int fileId, IBTreeLeafFrame leafFrame, IBTreeMetaDataFrame metaFrame) throws Exception {
@@ -317,12 +310,12 @@
 		return maxPage;
     }
     
-    public void printTree(IBTreeLeafFrame leafFrame, IBTreeInteriorFrame interiorFrame) throws Exception {
-        printTree(rootPage, null, false, leafFrame, interiorFrame);
+    public void printTree(IBTreeLeafFrame leafFrame, IBTreeInteriorFrame interiorFrame, ISerializerDeserializer[] fields) throws Exception {
+        printTree(rootPage, null, false, leafFrame, interiorFrame, fields);
     }
 
     public void printTree(int pageId, ICachedPage parent, boolean unpin, IBTreeLeafFrame leafFrame,
-            IBTreeInteriorFrame interiorFrame) throws Exception {
+            IBTreeInteriorFrame interiorFrame, ISerializerDeserializer[] fields) throws Exception {
         
         ICachedPage node = bufferCache.pin(FileInfo.getDiskPageId(fileId, pageId), false);
         pins++;
@@ -369,12 +362,12 @@
             String keyString;
             if(interiorFrame.isLeaf()) {
             	leafFrame.setPage(node);
-            	keyString = leafFrame.printKeys(cmp, cmp.getFields().length);
+            	keyString = leafFrame.printKeys(cmp, fields);
             }
             else {
-            	keyString = interiorFrame.printKeys(interiorCmp, cmp.getKeyLength());
+            	keyString = interiorFrame.printKeys(cmp, fields);
             }
-
+            
             System.out.format(keyString);
             if (!interiorFrame.isLeaf()) {
             	ArrayList<Integer> children = ((NSMInteriorFrame) (interiorFrame)).getChildren(cmp);
@@ -387,7 +380,7 @@
             	System.out.println();
             	            	
                 for (int i = 0; i < children.size(); i++) {
-                    printTree(children.get(i), node, i == children.size() - 1, leafFrame, interiorFrame);
+                    printTree(children.get(i), node, i == children.size() - 1, leafFrame, interiorFrame, fields);
                 }
             } else {
                 node.releaseReadLatch();
@@ -563,7 +556,7 @@
         ctx.metaFrame = metaFrame;
         ctx.pred = new RangePredicate(true, tuple, tuple, cmp);
         ctx.splitKey = new SplitKey();
-        ctx.splitKey.getTuple().setFieldCount(cmp.getKeyLength());
+        ctx.splitKey.getTuple().setFieldCount(cmp.getKeyFieldCount());
         ctx.smPages = new ArrayList<Integer>();
         ctx.pageLsns = new Stack<Integer>();
 
@@ -595,7 +588,7 @@
     
     private void insertLeaf(ICachedPage node, int pageId, ITupleReference tuple, BTreeOpContext ctx) throws Exception {
     	ctx.leafFrame.setPage(node);
-    	ctx.leafFrame.setPageTupleFieldCount(cmp.getFields().length);
+    	ctx.leafFrame.setPageTupleFieldCount(cmp.getFieldCount());
     	SpaceStatus spaceStatus = ctx.leafFrame.hasSpaceInsert(tuple, cmp);
     	switch (spaceStatus) {
     	
@@ -655,7 +648,7 @@
     					IBTreeLeafFrame rightFrame = leafFrameFactory.getFrame();
     					rightFrame.setPage(rightNode);
     					rightFrame.initBuffer((byte) 0);
-    					rightFrame.setPageTupleFieldCount(cmp.getFields().length);
+    					rightFrame.setPageTupleFieldCount(cmp.getFieldCount());
     					
     					int ret = ctx.leafFrame.split(rightFrame, tuple, cmp, ctx.splitKey);
 
@@ -726,7 +719,7 @@
 
     private void insertInterior(ICachedPage node, int pageId, ITupleReference tuple, BTreeOpContext ctx) throws Exception {
         ctx.interiorFrame.setPage(node);
-        ctx.interiorFrame.setPageTupleFieldCount(cmp.getKeyLength());
+        ctx.interiorFrame.setPageTupleFieldCount(cmp.getKeyFieldCount());
         SpaceStatus spaceStatus = ctx.interiorFrame.hasSpaceInsert(tuple, cmp);
         switch (spaceStatus) {
             case INSUFFICIENT_SPACE: {
@@ -740,7 +733,7 @@
                     IBTreeFrame rightFrame = interiorFrameFactory.getFrame();
                     rightFrame.setPage(rightNode);
                     rightFrame.initBuffer((byte) ctx.interiorFrame.getLevel());
-                    rightFrame.setPageTupleFieldCount(cmp.getKeyLength());
+                    rightFrame.setPageTupleFieldCount(cmp.getKeyFieldCount());
                     // instead of creating a new split key, use the existing
                     // splitKey
                     int ret = ctx.interiorFrame.split(rightFrame, ctx.splitKey.getTuple(), cmp, ctx.splitKey);
@@ -800,7 +793,7 @@
         ctx.metaFrame = metaFrame;
         ctx.pred = new RangePredicate(true, tuple, tuple, cmp);
         ctx.splitKey = new SplitKey();
-        ctx.splitKey.getTuple().setFieldCount(cmp.getKeyLength());
+        ctx.splitKey.getTuple().setFieldCount(cmp.getKeyFieldCount());
         ctx.smPages = new ArrayList<Integer>();
         ctx.pageLsns = new Stack<Integer>();
         ctx.freePages = new ArrayList<Integer>();
@@ -1244,7 +1237,7 @@
             frontier.pageId = getFreePage(metaFrame);
             frontier.page = bufferCache.pin(FileInfo.getDiskPageId(fileId, frontier.pageId), bulkNewPage);
             frontier.page.acquireWriteLatch();
-            frontier.lastTuple.setFieldCount(cmp.getKeyLength());
+            frontier.lastTuple.setFieldCount(cmp.getKeyFieldCount());
             interiorFrame.setPage(frontier.page);
             interiorFrame.initBuffer((byte) nodeFrontiers.size());
             nodeFrontiers.add(frontier);
@@ -1263,7 +1256,7 @@
         ctx.interiorFrame.setPage(frontier.page);
         
         ITupleReference tuple = ctx.splitKey.getTuple();        
-        int spaceNeeded = ctx.tupleWriter.bytesRequired(tuple, 0, cmp.getKeyLength()) + ctx.slotSize + 4;                
+        int spaceNeeded = ctx.tupleWriter.bytesRequired(tuple, 0, cmp.getKeyFieldCount()) + ctx.slotSize + 4;                
         int spaceUsed = ctx.interiorFrame.getBuffer().capacity() - ctx.interiorFrame.getTotalFreeSpace();
         if (spaceUsed + spaceNeeded > ctx.interiorMaxBytes) {
             //ctx.interiorFrame.deleteGreatest(cmp);
@@ -1272,9 +1265,9 @@
         	tuple = copyKey.getTuple();
         	
             frontier.lastTuple.resetByOffset(frontier.page.getBuffer(), ctx.interiorFrame.getTupleOffset(ctx.interiorFrame.getTupleCount()-1));            
-            int splitKeySize = ctx.tupleWriter.bytesRequired(frontier.lastTuple, 0, cmp.getKeyLength());
+            int splitKeySize = ctx.tupleWriter.bytesRequired(frontier.lastTuple, 0, cmp.getKeyFieldCount());
             ctx.splitKey.initData(splitKeySize);
-            ctx.tupleWriter.writeTupleFields(frontier.lastTuple, 0, cmp.getKeyLength(), ctx.splitKey.getBuffer(), 0);                  
+            ctx.tupleWriter.writeTupleFields(frontier.lastTuple, 0, cmp.getKeyFieldCount(), ctx.splitKey.getBuffer(), 0);                  
             ctx.splitKey.setLeftPage(frontier.pageId);
             
             ctx.interiorFrame.deleteGreatest(cmp);
@@ -1297,8 +1290,8 @@
     // assumes btree has been created and opened
     public BulkLoadContext beginBulkLoad(float fillFactor, IBTreeLeafFrame leafFrame, IBTreeInteriorFrame interiorFrame, IBTreeMetaDataFrame metaFrame) throws Exception {
         BulkLoadContext ctx = new BulkLoadContext(fillFactor, leafFrame, interiorFrame, metaFrame);
-        ctx.nodeFrontiers.get(0).lastTuple.setFieldCount(cmp.getFields().length);
-        ctx.splitKey.getTuple().setFieldCount(cmp.getKeyLength());
+        ctx.nodeFrontiers.get(0).lastTuple.setFieldCount(cmp.getFieldCount());
+        ctx.splitKey.getTuple().setFieldCount(cmp.getKeyFieldCount());
         return ctx;
     }
     
@@ -1317,9 +1310,9 @@
         
         if (spaceUsed + spaceNeeded > ctx.leafMaxBytes) {        	
         	leafFrontier.lastTuple.resetByOffset(leafFrontier.page.getBuffer(), leafFrame.getTupleOffset(leafFrame.getTupleCount()-1));        	
-        	int splitKeySize = ctx.tupleWriter.bytesRequired(leafFrontier.lastTuple, 0, cmp.getKeyLength());        	
+        	int splitKeySize = ctx.tupleWriter.bytesRequired(leafFrontier.lastTuple, 0, cmp.getKeyFieldCount());        	
             ctx.splitKey.initData(splitKeySize);
-            ctx.tupleWriter.writeTupleFields(leafFrontier.lastTuple, 0, cmp.getKeyLength(), ctx.splitKey.getBuffer(), 0);
+            ctx.tupleWriter.writeTupleFields(leafFrontier.lastTuple, 0, cmp.getKeyFieldCount(), ctx.splitKey.getBuffer(), 0);
             ctx.splitKey.setLeftPage(leafFrontier.pageId);
             int prevPageId = leafFrontier.pageId;
             leafFrontier.pageId = getFreePage(ctx.metaFrame);
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/DiskOrderScanCursor.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/DiskOrderScanCursor.java
index a9d8560..7605e1b 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/DiskOrderScanCursor.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/DiskOrderScanCursor.java
@@ -113,7 +113,7 @@
         frame.setPage(page);
         RangePredicate pred = (RangePredicate)searchPred;
 		MultiComparator cmp = pred.getComparator();
-		frameTuple.setFieldCount(cmp.getFields().length);
+		frameTuple.setFieldCount(cmp.getFieldCount());
         boolean leafExists = positionToNextLeaf(false);
         if(!leafExists) {
             throw new Exception("Failed to open disk-order scan cursor for B-tree. Traget B-tree has no leaves.");
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixSlotManager.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixSlotManager.java
index f510dad..9b172ba 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixSlotManager.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixSlotManager.java
@@ -67,7 +67,7 @@
 	public int findSlot(ITupleReference tuple, IBTreeTupleReference frameTuple, IBTreeTupleReference framePrefixTuple, MultiComparator multiCmp, boolean exact) {				
 		if(frame.getTupleCount() <= 0) encodeSlotFields(TUPLE_UNCOMPRESSED, GREATEST_SLOT);
 								
-		frameTuple.setFieldCount(multiCmp.getFields().length);
+		frameTuple.setFieldCount(multiCmp.getFieldCount());
 		
 	    int prefixMid;
 	    int prefixBegin = 0;
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/MultiComparator.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/MultiComparator.java
index fe8c25a..286dabe 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/MultiComparator.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/MultiComparator.java
@@ -23,7 +23,6 @@
 import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
 import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
 import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
-import edu.uci.ics.hyracks.storage.am.btree.api.IFieldAccessor;
 
 @SuppressWarnings("unchecked")
 public class MultiComparator {
@@ -31,37 +30,13 @@
 	private static final long serialVersionUID = 1L;
 	
 	private IBinaryComparator[] cmps = null;
-	private IFieldAccessor[] fields = null;
+	private int fieldCount;
 	
-	public MultiComparator(IBinaryComparator[] cmps, IFieldAccessor[] fields) {
-		this.cmps = cmps;
-		this.fields = fields;
-	}
-	
-	public IBinaryComparator[] getComparators() {
-	    return cmps;
-	}
-	
-	public int getKeyLength() {
-	    return cmps.length;
-	}
-	
-	public void setComparators(IBinaryComparator[] cmps) {
+	public MultiComparator(int fieldCount, IBinaryComparator[] cmps) {
+		this.fieldCount = fieldCount;
 		this.cmps = cmps;
 	}
-	
-	public int size() {
-		return cmps.length;
-	}
-	
-	public void setFields(IFieldAccessor[] fields) {
-		this.fields = fields;
-	}	
-	
-	public IFieldAccessor[] getFields() {
-		return fields;
-	}
-	
+		
 	public int compare(ITupleReference tupleA, ITupleReference tupleB) {			
 		for(int i = 0; i < cmps.length; i++) {						
 			int cmp = cmps[i].compare(tupleA.getFieldData(i), 
@@ -100,5 +75,25 @@
 			strBuilder.append(o.toString() + " ");
 		}
 		return strBuilder.toString();
-	}			
+	}		
+	
+	public IBinaryComparator[] getComparators() {
+	    return cmps;
+	}
+	
+	public int getKeyFieldCount() {
+	    return cmps.length;
+	}
+	
+	public void setComparators(IBinaryComparator[] cmps) {
+		this.cmps = cmps;
+	}
+	
+	public int size() {
+		return cmps.length;
+	}
+	
+	public int getFieldCount() {
+		return fieldCount;
+	}
 }
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/RangeSearchCursor.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/RangeSearchCursor.java
index 22687d4..ec99a90 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/RangeSearchCursor.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/RangeSearchCursor.java
@@ -134,7 +134,7 @@
 		// TODO: can be done more efficiently with binary search but this needs some thinking/refactoring
 		RangePredicate pred = (RangePredicate)searchPred;
 		MultiComparator cmp = pred.getComparator();
-		frameTuple.setFieldCount(cmp.getFields().length);
+		frameTuple.setFieldCount(cmp.getFieldCount());
 		if(searchPred.isForward()) {
 			ITupleReference lowKey = pred.getLowKey();			
 			
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/SelfDescTupleReference.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/SelfDescTupleReference.java
deleted file mode 100644
index b264c66..0000000
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/SelfDescTupleReference.java
+++ /dev/null
@@ -1,59 +0,0 @@
-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/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/types/Int32Accessor.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/types/Int32Accessor.java
deleted file mode 100644
index 0ab80cb..0000000
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/types/Int32Accessor.java
+++ /dev/null
@@ -1,34 +0,0 @@
-/*
- * Copyright 2009-2010 by The Regents of the University of California
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * you may obtain a copy of the License from
- * 
- *     http://www.apache.org/licenses/LICENSE-2.0
- * 
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package edu.uci.ics.hyracks.storage.am.btree.types;
-
-import java.nio.ByteBuffer;
-
-import edu.uci.ics.hyracks.storage.am.btree.api.IFieldAccessor;
-
-public class Int32Accessor implements IFieldAccessor {
-	
-	@Override
-	public int getLength(byte[] data, int offset) {
-		return 4;
-	}
-	
-	@Override
-	public String print(byte[] data, int offset) {
-	    ByteBuffer buf = ByteBuffer.wrap(data);
-	    return String.format("%6d ", buf.getInt(offset));
-	}
-}
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/types/Int32AccessorFactory.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/types/Int32AccessorFactory.java
deleted file mode 100644
index 4eceeb2..0000000
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/types/Int32AccessorFactory.java
+++ /dev/null
@@ -1,30 +0,0 @@
-/*
- * Copyright 2009-2010 by The Regents of the University of California
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * you may obtain a copy of the License from
- * 
- *     http://www.apache.org/licenses/LICENSE-2.0
- * 
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package edu.uci.ics.hyracks.storage.am.btree.types;
-
-import edu.uci.ics.hyracks.storage.am.btree.api.IFieldAccessor;
-import edu.uci.ics.hyracks.storage.am.btree.api.IFieldAccessorFactory;
-
-public class Int32AccessorFactory implements IFieldAccessorFactory {
-	
-	private static final long serialVersionUID = 1L;
-		
-	@Override
-	public IFieldAccessor getFieldAccessor() {	
-		return new Int32Accessor();
-	}
-	
-}
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/types/UTF8StringAccessor.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/types/UTF8StringAccessor.java
deleted file mode 100644
index e47ce42..0000000
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/types/UTF8StringAccessor.java
+++ /dev/null
@@ -1,48 +0,0 @@
-/*
- * Copyright 2009-2010 by The Regents of the University of California
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * you may obtain a copy of the License from
- * 
- *     http://www.apache.org/licenses/LICENSE-2.0
- * 
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package edu.uci.ics.hyracks.storage.am.btree.types;
-
-import java.io.DataInputStream;
-import java.nio.ByteBuffer;
-
-import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
-import edu.uci.ics.hyracks.dataflow.common.comm.util.ByteBufferInputStream;
-import edu.uci.ics.hyracks.dataflow.common.data.marshalling.UTF8StringSerializerDeserializer;
-import edu.uci.ics.hyracks.dataflow.common.data.util.StringUtils;
-import edu.uci.ics.hyracks.storage.am.btree.api.IFieldAccessor;
-
-public class UTF8StringAccessor implements IFieldAccessor {
-	
-	@Override
-	public int getLength(byte[] data, int offset) {
-		return StringUtils.getUTFLen(data, offset) + 2;
-	}
-		
-	@Override
-	public String print(byte[] data, int offset) {				
-		ByteBuffer buf = ByteBuffer.wrap(data);
-		ByteBufferInputStream bbis = new ByteBufferInputStream();
-		bbis.setByteBuffer(buf, offset);		
-		DataInputStream din = new DataInputStream(bbis);		
-		String field = new String();
-		try {
-			field = UTF8StringSerializerDeserializer.INSTANCE.deserialize(din);
-		} catch (HyracksDataException e) {
-			e.printStackTrace();
-		}		
-		return String.format("%10s ", field);
-	}   
-}
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/types/UTF8StringAccessorFactory.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/types/UTF8StringAccessorFactory.java
deleted file mode 100644
index 1359190..0000000
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/types/UTF8StringAccessorFactory.java
+++ /dev/null
@@ -1,29 +0,0 @@
-/*
- * Copyright 2009-2010 by The Regents of the University of California
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * you may obtain a copy of the License from
- * 
- *     http://www.apache.org/licenses/LICENSE-2.0
- * 
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package edu.uci.ics.hyracks.storage.am.btree.types;
-
-import edu.uci.ics.hyracks.storage.am.btree.api.IFieldAccessor;
-import edu.uci.ics.hyracks.storage.am.btree.api.IFieldAccessorFactory;
-
-public class UTF8StringAccessorFactory implements IFieldAccessorFactory {
-	
-	private static final long serialVersionUID = 1L;
-
-	@Override
-	public IFieldAccessor getFieldAccessor() {
-		return new UTF8StringAccessor();
-	}	
-}
diff --git a/hyracks/hyracks-storage-am-btree/src/test/java/edu/uci/ics/hyracks/storage/am/btree/BTreeFieldPrefixNSMTest.java b/hyracks/hyracks-storage-am-btree/src/test/java/edu/uci/ics/hyracks/storage/am/btree/BTreeFieldPrefixNSMTest.java
index 207ffa3..a17de21 100644
--- a/hyracks/hyracks-storage-am-btree/src/test/java/edu/uci/ics/hyracks/storage/am/btree/BTreeFieldPrefixNSMTest.java
+++ b/hyracks/hyracks-storage-am-btree/src/test/java/edu/uci/ics/hyracks/storage/am/btree/BTreeFieldPrefixNSMTest.java
@@ -38,13 +38,11 @@
 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;
 import edu.uci.ics.hyracks.storage.common.buffercache.BufferCache;
 import edu.uci.ics.hyracks.storage.common.buffercache.ClockPageReplacementStrategy;
 import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
@@ -128,19 +126,17 @@
         int fileId = 0;
         FileInfo fi = new FileInfo(fileId, raf);
         fileManager.registerFile(fi);
-                
-        int numFields = 3;
-        IFieldAccessor[] fields = new IFieldAccessor[numFields];
-        fields[0] = new Int32Accessor(); // first key field
-        fields[1] = new Int32Accessor(); // second key field
-        fields[2] = new Int32Accessor(); // third key field
         
-        int keyLen = 3;
-        IBinaryComparator[] cmps = new IBinaryComparator[keyLen];
+        int fieldCount = 3;
+        int keyFieldCount = 3;
+        IBinaryComparator[] cmps = new IBinaryComparator[keyFieldCount];
         cmps[0] = IntegerBinaryComparatorFactory.INSTANCE.createBinaryComparator();
         cmps[1] = IntegerBinaryComparatorFactory.INSTANCE.createBinaryComparator();
         cmps[2] = IntegerBinaryComparatorFactory.INSTANCE.createBinaryComparator();
-        MultiComparator cmp = new MultiComparator(cmps, fields);
+        MultiComparator cmp = new MultiComparator(fieldCount, cmps);
+        
+        // just for printing
+        ISerializerDeserializer[] sers = { IntegerSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE };
         
         Random rnd = new Random();
         rnd.setSeed(50);
@@ -190,16 +186,16 @@
         		savedFields[i][2] = c;
         		            
         		if(rnd.nextInt() % compactFreq == 0) {
-        			before = frame.printKeys(cmp, fields.length);
+        			before = frame.printKeys(cmp, sers);
         			frame.compact(cmp);
-        			after = frame.printKeys(cmp, fields.length);
+        			after = frame.printKeys(cmp, sers);
         			Assert.assertEquals(before, after);
         		}
         		
         		if(rnd.nextInt() % compressFreq == 0) {
-        			before = frame.printKeys(cmp, fields.length);
+        			before = frame.printKeys(cmp, sers);
         			frame.compress(cmp);
-        			after = frame.printKeys(cmp, fields.length);
+        			after = frame.printKeys(cmp, sers);
         			Assert.assertEquals(before, after);
         		}
         	}
@@ -217,16 +213,16 @@
         		}
         		
         		if(rnd.nextInt() % compactFreq == 0) {
-        			before = frame.printKeys(cmp, fields.length);
+        			before = frame.printKeys(cmp, sers);
         			frame.compact(cmp);
-        			after = frame.printKeys(cmp, fields.length);
+        			after = frame.printKeys(cmp, sers);
         			Assert.assertEquals(before, after);
         		}
         		
         		if(rnd.nextInt() % compressFreq == 0) {
-        			before = frame.printKeys(cmp, fields.length);
+        			before = frame.printKeys(cmp, sers);
         			frame.compress(cmp);
-        			after = frame.printKeys(cmp, fields.length);
+        			after = frame.printKeys(cmp, sers);
         			Assert.assertEquals(before, after);
         		}  
         	}
diff --git a/hyracks/hyracks-storage-am-btree/src/test/java/edu/uci/ics/hyracks/storage/am/btree/BTreeTest.java b/hyracks/hyracks-storage-am-btree/src/test/java/edu/uci/ics/hyracks/storage/am/btree/BTreeTest.java
index be4d4c2..5f2508b 100644
--- a/hyracks/hyracks-storage-am-btree/src/test/java/edu/uci/ics/hyracks/storage/am/btree/BTreeTest.java
+++ b/hyracks/hyracks-storage-am-btree/src/test/java/edu/uci/ics/hyracks/storage/am/btree/BTreeTest.java
@@ -16,7 +16,6 @@
 package edu.uci.ics.hyracks.storage.am.btree;
 
 import java.io.DataOutput;
-import java.io.DataOutputStream;
 import java.io.File;
 import java.io.RandomAccessFile;
 import java.nio.ByteBuffer;
@@ -31,7 +30,6 @@
 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;
@@ -47,7 +45,6 @@
 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.IBTreeMetaDataFrameFactory;
-import edu.uci.ics.hyracks.storage.am.btree.api.IFieldAccessor;
 import edu.uci.ics.hyracks.storage.am.btree.frames.MetaDataFrameFactory;
 import edu.uci.ics.hyracks.storage.am.btree.frames.NSMInteriorFrameFactory;
 import edu.uci.ics.hyracks.storage.am.btree.frames.NSMLeafFrameFactory;
@@ -57,9 +54,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.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;
 import edu.uci.ics.hyracks.storage.common.buffercache.ClockPageReplacementStrategy;
 import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
@@ -125,15 +119,12 @@
         IBTreeLeafFrame leafFrame = leafFrameFactory.getFrame();
         IBTreeInteriorFrame interiorFrame = interiorFrameFactory.getFrame();
         IBTreeMetaDataFrame metaFrame = metaFrameFactory.getFrame();        
-        
-        IFieldAccessor[] fields = new IFieldAccessor[2];
-        fields[0] = new Int32Accessor(); // key field
-        fields[1] = new Int32Accessor(); // value field
-        
-        int keyLen = 1;
-        IBinaryComparator[] cmps = new IBinaryComparator[keyLen];
+               
+        int fieldCount = 2;
+        int keyFieldCount = 1;
+        IBinaryComparator[] cmps = new IBinaryComparator[keyFieldCount];
         cmps[0] = IntegerBinaryComparatorFactory.INSTANCE.createBinaryComparator();
-        MultiComparator cmp = new MultiComparator(cmps, fields);
+        MultiComparator cmp = new MultiComparator(2, cmps);
         
         BTree btree = new BTree(bufferCache, interiorFrameFactory, leafFrameFactory, cmp);
         btree.create(fileId, leafFrame, metaFrame);
@@ -149,7 +140,7 @@
         IHyracksContext ctx = new HyracksContext(HYRACKS_FRAME_SIZE);        
         ByteBuffer frame = ctx.getResourceManager().allocateFrame();
 		FrameTupleAppender appender = new FrameTupleAppender(ctx);				
-		ArrayTupleBuilder tb = new ArrayTupleBuilder(cmp.getFields().length);
+		ArrayTupleBuilder tb = new ArrayTupleBuilder(cmp.getFieldCount());
 		DataOutput dos = tb.getDataOutput();
 		
 		ISerializerDeserializer[] recDescSers = { IntegerSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE};
@@ -247,27 +238,40 @@
  
         IBTreeCursor rangeCursor = new RangeSearchCursor(leafFrame);
 
-        ByteArrayAccessibleOutputStream lkbaaos = new ByteArrayAccessibleOutputStream();
-    	DataOutputStream lkdos = new DataOutputStream(lkbaaos);    	    	    	
-    	IntegerSerializerDeserializer.INSTANCE.serialize(-1000, lkdos);
+        // build low and high keys                       
+        ArrayTupleBuilder ktb = new ArrayTupleBuilder(cmp.getKeyFieldCount());
+		DataOutput kdos = ktb.getDataOutput();
+		
+		ISerializerDeserializer[] keyDescSers = { IntegerSerializerDeserializer.INSTANCE };
+		RecordDescriptor keyDesc = new RecordDescriptor(keyDescSers);
+		IFrameTupleAccessor keyAccessor = new FrameTupleAccessor(ctx, keyDesc);
+		keyAccessor.reset(frame);
+		
+		appender.reset(frame, true);
+		
+		// build and append low key
+		ktb.reset();
+    	IntegerSerializerDeserializer.INSTANCE.serialize(-1000, kdos);
+    	ktb.addFieldEndOffset();    	
+    	appender.append(ktb.getFieldEndOffsets(), ktb.getByteArray(), 0, ktb.getSize());
     	
-    	ByteArrayAccessibleOutputStream hkbaaos = new ByteArrayAccessibleOutputStream();
-    	DataOutputStream hkdos = new DataOutputStream(hkbaaos);    	    	    	
-    	IntegerSerializerDeserializer.INSTANCE.serialize(1000, hkdos);
-    	    	        
-        byte[] lowKeyBytes = lkbaaos.toByteArray();
-        ByteBuffer lowKeyBuf = ByteBuffer.wrap(lowKeyBytes);
-        SelfDescTupleReference lowKey = new SelfDescTupleReference(cmp.getFields());
-        lowKey.reset(lowKeyBuf, 0);
+    	// build and append high key
+    	ktb.reset();
+    	IntegerSerializerDeserializer.INSTANCE.serialize(1000, kdos);
+    	ktb.addFieldEndOffset();
+    	appender.append(ktb.getFieldEndOffsets(), ktb.getByteArray(), 0, ktb.getSize());
         
-        byte[] highKeyBytes = hkbaaos.toByteArray();
-        ByteBuffer highKeyBuf = ByteBuffer.wrap(highKeyBytes);
-        SelfDescTupleReference highKey = new SelfDescTupleReference(cmp.getFields());
-        highKey.reset(highKeyBuf, 0);
-        
+    	// create tuplereferences for search keys
+    	FrameTupleReference lowKey = new FrameTupleReference();
+    	lowKey.reset(keyAccessor, 0);
+    	
+		FrameTupleReference highKey = new FrameTupleReference();
+		highKey.reset(keyAccessor, 1);
+    	
+		
         IBinaryComparator[] searchCmps = new IBinaryComparator[1];
         searchCmps[0] = IntegerBinaryComparatorFactory.INSTANCE.createBinaryComparator();
-        MultiComparator searchCmp = new MultiComparator(searchCmps, fields);
+        MultiComparator searchCmp = new MultiComparator(fieldCount, searchCmps);
         
         RangePredicate rangePred = new RangePredicate(true, lowKey, highKey, searchCmp);
         btree.search(rangeCursor, rangePred, leafFrame, interiorFrame);
@@ -320,17 +324,13 @@
         IBTreeLeafFrame leafFrame = leafFrameFactory.getFrame();
         IBTreeInteriorFrame interiorFrame = interiorFrameFactory.getFrame();
         IBTreeMetaDataFrame metaFrame = metaFrameFactory.getFrame();   
-        
-        IFieldAccessor[] fields = new IFieldAccessor[3];
-        fields[0] = new Int32Accessor(); // key field 1
-        fields[1] = new Int32Accessor(); // key field 2
-        fields[2] = new Int32Accessor(); // value field        
-        
-        int keyLen = 2;
-        IBinaryComparator[] cmps = new IBinaryComparator[keyLen];
+         
+        int fieldCount = 3;
+        int keyFieldCount = 2;
+        IBinaryComparator[] cmps = new IBinaryComparator[keyFieldCount];
         cmps[0] = IntegerBinaryComparatorFactory.INSTANCE.createBinaryComparator();
         cmps[1] = IntegerBinaryComparatorFactory.INSTANCE.createBinaryComparator();       
-        MultiComparator cmp = new MultiComparator(cmps, fields);
+        MultiComparator cmp = new MultiComparator(fieldCount, cmps);
 
         BTree btree = new BTree(bufferCache, interiorFrameFactory, leafFrameFactory, cmp);
         btree.create(fileId, leafFrame, metaFrame);
@@ -346,7 +346,7 @@
         IHyracksContext ctx = new HyracksContext(HYRACKS_FRAME_SIZE);        
         ByteBuffer frame = ctx.getResourceManager().allocateFrame();
 		FrameTupleAppender appender = new FrameTupleAppender(ctx);				
-		ArrayTupleBuilder tb = new ArrayTupleBuilder(cmp.getFields().length);
+		ArrayTupleBuilder tb = new ArrayTupleBuilder(cmp.getFieldCount());
 		DataOutput dos = tb.getDataOutput();
 		
 		ISerializerDeserializer[] recDescSers = { IntegerSerializerDeserializer.INSTANCE, 
@@ -412,27 +412,41 @@
         print("RANGE SEARCH:\n");        
         IBTreeCursor rangeCursor = new RangeSearchCursor(leafFrame);
 
-        ByteArrayAccessibleOutputStream lkbaaos = new ByteArrayAccessibleOutputStream();
-    	DataOutputStream lkdos = new DataOutputStream(lkbaaos);    	    	    	
-    	IntegerSerializerDeserializer.INSTANCE.serialize(-3, lkdos);
-    	
-    	ByteArrayAccessibleOutputStream hkbaaos = new ByteArrayAccessibleOutputStream();
-    	DataOutputStream hkdos = new DataOutputStream(hkbaaos);    	    	    	
-    	IntegerSerializerDeserializer.INSTANCE.serialize(3, hkdos);
-    	
-        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);
+        // build low and high keys             
+        ArrayTupleBuilder ktb = new ArrayTupleBuilder(cmp.getKeyFieldCount());
+		DataOutput kdos = ktb.getDataOutput();
+		
+		ISerializerDeserializer[] keyDescSers = { IntegerSerializerDeserializer.INSTANCE };
+		RecordDescriptor keyDesc = new RecordDescriptor(keyDescSers);
+		IFrameTupleAccessor keyAccessor = new FrameTupleAccessor(ctx, keyDesc);
+		keyAccessor.reset(frame);
+		
+		appender.reset(frame, true);
+		
+		// build and append low key
+		ktb.reset();
+    	IntegerSerializerDeserializer.INSTANCE.serialize(-3, kdos);
+    	ktb.addFieldEndOffset();    	
+    	appender.append(ktb.getFieldEndOffsets(), ktb.getByteArray(), 0, ktb.getSize());
+    	
+    	// build and append high key
+    	ktb.reset();
+    	IntegerSerializerDeserializer.INSTANCE.serialize(3, kdos);
+    	ktb.addFieldEndOffset();
+    	appender.append(ktb.getFieldEndOffsets(), ktb.getByteArray(), 0, ktb.getSize());
         
+    	// create tuplereferences for search keys
+    	FrameTupleReference lowKey = new FrameTupleReference();
+    	lowKey.reset(keyAccessor, 0);
+    	
+		FrameTupleReference highKey = new FrameTupleReference();
+		highKey.reset(keyAccessor, 1);
+		
+		
         IBinaryComparator[] searchCmps = new IBinaryComparator[1];
         searchCmps[0] = IntegerBinaryComparatorFactory.INSTANCE.createBinaryComparator();       
-        MultiComparator searchCmp = new MultiComparator(searchCmps, fields); // use only a single comparator for searching
+        MultiComparator searchCmp = new MultiComparator(fieldCount, searchCmps); // use only a single comparator for searching
         
         RangePredicate rangePred = new RangePredicate(true, lowKey, highKey, searchCmp);
         btree.search(rangeCursor, rangePred, leafFrame, interiorFrame);
@@ -486,15 +500,12 @@
         IBTreeLeafFrame leafFrame = leafFrameFactory.getFrame();
         IBTreeInteriorFrame interiorFrame = interiorFrameFactory.getFrame();
         IBTreeMetaDataFrame metaFrame = metaFrameFactory.getFrame();   
-
-    	IFieldAccessor[] fields = new IFieldAccessor[2];
-    	fields[0] = new UTF8StringAccessor(); // key        
-    	fields[1] = new UTF8StringAccessor(); // value
-    	
-    	int keyLen = 1;
-    	IBinaryComparator[] cmps = new IBinaryComparator[keyLen];
+        
+    	int fieldCount = 2;
+    	int keyFieldCount = 1;
+    	IBinaryComparator[] cmps = new IBinaryComparator[keyFieldCount];
     	cmps[0] = UTF8StringBinaryComparatorFactory.INSTANCE.createBinaryComparator();
-    	MultiComparator cmp = new MultiComparator(cmps, fields);
+    	MultiComparator cmp = new MultiComparator(fieldCount, cmps);
     	
     	BTree btree = new BTree(bufferCache, interiorFrameFactory, leafFrameFactory, cmp);
     	btree.create(fileId, leafFrame, metaFrame);
@@ -506,7 +517,7 @@
     	IHyracksContext ctx = new HyracksContext(HYRACKS_FRAME_SIZE);        
         ByteBuffer frame = ctx.getResourceManager().allocateFrame();
 		FrameTupleAppender appender = new FrameTupleAppender(ctx);				
-		ArrayTupleBuilder tb = new ArrayTupleBuilder(cmp.getFields().length);
+		ArrayTupleBuilder tb = new ArrayTupleBuilder(cmp.getFieldCount());
 		DataOutput dos = tb.getDataOutput();
 		
 		ISerializerDeserializer[] recDescSers = { UTF8StringSerializerDeserializer.INSTANCE, UTF8StringSerializerDeserializer.INSTANCE };
@@ -569,27 +580,40 @@
         
         IBTreeCursor rangeCursor = new RangeSearchCursor(leafFrame);
                 
-        ByteArrayAccessibleOutputStream lkbaaos = new ByteArrayAccessibleOutputStream();
-    	DataOutputStream lkdos = new DataOutputStream(lkbaaos);    	    	    	
-    	UTF8StringSerializerDeserializer.INSTANCE.serialize("cbf", lkdos);
+        // build low and high keys                       
+        ArrayTupleBuilder ktb = new ArrayTupleBuilder(cmp.getKeyFieldCount());
+		DataOutput kdos = ktb.getDataOutput();
+		
+		ISerializerDeserializer[] keyDescSers = { UTF8StringSerializerDeserializer.INSTANCE };
+		RecordDescriptor keyDesc = new RecordDescriptor(keyDescSers);
+		IFrameTupleAccessor keyAccessor = new FrameTupleAccessor(ctx, keyDesc);
+		keyAccessor.reset(frame);
+		
+		appender.reset(frame, true);
+		
+		// build and append low key
+		ktb.reset();
+		UTF8StringSerializerDeserializer.INSTANCE.serialize("cbf", kdos);
+    	ktb.addFieldEndOffset();    	
+    	appender.append(ktb.getFieldEndOffsets(), ktb.getByteArray(), 0, ktb.getSize());
     	
-    	ByteArrayAccessibleOutputStream hkbaaos = new ByteArrayAccessibleOutputStream();
-    	DataOutputStream hkdos = new DataOutputStream(hkbaaos);    	    	    	
-    	UTF8StringSerializerDeserializer.INSTANCE.serialize("cc7", hkdos);
+    	// build and append high key
+    	ktb.reset();
+    	UTF8StringSerializerDeserializer.INSTANCE.serialize("cc7", kdos);
+    	ktb.addFieldEndOffset();
+    	appender.append(ktb.getFieldEndOffsets(), ktb.getByteArray(), 0, ktb.getSize());
         
-    	byte[] lowKeyBytes = lkbaaos.toByteArray();
-        ByteBuffer lowKeyBuf = ByteBuffer.wrap(lowKeyBytes);
-        SelfDescTupleReference lowKey = new SelfDescTupleReference(cmp.getFields());
-        lowKey.reset(lowKeyBuf, 0);
+    	// create tuplereferences for search keys
+    	FrameTupleReference lowKey = new FrameTupleReference();
+    	lowKey.reset(keyAccessor, 0);
+    	
+		FrameTupleReference highKey = new FrameTupleReference();
+		highKey.reset(keyAccessor, 1);
         
-        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();
-        MultiComparator searchCmp = new MultiComparator(searchCmps, fields);
+        MultiComparator searchCmp = new MultiComparator(fieldCount, searchCmps);
         
         RangePredicate rangePred = new RangePredicate(true, lowKey, highKey, searchCmp);
         btree.search(rangeCursor, rangePred, leafFrame, interiorFrame);
@@ -644,15 +668,12 @@
         IBTreeInteriorFrame interiorFrame = interiorFrameFactory.getFrame();
         IBTreeMetaDataFrame metaFrame = metaFrameFactory.getFrame();   
         
-        IFieldAccessor[] fields = new IFieldAccessor[2];
-        fields[0] = new UTF8StringAccessor(); // key        
-        fields[1] = new UTF8StringAccessor(); // value
-
-        int keyLen = 1;
-        IBinaryComparator[] cmps = new IBinaryComparator[keyLen];
+        int fieldCount = 2;
+        int keyFieldCount = 1;
+        IBinaryComparator[] cmps = new IBinaryComparator[keyFieldCount];
         cmps[0] = UTF8StringBinaryComparatorFactory.INSTANCE.createBinaryComparator();
         
-        MultiComparator cmp = new MultiComparator(cmps, fields);
+        MultiComparator cmp = new MultiComparator(fieldCount, cmps);
 
         BTree btree = new BTree(bufferCache, interiorFrameFactory, leafFrameFactory, cmp);
         btree.create(fileId, leafFrame, metaFrame);
@@ -664,7 +685,7 @@
         IHyracksContext ctx = new HyracksContext(HYRACKS_FRAME_SIZE);        
         ByteBuffer frame = ctx.getResourceManager().allocateFrame();
 		FrameTupleAppender appender = new FrameTupleAppender(ctx);				
-		ArrayTupleBuilder tb = new ArrayTupleBuilder(cmp.getFields().length);
+		ArrayTupleBuilder tb = new ArrayTupleBuilder(cmp.getFieldCount());
 		DataOutput dos = tb.getDataOutput();
 		
 		ISerializerDeserializer[] recDescSers = { UTF8StringSerializerDeserializer.INSTANCE, UTF8StringSerializerDeserializer.INSTANCE };
@@ -799,19 +820,15 @@
         
         IBTreeLeafFrame leafFrame = leafFrameFactory.getFrame();
         IBTreeInteriorFrame interiorFrame = interiorFrameFactory.getFrame();
-        IBTreeMetaDataFrame metaFrame = metaFrameFactory.getFrame();          
+        IBTreeMetaDataFrame metaFrame = metaFrameFactory.getFrame();       
         
-        IFieldAccessor[] fields = new IFieldAccessor[3];
-        fields[0] = new Int32Accessor();
-        fields[1] = new Int32Accessor();
-        fields[2] = new Int32Accessor();
-        
-        int keyLen = 2;
-        IBinaryComparator[] cmps = new IBinaryComparator[keyLen];
+        int fieldCount = 3;
+        int keyFieldCount = 2;
+        IBinaryComparator[] cmps = new IBinaryComparator[keyFieldCount];
         cmps[0] = IntegerBinaryComparatorFactory.INSTANCE.createBinaryComparator();
         cmps[1] = IntegerBinaryComparatorFactory.INSTANCE.createBinaryComparator();
         
-        MultiComparator cmp = new MultiComparator(cmps, fields);
+        MultiComparator cmp = new MultiComparator(fieldCount, cmps);
 
         BTree btree = new BTree(bufferCache, interiorFrameFactory, leafFrameFactory, cmp);
         btree.create(fileId, leafFrame, metaFrame);
@@ -823,7 +840,7 @@
         IHyracksContext ctx = new HyracksContext(HYRACKS_FRAME_SIZE);        
         ByteBuffer frame = ctx.getResourceManager().allocateFrame();
 		FrameTupleAppender appender = new FrameTupleAppender(ctx);				
-		ArrayTupleBuilder tb = new ArrayTupleBuilder(cmp.getFields().length);
+		ArrayTupleBuilder tb = new ArrayTupleBuilder(cmp.getFieldCount());
 		DataOutput dos = tb.getDataOutput();
 		
 		ISerializerDeserializer[] recDescSers = { IntegerSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE };
@@ -868,27 +885,40 @@
         print("RANGE SEARCH:\n");
         IBTreeCursor rangeCursor = new RangeSearchCursor(leafFrame);
         
-        ByteArrayAccessibleOutputStream lkbaaos = new ByteArrayAccessibleOutputStream();
-    	DataOutputStream lkdos = new DataOutputStream(lkbaaos);    	    	    	
-    	IntegerSerializerDeserializer.INSTANCE.serialize(44444, lkdos);
+        // build low and high keys                       
+        ArrayTupleBuilder ktb = new ArrayTupleBuilder(1);
+		DataOutput kdos = ktb.getDataOutput();
+		
+		ISerializerDeserializer[] keyDescSers = { IntegerSerializerDeserializer.INSTANCE };
+		RecordDescriptor keyDesc = new RecordDescriptor(keyDescSers);
+		IFrameTupleAccessor keyAccessor = new FrameTupleAccessor(ctx, keyDesc);
+		keyAccessor.reset(frame);
+		
+		appender.reset(frame, true);
+		
+		// build and append low key
+		ktb.reset();
+    	IntegerSerializerDeserializer.INSTANCE.serialize(44444, kdos);
+    	ktb.addFieldEndOffset();    	
+    	appender.append(ktb.getFieldEndOffsets(), ktb.getByteArray(), 0, ktb.getSize());
     	
-    	ByteArrayAccessibleOutputStream hkbaaos = new ByteArrayAccessibleOutputStream();
-    	DataOutputStream hkdos = new DataOutputStream(hkbaaos);    	    	    	
-    	IntegerSerializerDeserializer.INSTANCE.serialize(44500, hkdos);
-    	    	        
-    	byte[] lowKeyBytes = lkbaaos.toByteArray();
-        ByteBuffer lowKeyBuf = ByteBuffer.wrap(lowKeyBytes);
-        SelfDescTupleReference lowKey = new SelfDescTupleReference(cmp.getFields());
-        lowKey.reset(lowKeyBuf, 0);
+    	// build and append high key
+    	ktb.reset();
+    	IntegerSerializerDeserializer.INSTANCE.serialize(44500, kdos);
+    	ktb.addFieldEndOffset();
+    	appender.append(ktb.getFieldEndOffsets(), ktb.getByteArray(), 0, ktb.getSize());
         
-        byte[] highKeyBytes = hkbaaos.toByteArray();
-        ByteBuffer highKeyBuf = ByteBuffer.wrap(highKeyBytes);
-        SelfDescTupleReference highKey = new SelfDescTupleReference(cmp.getFields());
-        highKey.reset(highKeyBuf, 0);
+    	// create tuplereferences for search keys
+    	FrameTupleReference lowKey = new FrameTupleReference();
+    	lowKey.reset(keyAccessor, 0);
+    	
+		FrameTupleReference highKey = new FrameTupleReference();
+		highKey.reset(keyAccessor, 1);
+		              
     	        
         IBinaryComparator[] searchCmps = new IBinaryComparator[1];
         searchCmps[0] = IntegerBinaryComparatorFactory.INSTANCE.createBinaryComparator();
-        MultiComparator searchCmp = new MultiComparator(searchCmps, fields);
+        MultiComparator searchCmp = new MultiComparator(fieldCount, searchCmps);
         
         // TODO: check when searching backwards
         RangePredicate rangePred = new RangePredicate(true, lowKey, highKey, searchCmp);
@@ -940,18 +970,13 @@
         IBTreeLeafFrame leafFrame = leafFrameFactory.getFrame();
         IBTreeInteriorFrame interiorFrame = interiorFrameFactory.getFrame();
         IBTreeMetaDataFrame metaFrame = metaFrameFactory.getFrame();
-
-        int keyLen = 2;
         
-        IFieldAccessor[] fields = new IFieldAccessor[3];
-        fields[0] = new Int32Accessor();
-        fields[1] = new Int32Accessor();
-        fields[2] = new Int32Accessor();
-
-        IBinaryComparator[] cmps = new IBinaryComparator[keyLen];
+        int fieldCount = 3;
+        int keyFieldCount = 2;
+        IBinaryComparator[] cmps = new IBinaryComparator[keyFieldCount];
         cmps[0] = IntegerBinaryComparatorFactory.INSTANCE.createBinaryComparator();
         cmps[1] = IntegerBinaryComparatorFactory.INSTANCE.createBinaryComparator();        
-        MultiComparator cmp = new MultiComparator(cmps, fields);
+        MultiComparator cmp = new MultiComparator(fieldCount, cmps);
 
         BTree btree = new BTree(bufferCache, interiorFrameFactory, leafFrameFactory, cmp);
         btree.create(fileId, leafFrame, metaFrame);
@@ -963,7 +988,7 @@
         IHyracksContext ctx = new HyracksContext(HYRACKS_FRAME_SIZE);        
         ByteBuffer frame = ctx.getResourceManager().allocateFrame();
 		FrameTupleAppender appender = new FrameTupleAppender(ctx);				
-		ArrayTupleBuilder tb = new ArrayTupleBuilder(cmp.getFields().length);
+		ArrayTupleBuilder tb = new ArrayTupleBuilder(cmp.getFieldCount());
 		DataOutput dos = tb.getDataOutput();
 		
 		ISerializerDeserializer[] recDescSers = { IntegerSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE };
@@ -1066,30 +1091,45 @@
         print("RANGE SEARCH:\n");
         IBTreeCursor rangeCursor = new RangeSearchCursor(leafFrame);
 
-        ByteArrayAccessibleOutputStream lkbaaos = new ByteArrayAccessibleOutputStream();
-    	DataOutputStream lkdos = new DataOutputStream(lkbaaos);    	    	    	
-    	IntegerSerializerDeserializer.INSTANCE.serialize(12, lkdos);
-    	IntegerSerializerDeserializer.INSTANCE.serialize(12, lkdos);
+        // build low and high keys                       
+        ArrayTupleBuilder ktb = new ArrayTupleBuilder(cmp.getKeyFieldCount());
+		DataOutput kdos = ktb.getDataOutput();
+		
+		ISerializerDeserializer[] keyDescSers = { IntegerSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE };
+		RecordDescriptor keyDesc = new RecordDescriptor(keyDescSers);
+		IFrameTupleAccessor keyAccessor = new FrameTupleAccessor(ctx, keyDesc);
+		keyAccessor.reset(frame);
+		
+		appender.reset(frame, true);
+		
+		// build and append low key
+		ktb.reset();
+    	IntegerSerializerDeserializer.INSTANCE.serialize(12, kdos);
+    	ktb.addFieldEndOffset();
+    	IntegerSerializerDeserializer.INSTANCE.serialize(12, kdos);
+    	ktb.addFieldEndOffset();    	
+    	appender.append(ktb.getFieldEndOffsets(), ktb.getByteArray(), 0, ktb.getSize());
     	
-    	ByteArrayAccessibleOutputStream hkbaaos = new ByteArrayAccessibleOutputStream();
-    	DataOutputStream hkdos = new DataOutputStream(hkbaaos);    	    	    	
-    	IntegerSerializerDeserializer.INSTANCE.serialize(19, hkdos);
-    	IntegerSerializerDeserializer.INSTANCE.serialize(19, hkdos);    	        
-    	       
-        byte[] lowKeyBytes = lkbaaos.toByteArray();
-        ByteBuffer lowKeyBuf = ByteBuffer.wrap(lowKeyBytes);
-        SelfDescTupleReference lowKey = new SelfDescTupleReference(cmp.getFields());
-        lowKey.reset(lowKeyBuf, 0);
+    	// build and append high key
+    	ktb.reset();
+    	IntegerSerializerDeserializer.INSTANCE.serialize(19, kdos);
+    	ktb.addFieldEndOffset();
+    	IntegerSerializerDeserializer.INSTANCE.serialize(19, kdos);
+    	ktb.addFieldEndOffset();
+    	appender.append(ktb.getFieldEndOffsets(), ktb.getByteArray(), 0, ktb.getSize());
         
-        byte[] highKeyBytes = hkbaaos.toByteArray();
-        ByteBuffer highKeyBuf = ByteBuffer.wrap(highKeyBytes);
-        SelfDescTupleReference highKey = new SelfDescTupleReference(cmp.getFields());
-        highKey.reset(highKeyBuf, 0);
+    	// create tuplereferences for search keys
+    	FrameTupleReference lowKey = new FrameTupleReference();
+    	lowKey.reset(keyAccessor, 0);
+    	
+		FrameTupleReference highKey = new FrameTupleReference();
+		highKey.reset(keyAccessor, 1);
+		               
         
         IBinaryComparator[] searchCmps = new IBinaryComparator[2];
         searchCmps[0] = IntegerBinaryComparatorFactory.INSTANCE.createBinaryComparator();
         searchCmps[1] = IntegerBinaryComparatorFactory.INSTANCE.createBinaryComparator();
-        MultiComparator searchCmp = new MultiComparator(searchCmps, fields);
+        MultiComparator searchCmp = new MultiComparator(fieldCount, searchCmps);
                 
         //print("INDEX RANGE SEARCH ON: " + cmp.printKey(lowKey, 0) + " " + cmp.printKey(highKey, 0) + "\n");