diff --git a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/dataflow/value/ITypeTrait.java b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/dataflow/value/ITypeTrait.java
new file mode 100644
index 0000000..0f46f57
--- /dev/null
+++ b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/dataflow/value/ITypeTrait.java
@@ -0,0 +1,8 @@
+package edu.uci.ics.hyracks.api.dataflow.value;
+
+import java.io.Serializable;
+
+public interface ITypeTrait extends Serializable {
+	public static final int VARIABLE_LENGTH = -1;	
+	int getStaticallyKnownDataLength();
+}
diff --git a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/dataflow/value/RecordDescriptor.java b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/dataflow/value/RecordDescriptor.java
index 20dc034..4922723 100644
--- a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/dataflow/value/RecordDescriptor.java
+++ b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/dataflow/value/RecordDescriptor.java
@@ -21,12 +21,25 @@
     private static final long serialVersionUID = 1L;
 
     private final ISerializerDeserializer[] fields;
-
+    private final ITypeTrait[] typeTraits;
+    
+    // leaving this constructor for backwards-compatibility
     public RecordDescriptor(ISerializerDeserializer[] fields) {
         this.fields = fields;
+        this.typeTraits = null;
+    }
+    
+    // temporarily adding constructor to include type traits
+    public RecordDescriptor(ISerializerDeserializer[] fields, ITypeTrait[] typeTraits) {
+    	this.fields = fields;
+        this.typeTraits = typeTraits;
     }
 
     public ISerializerDeserializer[] getFields() {
         return fields;
     }
+    
+    public ITypeTrait[] getTypeTraits() {
+    	return typeTraits;
+    }
 }
\ No newline at end of file
diff --git a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/dataflow/value/TypeTrait.java b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/dataflow/value/TypeTrait.java
new file mode 100644
index 0000000..6be9552
--- /dev/null
+++ b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/dataflow/value/TypeTrait.java
@@ -0,0 +1,16 @@
+package edu.uci.ics.hyracks.api.dataflow.value;
+
+public class TypeTrait implements ITypeTrait {
+	
+	private static final long serialVersionUID = 1L;
+	private int length;
+	
+	public TypeTrait(int length) {
+		this.length = length;
+	}
+	
+	@Override
+	public int getStaticallyKnownDataLength() {
+		return length;
+	}	
+}
diff --git a/hyracks/hyracks-examples/btree-example/btreeclient/src/main/java/edu/uci/ics/hyracks/examples/btree/client/InsertPipelineExample.java b/hyracks/hyracks-examples/btree-example/btreeclient/src/main/java/edu/uci/ics/hyracks/examples/btree/client/InsertPipelineExample.java
index 7a27c77..80cc608 100644
--- a/hyracks/hyracks-examples/btree-example/btreeclient/src/main/java/edu/uci/ics/hyracks/examples/btree/client/InsertPipelineExample.java
+++ b/hyracks/hyracks-examples/btree-example/btreeclient/src/main/java/edu/uci/ics/hyracks/examples/btree/client/InsertPipelineExample.java
@@ -15,7 +15,9 @@
 import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparatorFactory;
 import edu.uci.ics.hyracks.api.dataflow.value.IBinaryHashFunctionFactory;
 import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
+import edu.uci.ics.hyracks.api.dataflow.value.ITypeTrait;
 import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
+import edu.uci.ics.hyracks.api.dataflow.value.TypeTrait;
 import edu.uci.ics.hyracks.api.job.JobSpecification;
 import edu.uci.ics.hyracks.dataflow.common.data.comparators.IntegerBinaryComparatorFactory;
 import edu.uci.ics.hyracks.dataflow.common.data.comparators.UTF8StringBinaryComparatorFactory;
@@ -40,6 +42,7 @@
 import edu.uci.ics.hyracks.storage.am.btree.frames.NSMInteriorFrameFactory;
 import edu.uci.ics.hyracks.storage.am.btree.frames.NSMLeafFrameFactory;
 import edu.uci.ics.hyracks.storage.am.btree.impls.BTreeOp;
+import edu.uci.ics.hyracks.storage.am.btree.tuples.TypeAwareTupleWriterFactory;
 
 // This example will insert tuples into the primary and secondary index using an insert pipeline
 
@@ -60,11 +63,11 @@
         @Option(name = "-num-tuples", usage = "Total number of tuples to to be generated for insertion", required = true)
         public int numTuples;
         
-        @Option(name = "-primary-name", usage = "B-Tree file name of primary index", required = true)
-        public String primaryBtreeName;
+        @Option(name = "-primary-btreename", usage = "B-Tree file name of primary index", required = true)
+        public String primaryBTreeName;
         
-        @Option(name = "-secondary-name", usage = "B-Tree file name of secondary index", required = true)
-        public String secondaryBtreeName;                
+        @Option(name = "-secondary-btreename", usage = "B-Tree file name of secondary index", required = true)
+        public String secondaryBTreeName;            
     }
 
     public static void main(String[] args) throws Exception {
@@ -105,10 +108,7 @@
         // run data generator on first nodecontroller given
         PartitionConstraint dataGenConstraint = new ExplicitPartitionConstraint(new LocationConstraint[] { new AbsoluteLocationConstraint(splitNCs[0]) });
         dataGen.setPartitionConstraint(dataGenConstraint);
-        
-        // create factories and providers for B-Trees
-        IBTreeInteriorFrameFactory interiorFrameFactory = new NSMInteriorFrameFactory();
-        IBTreeLeafFrameFactory leafFrameFactory = new NSMLeafFrameFactory();        
+                
         IBufferCacheProvider bufferCacheProvider = BufferCacheProvider.INSTANCE;
         IBTreeRegistryProvider btreeRegistryProvider = BTreeRegistryProvider.INSTANCE;
         IFileMappingProviderProvider fileMappingProviderProvider = FileMappingProviderProvider.INSTANCE;
@@ -116,29 +116,49 @@
         // prepare insertion into primary index
         // tuples to be put into B-Tree shall have 4 fields
         int primaryFieldCount = 4;
+        ITypeTrait[] primaryTypeTraits = new ITypeTrait[primaryFieldCount];
+        primaryTypeTraits[0] = new TypeTrait(4);
+        primaryTypeTraits[1] = new TypeTrait(ITypeTrait.VARIABLE_LENGTH);
+        primaryTypeTraits[2] = new TypeTrait(4);
+        primaryTypeTraits[3] = new TypeTrait(ITypeTrait.VARIABLE_LENGTH);
+        
+        // create factories and providers for secondary B-Tree
+        TypeAwareTupleWriterFactory primaryTupleWriterFactory = new TypeAwareTupleWriterFactory(primaryTypeTraits);
+        IBTreeInteriorFrameFactory primaryInteriorFrameFactory = new NSMInteriorFrameFactory(primaryTupleWriterFactory);
+        IBTreeLeafFrameFactory primaryLeafFrameFactory = new NSMLeafFrameFactory(primaryTupleWriterFactory);
+        
         // the B-Tree expects its keyfields to be at the front of its input tuple 
         int[] primaryFieldPermutation = { 2, 1, 3, 4 }; // map field 2 of input tuple to field 0 of B-Tree tuple, etc.
         // comparator factories for primary index
         IBinaryComparatorFactory[] primaryComparatorFactories = new IBinaryComparatorFactory[1];
 		primaryComparatorFactories[0] = IntegerBinaryComparatorFactory.INSTANCE;
-		IFileSplitProvider primarySplitProvider = JobHelper.createFileSplitProvider(splitNCs, options.primaryBtreeName);
+		IFileSplitProvider primarySplitProvider = JobHelper.createFileSplitProvider(splitNCs, options.primaryBTreeName);
 		// create operator descriptor
-        BTreeInsertUpdateDeleteOperatorDescriptor primaryInsert = new BTreeInsertUpdateDeleteOperatorDescriptor(spec, recDesc, bufferCacheProvider, btreeRegistryProvider, primarySplitProvider, fileMappingProviderProvider, interiorFrameFactory, leafFrameFactory, primaryFieldCount, primaryComparatorFactories, primaryFieldPermutation, BTreeOp.BTO_INSERT);
+        BTreeInsertUpdateDeleteOperatorDescriptor primaryInsert = new BTreeInsertUpdateDeleteOperatorDescriptor(spec, recDesc, bufferCacheProvider, btreeRegistryProvider, primarySplitProvider, fileMappingProviderProvider, primaryInteriorFrameFactory, primaryLeafFrameFactory, primaryTypeTraits, primaryComparatorFactories, primaryFieldPermutation, BTreeOp.BTO_INSERT);
         PartitionConstraint primaryInsertConstraint = JobHelper.createPartitionConstraint(splitNCs);
         primaryInsert.setPartitionConstraint(primaryInsertConstraint);
         
         // prepare insertion into secondary index
         // tuples to be put into B-Tree shall have 2 fields
-        int secondaryFieldCount = 2;
+        int secondaryFieldCount = 2;       
+        ITypeTrait[] secondaryTypeTraits = new ITypeTrait[secondaryFieldCount];
+        secondaryTypeTraits[0] = new TypeTrait(ITypeTrait.VARIABLE_LENGTH);
+        secondaryTypeTraits[1] = new TypeTrait(4); 
+        
+        // create factories and providers for secondary B-Tree
+        TypeAwareTupleWriterFactory secondaryTupleWriterFactory = new TypeAwareTupleWriterFactory(secondaryTypeTraits);
+        IBTreeInteriorFrameFactory secondaryInteriorFrameFactory = new NSMInteriorFrameFactory(secondaryTupleWriterFactory);
+        IBTreeLeafFrameFactory secondaryLeafFrameFactory = new NSMLeafFrameFactory(secondaryTupleWriterFactory);
+        
         // the B-Tree expects its keyfields to be at the front of its input tuple 
         int[] secondaryFieldPermutation = { 1, 2 };
         // comparator factories for primary index
         IBinaryComparatorFactory[] secondaryComparatorFactories = new IBinaryComparatorFactory[2];
 		secondaryComparatorFactories[0] = UTF8StringBinaryComparatorFactory.INSTANCE;
 		secondaryComparatorFactories[1] = IntegerBinaryComparatorFactory.INSTANCE;
-		IFileSplitProvider secondarySplitProvider = JobHelper.createFileSplitProvider(splitNCs, options.secondaryBtreeName);
+		IFileSplitProvider secondarySplitProvider = JobHelper.createFileSplitProvider(splitNCs, options.secondaryBTreeName);
 		// create operator descriptor
-        BTreeInsertUpdateDeleteOperatorDescriptor secondaryInsert = new BTreeInsertUpdateDeleteOperatorDescriptor(spec, recDesc, bufferCacheProvider, btreeRegistryProvider, secondarySplitProvider, fileMappingProviderProvider, interiorFrameFactory, leafFrameFactory, secondaryFieldCount, secondaryComparatorFactories, secondaryFieldPermutation, BTreeOp.BTO_INSERT);
+        BTreeInsertUpdateDeleteOperatorDescriptor secondaryInsert = new BTreeInsertUpdateDeleteOperatorDescriptor(spec, recDesc, bufferCacheProvider, btreeRegistryProvider, secondarySplitProvider, fileMappingProviderProvider, secondaryInteriorFrameFactory, secondaryLeafFrameFactory, secondaryTypeTraits, secondaryComparatorFactories, secondaryFieldPermutation, BTreeOp.BTO_INSERT);
         PartitionConstraint secondaryInsertConstraint = JobHelper.createPartitionConstraint(splitNCs);
         secondaryInsert.setPartitionConstraint(secondaryInsertConstraint);
         
diff --git a/hyracks/hyracks-examples/btree-example/btreeclient/src/main/java/edu/uci/ics/hyracks/examples/btree/client/PrimaryIndexBulkLoadExample.java b/hyracks/hyracks-examples/btree-example/btreeclient/src/main/java/edu/uci/ics/hyracks/examples/btree/client/PrimaryIndexBulkLoadExample.java
index dd0d11b..8a8f60e 100644
--- a/hyracks/hyracks-examples/btree-example/btreeclient/src/main/java/edu/uci/ics/hyracks/examples/btree/client/PrimaryIndexBulkLoadExample.java
+++ b/hyracks/hyracks-examples/btree-example/btreeclient/src/main/java/edu/uci/ics/hyracks/examples/btree/client/PrimaryIndexBulkLoadExample.java
@@ -29,7 +29,9 @@
 import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparatorFactory;
 import edu.uci.ics.hyracks.api.dataflow.value.IBinaryHashFunctionFactory;
 import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
+import edu.uci.ics.hyracks.api.dataflow.value.ITypeTrait;
 import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
+import edu.uci.ics.hyracks.api.dataflow.value.TypeTrait;
 import edu.uci.ics.hyracks.api.job.JobSpecification;
 import edu.uci.ics.hyracks.dataflow.common.data.comparators.IntegerBinaryComparatorFactory;
 import edu.uci.ics.hyracks.dataflow.common.data.hash.UTF8StringBinaryHashFunctionFactory;
@@ -52,6 +54,7 @@
 import edu.uci.ics.hyracks.storage.am.btree.dataflow.IFileMappingProviderProvider;
 import edu.uci.ics.hyracks.storage.am.btree.frames.NSMInteriorFrameFactory;
 import edu.uci.ics.hyracks.storage.am.btree.frames.NSMLeafFrameFactory;
+import edu.uci.ics.hyracks.storage.am.btree.tuples.TypeAwareTupleWriterFactory;
 
 // This example will load a primary index from randomly generated data
 
@@ -128,21 +131,28 @@
         PartitionConstraint sorterConstraint = JobHelper.createPartitionConstraint(splitNCs);
         sorter.setPartitionConstraint(sorterConstraint);
         
-        // create factories and providers for B-Tree
-        IBTreeInteriorFrameFactory interiorFrameFactory = new NSMInteriorFrameFactory();
-        IBTreeLeafFrameFactory leafFrameFactory = new NSMLeafFrameFactory();        
-        IBufferCacheProvider bufferCacheProvider = BufferCacheProvider.INSTANCE;
-        IBTreeRegistryProvider btreeRegistryProvider = BTreeRegistryProvider.INSTANCE;
-        IFileMappingProviderProvider fileMappingProviderProvider = FileMappingProviderProvider.INSTANCE;
-        
         // tuples to be put into B-Tree shall have 4 fields
         int fieldCount = 4;
+        ITypeTrait[] typeTraits = new ITypeTrait[fieldCount];
+        typeTraits[0] = new TypeTrait(4);
+        typeTraits[1] = new TypeTrait(ITypeTrait.VARIABLE_LENGTH);
+        typeTraits[2] = new TypeTrait(4);
+        typeTraits[3] = new TypeTrait(ITypeTrait.VARIABLE_LENGTH);
+        
+        // create factories and providers for B-Tree
+        TypeAwareTupleWriterFactory tupleWriterFactory = new TypeAwareTupleWriterFactory(typeTraits);
+        IBTreeInteriorFrameFactory interiorFrameFactory = new NSMInteriorFrameFactory(tupleWriterFactory);
+        IBTreeLeafFrameFactory leafFrameFactory = new NSMLeafFrameFactory(tupleWriterFactory);        
+        IBufferCacheProvider bufferCacheProvider = BufferCacheProvider.INSTANCE;
+        IBTreeRegistryProvider btreeRegistryProvider = BTreeRegistryProvider.INSTANCE;
+        IFileMappingProviderProvider fileMappingProviderProvider = FileMappingProviderProvider.INSTANCE;        
+        
         // the B-Tree expects its keyfields to be at the front of its input tuple 
         int[] fieldPermutation = { 2, 1, 3, 4 }; // map field 2 of input tuple to field 0 of B-Tree tuple, etc.
         IFileSplitProvider btreeSplitProvider = JobHelper.createFileSplitProvider(splitNCs, options.btreeName);
         BTreeBulkLoadOperatorDescriptor btreeBulkLoad = new BTreeBulkLoadOperatorDescriptor(spec, 
         		bufferCacheProvider, btreeRegistryProvider, btreeSplitProvider, fileMappingProviderProvider, interiorFrameFactory,
-                leafFrameFactory, fieldCount, comparatorFactories, fieldPermutation, 0.7f);
+                leafFrameFactory, typeTraits, comparatorFactories, fieldPermutation, 0.7f);
         PartitionConstraint bulkLoadConstraint = JobHelper.createPartitionConstraint(splitNCs);
         btreeBulkLoad.setPartitionConstraint(bulkLoadConstraint);
         
diff --git a/hyracks/hyracks-examples/btree-example/btreeclient/src/main/java/edu/uci/ics/hyracks/examples/btree/client/PrimaryIndexEnlistFilesExample.java b/hyracks/hyracks-examples/btree-example/btreeclient/src/main/java/edu/uci/ics/hyracks/examples/btree/client/PrimaryIndexEnlistFilesExample.java
index a1e4fbc..1bed3bf 100644
--- a/hyracks/hyracks-examples/btree-example/btreeclient/src/main/java/edu/uci/ics/hyracks/examples/btree/client/PrimaryIndexEnlistFilesExample.java
+++ b/hyracks/hyracks-examples/btree-example/btreeclient/src/main/java/edu/uci/ics/hyracks/examples/btree/client/PrimaryIndexEnlistFilesExample.java
@@ -24,7 +24,9 @@
 import edu.uci.ics.hyracks.api.constraints.PartitionConstraint;
 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.ITypeTrait;
 import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
+import edu.uci.ics.hyracks.api.dataflow.value.TypeTrait;
 import edu.uci.ics.hyracks.api.job.JobSpecification;
 import edu.uci.ics.hyracks.dataflow.common.data.comparators.IntegerBinaryComparatorFactory;
 import edu.uci.ics.hyracks.dataflow.common.data.marshalling.IntegerSerializerDeserializer;
@@ -41,6 +43,7 @@
 import edu.uci.ics.hyracks.storage.am.btree.dataflow.IFileMappingProviderProvider;
 import edu.uci.ics.hyracks.storage.am.btree.frames.NSMInteriorFrameFactory;
 import edu.uci.ics.hyracks.storage.am.btree.frames.NSMLeafFrameFactory;
+import edu.uci.ics.hyracks.storage.am.btree.tuples.TypeAwareTupleWriterFactory;
 
 // This example will enlist existing files as primary index
 
@@ -93,9 +96,17 @@
                 UTF8StringSerializerDeserializer.INSTANCE
                 });
         
+        int fieldCount = 4;
+        ITypeTrait[] typeTraits = new ITypeTrait[fieldCount];
+        typeTraits[0] = new TypeTrait(4);
+        typeTraits[1] = new TypeTrait(ITypeTrait.VARIABLE_LENGTH);
+        typeTraits[2] = new TypeTrait(4);
+        typeTraits[3] = new TypeTrait(ITypeTrait.VARIABLE_LENGTH);
+        
         // create factories and providers for B-Tree
-        IBTreeInteriorFrameFactory interiorFrameFactory = new NSMInteriorFrameFactory();
-        IBTreeLeafFrameFactory leafFrameFactory = new NSMLeafFrameFactory();        
+        TypeAwareTupleWriterFactory tupleWriterFactory = new TypeAwareTupleWriterFactory(typeTraits);
+        IBTreeInteriorFrameFactory interiorFrameFactory = new NSMInteriorFrameFactory(tupleWriterFactory);
+        IBTreeLeafFrameFactory leafFrameFactory = new NSMLeafFrameFactory(tupleWriterFactory);        
         IBufferCacheProvider bufferCacheProvider = BufferCacheProvider.INSTANCE;
         IBTreeRegistryProvider btreeRegistryProvider = BTreeRegistryProvider.INSTANCE;
         IFileMappingProviderProvider fileMappingProviderProvider = FileMappingProviderProvider.INSTANCE;
@@ -104,7 +115,7 @@
         comparatorFactories[0] = IntegerBinaryComparatorFactory.INSTANCE;
         
         IFileSplitProvider btreeSplitProvider = JobHelper.createFileSplitProvider(splitNCs, options.btreeName);
-        BTreeFileEnlistmentOperatorDescriptor fileEnlistmentOp = new BTreeFileEnlistmentOperatorDescriptor(spec, recDesc, bufferCacheProvider, btreeRegistryProvider, btreeSplitProvider, fileMappingProviderProvider, interiorFrameFactory, leafFrameFactory, recDesc.getFields().length, comparatorFactories);
+        BTreeFileEnlistmentOperatorDescriptor fileEnlistmentOp = new BTreeFileEnlistmentOperatorDescriptor(spec, recDesc, bufferCacheProvider, btreeRegistryProvider, btreeSplitProvider, fileMappingProviderProvider, interiorFrameFactory, leafFrameFactory, typeTraits, comparatorFactories);
         PartitionConstraint fileEnlistmentConstraint = JobHelper.createPartitionConstraint(splitNCs);
         fileEnlistmentOp.setPartitionConstraint(fileEnlistmentConstraint);                                            
         
diff --git a/hyracks/hyracks-examples/btree-example/btreeclient/src/main/java/edu/uci/ics/hyracks/examples/btree/client/PrimaryIndexSearchExample.java b/hyracks/hyracks-examples/btree-example/btreeclient/src/main/java/edu/uci/ics/hyracks/examples/btree/client/PrimaryIndexSearchExample.java
index 1d72cc3..928cf1f 100644
--- a/hyracks/hyracks-examples/btree-example/btreeclient/src/main/java/edu/uci/ics/hyracks/examples/btree/client/PrimaryIndexSearchExample.java
+++ b/hyracks/hyracks-examples/btree-example/btreeclient/src/main/java/edu/uci/ics/hyracks/examples/btree/client/PrimaryIndexSearchExample.java
@@ -25,7 +25,9 @@
 import edu.uci.ics.hyracks.api.constraints.PartitionConstraint;
 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.ITypeTrait;
 import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
+import edu.uci.ics.hyracks.api.dataflow.value.TypeTrait;
 import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
 import edu.uci.ics.hyracks.api.job.JobSpecification;
 import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
@@ -47,6 +49,7 @@
 import edu.uci.ics.hyracks.storage.am.btree.dataflow.IFileMappingProviderProvider;
 import edu.uci.ics.hyracks.storage.am.btree.frames.NSMInteriorFrameFactory;
 import edu.uci.ics.hyracks.storage.am.btree.frames.NSMLeafFrameFactory;
+import edu.uci.ics.hyracks.storage.am.btree.tuples.TypeAwareTupleWriterFactory;
 
 // This example will perform an ordered scan on the primary index
 // i.e. a range-search for [-infinity, +infinity]
@@ -92,9 +95,17 @@
 
     	String[] splitNCs = options.ncs.split(",");
     	
+    	int fieldCount = 4;
+        ITypeTrait[] typeTraits = new ITypeTrait[fieldCount];
+        typeTraits[0] = new TypeTrait(4);
+        typeTraits[1] = new TypeTrait(ITypeTrait.VARIABLE_LENGTH);
+        typeTraits[2] = new TypeTrait(4);
+        typeTraits[3] = new TypeTrait(ITypeTrait.VARIABLE_LENGTH);
+    	
         // create factories and providers for B-Tree
-        IBTreeInteriorFrameFactory interiorFrameFactory = new NSMInteriorFrameFactory();
-        IBTreeLeafFrameFactory leafFrameFactory = new NSMLeafFrameFactory();        
+        TypeAwareTupleWriterFactory tupleWriterFactory = new TypeAwareTupleWriterFactory(typeTraits);
+        IBTreeInteriorFrameFactory interiorFrameFactory = new NSMInteriorFrameFactory(tupleWriterFactory);
+        IBTreeLeafFrameFactory leafFrameFactory = new NSMLeafFrameFactory(tupleWriterFactory);        
         IBufferCacheProvider bufferCacheProvider = BufferCacheProvider.INSTANCE;
         IBTreeRegistryProvider btreeRegistryProvider = BTreeRegistryProvider.INSTANCE;
         IFileMappingProviderProvider fileMappingProviderProvider = FileMappingProviderProvider.INSTANCE;
@@ -127,13 +138,12 @@
 		ConstantTupleSourceOperatorDescriptor keyProviderOp = new ConstantTupleSourceOperatorDescriptor(spec, keyRecDesc, tb.getFieldEndOffsets(), tb.getByteArray(), tb.getSize());
 		PartitionConstraint keyProviderPartitionConstraint = JobHelper.createPartitionConstraint(splitNCs);
 		keyProviderOp.setPartitionConstraint(keyProviderPartitionConstraint);
-		
-		
+				
         int[] lowKeyFields = { 0 }; // low key is in field 0 of tuples going into search op 
         int[] highKeyFields = { 1 }; // low key is in field 1 of tuples going into search op
-        
+        		
         IFileSplitProvider btreeSplitProvider = JobHelper.createFileSplitProvider(splitNCs, options.btreeName);
-        BTreeSearchOperatorDescriptor btreeSearchOp = new BTreeSearchOperatorDescriptor(spec, recDesc, bufferCacheProvider, btreeRegistryProvider, btreeSplitProvider, fileMappingProviderProvider, interiorFrameFactory, leafFrameFactory, recDesc.getFields().length, comparatorFactories, true, lowKeyFields, highKeyFields);
+        BTreeSearchOperatorDescriptor btreeSearchOp = new BTreeSearchOperatorDescriptor(spec, recDesc, bufferCacheProvider, btreeRegistryProvider, btreeSplitProvider, fileMappingProviderProvider, interiorFrameFactory, leafFrameFactory, typeTraits, comparatorFactories, true, lowKeyFields, highKeyFields);
         PartitionConstraint btreeSearchConstraint = JobHelper.createPartitionConstraint(splitNCs);
         btreeSearchOp.setPartitionConstraint(btreeSearchConstraint);
         
diff --git a/hyracks/hyracks-examples/btree-example/btreeclient/src/main/java/edu/uci/ics/hyracks/examples/btree/client/SecondaryIndexBulkLoadExample.java b/hyracks/hyracks-examples/btree-example/btreeclient/src/main/java/edu/uci/ics/hyracks/examples/btree/client/SecondaryIndexBulkLoadExample.java
index dc6f420..7702090 100644
--- a/hyracks/hyracks-examples/btree-example/btreeclient/src/main/java/edu/uci/ics/hyracks/examples/btree/client/SecondaryIndexBulkLoadExample.java
+++ b/hyracks/hyracks-examples/btree-example/btreeclient/src/main/java/edu/uci/ics/hyracks/examples/btree/client/SecondaryIndexBulkLoadExample.java
@@ -24,7 +24,9 @@
 import edu.uci.ics.hyracks.api.constraints.PartitionConstraint;
 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.ITypeTrait;
 import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
+import edu.uci.ics.hyracks.api.dataflow.value.TypeTrait;
 import edu.uci.ics.hyracks.api.job.JobSpecification;
 import edu.uci.ics.hyracks.dataflow.common.data.comparators.IntegerBinaryComparatorFactory;
 import edu.uci.ics.hyracks.dataflow.common.data.comparators.UTF8StringBinaryComparatorFactory;
@@ -45,6 +47,7 @@
 import edu.uci.ics.hyracks.storage.am.btree.dataflow.IFileMappingProviderProvider;
 import edu.uci.ics.hyracks.storage.am.btree.frames.NSMInteriorFrameFactory;
 import edu.uci.ics.hyracks.storage.am.btree.frames.NSMLeafFrameFactory;
+import edu.uci.ics.hyracks.storage.am.btree.tuples.TypeAwareTupleWriterFactory;
 
 // This example will load a secondary index with <key, primary-index key> pairs
 // We require an existing primary index built with PrimaryIndexBulkLoadExample
@@ -64,10 +67,10 @@
         public String ncs;
                 
         @Option(name = "-primary-btreename", usage = "Name of primary-index B-Tree to load from", required = true)
-        public String primaryBtreeName;
+        public String primaryBTreeName;
         
         @Option(name = "-secondary-btreename", usage = "B-Tree file name for secondary index to be built", required = true)
-        public String btreeName;
+        public String secondaryBTreeName;
         
         @Option(name = "-sortbuffer-size", usage = "Sort buffer size in frames (default: 32768)", required = false)
         public int sbSize = 32768;
@@ -96,6 +99,10 @@
 
     	String[] splitNCs = options.ncs.split(",");
     	
+    	IBufferCacheProvider bufferCacheProvider = BufferCacheProvider.INSTANCE;
+        IBTreeRegistryProvider btreeRegistryProvider = BTreeRegistryProvider.INSTANCE;
+        IFileMappingProviderProvider fileMappingProviderProvider = FileMappingProviderProvider.INSTANCE;
+    	
     	// schema of tuples that we are retrieving from the primary index
         RecordDescriptor recDesc = new RecordDescriptor(new ISerializerDeserializer[] {                                    
                 IntegerSerializerDeserializer.INSTANCE, // we will use this as payload in secondary index
@@ -104,19 +111,24 @@
                 UTF8StringSerializerDeserializer.INSTANCE
                 });
     	
-        // create factories and providers for B-Tree(s)
-        IBTreeInteriorFrameFactory interiorFrameFactory = new NSMInteriorFrameFactory();
-        IBTreeLeafFrameFactory leafFrameFactory = new NSMLeafFrameFactory();
-        IBufferCacheProvider bufferCacheProvider = BufferCacheProvider.INSTANCE;
-        IBTreeRegistryProvider btreeRegistryProvider = BTreeRegistryProvider.INSTANCE;
-        IFileMappingProviderProvider fileMappingProviderProvider = FileMappingProviderProvider.INSTANCE;
-    	        
+        int primaryFieldCount = 4;
+        ITypeTrait[] primaryTypeTraits = new ITypeTrait[primaryFieldCount];
+        primaryTypeTraits[0] = new TypeTrait(4);
+        primaryTypeTraits[1] = new TypeTrait(ITypeTrait.VARIABLE_LENGTH);
+        primaryTypeTraits[2] = new TypeTrait(4);
+        primaryTypeTraits[3] = new TypeTrait(ITypeTrait.VARIABLE_LENGTH);
+        
+        // create factories and providers for primary B-Tree
+        TypeAwareTupleWriterFactory primaryTupleWriterFactory = new TypeAwareTupleWriterFactory(primaryTypeTraits);
+        IBTreeInteriorFrameFactory primaryInteriorFrameFactory = new NSMInteriorFrameFactory(primaryTupleWriterFactory);
+        IBTreeLeafFrameFactory primaryLeafFrameFactory = new NSMLeafFrameFactory(primaryTupleWriterFactory);
+        
         // use a disk-order scan to read primary index    	
-        IFileSplitProvider primarySplitProvider = JobHelper.createFileSplitProvider(splitNCs, options.primaryBtreeName);
-        BTreeDiskOrderScanOperatorDescriptor btreeScanOp = new BTreeDiskOrderScanOperatorDescriptor(spec, recDesc, bufferCacheProvider, btreeRegistryProvider, primarySplitProvider, fileMappingProviderProvider, interiorFrameFactory, leafFrameFactory, recDesc.getFields().length);		
+        IFileSplitProvider primarySplitProvider = JobHelper.createFileSplitProvider(splitNCs, options.primaryBTreeName);
+        BTreeDiskOrderScanOperatorDescriptor btreeScanOp = new BTreeDiskOrderScanOperatorDescriptor(spec, recDesc, bufferCacheProvider, btreeRegistryProvider, primarySplitProvider, fileMappingProviderProvider, primaryInteriorFrameFactory, primaryLeafFrameFactory, primaryTypeTraits);		
 		PartitionConstraint scanPartitionConstraint = JobHelper.createPartitionConstraint(splitNCs);
 		btreeScanOp.setPartitionConstraint(scanPartitionConstraint);
-		
+		        
         // sort the tuples as preparation for bulk load into secondary index
         // fields to sort on
         int[] sortFields = { 1, 0 };
@@ -129,13 +141,22 @@
         sorter.setPartitionConstraint(sorterConstraint);
         
         // tuples to be put into B-Tree shall have 2 fields
-        int fieldCount = 2;
+        int secondaryFieldCount = 2;       
+        ITypeTrait[] secondaryTypeTraits = new ITypeTrait[secondaryFieldCount];
+        secondaryTypeTraits[0] = new TypeTrait(ITypeTrait.VARIABLE_LENGTH);
+        secondaryTypeTraits[1] = new TypeTrait(4); 
+        
+        // create factories and providers for secondary B-Tree
+        TypeAwareTupleWriterFactory secondaryTupleWriterFactory = new TypeAwareTupleWriterFactory(secondaryTypeTraits);
+        IBTreeInteriorFrameFactory secondaryInteriorFrameFactory = new NSMInteriorFrameFactory(secondaryTupleWriterFactory);
+        IBTreeLeafFrameFactory secondaryLeafFrameFactory = new NSMLeafFrameFactory(secondaryTupleWriterFactory);
+        
         // the B-Tree expects its keyfields to be at the front of its input tuple 
         int[] fieldPermutation = { 1, 0 };
-        IFileSplitProvider btreeSplitProvider = JobHelper.createFileSplitProvider(splitNCs, options.btreeName);
+        IFileSplitProvider btreeSplitProvider = JobHelper.createFileSplitProvider(splitNCs, options.secondaryBTreeName);
         BTreeBulkLoadOperatorDescriptor btreeBulkLoad = new BTreeBulkLoadOperatorDescriptor(spec, 
-        		bufferCacheProvider, btreeRegistryProvider, btreeSplitProvider, fileMappingProviderProvider, interiorFrameFactory,
-                leafFrameFactory, fieldCount, comparatorFactories, fieldPermutation, 0.7f);
+        		bufferCacheProvider, btreeRegistryProvider, btreeSplitProvider, fileMappingProviderProvider, secondaryInteriorFrameFactory,
+                secondaryLeafFrameFactory, secondaryTypeTraits, comparatorFactories, fieldPermutation, 0.7f);
         PartitionConstraint bulkLoadConstraint = JobHelper.createPartitionConstraint(splitNCs);
         btreeBulkLoad.setPartitionConstraint(bulkLoadConstraint);        
         
diff --git a/hyracks/hyracks-examples/btree-example/btreeclient/src/main/java/edu/uci/ics/hyracks/examples/btree/client/SecondaryIndexSearchExample.java b/hyracks/hyracks-examples/btree-example/btreeclient/src/main/java/edu/uci/ics/hyracks/examples/btree/client/SecondaryIndexSearchExample.java
index e69641a..3de3a37 100644
--- a/hyracks/hyracks-examples/btree-example/btreeclient/src/main/java/edu/uci/ics/hyracks/examples/btree/client/SecondaryIndexSearchExample.java
+++ b/hyracks/hyracks-examples/btree-example/btreeclient/src/main/java/edu/uci/ics/hyracks/examples/btree/client/SecondaryIndexSearchExample.java
@@ -25,7 +25,9 @@
 import edu.uci.ics.hyracks.api.constraints.PartitionConstraint;
 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.ITypeTrait;
 import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
+import edu.uci.ics.hyracks.api.dataflow.value.TypeTrait;
 import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
 import edu.uci.ics.hyracks.api.job.JobSpecification;
 import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
@@ -47,6 +49,7 @@
 import edu.uci.ics.hyracks.storage.am.btree.dataflow.IFileMappingProviderProvider;
 import edu.uci.ics.hyracks.storage.am.btree.frames.NSMInteriorFrameFactory;
 import edu.uci.ics.hyracks.storage.am.btree.frames.NSMLeafFrameFactory;
+import edu.uci.ics.hyracks.storage.am.btree.tuples.TypeAwareTupleWriterFactory;
 
 // This example will perform range search on the secondary index
 // and then retrieve the corresponding source records from the primary index
@@ -95,12 +98,9 @@
 
     	String[] splitNCs = options.ncs.split(",");
     	
-        // create factories and providers for B-Tree
-        IBTreeInteriorFrameFactory interiorFrameFactory = new NSMInteriorFrameFactory();
-        IBTreeLeafFrameFactory leafFrameFactory = new NSMLeafFrameFactory();        
-        IBufferCacheProvider bufferCacheProvider = BufferCacheProvider.INSTANCE;
+    	IBufferCacheProvider bufferCacheProvider = BufferCacheProvider.INSTANCE;
         IBTreeRegistryProvider btreeRegistryProvider = BTreeRegistryProvider.INSTANCE;
-        IFileMappingProviderProvider fileMappingProviderProvider = FileMappingProviderProvider.INSTANCE;
+        IFileMappingProviderProvider fileMappingProviderProvider = FileMappingProviderProvider.INSTANCE;    	              
     	
     	// schema of tuples coming out of secondary index
         RecordDescriptor secondaryRecDesc = new RecordDescriptor(new ISerializerDeserializer[] {                
@@ -108,6 +108,16 @@
         		IntegerSerializerDeserializer.INSTANCE                
                 });
         
+        int secondaryFieldCount = 2;       
+        ITypeTrait[] secondaryTypeTraits = new ITypeTrait[secondaryFieldCount];
+        secondaryTypeTraits[0] = new TypeTrait(ITypeTrait.VARIABLE_LENGTH);
+        secondaryTypeTraits[1] = new TypeTrait(4); 
+        
+        // create factories and providers for secondary B-Tree
+        TypeAwareTupleWriterFactory secondaryTupleWriterFactory = new TypeAwareTupleWriterFactory(secondaryTypeTraits);
+        IBTreeInteriorFrameFactory secondaryInteriorFrameFactory = new NSMInteriorFrameFactory(secondaryTupleWriterFactory);
+        IBTreeLeafFrameFactory secondaryLeafFrameFactory = new NSMLeafFrameFactory(secondaryTupleWriterFactory);
+        
         // schema of tuples coming out of primary index
         RecordDescriptor primaryRecDesc = new RecordDescriptor(new ISerializerDeserializer[] {                
         		IntegerSerializerDeserializer.INSTANCE,
@@ -116,6 +126,18 @@
                 UTF8StringSerializerDeserializer.INSTANCE,
                 });
         
+        int primaryFieldCount = 4;
+        ITypeTrait[] primaryTypeTraits = new ITypeTrait[primaryFieldCount];
+        primaryTypeTraits[0] = new TypeTrait(4);
+        primaryTypeTraits[1] = new TypeTrait(ITypeTrait.VARIABLE_LENGTH);
+        primaryTypeTraits[2] = new TypeTrait(4);
+        primaryTypeTraits[3] = new TypeTrait(ITypeTrait.VARIABLE_LENGTH);
+        
+        // create factories and providers for secondary B-Tree
+        TypeAwareTupleWriterFactory primaryTupleWriterFactory = new TypeAwareTupleWriterFactory(primaryTypeTraits);
+        IBTreeInteriorFrameFactory primaryInteriorFrameFactory = new NSMInteriorFrameFactory(primaryTupleWriterFactory);
+        IBTreeLeafFrameFactory primaryLeafFrameFactory = new NSMLeafFrameFactory(primaryTupleWriterFactory);
+        
         // comparators for btree, note that we only need a comparator for the non-unique key
         // i.e. we will have a range condition on the first field only (implying [-infinity, +infinity] for the second field)
         IBinaryComparatorFactory[] comparatorFactories = new IBinaryComparatorFactory[1];
@@ -142,7 +164,7 @@
         int[] secondaryHighKeyFields = { 1 }; // high key is in field 1 of tuples going into secondary index search op
 		
         IFileSplitProvider secondarySplitProvider = JobHelper.createFileSplitProvider(splitNCs, options.secondaryBTreeName);
-        BTreeSearchOperatorDescriptor secondarySearchOp = new BTreeSearchOperatorDescriptor(spec, secondaryRecDesc, bufferCacheProvider, btreeRegistryProvider, secondarySplitProvider, fileMappingProviderProvider, interiorFrameFactory, leafFrameFactory, secondaryRecDesc.getFields().length, comparatorFactories, true, secondaryLowKeyFields, secondaryHighKeyFields);
+        BTreeSearchOperatorDescriptor secondarySearchOp = new BTreeSearchOperatorDescriptor(spec, secondaryRecDesc, bufferCacheProvider, btreeRegistryProvider, secondarySplitProvider, fileMappingProviderProvider, secondaryInteriorFrameFactory, secondaryLeafFrameFactory, secondaryTypeTraits, comparatorFactories, true, secondaryLowKeyFields, secondaryHighKeyFields);
         PartitionConstraint secondarySearchConstraint = JobHelper.createPartitionConstraint(splitNCs);
         secondarySearchOp.setPartitionConstraint(secondarySearchConstraint);
         
@@ -152,7 +174,7 @@
         int[] primaryHighKeyFields = { 1 }; // high key is in field 1 of tuples going into primary index search op
 		
         IFileSplitProvider primarySplitProvider = JobHelper.createFileSplitProvider(splitNCs, options.primaryBTreeName);
-        BTreeSearchOperatorDescriptor primarySearchOp = new BTreeSearchOperatorDescriptor(spec, primaryRecDesc, bufferCacheProvider, btreeRegistryProvider, primarySplitProvider, fileMappingProviderProvider, interiorFrameFactory, leafFrameFactory, primaryRecDesc.getFields().length, comparatorFactories, true, primaryLowKeyFields, primaryHighKeyFields);
+        BTreeSearchOperatorDescriptor primarySearchOp = new BTreeSearchOperatorDescriptor(spec, primaryRecDesc, bufferCacheProvider, btreeRegistryProvider, primarySplitProvider, fileMappingProviderProvider, primaryInteriorFrameFactory, primaryLeafFrameFactory, primaryTypeTraits, comparatorFactories, true, primaryLowKeyFields, primaryHighKeyFields);
         PartitionConstraint primarySearchConstraint = JobHelper.createPartitionConstraint(splitNCs);
         primarySearchOp.setPartitionConstraint(primarySearchConstraint);
         
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 7ec3883..37027f7 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
@@ -28,7 +28,9 @@
 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.ITypeTrait;
 import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
+import edu.uci.ics.hyracks.api.dataflow.value.TypeTrait;
 import edu.uci.ics.hyracks.api.job.JobSpecification;
 import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
 import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
@@ -67,6 +69,7 @@
 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.tuples.TypeAwareTupleWriterFactory;
 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;
@@ -107,26 +110,33 @@
                 new AbsoluteLocationConstraint(NC1_ID) });
         sorter.setPartitionConstraint(sortersPartitionConstraint);
         
-        IBTreeInteriorFrameFactory interiorFrameFactory = new NSMInteriorFrameFactory();
-		IBTreeLeafFrameFactory leafFrameFactory = new NSMLeafFrameFactory();
-		
-		IBufferCacheProvider bufferCacheProvider = new BufferCacheProvider();		
-		IBTreeRegistryProvider btreeRegistryProvider = new BTreeRegistryProvider();
-		IFileMappingProviderProvider fileMappingProviderProvider = new FileMappingProviderProvider();
-				
+        // declare fields
 		int fieldCount = 3;
+		ITypeTrait[] typeTraits = new ITypeTrait[fieldCount];
+        typeTraits[0] = new TypeTrait(ITypeTrait.VARIABLE_LENGTH);
+        typeTraits[1] = new TypeTrait(ITypeTrait.VARIABLE_LENGTH);
+        typeTraits[2] = new TypeTrait(ITypeTrait.VARIABLE_LENGTH);
+				
+        // declare keys
 		int keyFieldCount = 1;
 		IBinaryComparatorFactory[] comparatorFactories = new IBinaryComparatorFactory[keyFieldCount];
 		comparatorFactories[0] = UTF8StringBinaryComparatorFactory.INSTANCE;		
-				
-		int[] fieldPermutation = { 0, 4, 5 };
-        
+		
+		TypeAwareTupleWriterFactory tupleWriterFactory = new TypeAwareTupleWriterFactory(typeTraits);
+		//SimpleTupleWriterFactory tupleWriterFactory = new SimpleTupleWriterFactory();
+		IBTreeInteriorFrameFactory interiorFrameFactory = new NSMInteriorFrameFactory(tupleWriterFactory);
+		IBTreeLeafFrameFactory leafFrameFactory = new NSMLeafFrameFactory(tupleWriterFactory);		
+		IBufferCacheProvider bufferCacheProvider = new BufferCacheProvider();		
+		IBTreeRegistryProvider btreeRegistryProvider = new BTreeRegistryProvider();
+		IFileMappingProviderProvider fileMappingProviderProvider = new FileMappingProviderProvider();
+		
+		int[] fieldPermutation = { 0, 4, 5};
 		String btreeName = "btree.bin";
 		String nc1FileName = System.getProperty("java.io.tmpdir") + "/nc1/" + btreeName;
 		IFileSplitProvider btreeSplitProvider = new ConstantFileSplitProvider(
 				new FileSplit[] { new FileSplit(NC1_ID, new File(nc1FileName)) } );
 		
-		BTreeBulkLoadOperatorDescriptor btreeBulkLoad = new BTreeBulkLoadOperatorDescriptor(spec, bufferCacheProvider, btreeRegistryProvider, btreeSplitProvider, fileMappingProviderProvider, interiorFrameFactory, leafFrameFactory, fieldCount, comparatorFactories, fieldPermutation, 0.7f);		
+		BTreeBulkLoadOperatorDescriptor btreeBulkLoad = new BTreeBulkLoadOperatorDescriptor(spec, bufferCacheProvider, btreeRegistryProvider, btreeSplitProvider, fileMappingProviderProvider, interiorFrameFactory, leafFrameFactory, typeTraits, comparatorFactories, fieldPermutation, 0.7f);		
 		PartitionConstraint btreePartitionConstraintA = new ExplicitPartitionConstraint(new LocationConstraint[] { new AbsoluteLocationConstraint(NC1_ID) });
 		btreeBulkLoad.setPartitionConstraint(btreePartitionConstraintA);
 				
@@ -143,10 +153,10 @@
     		comparators[i] = comparatorFactories[i].createBinaryComparator();
     	}
     	
-        MultiComparator cmp = new MultiComparator(fieldCount, comparators);
+        MultiComparator cmp = new MultiComparator(typeTraits, comparators);
         
         // try an ordered scan on the bulk-loaded btree
-        int btreeFileId = 0; // TODO: this relies on the way FileMappingProvider assignds ids (in sequence starting from 0)
+        int btreeFileId = 0; // TODO: this relies on the way FileMappingProvider assigns ids (in sequence starting from 0)
         BTree btree = btreeRegistryProvider.getBTreeRegistry().get(btreeFileId);
         IBTreeCursor scanCursor = new RangeSearchCursor(leafFrameFactory.getFrame());
         RangePredicate nullPred = new RangePredicate(true, null, null, null);
@@ -162,7 +172,7 @@
         	e.printStackTrace();
         } finally {
         	scanCursor.close();
-        }                
+        }             
 	}	
 		
 	@Test
@@ -171,21 +181,30 @@
 		System.setProperty("NodeControllerDataPath", System.getProperty("java.io.tmpdir") + "/");
 		
 		JobSpecification spec = new JobSpecification();
-				
-		IBTreeInteriorFrameFactory interiorFrameFactory = new NSMInteriorFrameFactory();
-		IBTreeLeafFrameFactory leafFrameFactory = new NSMLeafFrameFactory();        
-				
+						
+		// declare fields
 		int fieldCount = 3;
+		ITypeTrait[] typeTraits = new ITypeTrait[fieldCount];
+        typeTraits[0] = new TypeTrait(ITypeTrait.VARIABLE_LENGTH);
+        typeTraits[1] = new TypeTrait(ITypeTrait.VARIABLE_LENGTH);
+        typeTraits[2] = new TypeTrait(ITypeTrait.VARIABLE_LENGTH);
+		
+        // declare keys
 		int keyFieldCount = 1;
 		IBinaryComparatorFactory[] comparatorFactories = new IBinaryComparatorFactory[keyFieldCount];
 		comparatorFactories[0] = UTF8StringBinaryComparatorFactory.INSTANCE;		
 		
+		TypeAwareTupleWriterFactory tupleWriterFactory = new TypeAwareTupleWriterFactory(typeTraits);
+		//SimpleTupleWriterFactory tupleWriterFactory = new SimpleTupleWriterFactory();
+		IBTreeInteriorFrameFactory interiorFrameFactory = new NSMInteriorFrameFactory(tupleWriterFactory);
+		IBTreeLeafFrameFactory leafFrameFactory = new NSMLeafFrameFactory(tupleWriterFactory);   				
+		
 		// 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(fieldCount, comparators);
+        MultiComparator cmp = new MultiComparator(typeTraits, comparators);
 		        
         // build tuple containing low and high search key
 		ArrayTupleBuilder tb = new ArrayTupleBuilder(cmp.getKeyFieldCount()*2); // high key and low key
@@ -216,7 +235,7 @@
 		IFileSplitProvider btreeSplitProvider = new ConstantFileSplitProvider(
 				new FileSplit[] { new FileSplit(NC1_ID, new File(nc1FileName)) } );
 		
-		BTreeSearchOperatorDescriptor btreeSearchOp = new BTreeSearchOperatorDescriptor(spec, recDesc, bufferCacheProvider, btreeRegistryProvider, btreeSplitProvider, fileMappingProviderProvider, interiorFrameFactory, leafFrameFactory, fieldCount, comparatorFactories, true, new int[]{0}, new int[]{1});
+		BTreeSearchOperatorDescriptor btreeSearchOp = new BTreeSearchOperatorDescriptor(spec, recDesc, bufferCacheProvider, btreeRegistryProvider, btreeSplitProvider, fileMappingProviderProvider, interiorFrameFactory, leafFrameFactory, typeTraits, comparatorFactories, true, new int[]{0}, new int[]{1});
 		//BTreeDiskOrderScanOperatorDescriptor btreeSearchOp = new BTreeDiskOrderScanOperatorDescriptor(spec, splitProvider, recDesc, bufferCacheProvider, btreeRegistryProvider, 0, "btreetest.bin", interiorFrameFactory, leafFrameFactory, cmp);
 		
 		PartitionConstraint btreePartitionConstraint = new ExplicitPartitionConstraint(new LocationConstraint[] { new AbsoluteLocationConstraint(NC1_ID) });
@@ -232,7 +251,7 @@
         spec.addRoot(printer);
         runTest(spec);
     }
-	
+		
 	@Test
 	public void insertTest() throws Exception {
 		// relies on the fact that NCs are run from same process
@@ -259,47 +278,65 @@
         PartitionConstraint ordersPartitionConstraint = new ExplicitPartitionConstraint(new LocationConstraint[] {
                 new AbsoluteLocationConstraint(NC1_ID) });
         ordScanner.setPartitionConstraint(ordersPartitionConstraint);
+                
+		// we will create a primary index and 2 secondary indexes
+		// first create comparators for primary index		
+		int primaryFieldCount = 6;
+		ITypeTrait[] primaryTypeTraits = new ITypeTrait[primaryFieldCount];
+        primaryTypeTraits[0] = new TypeTrait(ITypeTrait.VARIABLE_LENGTH);
+        primaryTypeTraits[1] = new TypeTrait(ITypeTrait.VARIABLE_LENGTH);
+        primaryTypeTraits[2] = new TypeTrait(ITypeTrait.VARIABLE_LENGTH);
+        primaryTypeTraits[3] = new TypeTrait(ITypeTrait.VARIABLE_LENGTH);
+        primaryTypeTraits[4] = new TypeTrait(ITypeTrait.VARIABLE_LENGTH);
+        primaryTypeTraits[5] = new TypeTrait(ITypeTrait.VARIABLE_LENGTH);
         
-        IBTreeInteriorFrameFactory interiorFrameFactory = new NSMInteriorFrameFactory();
-		IBTreeLeafFrameFactory leafFrameFactory = new NSMLeafFrameFactory();
+		int primaryKeyFieldCount = 1;
+		IBinaryComparatorFactory[] primaryComparatorFactories = new IBinaryComparatorFactory[primaryKeyFieldCount];
+		primaryComparatorFactories[0] = UTF8StringBinaryComparatorFactory.INSTANCE;		
 		
+		TypeAwareTupleWriterFactory primaryTupleWriterFactory = new TypeAwareTupleWriterFactory(primaryTypeTraits);
+		//SimpleTupleWriterFactory primaryTupleWriterFactory = new SimpleTupleWriterFactory();
+		IBTreeInteriorFrameFactory primaryInteriorFrameFactory = new NSMInteriorFrameFactory(primaryTupleWriterFactory);
+		IBTreeLeafFrameFactory primaryLeafFrameFactory = new NSMLeafFrameFactory(primaryTupleWriterFactory);		
+				
 		IBufferCacheProvider bufferCacheProvider = new BufferCacheProvider();		
 		IBTreeRegistryProvider btreeRegistryProvider = new BTreeRegistryProvider();		
 		
 		// trick to clear pages of old fileids
 		BufferCacheProvider tmp = (BufferCacheProvider)bufferCacheProvider;
 		tmp.reset();
-		
-		// we will create a primary index and 2 secondary indexes
-		// 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       
     	IBinaryComparator[] primaryComparators = new IBinaryComparator[primaryComparatorFactories.length];
     	for(int i = 0; i < primaryComparatorFactories.length; i++) {
     		primaryComparators[i] = primaryComparatorFactories[i].createBinaryComparator();
     	}
     	
-        MultiComparator primaryCmp = new MultiComparator(primaryFieldCount, primaryComparators);
-        
-        
+        MultiComparator primaryCmp = new MultiComparator(primaryTypeTraits, primaryComparators);
+                
         // now create comparators for secondary indexes		
 		int secondaryFieldCount = 2;
+		ITypeTrait[] secondaryTypeTraits = new ITypeTrait[secondaryFieldCount];
+        secondaryTypeTraits[0] = new TypeTrait(ITypeTrait.VARIABLE_LENGTH);
+        secondaryTypeTraits[1] = new TypeTrait(ITypeTrait.VARIABLE_LENGTH);        
+		
 		int secondaryKeyFieldCount = 2;
 		IBinaryComparatorFactory[] secondaryComparatorFactories = new IBinaryComparatorFactory[secondaryKeyFieldCount];
 		secondaryComparatorFactories[0] = UTF8StringBinaryComparatorFactory.INSTANCE;
 		secondaryComparatorFactories[1] = UTF8StringBinaryComparatorFactory.INSTANCE;		
 		
+		TypeAwareTupleWriterFactory secondaryTupleWriterFactory = new TypeAwareTupleWriterFactory(secondaryTypeTraits);
+		//SimpleTupleWriterFactory secondaryTupleWriterFactory = new SimpleTupleWriterFactory();
+		IBTreeInteriorFrameFactory secondaryInteriorFrameFactory = new NSMInteriorFrameFactory(secondaryTupleWriterFactory);
+		IBTreeLeafFrameFactory secondaryLeafFrameFactory = new NSMLeafFrameFactory(secondaryTupleWriterFactory);		
+		
 		// 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(secondaryFieldCount, secondaryComparators);
+        MultiComparator secondaryCmp = new MultiComparator(secondaryTypeTraits, secondaryComparators);
         
         // we create and register 3 btrees for in an insert pipeline being fed from a filescan op        
         IBufferCache bufferCache = bufferCacheProvider.getBufferCache();
@@ -313,8 +350,8 @@
         RandomAccessFile rafA = new RandomAccessFile(fA, "rw");                
         FileInfo fiA = new FileInfo(fileIdA, rafA);
         fileManager.registerFile(fiA);
-        BTree btreeA = new BTree(bufferCache, interiorFrameFactory, leafFrameFactory, primaryCmp);		
-		btreeA.create(fileIdA, leafFrameFactory.getFrame(), new MetaDataFrame());
+        BTree btreeA = new BTree(bufferCache, primaryInteriorFrameFactory, primaryLeafFrameFactory, primaryCmp);		
+		btreeA.create(fileIdA, primaryLeafFrameFactory.getFrame(), new MetaDataFrame());
         btreeA.open(fileIdA);
         btreeRegistry.register(fileIdA, btreeA);
         
@@ -324,8 +361,8 @@
         RandomAccessFile rafB = new RandomAccessFile(fB, "rw");   
         FileInfo fiB = new FileInfo(fileIdB, rafB);
         fileManager.registerFile(fiB);
-        BTree btreeB = new BTree(bufferCache, interiorFrameFactory, leafFrameFactory, secondaryCmp);		
-		btreeB.create(fileIdB, leafFrameFactory.getFrame(), new MetaDataFrame());
+        BTree btreeB = new BTree(bufferCache, secondaryInteriorFrameFactory, secondaryLeafFrameFactory, secondaryCmp);		
+		btreeB.create(fileIdB, secondaryLeafFrameFactory.getFrame(), new MetaDataFrame());
         btreeB.open(fileIdB);
         btreeRegistry.register(fileIdB, btreeB);
         
@@ -335,8 +372,8 @@
         RandomAccessFile rafC = new RandomAccessFile(fC, "rw");                
         FileInfo fiC = new FileInfo(fileIdC, rafC);
         fileManager.registerFile(fiC);
-        BTree btreeC = new BTree(bufferCache, interiorFrameFactory, leafFrameFactory, secondaryCmp);	
-		btreeC.create(fileIdC, leafFrameFactory.getFrame(), new MetaDataFrame());
+        BTree btreeC = new BTree(bufferCache, secondaryInteriorFrameFactory, secondaryLeafFrameFactory, secondaryCmp);	
+		btreeC.create(fileIdC, secondaryLeafFrameFactory.getFrame(), new MetaDataFrame());
         btreeC.open(fileIdC);
         btreeRegistry.register(fileIdC, btreeC);
                         
@@ -346,7 +383,7 @@
 		IFileSplitProvider btreeSplitProviderA = new ConstantFileSplitProvider(
 				new FileSplit[] { new FileSplit(NC1_ID, new File("/tmp/btreetestA.ix")) } );        
         int[] fieldPermutationA = { 0,1,2,3,4,5 };                       
-        BTreeInsertUpdateDeleteOperatorDescriptor insertOpA = new BTreeInsertUpdateDeleteOperatorDescriptor(spec, ordersDesc, bufferCacheProvider, btreeRegistryProvider, btreeSplitProviderA, fileMappingProviderProvider, interiorFrameFactory, leafFrameFactory, primaryFieldCount, primaryComparatorFactories, fieldPermutationA, BTreeOp.BTO_INSERT);
+        BTreeInsertUpdateDeleteOperatorDescriptor insertOpA = new BTreeInsertUpdateDeleteOperatorDescriptor(spec, ordersDesc, bufferCacheProvider, btreeRegistryProvider, btreeSplitProviderA, fileMappingProviderProvider, primaryInteriorFrameFactory, primaryLeafFrameFactory, primaryTypeTraits, primaryComparatorFactories, fieldPermutationA, BTreeOp.BTO_INSERT);
         PartitionConstraint insertPartitionConstraintA = new ExplicitPartitionConstraint(new LocationConstraint[] { new AbsoluteLocationConstraint(NC1_ID) });
         insertOpA.setPartitionConstraint(insertPartitionConstraintA);
         
@@ -354,7 +391,7 @@
         IFileSplitProvider btreeSplitProviderB = new ConstantFileSplitProvider(
 				new FileSplit[] { new FileSplit(NC1_ID, new File("/tmp/btreetestB.ix")) } );   
         int[] fieldPermutationB = { 3, 0 };                    
-        BTreeInsertUpdateDeleteOperatorDescriptor insertOpB = new BTreeInsertUpdateDeleteOperatorDescriptor(spec, ordersDesc, bufferCacheProvider, btreeRegistryProvider, btreeSplitProviderB, fileMappingProviderProvider, interiorFrameFactory, leafFrameFactory, secondaryFieldCount, secondaryComparatorFactories, fieldPermutationB, BTreeOp.BTO_INSERT);
+        BTreeInsertUpdateDeleteOperatorDescriptor insertOpB = new BTreeInsertUpdateDeleteOperatorDescriptor(spec, ordersDesc, bufferCacheProvider, btreeRegistryProvider, btreeSplitProviderB, fileMappingProviderProvider, secondaryInteriorFrameFactory, secondaryLeafFrameFactory, secondaryTypeTraits, secondaryComparatorFactories, fieldPermutationB, BTreeOp.BTO_INSERT);
         PartitionConstraint insertPartitionConstraintB = new ExplicitPartitionConstraint(new LocationConstraint[] { new AbsoluteLocationConstraint(NC1_ID) });
         insertOpB.setPartitionConstraint(insertPartitionConstraintB);
 		
@@ -362,7 +399,7 @@
         IFileSplitProvider btreeSplitProviderC = new ConstantFileSplitProvider(
 				new FileSplit[] { new FileSplit(NC1_ID, new File("/tmp/btreetestC.ix")) } );   
         int[] fieldPermutationC = { 4, 0 };                       
-        BTreeInsertUpdateDeleteOperatorDescriptor insertOpC = new BTreeInsertUpdateDeleteOperatorDescriptor(spec, ordersDesc, bufferCacheProvider, btreeRegistryProvider, btreeSplitProviderC, fileMappingProviderProvider, interiorFrameFactory, leafFrameFactory, secondaryFieldCount, secondaryComparatorFactories, fieldPermutationC, BTreeOp.BTO_INSERT);
+        BTreeInsertUpdateDeleteOperatorDescriptor insertOpC = new BTreeInsertUpdateDeleteOperatorDescriptor(spec, ordersDesc, bufferCacheProvider, btreeRegistryProvider, btreeSplitProviderC, fileMappingProviderProvider, secondaryInteriorFrameFactory, secondaryLeafFrameFactory, secondaryTypeTraits, secondaryComparatorFactories, fieldPermutationC, BTreeOp.BTO_INSERT);
         PartitionConstraint insertPartitionConstraintC = new ExplicitPartitionConstraint(new LocationConstraint[] { new AbsoluteLocationConstraint(NC1_ID) });
         insertOpC.setPartitionConstraint(insertPartitionConstraintC);
                 
@@ -383,9 +420,9 @@
         
         // scan primary index         
         System.out.println("PRINTING PRIMARY INDEX");
-        IBTreeCursor scanCursorA = new RangeSearchCursor(leafFrameFactory.getFrame());
+        IBTreeCursor scanCursorA = new RangeSearchCursor(primaryLeafFrameFactory.getFrame());
         RangePredicate nullPredA = new RangePredicate(true, null, null, null);
-        btreeA.search(scanCursorA, nullPredA, leafFrameFactory.getFrame(), interiorFrameFactory.getFrame());
+        btreeA.search(scanCursorA, nullPredA, primaryLeafFrameFactory.getFrame(), primaryInteriorFrameFactory.getFrame());
         try {
         	while (scanCursorA.hasNext()) {
         		scanCursorA.next();
@@ -402,9 +439,9 @@
         
         // scan first secondary index
         System.out.println("PRINTING FIRST SECONDARY INDEX");
-        IBTreeCursor scanCursorB = new RangeSearchCursor(leafFrameFactory.getFrame());
+        IBTreeCursor scanCursorB = new RangeSearchCursor(secondaryLeafFrameFactory.getFrame());
         RangePredicate nullPredB = new RangePredicate(true, null, null, null);
-        btreeB.search(scanCursorB, nullPredB, leafFrameFactory.getFrame(), interiorFrameFactory.getFrame());
+        btreeB.search(scanCursorB, nullPredB, secondaryLeafFrameFactory.getFrame(), secondaryInteriorFrameFactory.getFrame());
         try {
         	while (scanCursorB.hasNext()) {
         		scanCursorB.next();
@@ -421,9 +458,9 @@
         
         // scan second secondary index
         System.out.println("PRINTING SECOND SECONDARY INDEX");
-        IBTreeCursor scanCursorC = new RangeSearchCursor(leafFrameFactory.getFrame());
+        IBTreeCursor scanCursorC = new RangeSearchCursor(secondaryLeafFrameFactory.getFrame());
         RangePredicate nullPredC = new RangePredicate(true, null, null, null);
-        btreeC.search(scanCursorC, nullPredC, leafFrameFactory.getFrame(), interiorFrameFactory.getFrame());
+        btreeC.search(scanCursorC, nullPredC, secondaryLeafFrameFactory.getFrame(), secondaryInteriorFrameFactory.getFrame());
         try {
         	while (scanCursorC.hasNext()) {
         		scanCursorC.next();
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 018c42f..0ba6e1d 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
@@ -74,9 +74,7 @@
 	public void insertSorted(ITupleReference tuple, MultiComparator cmp) throws Exception;
 	
 	public int getSlotSize();
-	
-	public IBTreeTupleReference createTupleReference();
-	
+		
 	// TODO: should be removed after new tuple format		
 	public void setPageTupleFieldCount(int fieldCount);
 	
@@ -84,5 +82,6 @@
 	public int getFreeSpaceOff();
 	public void setFreeSpaceOff(int freeSpace);
 	
+	public IBTreeTupleWriter getTupleWriter();
 	
 }
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/ITupleWriter.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IBTreeTupleWriter.java
similarity index 88%
rename from hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/ITupleWriter.java
rename to hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IBTreeTupleWriter.java
index 52c3464..98c29f3 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/ITupleWriter.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IBTreeTupleWriter.java
@@ -4,7 +4,7 @@
 
 import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
 
-public interface ITupleWriter {	
+public interface IBTreeTupleWriter {	
 	public int writeTuple(ITupleReference tuple, ByteBuffer targetBuf, int targetOff);
 	public int bytesRequired(ITupleReference tuple);
 	
@@ -13,5 +13,5 @@
 	
 	// return a tuplereference instance that can read the tuple written by this writer
 	// the main idea is that the format of the written tuple may not be the same as the format written by this writer
-	public ITupleReference createTupleReference();
+	public IBTreeTupleReference createTupleReference();
 }
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IBTreeTupleWriterFactory.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IBTreeTupleWriterFactory.java
new file mode 100644
index 0000000..fb08a79
--- /dev/null
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IBTreeTupleWriterFactory.java
@@ -0,0 +1,7 @@
+package edu.uci.ics.hyracks.storage.am.btree.api;
+
+import java.io.Serializable;
+
+public interface IBTreeTupleWriterFactory extends Serializable {
+	public IBTreeTupleWriter createTupleWriter();
+}
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 d342e6f..953152e 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,13 +21,14 @@
 import java.util.Comparator;
 
 import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparator;
+import edu.uci.ics.hyracks.api.dataflow.value.ITypeTrait;
 import edu.uci.ics.hyracks.storage.am.btree.api.IFrameCompressor;
 import edu.uci.ics.hyracks.storage.am.btree.api.IPrefixSlotManager;
 import edu.uci.ics.hyracks.storage.am.btree.frames.FieldPrefixNSMLeafFrame;
 import edu.uci.ics.hyracks.storage.am.btree.impls.FieldPrefixSlotManager;
 import edu.uci.ics.hyracks.storage.am.btree.impls.FieldPrefixTupleReference;
 import edu.uci.ics.hyracks.storage.am.btree.impls.MultiComparator;
-import edu.uci.ics.hyracks.storage.am.btree.impls.SimpleTupleWriter;
+import edu.uci.ics.hyracks.storage.am.btree.tuples.TypeAwareTupleWriter;
 
 public class FieldPrefixCompressor implements IFrameCompressor {
 	
@@ -37,9 +38,12 @@
 	// minimum number of tuple matching field prefixes to consider compressing them 
 	private int occurrenceThreshold;
 	
-	public FieldPrefixCompressor(float ratioThreshold, int occurrenceThreshold) {
+	private ITypeTrait[] typeTraits;
+	
+	public FieldPrefixCompressor(ITypeTrait[] typeTraits, float ratioThreshold, int occurrenceThreshold) {
+		this.typeTraits = typeTraits;
 		this.ratioThreshold = ratioThreshold;
-		this.occurrenceThreshold = occurrenceThreshold;
+		this.occurrenceThreshold = occurrenceThreshold;		
 	}
 	
 	@Override
@@ -145,10 +149,10 @@
         int prefixTupleIndex = 0;
         uncompressedTupleCount = 0;
         
-        FieldPrefixTupleReference tupleToWrite = new FieldPrefixTupleReference();
-        tupleToWrite.setFieldCount(fieldCount);
+        TypeAwareTupleWriter tupleWriter = new TypeAwareTupleWriter(typeTraits);
         
-        SimpleTupleWriter tupleWriter = new SimpleTupleWriter();
+        FieldPrefixTupleReference tupleToWrite = new FieldPrefixTupleReference(tupleWriter.createTupleReference());
+        tupleToWrite.setFieldCount(fieldCount);                
         
         while(tupleIndex < tupleCount) {           
             if(kpIndex < keyPartitions.size()) {
@@ -163,10 +167,10 @@
             		
             		//System.out.println("PROCESSING KEYPARTITION: " + kpIndex + " RANGE: " + keyPartitions.get(kpIndex).firstRecSlotNum + " " + keyPartitions.get(kpIndex).lastRecSlotNum + " FIELDSTOCOMPRESS: " + numFieldsToCompress);
             		
-            		FieldPrefixTupleReference prevTuple = new FieldPrefixTupleReference();
+            		FieldPrefixTupleReference prevTuple = new FieldPrefixTupleReference(tupleWriter.createTupleReference());
             		prevTuple.setFieldCount(fieldCount);
             		
-            		FieldPrefixTupleReference tuple = new FieldPrefixTupleReference();
+            		FieldPrefixTupleReference tuple = new FieldPrefixTupleReference(tupleWriter.createTupleReference());
             		tuple.setFieldCount(fieldCount);
             		            	
                     for(int i = tupleIndex + 1; i <= keyPartitions.get(kpIndex).lastTupleIndex; i++) {
@@ -319,14 +323,14 @@
         KeyPartition kp = new KeyPartition(maxCmps);        
         keyPartitions.add(kp);
         
-        FieldPrefixTupleReference prevTuple = new FieldPrefixTupleReference();
+        TypeAwareTupleWriter tupleWriter = new TypeAwareTupleWriter(typeTraits);
+        
+        FieldPrefixTupleReference prevTuple = new FieldPrefixTupleReference(tupleWriter.createTupleReference());
 		prevTuple.setFieldCount(fieldCount);
 		
-		FieldPrefixTupleReference tuple = new FieldPrefixTupleReference();
+		FieldPrefixTupleReference tuple = new FieldPrefixTupleReference(tupleWriter.createTupleReference());
 		tuple.setFieldCount(fieldCount);
-		
-		SimpleTupleWriter tupleWriter = new SimpleTupleWriter();
-		
+				
         kp.firstTupleIndex = 0;        
         int tupleCount = frame.getTupleCount();
         for(int i = 1; i < tupleCount; i++) {        	        	
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 083afef..2ceb58a 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
@@ -16,6 +16,7 @@
 package edu.uci.ics.hyracks.storage.am.btree.dataflow;
 
 import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparatorFactory;
+import edu.uci.ics.hyracks.api.dataflow.value.ITypeTrait;
 import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
 import edu.uci.ics.hyracks.api.job.JobSpecification;
 import edu.uci.ics.hyracks.dataflow.std.base.AbstractSingleActivityOperatorDescriptor;
@@ -27,11 +28,10 @@
 	
 	private static final long serialVersionUID = 1L;
 	
-	protected IFileMappingProviderProvider fileMappingProviderProvider;
+	protected final IFileMappingProviderProvider fileMappingProviderProvider;
 	
 	protected final IFileSplitProvider fileSplitProvider;
 	
-	protected final int fieldCount;
 	protected final IBinaryComparatorFactory[] comparatorFactories;	
 	
 	protected final IBTreeInteriorFrameFactory interiorFrameFactory;
@@ -40,7 +40,9 @@
 	protected final IBufferCacheProvider bufferCacheProvider;
 	protected final IBTreeRegistryProvider btreeRegistryProvider;		
 	
-	public AbstractBTreeOperatorDescriptor(JobSpecification spec, int inputArity, int outputArity, RecordDescriptor recDesc, IBufferCacheProvider bufferCacheProvider, IBTreeRegistryProvider btreeRegistryProvider, IFileSplitProvider fileSplitProvider, IFileMappingProviderProvider fileMappingProviderProvider, IBTreeInteriorFrameFactory interiorFactory, IBTreeLeafFrameFactory leafFactory, int fieldCount, IBinaryComparatorFactory[] comparatorFactories) {
+	protected final ITypeTrait[] typeTraits;
+	
+	public AbstractBTreeOperatorDescriptor(JobSpecification spec, int inputArity, int outputArity, RecordDescriptor recDesc, IBufferCacheProvider bufferCacheProvider, IBTreeRegistryProvider btreeRegistryProvider, IFileSplitProvider fileSplitProvider, IFileMappingProviderProvider fileMappingProviderProvider, IBTreeInteriorFrameFactory interiorFactory, IBTreeLeafFrameFactory leafFactory, ITypeTrait[] typeTraits, IBinaryComparatorFactory[] comparatorFactories) {
         super(spec, inputArity, outputArity);
         this.fileSplitProvider = fileSplitProvider;
         this.fileMappingProviderProvider = fileMappingProviderProvider;
@@ -48,7 +50,7 @@
         this.btreeRegistryProvider = btreeRegistryProvider;        
         this.interiorFrameFactory = interiorFactory;
         this.leafFrameFactory = leafFactory;
-        this.fieldCount = fieldCount;
+        this.typeTraits = typeTraits;
         this.comparatorFactories = comparatorFactories;
         if(outputArity > 0) recordDescriptors[0] = recDesc;           
     }
@@ -64,11 +66,11 @@
 	public IBinaryComparatorFactory[] getComparatorFactories() {
 		return comparatorFactories;
 	}
-	
-	public int getFieldCount() {
-		return fieldCount;
-	}
 		
+	public ITypeTrait[] getTypeTraits() {
+		return typeTraits;
+	}
+	
 	public IBTreeInteriorFrameFactory getInteriorFactory() {
 		return interiorFrameFactory;
 	}
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 d96ae15..29a35d6 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
@@ -19,6 +19,7 @@
 import edu.uci.ics.hyracks.api.dataflow.IOperatorNodePushable;
 import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparatorFactory;
 import edu.uci.ics.hyracks.api.dataflow.value.IRecordDescriptorProvider;
+import edu.uci.ics.hyracks.api.dataflow.value.ITypeTrait;
 import edu.uci.ics.hyracks.api.job.IOperatorEnvironment;
 import edu.uci.ics.hyracks.api.job.JobSpecification;
 import edu.uci.ics.hyracks.dataflow.std.file.IFileSplitProvider;
@@ -36,12 +37,12 @@
 			IBufferCacheProvider bufferCacheProvider,
 			IBTreeRegistryProvider btreeRegistryProvider,
 			IFileSplitProvider fileSplitProvider, IFileMappingProviderProvider fileMappingProviderProvider, IBTreeInteriorFrameFactory interiorFactory,
-			IBTreeLeafFrameFactory leafFactory, int fieldCount, 
+			IBTreeLeafFrameFactory leafFactory, ITypeTrait[] typeTraits, 
 			IBinaryComparatorFactory[] comparatorFactories,			
 			int[] fieldPermutation, float fillFactor) {
 		super(spec, 1, 0, null, bufferCacheProvider,
 				btreeRegistryProvider, fileSplitProvider, fileMappingProviderProvider, interiorFactory,
-				leafFactory, fieldCount, comparatorFactories);
+				leafFactory, typeTraits, 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 f0dbf9c..730b5f3 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
@@ -18,6 +18,7 @@
 import edu.uci.ics.hyracks.api.context.IHyracksContext;
 import edu.uci.ics.hyracks.api.dataflow.IOperatorNodePushable;
 import edu.uci.ics.hyracks.api.dataflow.value.IRecordDescriptorProvider;
+import edu.uci.ics.hyracks.api.dataflow.value.ITypeTrait;
 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;
@@ -35,10 +36,10 @@
 			IBTreeRegistryProvider btreeRegistryProvider,
 			IFileSplitProvider fileSplitProvider, IFileMappingProviderProvider fileMappingProviderProvider, IBTreeInteriorFrameFactory interiorFactory,
 			IBTreeLeafFrameFactory leafFactory, 
-			int fieldCount) {
+			ITypeTrait[] typeTraits) {
 		super(spec, 0, 1, recDesc, bufferCacheProvider,
 				btreeRegistryProvider, fileSplitProvider, fileMappingProviderProvider, interiorFactory,
-				leafFactory, fieldCount, null);
+				leafFactory, typeTraits, null);
 	}
 	
 	@Override
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeFileEnlistmentOperatorDescriptor.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeFileEnlistmentOperatorDescriptor.java
index 035a530..fd7c5a6 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeFileEnlistmentOperatorDescriptor.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeFileEnlistmentOperatorDescriptor.java
@@ -4,6 +4,7 @@
 import edu.uci.ics.hyracks.api.dataflow.IOperatorNodePushable;
 import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparatorFactory;
 import edu.uci.ics.hyracks.api.dataflow.value.IRecordDescriptorProvider;
+import edu.uci.ics.hyracks.api.dataflow.value.ITypeTrait;
 import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
 import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
 import edu.uci.ics.hyracks.api.job.IOperatorEnvironment;
@@ -28,11 +29,11 @@
 			IFileSplitProvider fileSplitProvider,
 			IFileMappingProviderProvider fileMappingProviderProvider,
 			IBTreeInteriorFrameFactory interiorFactory,
-			IBTreeLeafFrameFactory leafFactory, int fieldCount,
+			IBTreeLeafFrameFactory leafFactory, ITypeTrait[] typeTraits,
 			IBinaryComparatorFactory[] comparatorFactories) {
 		super(spec, 0, 0, recDesc, bufferCacheProvider,
 				btreeRegistryProvider, fileSplitProvider, fileMappingProviderProvider,
-				interiorFactory, leafFactory, fieldCount, comparatorFactories);		
+				interiorFactory, leafFactory, typeTraits, comparatorFactories);		
 	}
 	
 	@Override
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 8a2e439..7fb4398 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
@@ -19,6 +19,7 @@
 import edu.uci.ics.hyracks.api.dataflow.IOperatorNodePushable;
 import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparatorFactory;
 import edu.uci.ics.hyracks.api.dataflow.value.IRecordDescriptorProvider;
+import edu.uci.ics.hyracks.api.dataflow.value.ITypeTrait;
 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;
@@ -41,12 +42,12 @@
 			IBTreeRegistryProvider btreeRegistryProvider,
 			IFileSplitProvider fileSplitProvider, IFileMappingProviderProvider fileMappingProviderProvider, 
 			IBTreeInteriorFrameFactory interiorFactory,
-			IBTreeLeafFrameFactory leafFactory, int fieldCount, 
+			IBTreeLeafFrameFactory leafFactory, ITypeTrait[] typeTraits, 
 			IBinaryComparatorFactory[] comparatorFactories,			
 			int[] fieldPermutation, BTreeOp op) {
 		super(spec, 1, 1, recDesc, bufferCacheProvider,
 				btreeRegistryProvider, fileSplitProvider, fileMappingProviderProvider, interiorFactory,
-				leafFactory, fieldCount, comparatorFactories);
+				leafFactory, typeTraits, 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 db4595a..0cd86c1 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
@@ -113,7 +113,7 @@
     	} break;
         }
     	
-    	btreeFileId = fileId;  
+    	btreeFileId = fileId;
     	
         if(mode == BTreeMode.CREATE_BTREE || mode == BTreeMode.ENLIST_BTREE) {
         	FileInfo fi = new FileInfo(btreeFileId, raf);
@@ -139,8 +139,8 @@
                     for (int i = 0; i < opDesc.getComparatorFactories().length; i++) {
                         comparators[i] = opDesc.getComparatorFactories()[i].createBinaryComparator();
                     }
-
-                    MultiComparator cmp = new MultiComparator(opDesc.getFieldCount(), comparators);
+                    
+                    MultiComparator cmp = new MultiComparator(opDesc.getTypeTraits(), comparators);
                     
                     btree = new BTree(bufferCache, opDesc.getInteriorFactory(), opDesc.getLeafFactory(), cmp);
                     if (mode == BTreeMode.CREATE_BTREE) {
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 4d84c41..d31766d 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
@@ -19,6 +19,7 @@
 import edu.uci.ics.hyracks.api.dataflow.IOperatorNodePushable;
 import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparatorFactory;
 import edu.uci.ics.hyracks.api.dataflow.value.IRecordDescriptorProvider;
+import edu.uci.ics.hyracks.api.dataflow.value.ITypeTrait;
 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;
@@ -34,8 +35,8 @@
 	private int[] lowKeyFields; // fields in input tuple to be used as low keys
 	private int[] highKeyFields; // fields in input tuple to be used as high keys
 	
-	public BTreeSearchOperatorDescriptor(JobSpecification spec, RecordDescriptor recDesc, IBufferCacheProvider bufferCacheProvider, IBTreeRegistryProvider btreeRegistryProvider, IFileSplitProvider fileSplitProvider, IFileMappingProviderProvider fileMappingProviderProvider, IBTreeInteriorFrameFactory interiorFactory, IBTreeLeafFrameFactory leafFactory, int fieldCount, IBinaryComparatorFactory[] comparatorFactories, boolean isForward, int[] lowKeyFields, int[] highKeyFields) {		
-		super(spec, 1, 1, recDesc, bufferCacheProvider, btreeRegistryProvider, fileSplitProvider, fileMappingProviderProvider, interiorFactory, leafFactory, fieldCount, comparatorFactories);
+	public BTreeSearchOperatorDescriptor(JobSpecification spec, RecordDescriptor recDesc, IBufferCacheProvider bufferCacheProvider, IBTreeRegistryProvider btreeRegistryProvider, IFileSplitProvider fileSplitProvider, IFileMappingProviderProvider fileMappingProviderProvider, IBTreeInteriorFrameFactory interiorFactory, IBTreeLeafFrameFactory leafFactory, ITypeTrait[] typeTraits, IBinaryComparatorFactory[] comparatorFactories, boolean isForward, int[] lowKeyFields, int[] highKeyFields) {		
+		super(spec, 1, 1, recDesc, bufferCacheProvider, btreeRegistryProvider, fileSplitProvider, fileMappingProviderProvider, interiorFactory, leafFactory, typeTraits, comparatorFactories);
 		this.isForward = isForward;
 		this.lowKeyFields = lowKeyFields;
 		this.highKeyFields = highKeyFields;
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 67e0c3b..86e5310 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
@@ -98,7 +98,7 @@
         for (int i = 0; i < numSearchFields; i++) {
         	searchComparators[i] = btree.getMultiComparator().getComparators()[i];
         }
-        searchCmp = new MultiComparator(btree.getMultiComparator().getFieldCount(), searchComparators);
+        searchCmp = new MultiComparator(btree.getMultiComparator().getTypeTraits(), searchComparators);
         
         rangePred = new RangePredicate(isForward, null, null, searchCmp);
                 
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/FileMappingProvider.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/FileMappingProvider.java
index 35eccc9..f4a68bf 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/FileMappingProvider.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/FileMappingProvider.java
@@ -12,7 +12,7 @@
 	private Map<String, Integer> map = new Hashtable<String, Integer>();
 		
 	@Override
-	public Integer mapNameToFileId(String name, boolean create) {		
+	public synchronized Integer mapNameToFileId(String name, boolean create) {		
 		Integer val = map.get(name);			
 		if(create) {
 			if(val == null) {
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 eb8263c..ee77467 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
@@ -23,11 +23,12 @@
 import java.util.Collections;
 
 import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
+import edu.uci.ics.hyracks.api.dataflow.value.ITypeTrait;
 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;
-import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeTupleReference;
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeTupleWriter;
 import edu.uci.ics.hyracks.storage.am.btree.api.IFrameCompressor;
 import edu.uci.ics.hyracks.storage.am.btree.api.IPrefixSlotManager;
 import edu.uci.ics.hyracks.storage.am.btree.api.ISlotManager;
@@ -37,12 +38,14 @@
 import edu.uci.ics.hyracks.storage.am.btree.impls.FieldPrefixSlotManager;
 import edu.uci.ics.hyracks.storage.am.btree.impls.FieldPrefixTupleReference;
 import edu.uci.ics.hyracks.storage.am.btree.impls.MultiComparator;
-import edu.uci.ics.hyracks.storage.am.btree.impls.SimpleTupleWriter;
 import edu.uci.ics.hyracks.storage.am.btree.impls.SlotOffTupleOff;
 import edu.uci.ics.hyracks.storage.am.btree.impls.SpaceStatus;
 import edu.uci.ics.hyracks.storage.am.btree.impls.SplitKey;
+import edu.uci.ics.hyracks.storage.am.btree.tuples.TypeAwareTupleWriter;
 import edu.uci.ics.hyracks.storage.common.buffercache.ICachedPage;
 
+// WARNING: only works when tupleWriter is an instance of TypeAwareTupleWriter
+
 public class FieldPrefixNSMLeafFrame implements IBTreeLeafFrame {
 	
     protected static final int pageLsnOff = 0;                              // 0
@@ -62,14 +65,18 @@
     public IFrameCompressor compressor;
     public IPrefixSlotManager slotManager; // TODO: should be protected, but will trigger some refactoring
     
-    private SimpleTupleWriter tupleWriter = new SimpleTupleWriter();
+    private IBTreeTupleWriter tupleWriter;
     
-    private FieldPrefixTupleReference frameTuple = new FieldPrefixTupleReference();            
-    private FieldPrefixPrefixTupleReference framePrefixTuple = new FieldPrefixPrefixTupleReference();
-          
-    public FieldPrefixNSMLeafFrame() {
-        this.slotManager = new FieldPrefixSlotManager();
-        this.compressor = new FieldPrefixCompressor(0.001f, 2);        
+    private FieldPrefixTupleReference frameTuple;            
+    private FieldPrefixPrefixTupleReference framePrefixTuple;
+    
+    public FieldPrefixNSMLeafFrame(IBTreeTupleWriter tupleWriter) {
+    	this.tupleWriter = tupleWriter;
+    	this.frameTuple = new FieldPrefixTupleReference(tupleWriter.createTupleReference());    	
+    	ITypeTrait[] typeTraits = ((TypeAwareTupleWriter)tupleWriter).getTypeTraits();
+    	this.framePrefixTuple = new FieldPrefixPrefixTupleReference(typeTraits); 
+    	this.slotManager = new FieldPrefixSlotManager();
+        this.compressor = new FieldPrefixCompressor(typeTraits, 0.001f, 2);        
     }
     
     @Override
@@ -135,16 +142,16 @@
         }
         Collections.sort(sortedTupleOffs);
                
-        for(int i = 0; i < sortedTupleOffs.size(); i++) {                    	
+        for(int i = 0; i < sortedTupleOffs.size(); i++) {        	
         	int tupleOff = sortedTupleOffs.get(i).tupleOff;
         	int tupleSlot = buf.getInt(sortedTupleOffs.get(i).slotOff);
             int prefixSlotNum = slotManager.decodeFirstSlotField(tupleSlot);            
-                        
+            
             frameTuple.resetByTupleIndex(this, sortedTupleOffs.get(i).tupleIndex);
             int tupleEndOff = frameTuple.getFieldStart(frameTuple.getFieldCount()-1) + frameTuple.getFieldLength(frameTuple.getFieldCount()-1);
             int tupleLength = tupleEndOff - tupleOff;
             System.arraycopy(buf.array(), tupleOff, buf.array(), freeSpace, tupleLength);
-                                    
+            
             slotManager.setSlot(sortedTupleOffs.get(i).slotOff, slotManager.encodeSlotFields(prefixSlotNum, freeSpace));
             freeSpace += tupleLength;
         }
@@ -585,14 +592,13 @@
 	public int getSlotSize() {
 		return slotManager.getSlotSize();
 	}
-	
-	@Override
-    public IBTreeTupleReference createTupleReference() {
-    	return new FieldPrefixTupleReference();
-    }
 		
 	@Override
 	public void setPageTupleFieldCount(int fieldCount) {
 		frameTuple.setFieldCount(fieldCount);
 	}	
+	
+	public IBTreeTupleWriter getTupleWriter() {
+    	return tupleWriter;
+    }
 }
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/FieldPrefixNSMLeafFrameFactory.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/FieldPrefixNSMLeafFrameFactory.java
index e50a470..21cfbcb 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/FieldPrefixNSMLeafFrameFactory.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/FieldPrefixNSMLeafFrameFactory.java
@@ -17,13 +17,18 @@
 
 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.IBTreeTupleWriterFactory;
 
 public class FieldPrefixNSMLeafFrameFactory implements IBTreeLeafFrameFactory {
 	
 	private static final long serialVersionUID = 1L;
-
+	private IBTreeTupleWriterFactory tupleWriterFactory;
+	
+	public FieldPrefixNSMLeafFrameFactory(IBTreeTupleWriterFactory tupleWriterFactory) {
+		this.tupleWriterFactory = tupleWriterFactory;
+	}
 	@Override
 	public IBTreeLeafFrame getFrame() {		
-		return new FieldPrefixNSMLeafFrame();
+		return new FieldPrefixNSMLeafFrame(tupleWriterFactory.createTupleWriter());
 	}
 }
\ No newline at end of file
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 a378a26..5030c45 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
@@ -25,14 +25,14 @@
 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.dataflow.common.data.marshalling.IntegerSerializerDeserializer;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeFrame;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeTupleReference;
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeTupleWriter;
 import edu.uci.ics.hyracks.storage.am.btree.api.ISlotManager;
 import edu.uci.ics.hyracks.storage.am.btree.impls.BTreeException;
 import edu.uci.ics.hyracks.storage.am.btree.impls.MultiComparator;
 import edu.uci.ics.hyracks.storage.am.btree.impls.OrderedSlotManager;
-import edu.uci.ics.hyracks.storage.am.btree.impls.SimpleTupleReference;
-import edu.uci.ics.hyracks.storage.am.btree.impls.SimpleTupleWriter;
 import edu.uci.ics.hyracks.storage.am.btree.impls.SlotOffTupleOff;
 import edu.uci.ics.hyracks.storage.am.btree.impls.SpaceStatus;
 import edu.uci.ics.hyracks.storage.common.buffercache.ICachedPage;
@@ -49,12 +49,14 @@
 	protected ICachedPage page = null;
 	protected ByteBuffer buf = null;
 	protected ISlotManager slotManager;
+		
+	protected IBTreeTupleWriter tupleWriter;
+	protected IBTreeTupleReference frameTuple;
 	
-	protected SimpleTupleWriter tupleWriter = new SimpleTupleWriter();
-	protected SimpleTupleReference frameTuple = new SimpleTupleReference();
-	
-	public NSMFrame() {
-		this.slotManager = new OrderedSlotManager();
+	public NSMFrame(IBTreeTupleWriter tupleWriter) {
+		this.tupleWriter = tupleWriter;
+		this.frameTuple = tupleWriter.createTupleReference();
+		this.slotManager = new OrderedSlotManager();		
 	}
 	
 	@Override
@@ -188,7 +190,7 @@
 	
 	@Override
 	public SpaceStatus hasSpaceInsert(ITupleReference tuple, MultiComparator cmp) {				
-		int bytesRequired = tupleWriter.bytesRequired(tuple);		
+		int bytesRequired = tupleWriter.bytesRequired(tuple);
 		if(bytesRequired + slotManager.getSlotSize() <= buf.capacity() - buf.getInt(freeSpaceOff) - (buf.getInt(tupleCountOff) * slotManager.getSlotSize()) ) return SpaceStatus.SUFFICIENT_CONTIGUOUS_SPACE;
 		else if(bytesRequired + slotManager.getSlotSize() <= buf.getInt(totalFreeSpaceOff)) return SpaceStatus.SUFFICIENT_SPACE;
 		else return SpaceStatus.INSUFFICIENT_SPACE;
@@ -242,10 +244,10 @@
 	public String printKeys(MultiComparator cmp, ISerializerDeserializer[] fields) throws HyracksDataException {		
 		StringBuilder strBuilder = new StringBuilder();		
 		int tupleCount = buf.getInt(tupleCountOff);
-		frameTuple.setFieldCount(fields.length);
+		frameTuple.setFieldCount(fields.length);	
 		for(int i = 0; i < tupleCount; i++) {						
 			frameTuple.resetByTupleIndex(this, i);												
-			for(int j = 0; j < cmp.getKeyFieldCount(); j++) {
+			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);
@@ -286,14 +288,13 @@
     public int getSlotSize() {
     	return slotManager.getSlotSize();
     }
-    
-    @Override
-    public IBTreeTupleReference createTupleReference() {
-    	return new SimpleTupleReference();
-    }
-    
+        
     @Override
 	public void setPageTupleFieldCount(int fieldCount) {
 		frameTuple.setFieldCount(fieldCount);
 	}
+    
+    public IBTreeTupleWriter getTupleWriter() {
+    	return tupleWriter;
+    }
 }
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 78424e6..8731e10 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
@@ -15,17 +15,24 @@
 
 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.dataflow.common.data.marshalling.IntegerSerializerDeserializer;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeFrame;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeInteriorFrame;
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeTupleReference;
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeTupleWriter;
 import edu.uci.ics.hyracks.storage.am.btree.impls.BTreeException;
 import edu.uci.ics.hyracks.storage.am.btree.impls.MultiComparator;
 import edu.uci.ics.hyracks.storage.am.btree.impls.RangePredicate;
-import edu.uci.ics.hyracks.storage.am.btree.impls.SimpleTupleReference;
 import edu.uci.ics.hyracks.storage.am.btree.impls.SlotOffTupleOff;
 import edu.uci.ics.hyracks.storage.am.btree.impls.SpaceStatus;
 import edu.uci.ics.hyracks.storage.am.btree.impls.SplitKey;
@@ -36,10 +43,13 @@
 	
 	private static final int childPtrSize = 4;
 	
-	private SimpleTupleReference cmpFrameTuple = new SimpleTupleReference();
+	//private SimpleTupleReference cmpFrameTuple = new SimpleTupleReference();
+	private IBTreeTupleReference cmpFrameTuple;
 	
-	public NSMInteriorFrame() {
-		super();
+	public NSMInteriorFrame(IBTreeTupleWriter tupleWriter) {
+		super(tupleWriter);
+		cmpFrameTuple = tupleWriter.createTupleReference();
+		
 	}
 	
 	private int getLeftChildPageOff(ITupleReference tuple, MultiComparator cmp) {		
@@ -61,7 +71,7 @@
 	}
 	
 	@Override
-	public void insert(ITupleReference tuple, MultiComparator cmp) throws Exception {
+	public void insert(ITupleReference tuple, MultiComparator cmp) throws Exception {		
 		frameTuple.setFieldCount(cmp.getKeyFieldCount());
 		int slotOff = slotManager.findSlot(tuple, frameTuple, cmp, false);
 		boolean isDuplicate = true;				
@@ -80,7 +90,7 @@
 			
 			int freeSpace = buf.getInt(freeSpaceOff);			
 			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);
+			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);
@@ -101,7 +111,7 @@
 					System.arraycopy(tuple.getFieldData(0), getLeftChildPageOff(tuple, cmp) + childPtrSize, buf.array(), getLeftChildPageOff(frameTuple, cmp), childPtrSize);
 				}				
 			}			
-		}		
+		}				
 	}
 	
 	@Override
@@ -157,7 +167,7 @@
 		buf.putInt(tupleCountOff, tuplesToLeft);
 				
 		// copy data to be inserted, we need this because creating the splitkey will overwrite the data param (data points to same memory as splitKey.getData())
-		SplitKey savedSplitKey = splitKey.duplicate();		
+		SplitKey savedSplitKey = splitKey.duplicate(tupleWriter.createTupleReference());
 		
 		// set split key to be highest value in left page	
 		int tupleOff = slotManager.getTupleOff(slotManager.getSlotEndOff());
@@ -165,6 +175,7 @@
 		int splitKeySize = tupleWriter.bytesRequired(frameTuple, 0, cmp.getKeyFieldCount());
 		splitKey.initData(splitKeySize);
 		tupleWriter.writeTupleFields(frameTuple, 0, cmp.getKeyFieldCount(), splitKey.getBuffer(), 0);			
+		splitKey.getTuple().resetByOffset(splitKey.getBuffer(), 0);
 		
 		int deleteTupleOff = slotManager.getTupleOff(slotManager.getSlotEndOff());
 		frameTuple.resetByOffset(buf, deleteTupleOff);
@@ -176,7 +187,7 @@
 		compact(cmp);
 		
 		// insert key
-		targetFrame.insert(savedSplitKey.getTuple(), cmp);					
+		targetFrame.insert(savedSplitKey.getTuple(), cmp);
 		
 		return 0;
 	}	
@@ -375,4 +386,23 @@
 	private int getInt(byte[] bytes, int offset) {
 		return ((bytes[offset] & 0xff) << 24) + ((bytes[offset + 1] & 0xff) << 16) + ((bytes[offset + 2] & 0xff) << 8) + ((bytes[offset + 3] & 0xff) << 0);
 	}
+	
+	@Override
+	public String printKeys(MultiComparator cmp, ISerializerDeserializer[] fields) throws HyracksDataException {		
+		StringBuilder strBuilder = new StringBuilder();		
+		int tupleCount = buf.getInt(tupleCountOff);
+		frameTuple.setFieldCount(cmp.getKeyFieldCount());	
+		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(" | ");				
+		}
+		strBuilder.append("\n");
+		return strBuilder.toString();
+	}
 }
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMInteriorFrameFactory.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMInteriorFrameFactory.java
index 81a6e34..8dd7175 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMInteriorFrameFactory.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMInteriorFrameFactory.java
@@ -17,13 +17,19 @@
 
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeInteriorFrame;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeInteriorFrameFactory;
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeTupleWriterFactory;
 
 public class NSMInteriorFrameFactory implements IBTreeInteriorFrameFactory {
 	
-	private static final long serialVersionUID = 1L;
-
+	private static final long serialVersionUID = 1L;	
+	private IBTreeTupleWriterFactory tupleWriterFactory;
+	
+	public NSMInteriorFrameFactory(IBTreeTupleWriterFactory tupleWriterFactory) {
+		this.tupleWriterFactory = tupleWriterFactory;
+	}
+	
 	@Override
 	public IBTreeInteriorFrame getFrame() {		
-		return new NSMInteriorFrame();
+		return new NSMInteriorFrame(tupleWriterFactory.createTupleWriter());
 	}	
 }
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 d82c47f..07e49c3 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
@@ -20,6 +20,8 @@
 import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeFrame;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeLeafFrame;
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeTupleReference;
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeTupleWriter;
 import edu.uci.ics.hyracks.storage.am.btree.impls.BTreeException;
 import edu.uci.ics.hyracks.storage.am.btree.impls.MultiComparator;
 import edu.uci.ics.hyracks.storage.am.btree.impls.SplitKey;
@@ -28,8 +30,8 @@
 	protected static final int prevLeafOff = smFlagOff + 1;
 	protected static final int nextLeafOff = prevLeafOff + 4;
 	
-	public NSMLeafFrame() {
-		super();
+	public NSMLeafFrame(IBTreeTupleWriter tupleWriter) {
+		super(tupleWriter);
 	}
 	
 	@Override
@@ -82,7 +84,7 @@
 			
 			buf.putInt(tupleCountOff, buf.getInt(tupleCountOff) + 1);
 			buf.putInt(freeSpaceOff, buf.getInt(freeSpaceOff) + bytesWritten);
-			buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) - bytesWritten - slotManager.getSlotSize());
+			buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) - bytesWritten - slotManager.getSlotSize());			
 		}	
 	}
 	
@@ -90,7 +92,7 @@
 	public void insertSorted(ITupleReference tuple, MultiComparator cmp) throws Exception {		
 		int freeSpace = buf.getInt(freeSpaceOff);
 		slotManager.insertSlot(-1, freeSpace);		
-		int bytesWritten = tupleWriter.writeTuple(tuple, buf, freeSpace);			
+		int bytesWritten = tupleWriter.writeTuple(tuple, buf, freeSpace);	
 		buf.putInt(tupleCountOff, buf.getInt(tupleCountOff) + 1);
 		buf.putInt(freeSpaceOff, buf.getInt(freeSpaceOff) + bytesWritten);
 		buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) - bytesWritten - slotManager.getSlotSize());
@@ -142,8 +144,8 @@
 			
 		// compact both pages
 		rightFrame.compact(cmp);		
-		compact(cmp);
-		
+		compact(cmp);		
+						
 		// insert last key
 		targetFrame.insert(tuple, cmp);	
 		
@@ -155,6 +157,7 @@
 		splitKey.initData(splitKeySize);				
 		tupleWriter.writeTupleFields(frameTuple, 0, cmp.getKeyFieldCount(), splitKey.getBuffer(), 0);
 		
+		splitKey.getTuple().resetByOffset(splitKey.getBuffer(), 0);				
 		return 0;
 	}
 
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMLeafFrameFactory.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMLeafFrameFactory.java
index 169327b..33316a9 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMLeafFrameFactory.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMLeafFrameFactory.java
@@ -17,13 +17,19 @@
 
 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.IBTreeTupleWriterFactory;
 
 public class NSMLeafFrameFactory implements IBTreeLeafFrameFactory {
 			
 	private static final long serialVersionUID = 1L;
-
+	private IBTreeTupleWriterFactory tupleWriterFactory;
+	
+	public NSMLeafFrameFactory(IBTreeTupleWriterFactory tupleWriterFactory) {
+		this.tupleWriterFactory = tupleWriterFactory;
+	}
+	
 	@Override
 	public IBTreeLeafFrame getFrame() {		
-		return new NSMLeafFrame();
+		return new NSMLeafFrame(tupleWriterFactory.createTupleWriter());
 	}
 }
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 1c346c6..ab7ef70 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
@@ -30,6 +30,7 @@
 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.IBTreeTupleWriter;
 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;
@@ -40,8 +41,7 @@
     private final static int RESTART_OP = Integer.MIN_VALUE;
     private final static int MAX_RESTARTS = 10;
     
-    private final int metaDataPage = 0; // page containing meta data, e.g.,
-                                        // maxPage
+    private final int metaDataPage = 0; // page containing meta data, e.g., maxPage
     private final int rootPage = 1; // the root page never changes
     
     private boolean created = false;        
@@ -335,26 +335,6 @@
             interiorFrame.setPage(node);
             int level = interiorFrame.getLevel();
             
-            
-//            if (GlobalConfig.ASTERIX_LOGGER.isLoggable(Level.FINEST)) {
-//                GlobalConfig.ASTERIX_LOGGER.finest(String.format("%1d ", level));
-//                GlobalConfig.ASTERIX_LOGGER.finest(String.format("%3d ", pageId));
-//                for (int i = 0; i < currentLevel - level; i++)
-//                    GlobalConfig.ASTERIX_LOGGER.finest("    ");
-//                
-//                String keyString;
-//                if(interiorFrame.isLeaf()) {
-//                	leafFrame.setPage(node);
-//                	keyString = leafFrame.printKeys(cmp);
-//                }
-//                else {
-//                	keyString = interiorFrame.printKeys(cmp);
-//                }
-//                
-//                GlobalConfig.ASTERIX_LOGGER.finest(keyString);
-//            }
-
-
             System.out.format("%1d ", level);
             System.out.format("%3d ", pageId);
             for (int i = 0; i < currentLevel - level; i++)
@@ -372,13 +352,6 @@
             System.out.format(keyString);
             if (!interiorFrame.isLeaf()) {
             	ArrayList<Integer> children = ((NSMInteriorFrame) (interiorFrame)).getChildren(cmp);
-            	
-            	// debug
-            	System.out.println("CHILDREN OF PAGE: " + interiorFrame.isLeaf() + " " + pageId);
-            	for (int i = 0; i < children.size(); i++) {
-                    System.out.print(children.get(i) + " ");
-                }
-            	System.out.println();
             	            	
                 for (int i = 0; i < children.size(); i++) {
                     printTree(children.get(i), node, i == children.size() - 1, leafFrame, interiorFrame, fields);
@@ -396,6 +369,7 @@
             
             bufferCache.unpin(node);
             unpins++;
+            e.printStackTrace();
         }
     }
 
@@ -556,7 +530,7 @@
         ctx.interiorFrame = interiorFrame;
         ctx.metaFrame = metaFrame;
         ctx.pred = new RangePredicate(true, tuple, tuple, cmp);
-        ctx.splitKey = new SplitKey();
+        ctx.splitKey = new SplitKey(leafFrame.getTupleWriter().createTupleReference());
         ctx.splitKey.getTuple().setFieldCount(cmp.getKeyFieldCount());
         ctx.smPages = new ArrayList<Integer>();
         ctx.pageLsns = new Stack<Integer>();
@@ -793,7 +767,7 @@
         ctx.interiorFrame = interiorFrame;
         ctx.metaFrame = metaFrame;
         ctx.pred = new RangePredicate(true, tuple, tuple, cmp);
-        ctx.splitKey = new SplitKey();
+        ctx.splitKey = new SplitKey(leafFrame.getTupleWriter().createTupleReference());
         ctx.splitKey.getTuple().setFieldCount(cmp.getKeyFieldCount());
         ctx.smPages = new ArrayList<Integer>();
         ctx.pageLsns = new Stack<Integer>();
@@ -1190,28 +1164,25 @@
     private boolean bulkNewPage = false;
 
     public final class BulkLoadContext {               
-        public int slotSize;
-        public int leafMaxBytes;
-        public int interiorMaxBytes;
-        public SplitKey splitKey = new SplitKey();
-        ArrayList<NodeFrontier> nodeFrontiers = new ArrayList<NodeFrontier>(); // we
-        // maintain
-        // a
-        // frontier
-        // of
-        // nodes
-        // for
-        // each
-        // level
-        IBTreeLeafFrame leafFrame;
-        IBTreeInteriorFrame interiorFrame;
-        IBTreeMetaDataFrame metaFrame;
-
-        SimpleTupleWriter tupleWriter = new SimpleTupleWriter();
+        public final int slotSize;
+        public final int leafMaxBytes;
+        public final int interiorMaxBytes;
+        public final SplitKey splitKey;
+        // we maintain a frontier of nodes for each level
+        private final ArrayList<NodeFrontier> nodeFrontiers = new ArrayList<NodeFrontier>();         
+        private final IBTreeLeafFrame leafFrame;
+        private final IBTreeInteriorFrame interiorFrame;
+        private final IBTreeMetaDataFrame metaFrame;
+        
+        private final IBTreeTupleWriter tupleWriter;
         
         public BulkLoadContext(float fillFactor, IBTreeLeafFrame leafFrame, IBTreeInteriorFrame interiorFrame,
                 IBTreeMetaDataFrame metaFrame) throws Exception {
-            NodeFrontier leafFrontier = new NodeFrontier();
+
+        	splitKey = new SplitKey(leafFrame.getTupleWriter().createTupleReference());
+        	tupleWriter = leafFrame.getTupleWriter();
+        	
+        	NodeFrontier leafFrontier = new NodeFrontier(tupleWriter.createTupleReference());
             leafFrontier.pageId = getFreePage(metaFrame);
             leafFrontier.page = bufferCache.pin(FileInfo.getDiskPageId(fileId, leafFrontier.pageId), bulkNewPage);
             leafFrontier.page.acquireWriteLatch();
@@ -1234,7 +1205,7 @@
         }
 
         private void addLevel() throws Exception {
-            NodeFrontier frontier = new NodeFrontier();
+            NodeFrontier frontier = new NodeFrontier(tupleWriter.createTupleReference());
             frontier.pageId = getFreePage(metaFrame);
             frontier.page = bufferCache.pin(FileInfo.getDiskPageId(fileId, frontier.pageId), bulkNewPage);
             frontier.page.acquireWriteLatch();
@@ -1262,13 +1233,14 @@
         if (spaceUsed + spaceNeeded > ctx.interiorMaxBytes) {
             //ctx.interiorFrame.deleteGreatest(cmp);
         	
-        	SplitKey copyKey = ctx.splitKey.duplicate();
+        	SplitKey copyKey = ctx.splitKey.duplicate(ctx.leafFrame.getTupleWriter().createTupleReference());
         	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.getKeyFieldCount());
             ctx.splitKey.initData(splitKeySize);
-            ctx.tupleWriter.writeTupleFields(frontier.lastTuple, 0, cmp.getKeyFieldCount(), ctx.splitKey.getBuffer(), 0);                  
+            ctx.tupleWriter.writeTupleFields(frontier.lastTuple, 0, cmp.getKeyFieldCount(), ctx.splitKey.getBuffer(), 0);
+            ctx.splitKey.getTuple().resetByOffset(ctx.splitKey.getBuffer(), 0);
             ctx.splitKey.setLeftPage(frontier.pageId);
             
             ctx.interiorFrame.deleteGreatest(cmp);
@@ -1317,6 +1289,7 @@
         	int splitKeySize = ctx.tupleWriter.bytesRequired(leafFrontier.lastTuple, 0, cmp.getKeyFieldCount());        	
             ctx.splitKey.initData(splitKeySize);
             ctx.tupleWriter.writeTupleFields(leafFrontier.lastTuple, 0, cmp.getKeyFieldCount(), ctx.splitKey.getBuffer(), 0);
+            ctx.splitKey.getTuple().resetByOffset(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 7605e1b..0784692 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
@@ -39,7 +39,7 @@
     
     public DiskOrderScanCursor(IBTreeLeafFrame frame) {
     	this.frame = frame;
-		this.frameTuple = frame.createTupleReference();
+		this.frameTuple = frame.getTupleWriter().createTupleReference();
     }
     
     @Override
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixPrefixTupleReference.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixPrefixTupleReference.java
index 0727cea..320025c 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixPrefixTupleReference.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixPrefixTupleReference.java
@@ -1,9 +1,15 @@
 package edu.uci.ics.hyracks.storage.am.btree.impls;
 
+import edu.uci.ics.hyracks.api.dataflow.value.ITypeTrait;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeFrame;
 import edu.uci.ics.hyracks.storage.am.btree.frames.FieldPrefixNSMLeafFrame;
+import edu.uci.ics.hyracks.storage.am.btree.tuples.TypeAwareTupleReference;
 
-public class FieldPrefixPrefixTupleReference extends SimpleTupleReference {
+public class FieldPrefixPrefixTupleReference extends TypeAwareTupleReference {
+	
+	public FieldPrefixPrefixTupleReference(ITypeTrait[] typeTraits) {
+		super(typeTraits);
+	}
 	
 	// assumes tuple index refers to prefix tuples
 	@Override
@@ -11,8 +17,9 @@
 		FieldPrefixNSMLeafFrame concreteFrame = (FieldPrefixNSMLeafFrame)frame;		
 		int prefixSlotOff = concreteFrame.slotManager.getPrefixSlotOff(tupleIndex);
 		int prefixSlot = concreteFrame.getBuffer().getInt(prefixSlotOff);
-		setFieldCount(concreteFrame.slotManager.decodeFirstSlotField(prefixSlot));
+		setFieldCount(concreteFrame.slotManager.decodeFirstSlotField(prefixSlot));		
 		tupleStartOff = concreteFrame.slotManager.decodeSecondSlotField(prefixSlot);
 		buf = concreteFrame.getBuffer();
+		resetByOffset(buf, tupleStartOff);
 	}	
 }
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixTupleReference.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixTupleReference.java
index c417b86..ba17ce7 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixTupleReference.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixTupleReference.java
@@ -13,7 +13,11 @@
 	private int suffixTupleStartOff;
 	private int numPrefixFields;
 	private int fieldCount;
-	private SimpleTupleReference helperTuple = new SimpleTupleReference();	
+	private IBTreeTupleReference helperTuple;
+	
+	public FieldPrefixTupleReference(IBTreeTupleReference helperTuple) {
+		this.helperTuple = helperTuple;
+	}
 	
 	@Override
 	public void resetByTupleIndex(IBTreeFrame frame, int tupleIndex) {
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/MultiComparator.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/MultiComparator.java
index 286dabe..486d732 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
@@ -21,6 +21,7 @@
 
 import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparator;
 import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
+import edu.uci.ics.hyracks.api.dataflow.value.ITypeTrait;
 import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
 import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
 
@@ -30,10 +31,10 @@
 	private static final long serialVersionUID = 1L;
 	
 	private IBinaryComparator[] cmps = null;
-	private int fieldCount;
+	private ITypeTrait[] typeTraits;
 	
-	public MultiComparator(int fieldCount, IBinaryComparator[] cmps) {
-		this.fieldCount = fieldCount;
+	public MultiComparator(ITypeTrait[] typeTraits, IBinaryComparator[] cmps) {
+		this.typeTraits = typeTraits;
 		this.cmps = cmps;
 	}
 		
@@ -94,6 +95,10 @@
 	}
 	
 	public int getFieldCount() {
-		return fieldCount;
+		return typeTraits.length;
+	}
+	
+	public ITypeTrait[] getTypeTraits() {
+		return typeTraits;
 	}
 }
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/NodeFrontier.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/NodeFrontier.java
index c578735..138c1fa 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/NodeFrontier.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/NodeFrontier.java
@@ -15,11 +15,16 @@
 
 package edu.uci.ics.hyracks.storage.am.btree.impls;
 
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeTupleReference;
 import edu.uci.ics.hyracks.storage.common.buffercache.ICachedPage;
 
 public class NodeFrontier {
 	public ICachedPage page;
 	public int pageId;
-	public SimpleTupleReference lastTuple = new SimpleTupleReference();
+	public IBTreeTupleReference lastTuple;
+	
+	public NodeFrontier(IBTreeTupleReference lastTuple) {
+		this.lastTuple = lastTuple;
+	}
 }
 
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 ec99a90..fd8a906 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
@@ -37,7 +37,7 @@
 			
 	public RangeSearchCursor(IBTreeLeafFrame frame) {
 		this.frame = frame;
-		this.frameTuple = frame.createTupleReference();
+		this.frameTuple = frame.getTupleWriter().createTupleReference();
 	}
 	
 	@Override
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/SplitKey.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/SplitKey.java
index 809e3da..67dfc92 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/SplitKey.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/SplitKey.java
@@ -17,12 +17,18 @@
 
 import java.nio.ByteBuffer;
 
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeTupleReference;
+
 public class SplitKey {		
 	public byte[] data = null;	
 	public ByteBuffer buf = null;
-	public SimpleTupleReference tuple = new SimpleTupleReference();
+	public IBTreeTupleReference tuple;
 	public int keySize = 0;
 	
+	public SplitKey(IBTreeTupleReference tuple) {
+		this.tuple = tuple;
+	}
+	
 	public void initData(int keySize) {
 		// try to reuse existing memory from a lower-level split if possible				
 		this.keySize = keySize;
@@ -49,7 +55,7 @@
 		return buf;
 	}
 	
-	public SimpleTupleReference getTuple() {
+	public IBTreeTupleReference getTuple() {
 		return tuple;
 	}
 	
@@ -74,13 +80,12 @@
 		buf.putInt(keySize + 4, rightPage);
 	}
 	
-	public SplitKey duplicate() {
-		SplitKey copy = new SplitKey();
+	public SplitKey duplicate(IBTreeTupleReference copyTuple) {
+		SplitKey copy = new SplitKey(copyTuple);
 		copy.data = data.clone();		
 		copy.buf = ByteBuffer.wrap(copy.data);
-		copy.tuple = new SimpleTupleReference();
 		copy.tuple.setFieldCount(tuple.getFieldCount());
-		copy.tuple.resetByOffset(copy.buf, 0);	
+		copy.tuple.resetByOffset(copy.buf, 0);
 		return copy;
 	}
 }
\ No newline at end of file
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/Test.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/Test.java
new file mode 100644
index 0000000..d905ac8
--- /dev/null
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/Test.java
@@ -0,0 +1,25 @@
+package edu.uci.ics.hyracks.storage.am.btree.impls;
+
+import edu.uci.ics.hyracks.storage.am.btree.tuples.VarLenIntEncoderDecoder;
+
+public class Test {
+	
+	/**
+	 * @param args
+	 */
+	public static void main(String[] args) {
+		
+		byte[] bytes = new byte[100];
+			
+		VarLenIntEncoderDecoder encDec = new VarLenIntEncoderDecoder();
+		
+		for(int i = 0; i < 1024; i++) {
+			encDec.reset(bytes, 0);
+			int bytesWritten = encDec.encode(i);
+			encDec.reset(bytes, 0);
+			int decoded = encDec.decode();		
+			int byteCount = encDec.getBytesRequired(i);			
+			System.out.println("ENCODED: " + i + " DECODED: " + decoded + " BYTES: " + bytesWritten + " " + byteCount);
+		}		
+	}		
+}
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/SimpleTupleReference.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/tuples/SimpleTupleReference.java
similarity index 95%
rename from hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/SimpleTupleReference.java
rename to hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/tuples/SimpleTupleReference.java
index d793e43..5a7ce72 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/SimpleTupleReference.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/tuples/SimpleTupleReference.java
@@ -1,4 +1,4 @@
-package edu.uci.ics.hyracks.storage.am.btree.impls;
+package edu.uci.ics.hyracks.storage.am.btree.tuples;
 
 import java.nio.ByteBuffer;
 
@@ -13,11 +13,18 @@
 	protected int nullFlagsBytes;
 	protected int fieldSlotsBytes;
 	
+	@Override
 	public void resetByOffset(ByteBuffer buf, int tupleStartOff) {
 		this.buf = buf;
 		this.tupleStartOff = tupleStartOff;
 	}
 	
+	@Override
+	public void resetByTupleIndex(IBTreeFrame frame, int tupleIndex) {
+		resetByOffset(frame.getBuffer(), frame.getTupleOffset(tupleIndex));		
+	}	
+	
+	@Override
 	public void setFieldCount(int fieldCount) {
 		this.fieldCount = fieldCount;
 		nullFlagsBytes = getNullFlagsBytes();
@@ -60,10 +67,5 @@
 	
 	protected int getFieldSlotsBytes() {
 		return fieldCount * 2;
-	}
-
-	@Override
-	public void resetByTupleIndex(IBTreeFrame frame, int tupleIndex) {
-		resetByOffset(frame.getBuffer(), frame.getTupleOffset(tupleIndex));		
 	}	
 }
\ No newline at end of file
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/SimpleTupleWriter.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/tuples/SimpleTupleWriter.java
similarity index 88%
rename from hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/SimpleTupleWriter.java
rename to hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/tuples/SimpleTupleWriter.java
index 8fa9677..0b1ccd0 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/SimpleTupleWriter.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/tuples/SimpleTupleWriter.java
@@ -1,11 +1,12 @@
-package edu.uci.ics.hyracks.storage.am.btree.impls;
+package edu.uci.ics.hyracks.storage.am.btree.tuples;
 
 import java.nio.ByteBuffer;
 
 import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
-import edu.uci.ics.hyracks.storage.am.btree.api.ITupleWriter;
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeTupleReference;
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeTupleWriter;
 
-public class SimpleTupleWriter implements ITupleWriter {
+public class SimpleTupleWriter implements IBTreeTupleWriter {
 
 	@Override
 	public int bytesRequired(ITupleReference tuple) {
@@ -26,9 +27,8 @@
 	}
 
 	@Override
-	public ITupleReference createTupleReference() {
-		// TODO Auto-generated method stub
-		return null;
+	public IBTreeTupleReference createTupleReference() {
+		return new SimpleTupleReference();
 	}
 	
 	@Override
@@ -73,7 +73,7 @@
 		
 		return runner - targetOff;				
 	}
-	
+		
 	private int getNullFlagsBytes(ITupleReference tuple) {
 		return (int)Math.ceil((double)tuple.getFieldCount() / 8.0);
 	}
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/tuples/SimpleTupleWriterFactory.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/tuples/SimpleTupleWriterFactory.java
new file mode 100644
index 0000000..31c3c40
--- /dev/null
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/tuples/SimpleTupleWriterFactory.java
@@ -0,0 +1,15 @@
+package edu.uci.ics.hyracks.storage.am.btree.tuples;
+
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeTupleWriter;
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeTupleWriterFactory;
+
+public class SimpleTupleWriterFactory implements IBTreeTupleWriterFactory {
+	
+	private static final long serialVersionUID = 1L;
+
+	@Override
+	public IBTreeTupleWriter createTupleWriter() {
+		return new SimpleTupleWriter();
+	}
+	
+}
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/tuples/TypeAwareTupleReference.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/tuples/TypeAwareTupleReference.java
new file mode 100644
index 0000000..5f8b6d8
--- /dev/null
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/tuples/TypeAwareTupleReference.java
@@ -0,0 +1,98 @@
+package edu.uci.ics.hyracks.storage.am.btree.tuples;
+
+import java.nio.ByteBuffer;
+
+import edu.uci.ics.hyracks.api.dataflow.value.ITypeTrait;
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeFrame;
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeTupleReference;
+
+public class TypeAwareTupleReference implements IBTreeTupleReference {
+	protected ByteBuffer buf;
+	protected int fieldCount;	
+	protected int tupleStartOff;
+	protected int nullFlagsBytes;
+	protected int dataStartOff;
+	
+	private ITypeTrait[] typeTraits;
+	private VarLenIntEncoderDecoder encDec = new VarLenIntEncoderDecoder();
+	private int[] decodedFieldSlots;
+	
+	public TypeAwareTupleReference(ITypeTrait[] typeTraits) {
+		this.typeTraits = typeTraits;
+	}
+	
+	@Override
+	public void resetByOffset(ByteBuffer buf, int tupleStartOff) {
+		this.buf = buf;
+		this.tupleStartOff = tupleStartOff;
+		
+		// decode field slots
+		int field = 0;
+		int cumul = 0;
+		encDec.reset(buf.array(), tupleStartOff + nullFlagsBytes);
+		for(int i = 0; i < fieldCount; i++) {
+			int staticDataLen = typeTraits[i].getStaticallyKnownDataLength();
+			if(staticDataLen == ITypeTrait.VARIABLE_LENGTH) {
+				cumul += encDec.decode();
+				decodedFieldSlots[field++] = cumul;
+			}
+			else {
+				cumul += staticDataLen;
+				decodedFieldSlots[field++] = cumul;
+			}
+		}
+		dataStartOff = encDec.getPos();
+	}
+	
+	@Override
+	public void resetByTupleIndex(IBTreeFrame frame, int tupleIndex) {		
+		resetByOffset(frame.getBuffer(), frame.getTupleOffset(tupleIndex));		
+	}
+	
+	public void setFieldCount(int fieldCount) {
+		this.fieldCount = fieldCount;
+		if(decodedFieldSlots == null) {
+			decodedFieldSlots = new int[fieldCount];
+		}
+		else {
+			if(fieldCount > decodedFieldSlots.length) {
+				decodedFieldSlots = new int[fieldCount];
+			}
+		}		
+		nullFlagsBytes = getNullFlagsBytes();
+	}
+	
+	@Override
+	public int getFieldCount() {
+		return fieldCount;
+	}
+
+	@Override
+	public byte[] getFieldData(int fIdx) {
+		return buf.array();
+	}
+
+	@Override
+	public int getFieldLength(int fIdx) {
+		if(fIdx == 0) {
+			return decodedFieldSlots[0];			
+		}
+		else {
+			return decodedFieldSlots[fIdx] - decodedFieldSlots[fIdx-1];
+		}
+	}
+
+	@Override
+	public int getFieldStart(int fIdx) {				
+		if(fIdx == 0) {			
+			return dataStartOff;
+		}
+		else {			
+			return dataStartOff + decodedFieldSlots[fIdx-1];
+		}		
+	}
+	
+	protected int getNullFlagsBytes() {
+		return (int)Math.ceil(fieldCount / 8.0);
+	}		
+}
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/tuples/TypeAwareTupleWriter.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/tuples/TypeAwareTupleWriter.java
new file mode 100644
index 0000000..e82e07f
--- /dev/null
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/tuples/TypeAwareTupleWriter.java
@@ -0,0 +1,126 @@
+package edu.uci.ics.hyracks.storage.am.btree.tuples;
+
+import java.nio.ByteBuffer;
+
+import edu.uci.ics.hyracks.api.dataflow.value.ITypeTrait;
+import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeTupleReference;
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeTupleWriter;
+
+public class TypeAwareTupleWriter implements IBTreeTupleWriter {
+
+	private ITypeTrait[] typeTraits;
+	private VarLenIntEncoderDecoder encDec = new VarLenIntEncoderDecoder();
+	
+	public TypeAwareTupleWriter(ITypeTrait[] typeTraits) {
+		this.typeTraits = typeTraits;
+	}
+	
+	@Override
+	public int bytesRequired(ITupleReference tuple) {
+		int bytes = getNullFlagsBytes(tuple) + getFieldSlotsBytes(tuple);
+		for(int i = 0; i < tuple.getFieldCount(); i++) {
+			bytes += tuple.getFieldLength(i);
+		}
+		return bytes;
+	}
+	
+	@Override
+	public int bytesRequired(ITupleReference tuple, int startField, int numFields) {		
+		int bytes = getNullFlagsBytes(tuple, startField, numFields) + getFieldSlotsBytes(tuple, startField, numFields);
+		for(int i = startField; i < startField + numFields; i++) {
+			bytes += tuple.getFieldLength(i);
+		}
+		return bytes;	
+	}
+
+	@Override
+	public IBTreeTupleReference createTupleReference() {
+		return new TypeAwareTupleReference(typeTraits);
+	}
+	
+	@Override
+	public int writeTuple(ITupleReference tuple, ByteBuffer targetBuf, int targetOff) {
+		int runner = targetOff;
+		int nullFlagsBytes = getNullFlagsBytes(tuple);
+		// write null indicator bits
+		for(int i = 0; i < nullFlagsBytes; i++) {
+			targetBuf.put(runner++, (byte)0);			
+		}
+		
+		// write field slots for variable length fields
+		encDec.reset(targetBuf.array(), runner);
+		for(int i = 0; i < tuple.getFieldCount(); i++) {
+			if(typeTraits[i].getStaticallyKnownDataLength() == ITypeTrait.VARIABLE_LENGTH) {
+				encDec.encode(tuple.getFieldLength(i));
+			}
+		}
+		runner = encDec.getPos();
+		
+		// write data fields
+		for(int i = 0; i < tuple.getFieldCount(); i++) {			
+			System.arraycopy(tuple.getFieldData(i), tuple.getFieldStart(i), targetBuf.array(), runner, tuple.getFieldLength(i));			
+			runner += tuple.getFieldLength(i);
+		}
+		
+		return runner - targetOff;
+	}
+
+	@Override
+	public int writeTupleFields(ITupleReference tuple, int startField, int numFields, ByteBuffer targetBuf, int targetOff) {	
+		int runner = targetOff;
+		int nullFlagsBytes = getNullFlagsBytes(tuple, startField, numFields);
+		// write null indicator bits
+		for(int i = 0; i < nullFlagsBytes; i++) {
+			targetBuf.put(runner++, (byte)0);
+		}
+		
+		// write field slots for variable length fields		
+		encDec.reset(targetBuf.array(), runner);
+		for(int i = startField; i < startField + numFields; i++) {
+			if(typeTraits[i].getStaticallyKnownDataLength() == ITypeTrait.VARIABLE_LENGTH) {
+				encDec.encode(tuple.getFieldLength(i));
+			}
+		}
+		runner = encDec.getPos();
+		
+		for(int i = startField; i < startField + numFields; i++) {			
+			System.arraycopy(tuple.getFieldData(i), tuple.getFieldStart(i), targetBuf.array(), runner, tuple.getFieldLength(i));
+			runner += tuple.getFieldLength(i);					
+		}
+		
+		return runner - targetOff;				
+	}
+	
+	private int getNullFlagsBytes(ITupleReference tuple) {
+		return (int)Math.ceil((double)tuple.getFieldCount() / 8.0);
+	}
+	
+	private int getFieldSlotsBytes(ITupleReference tuple) {
+		int fieldSlotBytes = 0;
+		for(int i = 0; i < tuple.getFieldCount(); i++) {
+			if(typeTraits[i].getStaticallyKnownDataLength() == ITypeTrait.VARIABLE_LENGTH) {
+				fieldSlotBytes += encDec.getBytesRequired(tuple.getFieldLength(i));
+			}
+		}				
+		return fieldSlotBytes;
+	}
+	
+	private int getNullFlagsBytes(ITupleReference tuple, int startField, int numFields) {
+		return (int)Math.ceil((double)numFields / 8.0);
+	}
+	
+	private int getFieldSlotsBytes(ITupleReference tuple, int startField, int numFields) {
+		int fieldSlotBytes = 0;
+		for(int i = startField; i < startField + numFields; i++) {
+			if(typeTraits[i].getStaticallyKnownDataLength() == ITypeTrait.VARIABLE_LENGTH) {
+				fieldSlotBytes += encDec.getBytesRequired(tuple.getFieldLength(i));
+			}			
+		}
+		return fieldSlotBytes;
+	}
+	
+	public ITypeTrait[] getTypeTraits() {
+		return typeTraits;
+	}
+}
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/tuples/TypeAwareTupleWriterFactory.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/tuples/TypeAwareTupleWriterFactory.java
new file mode 100644
index 0000000..12c7c1e
--- /dev/null
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/tuples/TypeAwareTupleWriterFactory.java
@@ -0,0 +1,21 @@
+package edu.uci.ics.hyracks.storage.am.btree.tuples;
+
+import edu.uci.ics.hyracks.api.dataflow.value.ITypeTrait;
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeTupleWriter;
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeTupleWriterFactory;
+
+public class TypeAwareTupleWriterFactory implements IBTreeTupleWriterFactory {
+	
+	private static final long serialVersionUID = 1L;
+	private ITypeTrait[] typeTraits;
+	
+	public TypeAwareTupleWriterFactory(ITypeTrait[] typeTraits) {
+		this.typeTraits = typeTraits;
+	}
+	
+	@Override
+	public IBTreeTupleWriter createTupleWriter() {
+		return new TypeAwareTupleWriter(typeTraits);
+	}
+	
+}
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/tuples/VarLenIntEncoderDecoder.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/tuples/VarLenIntEncoderDecoder.java
new file mode 100644
index 0000000..b3565ba
--- /dev/null
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/tuples/VarLenIntEncoderDecoder.java
@@ -0,0 +1,83 @@
+package edu.uci.ics.hyracks.storage.am.btree.tuples;
+
+// encodes positive integers in a variable-byte format
+
+public class VarLenIntEncoderDecoder {
+	public static final int ENCODE_MASK = 0x0000007F;
+	public static final byte CONTINUE_CHUNK = (byte)0x80;
+	public static final byte DECODE_MASK = (byte)0x7F;
+	
+	private byte[] encTmp = new byte[5];
+	
+	private int pos;
+	private byte[] bytes;	
+	
+	public void reset(byte[] bytes, int pos) {
+		this.bytes = bytes;
+		this.pos = pos;		
+	}
+		
+	public int encode(int val) {		
+		int origPos = 0;
+		int tmpPos = 0;
+		while(val > ENCODE_MASK) {			
+			encTmp[tmpPos++] = (byte)(val & ENCODE_MASK);		
+			val = val >>> 7;			
+		}
+		encTmp[tmpPos++] = (byte)(val);
+		
+		// reverse order to optimize for decoding speed
+		for(int i = 0; i < tmpPos-1; i++) {
+			bytes[pos++] = (byte)(encTmp[tmpPos-1-i] | CONTINUE_CHUNK);
+		}
+		bytes[pos++] = encTmp[0];
+		
+		return pos - origPos;
+	}
+	
+	public int decode() {
+		int sum = 0;
+		while( (bytes[pos] & CONTINUE_CHUNK) == CONTINUE_CHUNK) {
+			sum = (sum + (bytes[pos] & DECODE_MASK)) << 7;
+			pos++;
+		}
+		sum += bytes[pos++];
+		return sum;
+	}	
+	
+	// calculate the number of bytes needed for encoding
+	public int getBytesRequired(int val) {
+		int byteCount = 0;
+		while(val > ENCODE_MASK) {
+			val = val >>> 7;
+			byteCount++;
+		}		
+		return byteCount + 1;
+	}
+	
+	public int getPos() {
+		return pos;
+	}
+	
+	// fast encoding, slow decoding version
+	/*
+	public void encode(int val) {		
+		while(val > ENCODE_MASK) {			
+			bytes[pos++] = (byte)(((byte)(val & ENCODE_MASK)) | CONTINUE_CHUNK);			
+			val = val >>> 7;			
+		}
+		bytes[pos++] = (byte)(val);
+	}
+	
+	public int decode() {
+		int sum = 0;
+		int shift = 0;
+		while( (bytes[pos] & CONTINUE_CHUNK) == CONTINUE_CHUNK) {
+			sum = (sum + (bytes[pos] & DECODE_MASK)) << 7 * shift++;
+			pos++;
+		}
+		sum += bytes[pos++] << 7 * shift;
+		return sum;
+	}	
+	*/	
+}
\ No newline at end of file
diff --git a/hyracks/hyracks-storage-am-btree/src/test/java/edu/uci/ics/hyracks/storage/am/btree/BTreeFieldPrefixNSMTest.java b/hyracks/hyracks-storage-am-btree/src/test/java/edu/uci/ics/hyracks/storage/am/btree/BTreeFieldPrefixNSMTest.java
index 2746392..0730254 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
@@ -28,7 +28,9 @@
 import edu.uci.ics.hyracks.api.context.IHyracksContext;
 import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparator;
 import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
+import edu.uci.ics.hyracks.api.dataflow.value.ITypeTrait;
 import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
+import edu.uci.ics.hyracks.api.dataflow.value.TypeTrait;
 import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
 import edu.uci.ics.hyracks.control.nc.runtime.RootHyracksContext;
 import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
@@ -38,11 +40,13 @@
 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.IBTreeTupleWriter;
 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.tuples.TypeAwareTupleWriter;
 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;
@@ -129,13 +133,20 @@
         FileInfo fi = new FileInfo(fileId, raf);
         fileManager.registerFile(fi);
         
+        // declare fields
         int fieldCount = 3;
+        ITypeTrait[] typeTraits = new ITypeTrait[fieldCount];
+        typeTraits[0] = new TypeTrait(4);
+        typeTraits[1] = new TypeTrait(4);
+        typeTraits[2] = new TypeTrait(4);
+        
+        // declare keys
         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(fieldCount, cmps);
+        MultiComparator cmp = new MultiComparator(typeTraits, cmps);
         
         // just for printing
         ISerializerDeserializer[] sers = { IntegerSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE };
@@ -147,7 +158,8 @@
         try {
         	        	        	        	
             IPrefixSlotManager slotManager = new FieldPrefixSlotManager();
-            FieldPrefixNSMLeafFrame frame = new FieldPrefixNSMLeafFrame();                                    
+            IBTreeTupleWriter tupleWriter = new TypeAwareTupleWriter(typeTraits);
+            FieldPrefixNSMLeafFrame frame = new FieldPrefixNSMLeafFrame(tupleWriter);                               
             frame.setPage(page);            
             frame.initBuffer((byte)0);
             slotManager.setFrame(frame);          
@@ -165,7 +177,7 @@
         	
         	// insert records with random calls to compact and compress
         	for(int i = 0; i < numRecords; i++) {
-        		
+        		        		
         		if((i+1) % 100 == 0) print("INSERTING " + (i+1) + " / " + numRecords + "\n");        		
         		
         		int a = rnd.nextInt() % smallMax;
@@ -186,13 +198,13 @@
         		savedFields[i][0] = a;
         		savedFields[i][1] = b;
         		savedFields[i][2] = c;
-        		            
+        		        		
         		if(rnd.nextInt() % compactFreq == 0) {
         			before = frame.printKeys(cmp, sers);
         			frame.compact(cmp);
         			after = frame.printKeys(cmp, sers);
         			Assert.assertEquals(before, after);
-        		}
+        		}        		
         		
         		if(rnd.nextInt() % compressFreq == 0) {
         			before = frame.printKeys(cmp, sers);
@@ -200,11 +212,12 @@
         			after = frame.printKeys(cmp, sers);
         			Assert.assertEquals(before, after);
         		}
-        	}
+        		
+        	}        	
         	
         	// delete records with random calls to compact and compress        	
         	for(int i = 0; i < numRecords; i++) {        	            	    
-        		
+        		        		
         		if((i+1) % 100 == 0) print("DELETING " + (i+1) + " / " + numRecords + "\n");
         		
         		ITupleReference tuple = createTuple(savedFields[i][0], savedFields[i][1], savedFields[i][2], false);
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 b4ed2d4..0d2b256 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
@@ -27,7 +27,9 @@
 import edu.uci.ics.hyracks.api.context.IHyracksContext;
 import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparator;
 import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
+import edu.uci.ics.hyracks.api.dataflow.value.ITypeTrait;
 import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
+import edu.uci.ics.hyracks.api.dataflow.value.TypeTrait;
 import edu.uci.ics.hyracks.control.nc.runtime.RootHyracksContext;
 import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
 import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAccessor;
@@ -54,6 +56,8 @@
 import edu.uci.ics.hyracks.storage.am.btree.impls.MultiComparator;
 import edu.uci.ics.hyracks.storage.am.btree.impls.RangePredicate;
 import edu.uci.ics.hyracks.storage.am.btree.impls.RangeSearchCursor;
+import edu.uci.ics.hyracks.storage.am.btree.tuples.SimpleTupleWriterFactory;
+import edu.uci.ics.hyracks.storage.am.btree.tuples.TypeAwareTupleWriterFactory;
 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;
@@ -93,7 +97,7 @@
             return buffers;
         }
     }
-        
+    
     // FIXED-LENGTH KEY TEST
     // create a B-tree with one fixed-length "key" field and one fixed-length "value" field
     // fill B-tree with random values using insertions (not bulk load)
@@ -114,19 +118,28 @@
         FileInfo fi = new FileInfo(fileId, raf);
         fileManager.registerFile(fi);
         
-        IBTreeLeafFrameFactory leafFrameFactory = new NSMLeafFrameFactory();
-        IBTreeInteriorFrameFactory interiorFrameFactory = new NSMInteriorFrameFactory();
+        // declare fields
+        int fieldCount = 2;
+        ITypeTrait[] typeTraits = new ITypeTrait[fieldCount];
+        typeTraits[0] = new TypeTrait(4);
+        typeTraits[1] = new TypeTrait(4);
+        
+        // declare keys
+        int keyFieldCount = 1;
+        IBinaryComparator[] cmps = new IBinaryComparator[keyFieldCount];
+        cmps[0] = IntegerBinaryComparatorFactory.INSTANCE.createBinaryComparator();
+                
+        MultiComparator cmp = new MultiComparator(typeTraits, cmps);
+        
+        TypeAwareTupleWriterFactory tupleWriterFactory = new TypeAwareTupleWriterFactory(typeTraits);
+        //SimpleTupleWriterFactory tupleWriterFactory = new SimpleTupleWriterFactory();
+        IBTreeLeafFrameFactory leafFrameFactory = new NSMLeafFrameFactory(tupleWriterFactory);
+        IBTreeInteriorFrameFactory interiorFrameFactory = new NSMInteriorFrameFactory(tupleWriterFactory);
         IBTreeMetaDataFrameFactory metaFrameFactory = new MetaDataFrameFactory();
         
         IBTreeLeafFrame leafFrame = leafFrameFactory.getFrame();
         IBTreeInteriorFrame interiorFrame = interiorFrameFactory.getFrame();
         IBTreeMetaDataFrame metaFrame = metaFrameFactory.getFrame();        
-               
-        int fieldCount = 2;
-        int keyFieldCount = 1;
-        IBinaryComparator[] cmps = new IBinaryComparator[keyFieldCount];
-        cmps[0] = IntegerBinaryComparatorFactory.INSTANCE.createBinaryComparator();
-        MultiComparator cmp = new MultiComparator(2, cmps);
         
         BTree btree = new BTree(bufferCache, interiorFrameFactory, leafFrameFactory, cmp);
         btree.create(fileId, leafFrame, metaFrame);
@@ -153,7 +166,7 @@
 		
 		// 10000
         for (int i = 0; i < 10000; i++) {        	        	
-        	
+        	        	
         	int f0 = rnd.nextInt() % 10000;
         	int f1 = 5;
         	        	
@@ -188,8 +201,9 @@
         //btree.printTree(leafFrame, interiorFrame);
         //System.out.println();
         
-        print("TOTALSPACE: " + f.length() + "\n");
-        
+        int maxPage = btree.getMaxPage(metaFrame);
+        System.out.println("MAXPAGE: " + maxPage);
+                
         String stats = btree.printStats();
         print(stats);
         
@@ -273,7 +287,7 @@
 		
         IBinaryComparator[] searchCmps = new IBinaryComparator[1];
         searchCmps[0] = IntegerBinaryComparatorFactory.INSTANCE.createBinaryComparator();
-        MultiComparator searchCmp = new MultiComparator(fieldCount, searchCmps);
+        MultiComparator searchCmp = new MultiComparator(typeTraits, searchCmps);
         
         RangePredicate rangePred = new RangePredicate(true, lowKey, highKey, searchCmp);
         btree.search(rangeCursor, rangePred, leafFrame, interiorFrame);
@@ -319,20 +333,30 @@
         FileInfo fi = new FileInfo(fileId, raf);
         fileManager.registerFile(fi);
                 
-        IBTreeLeafFrameFactory leafFrameFactory = new NSMLeafFrameFactory();
-        IBTreeInteriorFrameFactory interiorFrameFactory = new NSMInteriorFrameFactory();
-        IBTreeMetaDataFrameFactory metaFrameFactory = new MetaDataFrameFactory();
-        
-        IBTreeLeafFrame leafFrame = leafFrameFactory.getFrame();
-        IBTreeInteriorFrame interiorFrame = interiorFrameFactory.getFrame();
-        IBTreeMetaDataFrame metaFrame = metaFrameFactory.getFrame();   
-         
+        // declare fields
         int fieldCount = 3;
+        ITypeTrait[] typeTraits = new ITypeTrait[fieldCount];
+        typeTraits[0] = new TypeTrait(4);
+        typeTraits[1] = new TypeTrait(4);
+        typeTraits[2] = new TypeTrait(4);
+        
+        // declare keys
         int keyFieldCount = 2;
         IBinaryComparator[] cmps = new IBinaryComparator[keyFieldCount];
         cmps[0] = IntegerBinaryComparatorFactory.INSTANCE.createBinaryComparator();
         cmps[1] = IntegerBinaryComparatorFactory.INSTANCE.createBinaryComparator();       
-        MultiComparator cmp = new MultiComparator(fieldCount, cmps);
+                
+        MultiComparator cmp = new MultiComparator(typeTraits, cmps);
+        
+        TypeAwareTupleWriterFactory tupleWriterFactory = new TypeAwareTupleWriterFactory(typeTraits);
+        //SimpleTupleWriterFactory tupleWriterFactory = new SimpleTupleWriterFactory();
+        IBTreeLeafFrameFactory leafFrameFactory = new NSMLeafFrameFactory(tupleWriterFactory);
+        IBTreeInteriorFrameFactory interiorFrameFactory = new NSMInteriorFrameFactory(tupleWriterFactory);
+        IBTreeMetaDataFrameFactory metaFrameFactory = new MetaDataFrameFactory();
+        
+        IBTreeLeafFrame leafFrame = leafFrameFactory.getFrame();
+        IBTreeInteriorFrame interiorFrame = interiorFrameFactory.getFrame();
+        IBTreeMetaDataFrame metaFrame = metaFrameFactory.getFrame();                    
 
         BTree btree = new BTree(bufferCache, interiorFrameFactory, leafFrameFactory, cmp);
         btree.create(fileId, leafFrame, metaFrame);
@@ -448,7 +472,7 @@
 		
         IBinaryComparator[] searchCmps = new IBinaryComparator[1];
         searchCmps[0] = IntegerBinaryComparatorFactory.INSTANCE.createBinaryComparator();       
-        MultiComparator searchCmp = new MultiComparator(fieldCount, searchCmps); // use only a single comparator for searching
+        MultiComparator searchCmp = new MultiComparator(typeTraits, searchCmps); // use only a single comparator for searching
         
         RangePredicate rangePred = new RangePredicate(true, lowKey, highKey, searchCmp);
         btree.search(rangeCursor, rangePred, leafFrame, interiorFrame);
@@ -472,8 +496,7 @@
         fileManager.close();
         
         print("\n");
-    }    
-    
+    }
     
     // VARIABLE-LENGTH TEST
     // create a B-tree with one variable-length "key" field and one variable-length "value" field
@@ -495,20 +518,29 @@
     	FileInfo fi = new FileInfo(fileId, raf);
     	fileManager.registerFile(fi);
     	
-        IBTreeLeafFrameFactory leafFrameFactory = new NSMLeafFrameFactory();
-        IBTreeInteriorFrameFactory interiorFrameFactory = new NSMInteriorFrameFactory();
+    	// declare fields
+    	int fieldCount = 2;
+    	ITypeTrait[] typeTraits = new ITypeTrait[fieldCount];
+        typeTraits[0] = new TypeTrait(ITypeTrait.VARIABLE_LENGTH);
+        typeTraits[1] = new TypeTrait(ITypeTrait.VARIABLE_LENGTH);
+    	
+        // declare keys
+    	int keyFieldCount = 1;
+    	IBinaryComparator[] cmps = new IBinaryComparator[keyFieldCount];
+    	cmps[0] = UTF8StringBinaryComparatorFactory.INSTANCE.createBinaryComparator();
+    	
+    	MultiComparator cmp = new MultiComparator(typeTraits, cmps);
+    	
+    	SimpleTupleWriterFactory tupleWriterFactory = new SimpleTupleWriterFactory();
+    	//TypeAwareTupleWriterFactory tupleWriterFactory = new TypeAwareTupleWriterFactory(typeTraits);
+        IBTreeLeafFrameFactory leafFrameFactory = new NSMLeafFrameFactory(tupleWriterFactory);
+        IBTreeInteriorFrameFactory interiorFrameFactory = new NSMInteriorFrameFactory(tupleWriterFactory);
         IBTreeMetaDataFrameFactory metaFrameFactory = new MetaDataFrameFactory();
         
         IBTreeLeafFrame leafFrame = leafFrameFactory.getFrame();
         IBTreeInteriorFrame interiorFrame = interiorFrameFactory.getFrame();
         IBTreeMetaDataFrame metaFrame = metaFrameFactory.getFrame();   
         
-    	int fieldCount = 2;
-    	int keyFieldCount = 1;
-    	IBinaryComparator[] cmps = new IBinaryComparator[keyFieldCount];
-    	cmps[0] = UTF8StringBinaryComparatorFactory.INSTANCE.createBinaryComparator();
-    	MultiComparator cmp = new MultiComparator(fieldCount, cmps);
-    	
     	BTree btree = new BTree(bufferCache, interiorFrameFactory, leafFrameFactory, cmp);
     	btree.create(fileId, leafFrame, metaFrame);
     	btree.open(fileId);
@@ -530,6 +562,7 @@
     	
     	int maxLength = 10; // max string length to be generated
     	for (int i = 0; i < 10000; i++) {
+    		    		
     		String f0 = randomString(Math.abs(rnd.nextInt()) % maxLength + 1, rnd);
     		String f1 = randomString(Math.abs(rnd.nextInt()) % maxLength + 1, rnd);
     		
@@ -553,11 +586,12 @@
     			btree.insert(tuple, leafFrame, interiorFrame, metaFrame);
     		} catch (Exception e) {
     			//e.printStackTrace();
-    		}
-    		    		
+    		}    		    	    		
     	}     
     	// btree.printTree();
 
+    	System.out.println("DONE INSERTING");
+    	
     	// ordered scan
         print("ORDERED SCAN:\n");        
         IBTreeCursor scanCursor = new RangeSearchCursor(leafFrame);
@@ -567,7 +601,7 @@
         try {
             while (scanCursor.hasNext()) {
                 scanCursor.next();
-                ITupleReference frameTuple = scanCursor.getTuple();                                
+                ITupleReference frameTuple = scanCursor.getTuple();                         
                 String rec = cmp.printTuple(frameTuple, recDescSers);
                 print(rec + "\n");  
             }
@@ -615,7 +649,7 @@
 		
         IBinaryComparator[] searchCmps = new IBinaryComparator[1];
         searchCmps[0] = UTF8StringBinaryComparatorFactory.INSTANCE.createBinaryComparator();
-        MultiComparator searchCmp = new MultiComparator(fieldCount, searchCmps);
+        MultiComparator searchCmp = new MultiComparator(typeTraits, searchCmps);
         
         RangePredicate rangePred = new RangePredicate(true, lowKey, highKey, searchCmp);
         btree.search(rangeCursor, rangePred, leafFrame, interiorFrame);
@@ -640,7 +674,7 @@
         
         print("\n");
     }
-        
+    
     
     // DELETION TEST
     // create a B-tree with one variable-length "key" field and one variable-length "value" field
@@ -662,21 +696,29 @@
         FileInfo fi = new FileInfo(fileId, raf);
         fileManager.registerFile(fi);
                 
-        IBTreeLeafFrameFactory leafFrameFactory = new NSMLeafFrameFactory();
-        IBTreeInteriorFrameFactory interiorFrameFactory = new NSMInteriorFrameFactory();
+        // declare fields
+        int fieldCount = 2;
+        ITypeTrait[] typeTraits = new ITypeTrait[fieldCount];
+        typeTraits[0] = new TypeTrait(ITypeTrait.VARIABLE_LENGTH);
+        typeTraits[1] = new TypeTrait(ITypeTrait.VARIABLE_LENGTH);
+        
+        // declare keys
+        int keyFieldCount = 1;
+        IBinaryComparator[] cmps = new IBinaryComparator[keyFieldCount];
+        cmps[0] = UTF8StringBinaryComparatorFactory.INSTANCE.createBinaryComparator();
+        
+        MultiComparator cmp = new MultiComparator(typeTraits, cmps);
+        
+        //SimpleTupleWriterFactory tupleWriterFactory = new SimpleTupleWriterFactory();
+        TypeAwareTupleWriterFactory tupleWriterFactory = new TypeAwareTupleWriterFactory(typeTraits);
+        IBTreeLeafFrameFactory leafFrameFactory = new NSMLeafFrameFactory(tupleWriterFactory);
+        IBTreeInteriorFrameFactory interiorFrameFactory = new NSMInteriorFrameFactory(tupleWriterFactory);
         IBTreeMetaDataFrameFactory metaFrameFactory = new MetaDataFrameFactory();
         
         IBTreeLeafFrame leafFrame = leafFrameFactory.getFrame();
         IBTreeInteriorFrame interiorFrame = interiorFrameFactory.getFrame();
         IBTreeMetaDataFrame metaFrame = metaFrameFactory.getFrame();   
-        
-        int fieldCount = 2;
-        int keyFieldCount = 1;
-        IBinaryComparator[] cmps = new IBinaryComparator[keyFieldCount];
-        cmps[0] = UTF8StringBinaryComparatorFactory.INSTANCE.createBinaryComparator();
-        
-        MultiComparator cmp = new MultiComparator(fieldCount, cmps);
-
+                
         BTree btree = new BTree(bufferCache, interiorFrameFactory, leafFrameFactory, cmp);
         btree.create(fileId, leafFrame, metaFrame);
         btree.open(fileId);
@@ -795,7 +837,6 @@
         print("\n");
     }
         
-    
     // BULK LOAD TEST
     // insert 100,000 records in bulk
     // B-tree has a composite key to "simulate" non-unique index creation
@@ -816,21 +857,30 @@
         FileInfo fi = new FileInfo(fileId, raf);
         fileManager.registerFile(fi);
         
-        IBTreeLeafFrameFactory leafFrameFactory = new NSMLeafFrameFactory();
-        IBTreeInteriorFrameFactory interiorFrameFactory = new NSMInteriorFrameFactory();
-        IBTreeMetaDataFrameFactory metaFrameFactory = new MetaDataFrameFactory();
-        
-        IBTreeLeafFrame leafFrame = leafFrameFactory.getFrame();
-        IBTreeInteriorFrame interiorFrame = interiorFrameFactory.getFrame();
-        IBTreeMetaDataFrame metaFrame = metaFrameFactory.getFrame();       
-        
+        // declare fields
         int fieldCount = 3;
+        ITypeTrait[] typeTraits = new ITypeTrait[fieldCount];
+        typeTraits[0] = new TypeTrait(4);
+        typeTraits[1] = new TypeTrait(4);
+        typeTraits[2] = new TypeTrait(4);
+        
+        // declare keys
         int keyFieldCount = 2;
         IBinaryComparator[] cmps = new IBinaryComparator[keyFieldCount];
         cmps[0] = IntegerBinaryComparatorFactory.INSTANCE.createBinaryComparator();
         cmps[1] = IntegerBinaryComparatorFactory.INSTANCE.createBinaryComparator();
         
-        MultiComparator cmp = new MultiComparator(fieldCount, cmps);
+        MultiComparator cmp = new MultiComparator(typeTraits, cmps);
+        
+        //SimpleTupleWriterFactory tupleWriterFactory = new SimpleTupleWriterFactory();
+        TypeAwareTupleWriterFactory tupleWriterFactory = new TypeAwareTupleWriterFactory(typeTraits);
+        IBTreeLeafFrameFactory leafFrameFactory = new NSMLeafFrameFactory(tupleWriterFactory);
+        IBTreeInteriorFrameFactory interiorFrameFactory = new NSMInteriorFrameFactory(tupleWriterFactory);
+        IBTreeMetaDataFrameFactory metaFrameFactory = new MetaDataFrameFactory();
+        
+        IBTreeLeafFrame leafFrame = leafFrameFactory.getFrame();
+        IBTreeInteriorFrame interiorFrame = interiorFrameFactory.getFrame();
+        IBTreeMetaDataFrame metaFrame = metaFrameFactory.getFrame();                       
 
         BTree btree = new BTree(bufferCache, interiorFrameFactory, leafFrameFactory, cmp);
         btree.create(fileId, leafFrame, metaFrame);
@@ -872,7 +922,7 @@
         	
         	tuple.reset(accessor, 0);
         	
-        	btree.bulkLoadAddTuple(bulkLoadCtx, tuple);        	
+        	btree.bulkLoadAddTuple(bulkLoadCtx, tuple);	
         }
 
         btree.endBulkLoad(bulkLoadCtx);
@@ -920,7 +970,7 @@
     	        
         IBinaryComparator[] searchCmps = new IBinaryComparator[1];
         searchCmps[0] = IntegerBinaryComparatorFactory.INSTANCE.createBinaryComparator();
-        MultiComparator searchCmp = new MultiComparator(fieldCount, searchCmps);
+        MultiComparator searchCmp = new MultiComparator(typeTraits, searchCmps);
         
         // TODO: check when searching backwards
         RangePredicate rangePred = new RangePredicate(true, lowKey, highKey, searchCmp);
@@ -964,21 +1014,30 @@
         int fileId = 0;
         FileInfo fi = new FileInfo(fileId, raf);
         fileManager.registerFile(fi);       
-                
-        IBTreeLeafFrameFactory leafFrameFactory = new NSMLeafFrameFactory();
-        IBTreeInteriorFrameFactory interiorFrameFactory = new NSMInteriorFrameFactory();
-        IBTreeMetaDataFrameFactory metaFrameFactory = new MetaDataFrameFactory();
         
-        IBTreeLeafFrame leafFrame = leafFrameFactory.getFrame();
-        IBTreeInteriorFrame interiorFrame = interiorFrameFactory.getFrame();
-        IBTreeMetaDataFrame metaFrame = metaFrameFactory.getFrame();
-        
+        // declare fields
         int fieldCount = 3;
+        ITypeTrait[] typeTraits = new ITypeTrait[fieldCount];
+        typeTraits[0] = new TypeTrait(4);
+        typeTraits[1] = new TypeTrait(4);
+        typeTraits[2] = new TypeTrait(4);
+        
+        // declare keys
         int keyFieldCount = 2;
         IBinaryComparator[] cmps = new IBinaryComparator[keyFieldCount];
         cmps[0] = IntegerBinaryComparatorFactory.INSTANCE.createBinaryComparator();
         cmps[1] = IntegerBinaryComparatorFactory.INSTANCE.createBinaryComparator();        
-        MultiComparator cmp = new MultiComparator(fieldCount, cmps);
+        MultiComparator cmp = new MultiComparator(typeTraits, cmps);
+        
+        //SimpleTupleWriterFactory tupleWriterFactory = new SimpleTupleWriterFactory();
+        TypeAwareTupleWriterFactory tupleWriterFactory = new TypeAwareTupleWriterFactory(typeTraits);
+        IBTreeLeafFrameFactory leafFrameFactory = new NSMLeafFrameFactory(tupleWriterFactory);
+        IBTreeInteriorFrameFactory interiorFrameFactory = new NSMInteriorFrameFactory(tupleWriterFactory);
+        IBTreeMetaDataFrameFactory metaFrameFactory = new MetaDataFrameFactory();
+        
+        IBTreeLeafFrame leafFrame = leafFrameFactory.getFrame();
+        IBTreeInteriorFrame interiorFrame = interiorFrameFactory.getFrame();
+        IBTreeMetaDataFrame metaFrame = metaFrameFactory.getFrame();               
 
         BTree btree = new BTree(bufferCache, interiorFrameFactory, leafFrameFactory, cmp);
         btree.create(fileId, leafFrame, metaFrame);
@@ -1131,7 +1190,7 @@
         IBinaryComparator[] searchCmps = new IBinaryComparator[2];
         searchCmps[0] = IntegerBinaryComparatorFactory.INSTANCE.createBinaryComparator();
         searchCmps[1] = IntegerBinaryComparatorFactory.INSTANCE.createBinaryComparator();
-        MultiComparator searchCmp = new MultiComparator(fieldCount, searchCmps);
+        MultiComparator searchCmp = new MultiComparator(typeTraits, searchCmps);
                 
         //print("INDEX RANGE SEARCH ON: " + cmp.printKey(lowKey, 0) + " " + cmp.printKey(highKey, 0) + "\n");                
         
@@ -1158,7 +1217,7 @@
         
         print("\n");
     }
-    
+        
     public static String randomString(int length, Random random) {
         String s = Long.toHexString(Double.doubleToLongBits(random.nextDouble()));
         StringBuilder strBuilder = new StringBuilder();