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 9f34737..0725929 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
@@ -61,7 +61,7 @@
 
         @Option(name = "-port", usage = "Hyracks Cluster Controller Port (default: 1099)")
         public int port = 1099;
-
+        
         @Option(name = "-app", usage = "Hyracks Application name", required = true)
         public String app;
         
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 c84ed4e..0d419e6 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
@@ -14,6 +14,7 @@
  */
 package edu.uci.ics.hyracks.examples.btree.client;
 
+import java.io.DataOutput;
 import java.util.UUID;
 
 import org.kohsuke.args4j.CmdLineParser;
@@ -28,7 +29,9 @@
 import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparatorFactory;
 import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
 import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
+import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
 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.comparators.IntegerBinaryComparatorFactory;
 import edu.uci.ics.hyracks.dataflow.common.data.marshalling.IntegerSerializerDeserializer;
 import edu.uci.ics.hyracks.dataflow.common.data.marshalling.UTF8StringSerializerDeserializer;
@@ -40,11 +43,10 @@
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeInteriorFrameFactory;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeLeafFrameFactory;
 import edu.uci.ics.hyracks.storage.am.btree.dataflow.BTreeSearchOperatorDescriptor;
+import edu.uci.ics.hyracks.storage.am.btree.dataflow.ConstantTupleSourceOperatorDescriptor;
 import edu.uci.ics.hyracks.storage.am.btree.dataflow.IBTreeRegistryProvider;
 import edu.uci.ics.hyracks.storage.am.btree.dataflow.IBufferCacheProvider;
 import edu.uci.ics.hyracks.storage.am.btree.dataflow.IFileMappingProviderProvider;
-import edu.uci.ics.hyracks.storage.am.btree.dataflow.ITupleReferenceFactory;
-import edu.uci.ics.hyracks.storage.am.btree.dataflow.NullTupleReferenceFactory;
 import edu.uci.ics.hyracks.storage.am.btree.frames.NSMInteriorFrameFactory;
 import edu.uci.ics.hyracks.storage.am.btree.frames.NSMLeafFrameFactory;
 
@@ -73,7 +75,7 @@
         Options options = new Options();
         CmdLineParser parser = new CmdLineParser(options);
         parser.parseArgument(args);
-
+        
         IHyracksClientConnection hcc = new HyracksRMIConnection(options.host, options.port);
 
         JobSpecification job = createJob(options);
@@ -86,7 +88,7 @@
         System.err.println(start + " " + end + " " + (end - start));
     }
     
-    private static JobSpecification createJob(Options options) {
+    private static JobSpecification createJob(Options options) throws HyracksDataException {
     	
     	JobSpecification spec = new JobSpecification();
 
@@ -111,12 +113,28 @@
         IBinaryComparatorFactory[] comparatorFactories = new IBinaryComparatorFactory[1];
         comparatorFactories[0] = IntegerBinaryComparatorFactory.INSTANCE;     
         
-        // build search key factories
-    	ITupleReferenceFactory[] searchKeys = new ITupleReferenceFactory[2]; 
-    	searchKeys[0] = new NullTupleReferenceFactory(); // equivalent to -infinity
-    	searchKeys[1] = new NullTupleReferenceFactory(); // equivalent to +infinity
+        // build tuple containing low and high search keys
+		ArrayTupleBuilder tb = new ArrayTupleBuilder(comparatorFactories.length*2); // high key and low key
+		DataOutput dos = tb.getDataOutput();
+				
+		tb.reset();				
+		IntegerSerializerDeserializer.INSTANCE.serialize(100, dos); // low key
+		tb.addFieldEndOffset();		
+		IntegerSerializerDeserializer.INSTANCE.serialize(200, dos); // build high key
+		tb.addFieldEndOffset();
+		
+		ISerializerDeserializer[] keyRecDescSers = { UTF8StringSerializerDeserializer.INSTANCE, UTF8StringSerializerDeserializer.INSTANCE };
+		RecordDescriptor keyRecDesc = new RecordDescriptor(keyRecDescSers);
+		
+		ConstantTupleSourceOperatorDescriptor keyProviderOp = new ConstantTupleSourceOperatorDescriptor(spec, keyRecDesc, tb.getFieldEndOffsets(), tb.getByteArray(), tb.getSize());
+		PartitionConstraint keyProviderPartitionConstraint = 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
         
-        BTreeSearchOperatorDescriptor btreeSearchOp = new BTreeSearchOperatorDescriptor(spec, recDesc, bufferCacheProvider, btreeRegistryProvider, options.btreeName, fileMappingProviderProvider, interiorFrameFactory, leafFrameFactory, recDesc.getFields().length, comparatorFactories, true, searchKeys, comparatorFactories.length);
+        BTreeSearchOperatorDescriptor btreeSearchOp = new BTreeSearchOperatorDescriptor(spec, recDesc, bufferCacheProvider, btreeRegistryProvider, options.btreeName, fileMappingProviderProvider, interiorFrameFactory, leafFrameFactory, recDesc.getFields().length, comparatorFactories, true, lowKeyFields, highKeyFields);
         PartitionConstraint btreeSearchConstraint = createPartitionConstraint(splitNCs);
         btreeSearchOp.setPartitionConstraint(btreeSearchConstraint);
         
@@ -125,6 +143,8 @@
         PartitionConstraint printerConstraint = createPartitionConstraint(splitNCs);
         printer.setPartitionConstraint(printerConstraint);
         
+        spec.connect(new OneToOneConnectorDescriptor(spec), keyProviderOp, 0, btreeSearchOp, 0);
+        
         spec.connect(new OneToOneConnectorDescriptor(spec), btreeSearchOp, 0, printer, 0);
         
         spec.addRoot(printer);
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 a39126d..a2efc18 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
@@ -68,7 +68,7 @@
         @Option(name = "-primary-btreename", usage = "Name of primary-index B-Tree to load from", required = true)
         public String primaryBtreeName;
         
-        @Option(name = "-btreename", usage = "B-Tree file name for secondary index to be built", required = true)
+        @Option(name = "-secondary-btreename", usage = "B-Tree file name for secondary index to be built", required = true)
         public String btreeName;
         
         @Option(name = "-sortbuffer-size", usage = "Sort buffer size in frames (default: 32768)", required = false)
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 a74d7a0..3302674 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
@@ -15,7 +15,6 @@
 package edu.uci.ics.hyracks.examples.btree.client;
 
 import java.io.DataOutput;
-import java.nio.ByteBuffer;
 import java.util.UUID;
 
 import org.kohsuke.args4j.CmdLineParser;
@@ -27,15 +26,12 @@
 import edu.uci.ics.hyracks.api.constraints.ExplicitPartitionConstraint;
 import edu.uci.ics.hyracks.api.constraints.LocationConstraint;
 import edu.uci.ics.hyracks.api.constraints.PartitionConstraint;
-import edu.uci.ics.hyracks.api.context.IHyracksContext;
 import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparatorFactory;
 import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
 import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
 import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
 import edu.uci.ics.hyracks.api.job.JobSpecification;
-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.FrameTupleAppender;
 import edu.uci.ics.hyracks.dataflow.common.data.comparators.UTF8StringBinaryComparatorFactory;
 import edu.uci.ics.hyracks.dataflow.common.data.marshalling.IntegerSerializerDeserializer;
 import edu.uci.ics.hyracks.dataflow.common.data.marshalling.UTF8StringSerializerDeserializer;
@@ -47,15 +43,15 @@
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeInteriorFrameFactory;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeLeafFrameFactory;
 import edu.uci.ics.hyracks.storage.am.btree.dataflow.BTreeSearchOperatorDescriptor;
-import edu.uci.ics.hyracks.storage.am.btree.dataflow.FrameTupleReferenceFactory;
+import edu.uci.ics.hyracks.storage.am.btree.dataflow.ConstantTupleSourceOperatorDescriptor;
 import edu.uci.ics.hyracks.storage.am.btree.dataflow.IBTreeRegistryProvider;
 import edu.uci.ics.hyracks.storage.am.btree.dataflow.IBufferCacheProvider;
 import edu.uci.ics.hyracks.storage.am.btree.dataflow.IFileMappingProviderProvider;
-import edu.uci.ics.hyracks.storage.am.btree.dataflow.ITupleReferenceFactory;
 import edu.uci.ics.hyracks.storage.am.btree.frames.NSMInteriorFrameFactory;
 import edu.uci.ics.hyracks.storage.am.btree.frames.NSMLeafFrameFactory;
 
 // This example will perform range search on the secondary index
+// and then retrieve the corresponding source records from the primary index
 
 public class SecondaryIndexSearchExample {
     private static class Options {
@@ -71,8 +67,11 @@
         @Option(name = "-target-ncs", usage = "Comma separated list of node-controller names to use", required = true)
         public String ncs;
                 
-        @Option(name = "-btreename", usage = "B-Tree file name to search", required = true)
-        public String btreeName;               
+        @Option(name = "-primary-btreename", usage = "Primary B-Tree file name", required = true)
+        public String primaryBTreeName;               
+        
+        @Option(name = "-secondary-btreename", usage = "Secondary B-Tree file name to search", required = true)
+        public String secondaryBTreeName;        
     }
 
     public static void main(String[] args) throws Exception {
@@ -106,60 +105,67 @@
         IFileMappingProviderProvider fileMappingProviderProvider = FileMappingProviderProvider.INSTANCE;
     	
     	// schema of tuples coming out of secondary index
-        RecordDescriptor recDesc = new RecordDescriptor(new ISerializerDeserializer[] {                
+        RecordDescriptor secondaryRecDesc = new RecordDescriptor(new ISerializerDeserializer[] {                
         		UTF8StringSerializerDeserializer.INSTANCE,
         		IntegerSerializerDeserializer.INSTANCE                
                 });
         
+        // schema of tuples coming out of primary index
+        RecordDescriptor primaryRecDesc = new RecordDescriptor(new ISerializerDeserializer[] {                
+        		IntegerSerializerDeserializer.INSTANCE,
+                UTF8StringSerializerDeserializer.INSTANCE,    
+                IntegerSerializerDeserializer.INSTANCE,                           
+                UTF8StringSerializerDeserializer.INSTANCE,
+                });
+        
         // 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];
         comparatorFactories[0] = UTF8StringBinaryComparatorFactory.INSTANCE;        
-        
-        
-        // build search keys (which must be of type ITupleReference)
-        // put search keys into frame and create tuplereference factories
-        IHyracksContext ctx = new RootHyracksContext(32768); // WARNING: make sure frame size is same as on NCs
-        ByteBuffer keyFrame = ctx.getResourceManager().allocateFrame();
-		FrameTupleAppender appender = new FrameTupleAppender(ctx);				
-		appender.reset(keyFrame, true);
-		ArrayTupleBuilder tb = new ArrayTupleBuilder(comparatorFactories.length);
+                
+        // build tuple containing low and high search keys
+		ArrayTupleBuilder tb = new ArrayTupleBuilder(comparatorFactories.length*2); // low and high key
 		DataOutput dos = tb.getDataOutput();
-		
-		ISerializerDeserializer[] keyRecDescSers = { UTF8StringSerializerDeserializer.INSTANCE };
-		RecordDescriptor keyRecDesc = new RecordDescriptor(keyRecDescSers);
-		
-		// build low key
-		tb.reset();
-		UTF8StringSerializerDeserializer.INSTANCE.serialize("a", dos);
+						
+		tb.reset();		
+		UTF8StringSerializerDeserializer.INSTANCE.serialize("0", dos); // low key
     	tb.addFieldEndOffset();    	  
-    	        	    	
-    	appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
-    	    	
-    	// build high key
-    	tb.reset();
-    	UTF8StringSerializerDeserializer.INSTANCE.serialize("f", dos);
+    	UTF8StringSerializerDeserializer.INSTANCE.serialize("f", dos); // high key
     	tb.addFieldEndOffset();
+    	    	
+    	ISerializerDeserializer[] keyRecDescSers = { UTF8StringSerializerDeserializer.INSTANCE, UTF8StringSerializerDeserializer.INSTANCE };
+		RecordDescriptor keyRecDesc = new RecordDescriptor(keyRecDescSers);
     	
-    	appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
-    	
-    	// build search key factories
-    	ITupleReferenceFactory[] searchKeys = new ITupleReferenceFactory[2]; 
-    	// the low key is tuple 0 in the keyFrame
-    	searchKeys[0] = new FrameTupleReferenceFactory(keyFrame.array(), 0, keyRecDesc);
-    	// the high key is tuple 1 in the keyFrame
-    	searchKeys[1] = new FrameTupleReferenceFactory(keyFrame.array(), 1, keyRecDesc);
-    	
-        BTreeSearchOperatorDescriptor btreeSearchOp = new BTreeSearchOperatorDescriptor(spec, recDesc, bufferCacheProvider, btreeRegistryProvider, options.btreeName, fileMappingProviderProvider, interiorFrameFactory, leafFrameFactory, recDesc.getFields().length, comparatorFactories, true, searchKeys, comparatorFactories.length);
-        PartitionConstraint btreeSearchConstraint = createPartitionConstraint(splitNCs);
-        btreeSearchOp.setPartitionConstraint(btreeSearchConstraint);
+    	ConstantTupleSourceOperatorDescriptor keyProviderOp = new ConstantTupleSourceOperatorDescriptor(spec, keyRecDesc, tb.getFieldEndOffsets(), tb.getByteArray(), tb.getSize());
+		PartitionConstraint keyProviderPartitionConstraint = createPartitionConstraint(splitNCs);
+		keyProviderOp.setPartitionConstraint(keyProviderPartitionConstraint);
+				
+		int[] secondaryLowKeyFields = { 0 }; // low key is in field 0 of tuples going into secondary index search op
+        int[] secondaryHighKeyFields = { 1 }; // high key is in field 1 of tuples going into secondary index search op
+		
+        BTreeSearchOperatorDescriptor secondarySearchOp = new BTreeSearchOperatorDescriptor(spec, secondaryRecDesc, bufferCacheProvider, btreeRegistryProvider, options.secondaryBTreeName, fileMappingProviderProvider, interiorFrameFactory, leafFrameFactory, secondaryRecDesc.getFields().length, comparatorFactories, true, secondaryLowKeyFields, secondaryHighKeyFields);
+        PartitionConstraint secondarySearchConstraint = createPartitionConstraint(splitNCs);
+        secondarySearchOp.setPartitionConstraint(secondarySearchConstraint);
+        
+        // secondary index will output tuples with [UTF8String, Integer]
+        // the Integer field refers to the key in the primary index of the source data records
+        int[] primaryLowKeyFields = { 1 }; // low key is in field 0 of tuples going into primary index search op
+        int[] primaryHighKeyFields = { 1 }; // high key is in field 1 of tuples going into primary index search op
+		
+        BTreeSearchOperatorDescriptor primarySearchOp = new BTreeSearchOperatorDescriptor(spec, primaryRecDesc, bufferCacheProvider, btreeRegistryProvider, options.primaryBTreeName, fileMappingProviderProvider, interiorFrameFactory, leafFrameFactory, primaryRecDesc.getFields().length, comparatorFactories, true, primaryLowKeyFields, primaryHighKeyFields);
+        PartitionConstraint primarySearchConstraint = createPartitionConstraint(splitNCs);
+        primarySearchOp.setPartitionConstraint(primarySearchConstraint);
         
         // have each node print the results of its respective B-Tree
         PrinterOperatorDescriptor printer = new PrinterOperatorDescriptor(spec);
         PartitionConstraint printerConstraint = createPartitionConstraint(splitNCs);
         printer.setPartitionConstraint(printerConstraint);
         
-        spec.connect(new OneToOneConnectorDescriptor(spec), btreeSearchOp, 0, printer, 0);
+        spec.connect(new OneToOneConnectorDescriptor(spec), keyProviderOp, 0, secondarySearchOp, 0);
+        
+        spec.connect(new OneToOneConnectorDescriptor(spec), secondarySearchOp, 0, primarySearchOp, 0);
+        
+        spec.connect(new OneToOneConnectorDescriptor(spec), primarySearchOp, 0, printer, 0);
         
         spec.addRoot(printer);
     	    	
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 7fca1d3..cf7c847 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
@@ -18,7 +18,6 @@
 import java.io.DataOutput;
 import java.io.File;
 import java.io.RandomAccessFile;
-import java.nio.ByteBuffer;
 
 import org.junit.Test;
 
@@ -26,16 +25,12 @@
 import edu.uci.ics.hyracks.api.constraints.ExplicitPartitionConstraint;
 import edu.uci.ics.hyracks.api.constraints.LocationConstraint;
 import edu.uci.ics.hyracks.api.constraints.PartitionConstraint;
-import edu.uci.ics.hyracks.api.context.IHyracksContext;
-import edu.uci.ics.hyracks.api.dataflow.IConnectorDescriptor;
 import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparator;
 import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparatorFactory;
 import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
 import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
 import edu.uci.ics.hyracks.api.job.JobSpecification;
-import edu.uci.ics.hyracks.control.nc.runtime.RootHyracksContext;
 import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
-import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAppender;
 import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
 import edu.uci.ics.hyracks.dataflow.common.data.comparators.UTF8StringBinaryComparatorFactory;
 import edu.uci.ics.hyracks.dataflow.common.data.marshalling.UTF8StringSerializerDeserializer;
@@ -59,12 +54,11 @@
 import edu.uci.ics.hyracks.storage.am.btree.dataflow.BTreeRegistryProvider;
 import edu.uci.ics.hyracks.storage.am.btree.dataflow.BTreeSearchOperatorDescriptor;
 import edu.uci.ics.hyracks.storage.am.btree.dataflow.BufferCacheProvider;
+import edu.uci.ics.hyracks.storage.am.btree.dataflow.ConstantTupleSourceOperatorDescriptor;
 import edu.uci.ics.hyracks.storage.am.btree.dataflow.FileMappingProviderProvider;
-import edu.uci.ics.hyracks.storage.am.btree.dataflow.FrameTupleReferenceFactory;
 import edu.uci.ics.hyracks.storage.am.btree.dataflow.IBTreeRegistryProvider;
 import edu.uci.ics.hyracks.storage.am.btree.dataflow.IBufferCacheProvider;
 import edu.uci.ics.hyracks.storage.am.btree.dataflow.IFileMappingProviderProvider;
-import edu.uci.ics.hyracks.storage.am.btree.dataflow.ITupleReferenceFactory;
 import edu.uci.ics.hyracks.storage.am.btree.frames.MetaDataFrame;
 import edu.uci.ics.hyracks.storage.am.btree.frames.NSMInteriorFrameFactory;
 import edu.uci.ics.hyracks.storage.am.btree.frames.NSMLeafFrameFactory;
@@ -83,7 +77,7 @@
 	@Test
 	public void bulkLoadTest() throws Exception {		
 		// relies on the fact that NCs are run from same process
-		System.setProperty("NodeControllerDataPath", "/tmp/");
+		System.setProperty("NodeControllerDataPath", System.getProperty("java.io.tmpdir") + "/");
 		
 		JobSpecification spec = new JobSpecification();
 		
@@ -170,7 +164,7 @@
 	@Test
 	public void btreeSearchTest() throws Exception {
 		// relies on the fact that NCs are run from same process
-		System.setProperty("NodeControllerDataPath", "/tmp/");
+		System.setProperty("NodeControllerDataPath", System.getProperty("java.io.tmpdir") + "/");
 		
 		JobSpecification spec = new JobSpecification();
 		
@@ -191,51 +185,32 @@
     		comparators[i] = comparatorFactories[i].createBinaryComparator();
     	}    	
         MultiComparator cmp = new MultiComparator(fieldCount, comparators);
-		
-        
-        // put search keys into frame and create tuplereference factories
-        IHyracksContext ctx = new RootHyracksContext(32768); // WARNING: make sure frame size is same as on NCs
-        ByteBuffer keyFrame = ctx.getResourceManager().allocateFrame();
-		FrameTupleAppender appender = new FrameTupleAppender(ctx);				
-		appender.reset(keyFrame, true);
-		ArrayTupleBuilder tb = new ArrayTupleBuilder(cmp.getKeyFieldCount());
+		        
+        // build tuple containing low and high search key
+		ArrayTupleBuilder tb = new ArrayTupleBuilder(cmp.getKeyFieldCount()*2); // high key and low key
 		DataOutput dos = tb.getDataOutput();
 		
-		ISerializerDeserializer[] keyRecDescSers = { UTF8StringSerializerDeserializer.INSTANCE};
+		tb.reset();		
+		UTF8StringSerializerDeserializer.INSTANCE.serialize("100", dos); // low key
+		tb.addFieldEndOffset();		
+		UTF8StringSerializerDeserializer.INSTANCE.serialize("200", dos); // high key
+		tb.addFieldEndOffset();		
+    	
+		ISerializerDeserializer[] keyRecDescSers = { UTF8StringSerializerDeserializer.INSTANCE, UTF8StringSerializerDeserializer.INSTANCE };
 		RecordDescriptor keyRecDesc = new RecordDescriptor(keyRecDescSers);
 		
-		// build low key
-		tb.reset();
-		UTF8StringSerializerDeserializer.INSTANCE.serialize("100", dos);
-    	tb.addFieldEndOffset();    	  
-    	        	    	
-    	appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
-    	    	
-    	// build high key
-    	tb.reset();
-    	UTF8StringSerializerDeserializer.INSTANCE.serialize("200", dos);
-    	tb.addFieldEndOffset();
-    	
-    	appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
-    	
-    	// build search key factories
-    	ITupleReferenceFactory[] searchKeys = new ITupleReferenceFactory[2]; 
-    	searchKeys[0] = new FrameTupleReferenceFactory(keyFrame.array(), 0, keyRecDesc);
-    	searchKeys[1] = new FrameTupleReferenceFactory(keyFrame.array(), 1, keyRecDesc);
+    	ConstantTupleSourceOperatorDescriptor keyProviderOp = new ConstantTupleSourceOperatorDescriptor(spec, keyRecDesc, tb.getFieldEndOffsets(), tb.getByteArray(), tb.getSize());
+		PartitionConstraint keyProviderPartitionConstraint = new ExplicitPartitionConstraint(new LocationConstraint[] { new AbsoluteLocationConstraint(NC1_ID) });
+		keyProviderOp.setPartitionConstraint(keyProviderPartitionConstraint);
     	    	    	
 		IBufferCacheProvider bufferCacheProvider = new BufferCacheProvider();
 		IBTreeRegistryProvider btreeRegistryProvider = new BTreeRegistryProvider();		
 		IFileMappingProviderProvider fileMappingProviderProvider = new FileMappingProviderProvider();
 		
 		RecordDescriptor recDesc = new RecordDescriptor(
-                new ISerializerDeserializer[] { UTF8StringSerializerDeserializer.INSTANCE, UTF8StringSerializerDeserializer.INSTANCE, UTF8StringSerializerDeserializer.INSTANCE });
-		
-		// TODO: hacky, manually adding file
-		//String ncDataPath = System.getProperty("NodeControllerDataPath");
-        //String fileName = ncDataPath + "btreetest.bin";		
-		//int blubb = fileMappingProviderProvider.getFileMappingProvider().mapNameToFileId(fileName, true);
-		
-		BTreeSearchOperatorDescriptor btreeSearchOp = new BTreeSearchOperatorDescriptor(spec, recDesc, bufferCacheProvider, btreeRegistryProvider, "btreetest.bin", fileMappingProviderProvider, interiorFrameFactory, leafFrameFactory, fieldCount, comparatorFactories, true, searchKeys, comparatorFactories.length);
+                new ISerializerDeserializer[] { UTF8StringSerializerDeserializer.INSTANCE, UTF8StringSerializerDeserializer.INSTANCE, UTF8StringSerializerDeserializer.INSTANCE });		
+				
+		BTreeSearchOperatorDescriptor btreeSearchOp = new BTreeSearchOperatorDescriptor(spec, recDesc, bufferCacheProvider, btreeRegistryProvider, "btreetest.bin", fileMappingProviderProvider, interiorFrameFactory, leafFrameFactory, fieldCount, 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) });
@@ -244,9 +219,9 @@
 		PrinterOperatorDescriptor printer = new PrinterOperatorDescriptor(spec);
 		PartitionConstraint printerPartitionConstraint = new ExplicitPartitionConstraint(new LocationConstraint[] { new AbsoluteLocationConstraint(NC1_ID) });
         printer.setPartitionConstraint(printerPartitionConstraint);
-        
-        IConnectorDescriptor conn = new OneToOneConnectorDescriptor(spec);
-        spec.connect(conn, btreeSearchOp, 0, printer, 0);
+                
+        spec.connect(new OneToOneConnectorDescriptor(spec), keyProviderOp, 0, btreeSearchOp, 0);
+        spec.connect(new OneToOneConnectorDescriptor(spec), btreeSearchOp, 0, printer, 0);
         
         spec.addRoot(printer);
         runTest(spec);
@@ -255,7 +230,7 @@
 	@Test
 	public void insertTest() throws Exception {
 		// relies on the fact that NCs are run from same process
-		System.setProperty("NodeControllerDataPath", "/tmp/");
+		System.setProperty("NodeControllerDataPath", System.getProperty("java.io.tmpdir") + "/");
 		
 		JobSpecification spec = new JobSpecification();
 		
@@ -450,5 +425,5 @@
         	scanCursorC.close();
         }        
         System.out.println();
-	}	
+	}
 }
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 fa3a748..03f60a0 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
@@ -19,10 +19,8 @@
 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;
-import edu.uci.ics.hyracks.dataflow.std.file.IFileSplitProvider;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeInteriorFrameFactory;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeLeafFrameFactory;
-import edu.uci.ics.hyracks.storage.common.file.IFileMappingProvider;
 
 public abstract class AbstractBTreeOperatorDescriptor extends AbstractSingleActivityOperatorDescriptor {
 	
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeInsertUpdateDeleteOperatorNodePushable.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeInsertUpdateDeleteOperatorNodePushable.java
index 5aa239b..e1fedda 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeInsertUpdateDeleteOperatorNodePushable.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeInsertUpdateDeleteOperatorNodePushable.java
@@ -43,6 +43,8 @@
 
     private PermutingFrameTupleReference tuple = new PermutingFrameTupleReference();
 
+    private ByteBuffer writeBuffer;
+    
     public BTreeInsertUpdateDeleteOperatorNodePushable(AbstractBTreeOperatorDescriptor opDesc, IHyracksContext ctx,
             int[] fieldPermutation, IRecordDescriptorProvider recordDescProvider, BTreeOp op) {
         btreeOpHelper = new BTreeOpHelper(opDesc, ctx, false);
@@ -93,7 +95,8 @@
         }
 
         // pass a copy of the frame to next op
-        FrameUtils.flushFrame(buffer.duplicate(), writer);
+        System.arraycopy(buffer.array(), 0, writeBuffer.array(), 0, buffer.capacity());
+        FrameUtils.flushFrame(writeBuffer, writer);
     }
 
     @Override
@@ -101,6 +104,7 @@
         AbstractBTreeOperatorDescriptor opDesc = btreeOpHelper.getOperatorDescriptor();
         RecordDescriptor recDesc = recordDescProvider.getInputRecordDescriptor(opDesc.getOperatorId(), 0);
         accessor = new FrameTupleAccessor(btreeOpHelper.getHyracksContext(), recDesc);
+        writeBuffer = btreeOpHelper.getHyracksContext().getResourceManager().allocateFrame();
         try {
             btreeOpHelper.init();
             btreeOpHelper.getBTree().open(btreeOpHelper.getBTreeFileId());
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 1e4ffee..b75b3a9 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
@@ -30,19 +30,19 @@
 	private static final long serialVersionUID = 1L;
 
 	private boolean isForward;
-	private ITupleReferenceFactory[] searchKeys; // create tuples for low and high keys	
-	private int searchKeyFieldCount;
+	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, String btreeFileName, IFileMappingProviderProvider fileMappingProviderProvider, IBTreeInteriorFrameFactory interiorFactory, IBTreeLeafFrameFactory leafFactory, int fieldCount, IBinaryComparatorFactory[] comparatorFactories, boolean isForward, ITupleReferenceFactory[] searchKeys, int searchKeyFields) {
-		super(spec, 0, 1, recDesc, bufferCacheProvider, btreeRegistryProvider, btreeFileName, fileMappingProviderProvider, interiorFactory, leafFactory, fieldCount, comparatorFactories);
+	public BTreeSearchOperatorDescriptor(JobSpecification spec, RecordDescriptor recDesc, IBufferCacheProvider bufferCacheProvider, IBTreeRegistryProvider btreeRegistryProvider, String btreeFileName, IFileMappingProviderProvider fileMappingProviderProvider, IBTreeInteriorFrameFactory interiorFactory, IBTreeLeafFrameFactory leafFactory, int fieldCount, IBinaryComparatorFactory[] comparatorFactories, boolean isForward, int[] lowKeyFields, int[] highKeyFields) {		
+		super(spec, 1, 1, recDesc, bufferCacheProvider, btreeRegistryProvider, btreeFileName, fileMappingProviderProvider, interiorFactory, leafFactory, fieldCount, comparatorFactories);
 		this.isForward = isForward;
-		this.searchKeys = searchKeys;
-		this.searchKeyFieldCount = searchKeyFields;
+		this.lowKeyFields = lowKeyFields;
+		this.highKeyFields = highKeyFields;
 	}
 	
 	@Override
 	public IOperatorNodePushable createPushRuntime(final IHyracksContext ctx, final IOperatorEnvironment env,
 			IRecordDescriptorProvider recordDescProvider, int partition, int nPartitions) {
-		return new BTreeSearchOperatorNodePushable(this, ctx, isForward, searchKeys, searchKeyFieldCount);
+		return new BTreeSearchOperatorNodePushable(this, ctx, recordDescProvider, isForward, lowKeyFields, 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 5f5d916..80b04ac 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
@@ -19,12 +19,15 @@
 
 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.IRecordDescriptorProvider;
+import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
 import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
 import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
+import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAccessor;
 import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAppender;
 import edu.uci.ics.hyracks.dataflow.common.comm.util.FrameUtils;
 import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
-import edu.uci.ics.hyracks.dataflow.std.base.AbstractUnaryOutputSourceOperatorNodePushable;
+import edu.uci.ics.hyracks.dataflow.std.base.AbstractUnaryInputUnaryOutputOperatorNodePushable;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeCursor;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeInteriorFrame;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeLeafFrame;
@@ -33,96 +36,139 @@
 import edu.uci.ics.hyracks.storage.am.btree.impls.RangePredicate;
 import edu.uci.ics.hyracks.storage.am.btree.impls.RangeSearchCursor;
 
-public class BTreeSearchOperatorNodePushable extends AbstractUnaryOutputSourceOperatorNodePushable {
-    private BTreeOpHelper btreeOpHelper;
-    private boolean isForward;
-    private ITupleReferenceFactory[] searchKeys;
-    private int searchKeyFieldCount;
-
-    public BTreeSearchOperatorNodePushable(AbstractBTreeOperatorDescriptor opDesc, IHyracksContext ctx,
-            boolean isForward, ITupleReferenceFactory[] searchKeys, int searchKeyFields) {
+public class BTreeSearchOperatorNodePushable extends AbstractUnaryInputUnaryOutputOperatorNodePushable {
+	private BTreeOpHelper btreeOpHelper;
+    private boolean isForward;    
+	private FrameTupleAccessor accessor;
+        
+    private ByteBuffer writeBuffer;
+    private FrameTupleAppender appender;
+    private ArrayTupleBuilder tb;
+    private DataOutput dos;
+    
+    private BTree btree;    
+    private PermutingFrameTupleReference lowKey; 
+    private PermutingFrameTupleReference highKey;
+    private RangePredicate rangePred;
+    private MultiComparator searchCmp;
+    private IBTreeCursor cursor;
+    private IBTreeLeafFrame leafFrame;
+    private IBTreeLeafFrame cursorFrame;
+    private IBTreeInteriorFrame interiorFrame;        
+    
+    private RecordDescriptor recDesc;       
+        
+    public BTreeSearchOperatorNodePushable(AbstractBTreeOperatorDescriptor opDesc, IHyracksContext ctx, IRecordDescriptorProvider recordDescProvider, boolean isForward, int[] lowKeyFields, int[] highKeyFields) {
         btreeOpHelper = new BTreeOpHelper(opDesc, ctx, false);
-        this.isForward = isForward;
-        this.searchKeys = searchKeys;
-        this.searchKeyFieldCount = searchKeyFields;
+        this.isForward = isForward;        
+        this.recDesc = recordDescProvider.getInputRecordDescriptor(opDesc.getOperatorId(), 0);        
+        if(lowKeyFields != null && lowKeyFields.length > 0) {
+        	lowKey = new PermutingFrameTupleReference();
+        	lowKey.setFieldPermutation(lowKeyFields);
+        }
+        if(highKeyFields != null && highKeyFields.length > 0) {
+        	highKey = new PermutingFrameTupleReference();
+        	highKey.setFieldPermutation(highKeyFields);
+        }
     }
-
+    
     @Override
-    public void initialize() throws HyracksDataException {
-        AbstractBTreeOperatorDescriptor opDesc = btreeOpHelper.getOperatorDescriptor();
-        IHyracksContext ctx = btreeOpHelper.getHyracksContext();
-
-        IBTreeLeafFrame cursorFrame = opDesc.getLeafFactory().getFrame();
-        IBTreeCursor cursor = new RangeSearchCursor(cursorFrame);
-
-        BTree btree = null;
+	public void open() throws HyracksDataException {
+		AbstractBTreeOperatorDescriptor opDesc = btreeOpHelper.getOperatorDescriptor();          
+        accessor = new FrameTupleAccessor(btreeOpHelper.getHyracksContext(), recDesc);
+        
+        cursorFrame = opDesc.getLeafFactory().getFrame();
+        cursor = new RangeSearchCursor(cursorFrame);
+        
         try {
-            btreeOpHelper.init();
-            //btreeOpHelper.fill();
-            btree = btreeOpHelper.getBTree();
+			btreeOpHelper.init();
+		} catch (Exception e) {
+			throw new HyracksDataException(e.getMessage());
+		}
+        btree = btreeOpHelper.getBTree();
 
-            IBTreeLeafFrame leafFrame = btreeOpHelper.getLeafFrame();
-            IBTreeInteriorFrame interiorFrame = btreeOpHelper.getInteriorFrame();
-
-            // construct range predicate
-            assert (searchKeyFieldCount <= btree.getMultiComparator().getKeyFieldCount());
-            IBinaryComparator[] searchComparators = new IBinaryComparator[searchKeyFieldCount];
-            for (int i = 0; i < searchKeyFieldCount; i++) {
-                searchComparators[i] = btree.getMultiComparator().getComparators()[i];
-            }
-            MultiComparator searchCmp = new MultiComparator(btree.getMultiComparator().getFieldCount(),
-                    searchComparators);
-
-            ITupleReference lowKey = searchKeys[0].createTuple(ctx);
-            ITupleReference highKey = searchKeys[1].createTuple(ctx);
-
-            RangePredicate rangePred = new RangePredicate(isForward, lowKey, highKey, searchCmp);
-
-            btree.search(cursor, rangePred, leafFrame, interiorFrame);
-        } catch (Exception e) {
-            throw new HyracksDataException(e);
+        leafFrame = btreeOpHelper.getLeafFrame();
+        interiorFrame = btreeOpHelper.getInteriorFrame();
+        
+        // construct range predicate
+        
+        int numSearchFields = btree.getMultiComparator().getComparators().length;
+        if(lowKey != null) numSearchFields = lowKey.getFieldCount();        
+        IBinaryComparator[] searchComparators = new IBinaryComparator[numSearchFields];
+        for (int i = 0; i < numSearchFields; i++) {
+        	searchComparators[i] = btree.getMultiComparator().getComparators()[i];
         }
-
-        MultiComparator cmp = btree.getMultiComparator();
-        ByteBuffer frame = ctx.getResourceManager().allocateFrame();
-        FrameTupleAppender appender = new FrameTupleAppender(ctx);
-        appender.reset(frame, true);
-        ArrayTupleBuilder tb = new ArrayTupleBuilder(cmp.getFieldCount());
-        DataOutput dos = tb.getDataOutput();
-
-        try {
-            while (cursor.hasNext()) {
-                tb.reset();
-                cursor.next();
-
-                ITupleReference frameTuple = cursor.getTuple();
-                for (int i = 0; i < frameTuple.getFieldCount(); i++) {
-                    dos.write(frameTuple.getFieldData(i), frameTuple.getFieldStart(i), frameTuple.getFieldLength(i));
-                    tb.addFieldEndOffset();
-                }
-
-                if (!appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize())) {
-                    FrameUtils.flushFrame(frame, writer);
-                    appender.reset(frame, true);
-                    if (!appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize())) {
-                        throw new IllegalStateException();
-                    }
-                }
-
-                //int recOffset = cursor.getOffset();                
-                //String rec = cmp.printRecord(array, recOffset);
-                //System.out.println(rec);
-            }
-
-            if (appender.getTupleCount() > 0) {
-                FrameUtils.flushFrame(frame, writer);
-            }
-
-            cursor.close();
-            writer.close();
-
-        } catch (Exception e) {
-            throw new HyracksDataException(e);
-        }
+        searchCmp = new MultiComparator(btree.getMultiComparator().getFieldCount(), searchComparators);
+        
+        rangePred = new RangePredicate(isForward, null, null, searchCmp);
+                
+        accessor = new FrameTupleAccessor(btreeOpHelper.getHyracksContext(), recDesc);
+        
+        writeBuffer = btreeOpHelper.getHyracksContext().getResourceManager().allocateFrame();
+        tb = new ArrayTupleBuilder(btree.getMultiComparator().getFieldCount());
+        dos = tb.getDataOutput();
+        appender = new FrameTupleAppender(btreeOpHelper.getHyracksContext());                
     }
+    	
+    private void writeSearchResults() throws Exception {
+    	while (cursor.hasNext()) {
+    		tb.reset();
+    		cursor.next();
+
+    		ITupleReference frameTuple = cursor.getTuple();
+    		for (int i = 0; i < frameTuple.getFieldCount(); i++) {
+    			dos.write(frameTuple.getFieldData(i), frameTuple.getFieldStart(i), frameTuple.getFieldLength(i));
+    			tb.addFieldEndOffset();
+    		}
+
+    		if (!appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize())) {
+    			FrameUtils.flushFrame(writeBuffer, writer);
+    			appender.reset(writeBuffer, true);
+    			if (!appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize())) {
+    				throw new IllegalStateException();
+    			}
+    		}             
+    	}    	
+    }
+    
+	@Override
+	public void nextFrame(ByteBuffer buffer) throws HyracksDataException {
+		accessor.reset(buffer);
+		
+        int tupleCount = accessor.getTupleCount();        
+        try {        	
+        	for (int i = 0; i < tupleCount; i++) {
+                if(lowKey != null) lowKey.reset(accessor, i);
+                if(highKey != null) highKey.reset(accessor, i);
+                rangePred.setLowKey(lowKey);
+                rangePred.setHighKey(highKey);
+                
+                cursor.reset();
+                btree.search(cursor, rangePred, leafFrame, interiorFrame);
+                appender.reset(writeBuffer, true);
+                writeSearchResults();                                                         
+            }
+        	
+        	if (appender.getTupleCount() > 0) {
+    			FrameUtils.flushFrame(writeBuffer, writer);
+    		}
+        	
+        } catch (Exception e) {
+        	throw new HyracksDataException(e.getMessage());
+        }
+	}	
+	
+	@Override
+	public void close() throws HyracksDataException {
+		writer.close();
+		try {
+			cursor.close();
+		} catch (Exception e) {
+			throw new HyracksDataException(e.getMessage());
+		}    	
+	}
+
+	@Override
+	public void flush() throws HyracksDataException {		
+	}	
 }
\ No newline at end of file
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/ConstantTupleSourceOperatorDescriptor.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/ConstantTupleSourceOperatorDescriptor.java
new file mode 100644
index 0000000..9ef860b
--- /dev/null
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/ConstantTupleSourceOperatorDescriptor.java
@@ -0,0 +1,36 @@
+package edu.uci.ics.hyracks.storage.am.btree.dataflow;
+
+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.RecordDescriptor;
+import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
+import edu.uci.ics.hyracks.api.job.IOperatorEnvironment;
+import edu.uci.ics.hyracks.api.job.JobSpecification;
+import edu.uci.ics.hyracks.dataflow.std.base.AbstractSingleActivityOperatorDescriptor;
+
+public class ConstantTupleSourceOperatorDescriptor extends AbstractSingleActivityOperatorDescriptor {
+	
+	private static final long serialVersionUID = 1L;
+	
+	private int[] fieldSlots;
+	private byte[] tupleData;	
+	private int tupleSize;
+	
+	public ConstantTupleSourceOperatorDescriptor(JobSpecification spec, RecordDescriptor recDesc, int[] fieldSlots, byte[] tupleData, int tupleSize) {
+		super(spec, 0, 1);
+		this.tupleData = tupleData;
+		this.fieldSlots = fieldSlots;
+		this.tupleSize = tupleSize;
+		recordDescriptors[0] = recDesc;
+	}
+	
+	@Override
+	public IOperatorNodePushable createPushRuntime(IHyracksContext ctx,
+			IOperatorEnvironment env,
+			IRecordDescriptorProvider recordDescProvider, int partition,
+			int nPartitions) throws HyracksDataException {
+		return new ConstantTupleSourceOperatorNodePushable(ctx, fieldSlots, tupleData, tupleSize);
+	}
+
+}
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/ConstantTupleSourceOperatorNodePushable.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/ConstantTupleSourceOperatorNodePushable.java
new file mode 100644
index 0000000..cbdb713
--- /dev/null
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/ConstantTupleSourceOperatorNodePushable.java
@@ -0,0 +1,37 @@
+package edu.uci.ics.hyracks.storage.am.btree.dataflow;
+
+import java.nio.ByteBuffer;
+
+import edu.uci.ics.hyracks.api.context.IHyracksContext;
+import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
+import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAppender;
+import edu.uci.ics.hyracks.dataflow.common.comm.util.FrameUtils;
+import edu.uci.ics.hyracks.dataflow.std.base.AbstractUnaryOutputSourceOperatorNodePushable;
+
+public class ConstantTupleSourceOperatorNodePushable extends AbstractUnaryOutputSourceOperatorNodePushable {
+	
+	private IHyracksContext ctx;
+	
+	private int[] fieldSlots;
+	private byte[] tupleData;	
+	private int tupleSize;	
+	
+	
+	public ConstantTupleSourceOperatorNodePushable(IHyracksContext ctx, int[] fieldSlots, byte[] tupleData, int tupleSize) {
+		super();
+		this.fieldSlots = fieldSlots;
+		this.tupleData = tupleData;		
+		this.tupleSize = tupleSize;
+		this.ctx = ctx;
+	}
+	
+	@Override
+    public void initialize() throws HyracksDataException {		
+		ByteBuffer writeBuffer = ctx.getResourceManager().allocateFrame();
+		FrameTupleAppender appender = new FrameTupleAppender(ctx);
+		appender.reset(writeBuffer, true);
+		appender.append(fieldSlots, tupleData, 0, tupleSize);		
+		FrameUtils.flushFrame(writeBuffer, writer);
+		writer.close();
+	}
+}
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/FrameTupleReferenceFactory.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/FrameTupleReferenceFactory.java
deleted file mode 100644
index 0702c29..0000000
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/FrameTupleReferenceFactory.java
+++ /dev/null
@@ -1,35 +0,0 @@
-package edu.uci.ics.hyracks.storage.am.btree.dataflow;
-
-import java.nio.ByteBuffer;
-
-import edu.uci.ics.hyracks.api.comm.IFrameTupleAccessor;
-import edu.uci.ics.hyracks.api.context.IHyracksContext;
-import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
-import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAccessor;
-import edu.uci.ics.hyracks.dataflow.common.data.accessors.FrameTupleReference;
-import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
-
-public class FrameTupleReferenceFactory implements ITupleReferenceFactory {
-	
-	private static final long serialVersionUID = 1L;
-
-	private byte[] frame;
-	private int tupleIndex;
-	private RecordDescriptor recDesc;
-	
-	public FrameTupleReferenceFactory(byte[] frame, int tupleIndex, RecordDescriptor recDesc) {
-		this.frame = frame;
-		this.tupleIndex = tupleIndex;		
-		this.recDesc = recDesc;
-	}
-	
-	// TODO: lots of object creation, fix later
-	@Override
-	public ITupleReference createTuple(IHyracksContext ctx) {				
-		IFrameTupleAccessor accessor = new FrameTupleAccessor(ctx, recDesc);
-		accessor.reset(ByteBuffer.wrap(frame));
-		FrameTupleReference tuple = new FrameTupleReference();
-		tuple.reset(accessor, tupleIndex);
-		return tuple;
-	}
-}
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/ITupleReferenceFactory.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/ITupleReferenceFactory.java
deleted file mode 100644
index a370f84..0000000
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/ITupleReferenceFactory.java
+++ /dev/null
@@ -1,10 +0,0 @@
-package edu.uci.ics.hyracks.storage.am.btree.dataflow;
-
-import java.io.Serializable;
-
-import edu.uci.ics.hyracks.api.context.IHyracksContext;
-import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
-
-public interface ITupleReferenceFactory extends Serializable {
-	ITupleReference createTuple(IHyracksContext ctx);
-}
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/NullTupleReferenceFactory.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/NullTupleReferenceFactory.java
deleted file mode 100644
index f2163a1..0000000
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/NullTupleReferenceFactory.java
+++ /dev/null
@@ -1,15 +0,0 @@
-package edu.uci.ics.hyracks.storage.am.btree.dataflow;
-
-import edu.uci.ics.hyracks.api.context.IHyracksContext;
-import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
-
-public class NullTupleReferenceFactory implements ITupleReferenceFactory {
-	
-	private static final long serialVersionUID = 1L;
-
-	@Override
-	public ITupleReference createTuple(IHyracksContext ctx) {
-		return null;
-	}
-
-}
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/RangePredicate.java b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/RangePredicate.java
index 74f385c..02928be 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/RangePredicate.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/RangePredicate.java
@@ -59,4 +59,12 @@
 	public ITupleReference getHighKey() {
 		return highKey;
 	}
+	
+	public void setLowKey(ITupleReference lowKey) {
+		this.lowKey = lowKey;
+	}
+	
+	public void setHighKey(ITupleReference highKey) {
+		this.highKey = highKey;
+	}
 }
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 77ef504..2746392 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
@@ -62,6 +62,8 @@
     private static final int NUM_PAGES = 40;
     private static final int HYRACKS_FRAME_SIZE = 128;
     
+    private String tmpDir = System.getProperty("java.io.tmpdir");
+    
     // to help with the logger madness
     private void print(String str) {
     	System.out.print(str);
@@ -121,7 +123,7 @@
         IPageReplacementStrategy prs = new ClockPageReplacementStrategy();
         IBufferCache bufferCache = new BufferCache(allocator, prs, fileManager, PAGE_SIZE, NUM_PAGES);
         
-        File f = new File("/tmp/btreetest.bin");
+        File f = new File(tmpDir + "/" + "btreetest.bin");
         RandomAccessFile raf = new RandomAccessFile(f, "rw");
         int fileId = 0;
         FileInfo fi = new FileInfo(fileId, raf);
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 9d4f390..b4ed2d4 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
@@ -72,6 +72,8 @@
     private static final int NUM_PAGES = 10;
     private static final int HYRACKS_FRAME_SIZE = 128;
     
+    private String tmpDir = System.getProperty("java.io.tmpdir");
+    
     // to help with the logger madness
     private void print(String str) {
     	System.out.print(str);
@@ -106,7 +108,7 @@
         IPageReplacementStrategy prs = new ClockPageReplacementStrategy();
         IBufferCache bufferCache = new BufferCache(allocator, prs, fileManager, PAGE_SIZE, NUM_PAGES);
         
-        File f = new File("/tmp/btreetest.bin");
+        File f = new File(tmpDir + "/" + "btreetest.bin");
         RandomAccessFile raf = new RandomAccessFile(f, "rw");
         int fileId = 0;
         FileInfo fi = new FileInfo(fileId, raf);
@@ -311,7 +313,7 @@
         IPageReplacementStrategy prs = new ClockPageReplacementStrategy();
         IBufferCache bufferCache = new BufferCache(allocator, prs, fileManager, PAGE_SIZE, NUM_PAGES);
 
-        File f = new File("/tmp/btreetest.bin");
+        File f = new File(tmpDir + "/" + "btreetest.bin");
         RandomAccessFile raf = new RandomAccessFile(f, "rw");
         int fileId = 0;
         FileInfo fi = new FileInfo(fileId, raf);
@@ -487,7 +489,7 @@
     	IPageReplacementStrategy prs = new ClockPageReplacementStrategy();
     	IBufferCache bufferCache = new BufferCache(allocator, prs, fileManager, PAGE_SIZE, NUM_PAGES);
 
-    	File f = new File("/tmp/btreetest.bin");
+    	File f = new File(tmpDir + "/" + "btreetest.bin");
     	RandomAccessFile raf = new RandomAccessFile(f, "rw");
     	int fileId = 0;
     	FileInfo fi = new FileInfo(fileId, raf);
@@ -654,7 +656,7 @@
         IPageReplacementStrategy prs = new ClockPageReplacementStrategy();
         IBufferCache bufferCache = new BufferCache(allocator, prs, fileManager, PAGE_SIZE, NUM_PAGES);
 
-        File f = new File("/tmp/btreetest.bin");
+        File f = new File(tmpDir + "/" + "btreetest.bin");
         RandomAccessFile raf = new RandomAccessFile(f, "rw");
         int fileId = 0;
         FileInfo fi = new FileInfo(fileId, raf);
@@ -808,7 +810,7 @@
         IPageReplacementStrategy prs = new ClockPageReplacementStrategy();
         IBufferCache bufferCache = new BufferCache(allocator, prs, fileManager, PAGE_SIZE, NUM_PAGES);
 
-        File f = new File("/tmp/btreetest.bin");
+        File f = new File(tmpDir + "/" + "btreetest.bin");
         RandomAccessFile raf = new RandomAccessFile(f, "rw");
         int fileId = 0;
         FileInfo fi = new FileInfo(fileId, raf);
@@ -957,7 +959,7 @@
         IPageReplacementStrategy prs = new ClockPageReplacementStrategy();
         IBufferCache bufferCache = new BufferCache(allocator, prs, fileManager, PAGE_SIZE, NUM_PAGES);
 
-        File f = new File("/tmp/btreetest.bin");
+        File f = new File(tmpDir + "/" + "btreetest.bin");
         RandomAccessFile raf = new RandomAccessFile(f, "rw");
         int fileId = 0;
         FileInfo fi = new FileInfo(fileId, raf);