diff --git a/hyracks-examples/hyracks-integration-tests/pom.xml b/hyracks-examples/hyracks-integration-tests/pom.xml
index 8cf1db7..882c7da 100644
--- a/hyracks-examples/hyracks-integration-tests/pom.xml
+++ b/hyracks-examples/hyracks-integration-tests/pom.xml
@@ -2,12 +2,12 @@
   <modelVersion>4.0.0</modelVersion>
   <groupId>edu.uci.ics.hyracks.examples</groupId>
   <artifactId>hyracks-integration-tests</artifactId>
-  <version>0.1.4-SNAPSHOT</version>
+  <version>0.1.5-SNAPSHOT</version>
 
   <parent>
     <groupId>edu.uci.ics.hyracks</groupId>
     <artifactId>hyracks-examples</artifactId>
-    <version>0.1.4-SNAPSHOT</version>
+    <version>0.1.5-SNAPSHOT</version>
   </parent>
 
   <build>
@@ -42,42 +42,42 @@
   	<dependency>
   		<groupId>edu.uci.ics.hyracks</groupId>
   		<artifactId>hyracks-dataflow-std</artifactId>
-  		<version>0.1.4-SNAPSHOT</version>
+  		<version>0.1.5-SNAPSHOT</version>
   		<type>jar</type>
   		<scope>compile</scope>
   	</dependency>
   	<dependency>
   		<groupId>edu.uci.ics.hyracks</groupId>
   		<artifactId>hyracks-control-cc</artifactId>
-  		<version>0.1.4-SNAPSHOT</version>
+  		<version>0.1.5-SNAPSHOT</version>
   		<type>jar</type>
   		<scope>compile</scope>
   	</dependency>
   	<dependency>
   		<groupId>edu.uci.ics.hyracks</groupId>
   		<artifactId>hyracks-control-nc</artifactId>
-  		<version>0.1.4-SNAPSHOT</version>
+  		<version>0.1.5-SNAPSHOT</version>
   		<type>jar</type>
   		<scope>compile</scope>
   	</dependency>
   	<dependency>
   		<groupId>edu.uci.ics.hyracks</groupId>
   		<artifactId>hyracks-storage-am-btree</artifactId>
-  		<version>0.1.4-SNAPSHOT</version>
+  		<version>0.1.5-SNAPSHOT</version>
   		<type>jar</type>
   		<scope>compile</scope>
   	</dependency>
   	<dependency>
   		<groupId>edu.uci.ics.hyracks</groupId>
   		<artifactId>hyracks-storage-am-invertedindex</artifactId>
-  		<version>0.1.4-SNAPSHOT</version>
+  		<version>0.1.5-SNAPSHOT</version>
   		<type>jar</type>
   		<scope>compile</scope>
   	</dependency>
   	<dependency>
   		<groupId>edu.uci.ics.hyracks</groupId>
   		<artifactId>hyracks-test-support</artifactId>
-  		<version>0.1.4-SNAPSHOT</version>
+  		<version>0.1.5-SNAPSHOT</version>
   		<type>jar</type>
   		<scope>test</scope>
   	</dependency>
diff --git a/hyracks-examples/hyracks-integration-tests/src/test/java/edu/uci/ics/hyracks/tests/btree/BTreeOperatorsTest.java b/hyracks-examples/hyracks-integration-tests/src/test/java/edu/uci/ics/hyracks/tests/btree/BTreeOperatorsTest.java
index 33aebc1..3109c33 100644
--- a/hyracks-examples/hyracks-integration-tests/src/test/java/edu/uci/ics/hyracks/tests/btree/BTreeOperatorsTest.java
+++ b/hyracks-examples/hyracks-integration-tests/src/test/java/edu/uci/ics/hyracks/tests/btree/BTreeOperatorsTest.java
@@ -17,11 +17,14 @@
 
 import java.io.DataOutput;
 import java.io.File;
+import java.text.SimpleDateFormat;
+import java.util.Date;
 
+import org.junit.AfterClass;
+import org.junit.Before;
 import org.junit.Test;
 
 import edu.uci.ics.hyracks.api.constraints.PartitionConstraintHelper;
-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;
@@ -30,7 +33,6 @@
 import edu.uci.ics.hyracks.api.io.FileReference;
 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;
 import edu.uci.ics.hyracks.dataflow.common.data.comparators.UTF8StringBinaryComparatorFactory;
 import edu.uci.ics.hyracks.dataflow.common.data.marshalling.UTF8StringSerializerDeserializer;
 import edu.uci.ics.hyracks.dataflow.common.data.parsers.IValueParserFactory;
@@ -44,454 +46,568 @@
 import edu.uci.ics.hyracks.dataflow.std.misc.ConstantTupleSourceOperatorDescriptor;
 import edu.uci.ics.hyracks.dataflow.std.misc.NullSinkOperatorDescriptor;
 import edu.uci.ics.hyracks.dataflow.std.misc.PrinterOperatorDescriptor;
-import edu.uci.ics.hyracks.dataflow.std.sort.InMemorySortOperatorDescriptor;
-import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeCursor;
+import edu.uci.ics.hyracks.dataflow.std.sort.ExternalSortOperatorDescriptor;
 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.BTreeBulkLoadOperatorDescriptor;
 import edu.uci.ics.hyracks.storage.am.btree.dataflow.BTreeInsertUpdateDeleteOperatorDescriptor;
-import edu.uci.ics.hyracks.storage.am.btree.dataflow.BTreeRegistry;
 import edu.uci.ics.hyracks.storage.am.btree.dataflow.BTreeSearchOperatorDescriptor;
 import edu.uci.ics.hyracks.storage.am.btree.dataflow.IBTreeRegistryProvider;
-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;
-import edu.uci.ics.hyracks.storage.am.btree.impls.BTree;
 import edu.uci.ics.hyracks.storage.am.btree.impls.BTreeOp;
-import edu.uci.ics.hyracks.storage.am.btree.impls.BTreeOpContext;
-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.IStorageManagerInterface;
-import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
-import edu.uci.ics.hyracks.storage.common.file.IFileMapProvider;
 import edu.uci.ics.hyracks.test.support.TestBTreeRegistryProvider;
 import edu.uci.ics.hyracks.test.support.TestStorageManagerComponentHolder;
 import edu.uci.ics.hyracks.test.support.TestStorageManagerInterface;
 import edu.uci.ics.hyracks.tests.integration.AbstractIntegrationTest;
 
 public class BTreeOperatorsTest extends AbstractIntegrationTest {
-    static {
-        TestStorageManagerComponentHolder.init(8192, 20);
-    }
+	static {
+		TestStorageManagerComponentHolder.init(8192, 20);
+	}
 
-    @Test
-    public void bulkLoadTest() throws Exception {
-        JobSpecification spec = new JobSpecification();
+	private IStorageManagerInterface storageManager = new TestStorageManagerInterface();
+	private IBTreeRegistryProvider btreeRegistryProvider = new TestBTreeRegistryProvider();
 
-        FileSplit[] ordersSplits = new FileSplit[] { new FileSplit(NC1_ID, new FileReference(new File(
-                "data/tpch0.001/orders-part1.tbl"))) };
-        IFileSplitProvider ordersSplitProvider = new ConstantFileSplitProvider(ordersSplits);
-        RecordDescriptor ordersDesc = new RecordDescriptor(new ISerializerDeserializer[] {
-                UTF8StringSerializerDeserializer.INSTANCE, UTF8StringSerializerDeserializer.INSTANCE,
-                UTF8StringSerializerDeserializer.INSTANCE, UTF8StringSerializerDeserializer.INSTANCE,
-                UTF8StringSerializerDeserializer.INSTANCE, UTF8StringSerializerDeserializer.INSTANCE,
-                UTF8StringSerializerDeserializer.INSTANCE, UTF8StringSerializerDeserializer.INSTANCE,
-                UTF8StringSerializerDeserializer.INSTANCE });
+	private final static SimpleDateFormat simpleDateFormat = new SimpleDateFormat(
+			"ddMMyy-hhmmssSS");
+	private final static String sep = System.getProperty("file.separator");
+	
+	// field, type and key declarations for primary index
+	private int primaryFieldCount = 6;
+	private ITypeTrait[] primaryTypeTraits = new ITypeTrait[primaryFieldCount];
+	private int primaryKeyFieldCount = 1;
+	private IBinaryComparatorFactory[] primaryComparatorFactories = new IBinaryComparatorFactory[primaryKeyFieldCount];
+	private TypeAwareTupleWriterFactory primaryTupleWriterFactory = new TypeAwareTupleWriterFactory(
+			primaryTypeTraits);
+	private IBTreeInteriorFrameFactory primaryInteriorFrameFactory = new NSMInteriorFrameFactory(
+			primaryTupleWriterFactory);
+	private IBTreeLeafFrameFactory primaryLeafFrameFactory = new NSMLeafFrameFactory(
+			primaryTupleWriterFactory);
 
-        FileScanOperatorDescriptor ordScanner = new FileScanOperatorDescriptor(spec, ordersSplitProvider,
-                new DelimitedDataTupleParserFactory(new IValueParserFactory[] { UTF8StringParserFactory.INSTANCE,
-                        UTF8StringParserFactory.INSTANCE, UTF8StringParserFactory.INSTANCE,
-                        UTF8StringParserFactory.INSTANCE, UTF8StringParserFactory.INSTANCE,
-                        UTF8StringParserFactory.INSTANCE, UTF8StringParserFactory.INSTANCE,
-                        UTF8StringParserFactory.INSTANCE, UTF8StringParserFactory.INSTANCE }, '|'), ordersDesc);
-        PartitionConstraintHelper.addAbsoluteLocationConstraint(spec, ordScanner, NC1_ID);
+	private static String primaryBtreeName = "primary"
+			+ simpleDateFormat.format(new Date());
+	private static String primaryFileName = System
+			.getProperty("java.io.tmpdir")
+			+ sep + "nc1" + sep + primaryBtreeName;
 
-        InMemorySortOperatorDescriptor sorter = new InMemorySortOperatorDescriptor(spec, new int[] { 0 },
-                new IBinaryComparatorFactory[] { UTF8StringBinaryComparatorFactory.INSTANCE }, ordersDesc);
-        PartitionConstraintHelper.addAbsoluteLocationConstraint(spec, sorter, NC1_ID);
+	private IFileSplitProvider primaryBtreeSplitProvider = new ConstantFileSplitProvider(
+			new FileSplit[] { new FileSplit(NC1_ID, new FileReference(new File(
+					primaryFileName))) });
 
-        // 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);
+	private RecordDescriptor primaryRecDesc = new RecordDescriptor(
+			new ISerializerDeserializer[] {
+					UTF8StringSerializerDeserializer.INSTANCE,
+					UTF8StringSerializerDeserializer.INSTANCE,
+					UTF8StringSerializerDeserializer.INSTANCE,
+					UTF8StringSerializerDeserializer.INSTANCE,
+					UTF8StringSerializerDeserializer.INSTANCE,
+					UTF8StringSerializerDeserializer.INSTANCE });
 
-        // declare keys
-        int keyFieldCount = 1;
-        IBinaryComparatorFactory[] comparatorFactories = new IBinaryComparatorFactory[keyFieldCount];
-        comparatorFactories[0] = UTF8StringBinaryComparatorFactory.INSTANCE;
+	// field, type and key declarations for secondary indexes
+	private int secondaryFieldCount = 2;
+	private ITypeTrait[] secondaryTypeTraits = new ITypeTrait[secondaryFieldCount];
+	private int secondaryKeyFieldCount = 2;
+	private IBinaryComparatorFactory[] secondaryComparatorFactories = new IBinaryComparatorFactory[secondaryKeyFieldCount];
+	private TypeAwareTupleWriterFactory secondaryTupleWriterFactory = new TypeAwareTupleWriterFactory(
+			secondaryTypeTraits);
+	private IBTreeInteriorFrameFactory secondaryInteriorFrameFactory = new NSMInteriorFrameFactory(
+			secondaryTupleWriterFactory);
+	private IBTreeLeafFrameFactory secondaryLeafFrameFactory = new NSMLeafFrameFactory(
+			secondaryTupleWriterFactory);
 
-        TypeAwareTupleWriterFactory tupleWriterFactory = new TypeAwareTupleWriterFactory(typeTraits);
-        // SimpleTupleWriterFactory tupleWriterFactory = new
-        // SimpleTupleWriterFactory();
-        IBTreeInteriorFrameFactory interiorFrameFactory = new NSMInteriorFrameFactory(tupleWriterFactory);
-        IBTreeLeafFrameFactory leafFrameFactory = new NSMLeafFrameFactory(tupleWriterFactory);
-        IBTreeRegistryProvider btreeRegistryProvider = new TestBTreeRegistryProvider();
+	private static String secondaryBtreeName = "secondary"
+			+ simpleDateFormat.format(new Date());
+	private static String secondaryFileName = System
+			.getProperty("java.io.tmpdir")
+			+ sep + "nc1" + sep + secondaryBtreeName;
 
-        int[] fieldPermutation = { 0, 4, 5 };
-        String btreeName = "btree.bin";
-        FileReference nc1File = new FileReference(new File(System.getProperty("java.io.tmpdir") + "/nc1/" + btreeName));
-        IFileSplitProvider btreeSplitProvider = new ConstantFileSplitProvider(new FileSplit[] { new FileSplit(NC1_ID,
-                nc1File) });
+	private IFileSplitProvider secondaryBtreeSplitProvider = new ConstantFileSplitProvider(
+			new FileSplit[] { new FileSplit(NC1_ID, new FileReference(new File(
+					secondaryFileName))) });
 
-        IStorageManagerInterface storageManager = new TestStorageManagerInterface();
-        BTreeBulkLoadOperatorDescriptor btreeBulkLoad = new BTreeBulkLoadOperatorDescriptor(spec, storageManager,
-                btreeRegistryProvider, btreeSplitProvider, interiorFrameFactory, leafFrameFactory, typeTraits,
-                comparatorFactories, fieldPermutation, 0.7f);
-        PartitionConstraintHelper.addAbsoluteLocationConstraint(spec, btreeBulkLoad, NC1_ID);
+	private RecordDescriptor secondaryRecDesc = new RecordDescriptor(
+			new ISerializerDeserializer[] {
+					UTF8StringSerializerDeserializer.INSTANCE,
+					UTF8StringSerializerDeserializer.INSTANCE });
 
-        spec.connect(new OneToOneConnectorDescriptor(spec), ordScanner, 0, sorter, 0);
+	@Before
+	public void setup() {
+		// field, type and key declarations for primary index
+		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);
+		primaryComparatorFactories[0] = UTF8StringBinaryComparatorFactory.INSTANCE;
 
-        spec.connect(new OneToOneConnectorDescriptor(spec), sorter, 0, btreeBulkLoad, 0);
+		// field, type and key declarations for secondary indexes
+		secondaryTypeTraits[0] = new TypeTrait(ITypeTrait.VARIABLE_LENGTH);
+		secondaryTypeTraits[1] = new TypeTrait(ITypeTrait.VARIABLE_LENGTH);
+		secondaryComparatorFactories[0] = UTF8StringBinaryComparatorFactory.INSTANCE;
+		secondaryComparatorFactories[1] = UTF8StringBinaryComparatorFactory.INSTANCE;
+	}
 
-        spec.addRoot(btreeBulkLoad);
-        runTest(spec);
+	@Test
+	public void loadPrimaryIndexTest() throws Exception {
+		JobSpecification spec = new JobSpecification();
 
-        // 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();
-        }
+		FileSplit[] ordersSplits = new FileSplit[] { new FileSplit(NC1_ID,
+				new FileReference(new File("data/tpch0.001/orders-part1.tbl"))) };
+		IFileSplitProvider ordersSplitProvider = new ConstantFileSplitProvider(
+				ordersSplits);
+		RecordDescriptor ordersDesc = new RecordDescriptor(
+				new ISerializerDeserializer[] {
+						UTF8StringSerializerDeserializer.INSTANCE,
+						UTF8StringSerializerDeserializer.INSTANCE,
+						UTF8StringSerializerDeserializer.INSTANCE,
+						UTF8StringSerializerDeserializer.INSTANCE,
+						UTF8StringSerializerDeserializer.INSTANCE,
+						UTF8StringSerializerDeserializer.INSTANCE,
+						UTF8StringSerializerDeserializer.INSTANCE,
+						UTF8StringSerializerDeserializer.INSTANCE,
+						UTF8StringSerializerDeserializer.INSTANCE });
 
-        MultiComparator cmp = new MultiComparator(typeTraits, comparators);
+		FileScanOperatorDescriptor ordScanner = new FileScanOperatorDescriptor(
+				spec, ordersSplitProvider, new DelimitedDataTupleParserFactory(
+						new IValueParserFactory[] {
+								UTF8StringParserFactory.INSTANCE,
+								UTF8StringParserFactory.INSTANCE,
+								UTF8StringParserFactory.INSTANCE,
+								UTF8StringParserFactory.INSTANCE,
+								UTF8StringParserFactory.INSTANCE,
+								UTF8StringParserFactory.INSTANCE,
+								UTF8StringParserFactory.INSTANCE,
+								UTF8StringParserFactory.INSTANCE,
+								UTF8StringParserFactory.INSTANCE }, '|'),
+				ordersDesc);
+		PartitionConstraintHelper.addAbsoluteLocationConstraint(spec,
+				ordScanner, NC1_ID);
 
-        // try an ordered scan on the bulk-loaded btree
-        int btreeFileId = storageManager.getFileMapProvider(null).lookupFileId(nc1File);
-        storageManager.getBufferCache(null).openFile(btreeFileId);
-        BTree btree = btreeRegistryProvider.getBTreeRegistry(null).get(btreeFileId);
-        IBTreeCursor scanCursor = new RangeSearchCursor(leafFrameFactory.getFrame());
-        RangePredicate nullPred = new RangePredicate(true, null, null, true, true, null, null);
-        BTreeOpContext opCtx = btree.createOpContext(BTreeOp.BTO_SEARCH, leafFrameFactory.getFrame(),
-                interiorFrameFactory.getFrame(), null);
-        btree.search(scanCursor, nullPred, opCtx);
-        try {
-            while (scanCursor.hasNext()) {
-                scanCursor.next();
-                ITupleReference frameTuple = scanCursor.getTuple();
-                String rec = cmp.printTuple(frameTuple, ordersDesc.getFields());
-                System.out.println(rec);
-            }
-        } catch (Exception e) {
-            e.printStackTrace();
-        } finally {
-            scanCursor.close();
-        }
-        storageManager.getBufferCache(null).closeFile(btreeFileId);
-    }
+		ExternalSortOperatorDescriptor sorter = new ExternalSortOperatorDescriptor(
+				spec,
+				1000,
+				new int[] { 0 },
+				new IBinaryComparatorFactory[] { UTF8StringBinaryComparatorFactory.INSTANCE },
+				ordersDesc);
+		PartitionConstraintHelper.addAbsoluteLocationConstraint(spec, sorter,
+				NC1_ID);
 
-    @Test
-    public void btreeSearchTest() throws Exception {
-        JobSpecification spec = new JobSpecification();
+		int[] fieldPermutation = { 0, 1, 2, 4, 5, 7 };
+		BTreeBulkLoadOperatorDescriptor primaryBtreeBulkLoad = new BTreeBulkLoadOperatorDescriptor(
+				spec, storageManager, btreeRegistryProvider,
+				primaryBtreeSplitProvider, primaryInteriorFrameFactory,
+				primaryLeafFrameFactory, primaryTypeTraits,
+				primaryComparatorFactories, fieldPermutation, 0.7f);
+		PartitionConstraintHelper.addAbsoluteLocationConstraint(spec,
+				primaryBtreeBulkLoad, NC1_ID);
 
-        // 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);
+		spec.connect(new OneToOneConnectorDescriptor(spec), ordScanner, 0,
+				sorter, 0);
 
-        // declare keys
-        int keyFieldCount = 1;
-        IBinaryComparatorFactory[] comparatorFactories = new IBinaryComparatorFactory[keyFieldCount];
-        comparatorFactories[0] = UTF8StringBinaryComparatorFactory.INSTANCE;
+		spec.connect(new OneToOneConnectorDescriptor(spec), sorter, 0,
+				primaryBtreeBulkLoad, 0);
 
-        TypeAwareTupleWriterFactory tupleWriterFactory = new TypeAwareTupleWriterFactory(typeTraits);
-        // SimpleTupleWriterFactory tupleWriterFactory = new
-        // SimpleTupleWriterFactory();
-        IBTreeInteriorFrameFactory interiorFrameFactory = new NSMInteriorFrameFactory(tupleWriterFactory);
-        IBTreeLeafFrameFactory leafFrameFactory = new NSMLeafFrameFactory(tupleWriterFactory);
+		spec.addRoot(primaryBtreeBulkLoad);
+		runTest(spec);
+	}
 
-        // 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(typeTraits, comparators);
+	@Test
+	public void scanPrimaryIndexTest() throws Exception {
+		JobSpecification spec = new JobSpecification();
 
-        // build tuple containing low and high search key
-        ArrayTupleBuilder tb = new ArrayTupleBuilder(cmp.getKeyFieldCount() * 2); // high
-                                                                                  // key
-                                                                                  // and
-                                                                                  // low
-                                                                                  // key
-        DataOutput dos = tb.getDataOutput();
+		// build dummy tuple containing nothing
+		ArrayTupleBuilder tb = new ArrayTupleBuilder(primaryKeyFieldCount * 2);
+		DataOutput dos = tb.getDataOutput();
 
-        tb.reset();
-        UTF8StringSerializerDeserializer.INSTANCE.serialize("100", dos); // low
-                                                                         // key
-        tb.addFieldEndOffset();
-        UTF8StringSerializerDeserializer.INSTANCE.serialize("200", dos); // high
-                                                                         // key
-        tb.addFieldEndOffset();
+		tb.reset();
+		UTF8StringSerializerDeserializer.INSTANCE.serialize("0", dos);
+		tb.addFieldEndOffset();
 
-        ISerializerDeserializer[] keyRecDescSers = { UTF8StringSerializerDeserializer.INSTANCE,
-                UTF8StringSerializerDeserializer.INSTANCE };
-        RecordDescriptor keyRecDesc = new RecordDescriptor(keyRecDescSers);
+		ISerializerDeserializer[] keyRecDescSers = {
+				UTF8StringSerializerDeserializer.INSTANCE,
+				UTF8StringSerializerDeserializer.INSTANCE };
+		RecordDescriptor keyRecDesc = new RecordDescriptor(keyRecDescSers);
 
-        ConstantTupleSourceOperatorDescriptor keyProviderOp = new ConstantTupleSourceOperatorDescriptor(spec,
-                keyRecDesc, tb.getFieldEndOffsets(), tb.getByteArray(), tb.getSize());
-        PartitionConstraintHelper.addAbsoluteLocationConstraint(spec, keyProviderOp, NC1_ID);
-        IBTreeRegistryProvider btreeRegistryProvider = new TestBTreeRegistryProvider();
+		ConstantTupleSourceOperatorDescriptor keyProviderOp = new ConstantTupleSourceOperatorDescriptor(
+				spec, keyRecDesc, tb.getFieldEndOffsets(), tb.getByteArray(),
+				tb.getSize());
+		PartitionConstraintHelper.addAbsoluteLocationConstraint(spec,
+				keyProviderOp, NC1_ID);
 
-        RecordDescriptor recDesc = new RecordDescriptor(new ISerializerDeserializer[] {
-                UTF8StringSerializerDeserializer.INSTANCE, UTF8StringSerializerDeserializer.INSTANCE,
-                UTF8StringSerializerDeserializer.INSTANCE });
+		int[] lowKeyFields = null; // - infinity
+		int[] highKeyFields = null; // + infinity
 
-        String btreeName = "btree.bin";
-        FileReference nc1File = new FileReference(new File(System.getProperty("java.io.tmpdir") + "/nc1/" + btreeName));
-        IFileSplitProvider btreeSplitProvider = new ConstantFileSplitProvider(new FileSplit[] { new FileSplit(NC1_ID,
-                nc1File) });
+		BTreeSearchOperatorDescriptor primaryBtreeSearchOp = new BTreeSearchOperatorDescriptor(
+				spec, primaryRecDesc, storageManager, btreeRegistryProvider,
+				primaryBtreeSplitProvider, primaryInteriorFrameFactory,
+				primaryLeafFrameFactory, primaryTypeTraits,
+				primaryComparatorFactories, true, lowKeyFields, highKeyFields,
+				true, true);
+		PartitionConstraintHelper.addAbsoluteLocationConstraint(spec,
+				primaryBtreeSearchOp, NC1_ID);
 
-        IStorageManagerInterface storageManager = new TestStorageManagerInterface();
-        BTreeSearchOperatorDescriptor btreeSearchOp = new BTreeSearchOperatorDescriptor(spec, recDesc, storageManager,
-                btreeRegistryProvider, btreeSplitProvider, interiorFrameFactory, leafFrameFactory, typeTraits,
-                comparatorFactories, true, new int[] { 0 }, new int[] { 1 }, true, true);
-        // BTreeDiskOrderScanOperatorDescriptor btreeSearchOp = new
-        // BTreeDiskOrderScanOperatorDescriptor(spec, splitProvider, recDesc,
-        // bufferCacheProvider, btreeRegistryProvider, 0, "btreetest.bin",
-        // interiorFrameFactory, leafFrameFactory, cmp);
+		PrinterOperatorDescriptor printer = new PrinterOperatorDescriptor(spec);
+		PartitionConstraintHelper.addAbsoluteLocationConstraint(spec, printer,
+				NC1_ID);
 
-        PartitionConstraintHelper.addAbsoluteLocationConstraint(spec, btreeSearchOp, NC1_ID);
+		spec.connect(new OneToOneConnectorDescriptor(spec), keyProviderOp, 0,
+				primaryBtreeSearchOp, 0);
+		spec.connect(new OneToOneConnectorDescriptor(spec),
+				primaryBtreeSearchOp, 0, printer, 0);
 
-        PrinterOperatorDescriptor printer = new PrinterOperatorDescriptor(spec);
-        PartitionConstraintHelper.addAbsoluteLocationConstraint(spec, printer, NC1_ID);
+		spec.addRoot(printer);
+		runTest(spec);
+	}
 
-        spec.connect(new OneToOneConnectorDescriptor(spec), keyProviderOp, 0, btreeSearchOp, 0);
-        spec.connect(new OneToOneConnectorDescriptor(spec), btreeSearchOp, 0, printer, 0);
+	@Test
+	public void searchPrimaryIndexTest() throws Exception {
+		JobSpecification spec = new JobSpecification();
 
-        spec.addRoot(printer);
-        runTest(spec);
-    }
+		// build tuple containing low and high search key
+		// high key and low key
+		ArrayTupleBuilder tb = new ArrayTupleBuilder(primaryKeyFieldCount * 2);
+		DataOutput dos = tb.getDataOutput();
 
-    @Test
-    public void insertTest() throws Exception {
-        // relies on the fact that NCs are run from same process
-        System.setProperty("NodeControllerDataPath", System.getProperty("java.io.tmpdir") + "/");
+		tb.reset();
+		// low key
+		UTF8StringSerializerDeserializer.INSTANCE.serialize("100", dos);
+		tb.addFieldEndOffset();
+		// high key
+		UTF8StringSerializerDeserializer.INSTANCE.serialize("200", dos);
+		tb.addFieldEndOffset();
 
-        JobSpecification spec = new JobSpecification();
+		ISerializerDeserializer[] keyRecDescSers = {
+				UTF8StringSerializerDeserializer.INSTANCE,
+				UTF8StringSerializerDeserializer.INSTANCE };
+		RecordDescriptor keyRecDesc = new RecordDescriptor(keyRecDescSers);
 
-        FileSplit[] ordersSplits = new FileSplit[] { new FileSplit(NC1_ID, new FileReference(new File(
-                "data/tpch0.001/orders-part1.tbl"))) };
-        IFileSplitProvider ordersSplitProvider = new ConstantFileSplitProvider(ordersSplits);
-        RecordDescriptor ordersDesc = new RecordDescriptor(new ISerializerDeserializer[] {
-                UTF8StringSerializerDeserializer.INSTANCE, UTF8StringSerializerDeserializer.INSTANCE,
-                UTF8StringSerializerDeserializer.INSTANCE, UTF8StringSerializerDeserializer.INSTANCE,
-                UTF8StringSerializerDeserializer.INSTANCE, UTF8StringSerializerDeserializer.INSTANCE,
-                UTF8StringSerializerDeserializer.INSTANCE, UTF8StringSerializerDeserializer.INSTANCE,
-                UTF8StringSerializerDeserializer.INSTANCE });
+		ConstantTupleSourceOperatorDescriptor keyProviderOp = new ConstantTupleSourceOperatorDescriptor(
+				spec, keyRecDesc, tb.getFieldEndOffsets(), tb.getByteArray(),
+				tb.getSize());
+		PartitionConstraintHelper.addAbsoluteLocationConstraint(spec,
+				keyProviderOp, NC1_ID);
 
-        FileScanOperatorDescriptor ordScanner = new FileScanOperatorDescriptor(spec, ordersSplitProvider,
-                new DelimitedDataTupleParserFactory(new IValueParserFactory[] { UTF8StringParserFactory.INSTANCE,
-                        UTF8StringParserFactory.INSTANCE, UTF8StringParserFactory.INSTANCE,
-                        UTF8StringParserFactory.INSTANCE, UTF8StringParserFactory.INSTANCE,
-                        UTF8StringParserFactory.INSTANCE, UTF8StringParserFactory.INSTANCE,
-                        UTF8StringParserFactory.INSTANCE, UTF8StringParserFactory.INSTANCE }, '|'), ordersDesc);
-        PartitionConstraintHelper.addAbsoluteLocationConstraint(spec, ordScanner, NC1_ID);
+		int[] lowKeyFields = { 0 };
+		int[] highKeyFields = { 1 };
 
-        // 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);
+		BTreeSearchOperatorDescriptor primaryBtreeSearchOp = new BTreeSearchOperatorDescriptor(
+				spec, primaryRecDesc, storageManager, btreeRegistryProvider,
+				primaryBtreeSplitProvider, primaryInteriorFrameFactory,
+				primaryLeafFrameFactory, primaryTypeTraits,
+				primaryComparatorFactories, true, lowKeyFields, highKeyFields,
+				true, true);
+		PartitionConstraintHelper.addAbsoluteLocationConstraint(spec,
+				primaryBtreeSearchOp, NC1_ID);
 
-        int primaryKeyFieldCount = 1;
-        IBinaryComparatorFactory[] primaryComparatorFactories = new IBinaryComparatorFactory[primaryKeyFieldCount];
-        primaryComparatorFactories[0] = UTF8StringBinaryComparatorFactory.INSTANCE;
+		PrinterOperatorDescriptor printer = new PrinterOperatorDescriptor(spec);
+		PartitionConstraintHelper.addAbsoluteLocationConstraint(spec, printer,
+				NC1_ID);
 
-        TypeAwareTupleWriterFactory primaryTupleWriterFactory = new TypeAwareTupleWriterFactory(primaryTypeTraits);
-        // SimpleTupleWriterFactory primaryTupleWriterFactory = new
-        // SimpleTupleWriterFactory();
-        IBTreeInteriorFrameFactory primaryInteriorFrameFactory = new NSMInteriorFrameFactory(primaryTupleWriterFactory);
-        IBTreeLeafFrameFactory primaryLeafFrameFactory = new NSMLeafFrameFactory(primaryTupleWriterFactory);
-        IBTreeRegistryProvider btreeRegistryProvider = new TestBTreeRegistryProvider();
+		spec.connect(new OneToOneConnectorDescriptor(spec), keyProviderOp, 0,
+				primaryBtreeSearchOp, 0);
+		spec.connect(new OneToOneConnectorDescriptor(spec),
+				primaryBtreeSearchOp, 0, printer, 0);
 
-        // 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();
-        }
+		spec.addRoot(printer);
+		runTest(spec);
+	}
 
-        MultiComparator primaryCmp = new MultiComparator(primaryTypeTraits, primaryComparators);
+	@Test
+	public void loadSecondaryIndexTest() throws Exception {
+		JobSpecification spec = new JobSpecification();
 
-        // 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);
+		// build dummy tuple containing nothing
+		ArrayTupleBuilder tb = new ArrayTupleBuilder(primaryKeyFieldCount * 2);
+		DataOutput dos = tb.getDataOutput();
 
-        int secondaryKeyFieldCount = 2;
-        IBinaryComparatorFactory[] secondaryComparatorFactories = new IBinaryComparatorFactory[secondaryKeyFieldCount];
-        secondaryComparatorFactories[0] = UTF8StringBinaryComparatorFactory.INSTANCE;
-        secondaryComparatorFactories[1] = UTF8StringBinaryComparatorFactory.INSTANCE;
+		tb.reset();
+		UTF8StringSerializerDeserializer.INSTANCE.serialize("0", dos);
+		tb.addFieldEndOffset();
 
-        TypeAwareTupleWriterFactory secondaryTupleWriterFactory = new TypeAwareTupleWriterFactory(secondaryTypeTraits);
-        // SimpleTupleWriterFactory secondaryTupleWriterFactory = new
-        // SimpleTupleWriterFactory();
-        IBTreeInteriorFrameFactory secondaryInteriorFrameFactory = new NSMInteriorFrameFactory(
-                secondaryTupleWriterFactory);
-        IBTreeLeafFrameFactory secondaryLeafFrameFactory = new NSMLeafFrameFactory(secondaryTupleWriterFactory);
+		ISerializerDeserializer[] keyRecDescSers = {
+				UTF8StringSerializerDeserializer.INSTANCE,
+				UTF8StringSerializerDeserializer.INSTANCE };
+		RecordDescriptor keyRecDesc = new RecordDescriptor(keyRecDescSers);
 
-        // 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();
-        }
+		ConstantTupleSourceOperatorDescriptor keyProviderOp = new ConstantTupleSourceOperatorDescriptor(
+				spec, keyRecDesc, tb.getFieldEndOffsets(), tb.getByteArray(),
+				tb.getSize());
+		PartitionConstraintHelper.addAbsoluteLocationConstraint(spec,
+				keyProviderOp, NC1_ID);
 
-        MultiComparator secondaryCmp = new MultiComparator(secondaryTypeTraits, secondaryComparators);
+		int[] lowKeyFields = null; // - infinity
+		int[] highKeyFields = null; // + infinity
 
-        // we create and register 3 btrees for in an insert pipeline being fed
-        // from a filescan op
-        BTreeRegistry btreeRegistry = btreeRegistryProvider.getBTreeRegistry(null);
-        IStorageManagerInterface storageManager = new TestStorageManagerInterface();
-        IBufferCache bufferCache = storageManager.getBufferCache(null);
-        IFileMapProvider fileMapProvider = storageManager.getFileMapProvider(null);
+		// scan primary index
+		BTreeSearchOperatorDescriptor primaryBtreeSearchOp = new BTreeSearchOperatorDescriptor(
+				spec, primaryRecDesc, storageManager, btreeRegistryProvider,
+				primaryBtreeSplitProvider, primaryInteriorFrameFactory,
+				primaryLeafFrameFactory, primaryTypeTraits,
+				primaryComparatorFactories, true, lowKeyFields, highKeyFields,
+				true, true);
+		PartitionConstraintHelper.addAbsoluteLocationConstraint(spec,
+				primaryBtreeSearchOp, NC1_ID);
 
-        // primary index
-        FileReference fileA = new FileReference(new File("/tmp/btreetestA.ix"));
-        bufferCache.createFile(fileA);
-        int fileIdA = fileMapProvider.lookupFileId(fileA);
-        bufferCache.openFile(fileIdA);
-        BTree btreeA = new BTree(bufferCache, primaryInteriorFrameFactory, primaryLeafFrameFactory, primaryCmp);
-        btreeA.create(fileIdA, primaryLeafFrameFactory.getFrame(), new MetaDataFrame());
-        btreeA.open(fileIdA);
-        btreeRegistry.register(fileIdA, btreeA);
-        bufferCache.closeFile(fileIdA);
+		// sort based on secondary keys
+		ExternalSortOperatorDescriptor sorter = new ExternalSortOperatorDescriptor(
+				spec,
+				1000,
+				new int[] { 3, 0 },
+				new IBinaryComparatorFactory[] { UTF8StringBinaryComparatorFactory.INSTANCE },
+				primaryRecDesc);
+		PartitionConstraintHelper.addAbsoluteLocationConstraint(spec, sorter,
+				NC1_ID);
 
-        // first secondary index
-        FileReference fileB = new FileReference(new File("/tmp/btreetestB.ix"));
-        bufferCache.createFile(fileB);
-        int fileIdB = fileMapProvider.lookupFileId(fileB);
-        bufferCache.openFile(fileIdB);
-        BTree btreeB = new BTree(bufferCache, secondaryInteriorFrameFactory, secondaryLeafFrameFactory, secondaryCmp);
-        btreeB.create(fileIdB, secondaryLeafFrameFactory.getFrame(), new MetaDataFrame());
-        btreeB.open(fileIdB);
-        btreeRegistry.register(fileIdB, btreeB);
-        bufferCache.closeFile(fileIdB);
+		// load secondary index
+		int[] fieldPermutation = { 3, 0 };
+		BTreeBulkLoadOperatorDescriptor secondaryBtreeBulkLoad = new BTreeBulkLoadOperatorDescriptor(
+				spec, storageManager, btreeRegistryProvider,
+				secondaryBtreeSplitProvider, secondaryInteriorFrameFactory,
+				secondaryLeafFrameFactory, secondaryTypeTraits,
+				secondaryComparatorFactories, fieldPermutation, 0.7f);
+		PartitionConstraintHelper.addAbsoluteLocationConstraint(spec,
+				secondaryBtreeBulkLoad, NC1_ID);
 
-        // second secondary index
-        FileReference fileC = new FileReference(new File("/tmp/btreetestC.ix"));
-        bufferCache.createFile(fileC);
-        int fileIdC = fileMapProvider.lookupFileId(fileC);
-        bufferCache.openFile(fileIdC);
-        BTree btreeC = new BTree(bufferCache, secondaryInteriorFrameFactory, secondaryLeafFrameFactory, secondaryCmp);
-        btreeC.create(fileIdC, secondaryLeafFrameFactory.getFrame(), new MetaDataFrame());
-        btreeC.open(fileIdC);
-        btreeRegistry.register(fileIdC, btreeC);
-        bufferCache.closeFile(fileIdC);
+		spec.connect(new OneToOneConnectorDescriptor(spec), keyProviderOp, 0,
+				primaryBtreeSearchOp, 0);
+		spec.connect(new OneToOneConnectorDescriptor(spec),
+				primaryBtreeSearchOp, 0, sorter, 0);
+		spec.connect(new OneToOneConnectorDescriptor(spec), sorter, 0,
+				secondaryBtreeBulkLoad, 0);
 
-        // create insert operators
+		spec.addRoot(secondaryBtreeBulkLoad);
+		runTest(spec);
+	}
 
-        // primary index
-        IFileSplitProvider btreeSplitProviderA = new ConstantFileSplitProvider(new FileSplit[] { new FileSplit(NC1_ID,
-                new FileReference(new File("/tmp/btreetestA.ix"))) });
-        int[] fieldPermutationA = { 0, 1, 2, 3, 4, 5 };
-        BTreeInsertUpdateDeleteOperatorDescriptor insertOpA = new BTreeInsertUpdateDeleteOperatorDescriptor(spec,
-                ordersDesc, storageManager, btreeRegistryProvider, btreeSplitProviderA, primaryInteriorFrameFactory,
-                primaryLeafFrameFactory, primaryTypeTraits, primaryComparatorFactories, fieldPermutationA,
-                BTreeOp.BTO_INSERT);
-        PartitionConstraintHelper.addAbsoluteLocationConstraint(spec, insertOpA, NC1_ID);
+	@Test
+	public void searchSecondaryIndexTest() throws Exception {
+		JobSpecification spec = new JobSpecification();
 
-        // first secondary index
-        IFileSplitProvider btreeSplitProviderB = new ConstantFileSplitProvider(new FileSplit[] { new FileSplit(NC1_ID,
-                new FileReference(new File("/tmp/btreetestB.ix"))) });
-        int[] fieldPermutationB = { 3, 0 };
-        BTreeInsertUpdateDeleteOperatorDescriptor insertOpB = new BTreeInsertUpdateDeleteOperatorDescriptor(spec,
-                ordersDesc, storageManager, btreeRegistryProvider, btreeSplitProviderB, secondaryInteriorFrameFactory,
-                secondaryLeafFrameFactory, secondaryTypeTraits, secondaryComparatorFactories, fieldPermutationB,
-                BTreeOp.BTO_INSERT);
-        PartitionConstraintHelper.addAbsoluteLocationConstraint(spec, insertOpB, NC1_ID);
+		// build tuple containing search keys (only use the first key as search
+		// key)
+		ArrayTupleBuilder tb = new ArrayTupleBuilder(secondaryKeyFieldCount);
+		DataOutput dos = tb.getDataOutput();
 
-        // second secondary index
-        IFileSplitProvider btreeSplitProviderC = new ConstantFileSplitProvider(new FileSplit[] { new FileSplit(NC1_ID,
-                new FileReference(new File("/tmp/btreetestC.ix"))) });
-        int[] fieldPermutationC = { 4, 0 };
-        BTreeInsertUpdateDeleteOperatorDescriptor insertOpC = new BTreeInsertUpdateDeleteOperatorDescriptor(spec,
-                ordersDesc, storageManager, btreeRegistryProvider, btreeSplitProviderC, secondaryInteriorFrameFactory,
-                secondaryLeafFrameFactory, secondaryTypeTraits, secondaryComparatorFactories, fieldPermutationC,
-                BTreeOp.BTO_INSERT);
-        PartitionConstraintHelper.addAbsoluteLocationConstraint(spec, insertOpC, NC1_ID);
+		tb.reset();
+		// low key
+		UTF8StringSerializerDeserializer.INSTANCE.serialize("1998-07-21", dos);
+		tb.addFieldEndOffset();
+		// high key
+		UTF8StringSerializerDeserializer.INSTANCE.serialize("2000-10-18", dos);
+		tb.addFieldEndOffset();
 
-        NullSinkOperatorDescriptor nullSink = new NullSinkOperatorDescriptor(spec);
-        PartitionConstraintHelper.addAbsoluteLocationConstraint(spec, nullSink, NC1_ID);
+		ISerializerDeserializer[] keyRecDescSers = {
+				UTF8StringSerializerDeserializer.INSTANCE,
+				UTF8StringSerializerDeserializer.INSTANCE };
+		RecordDescriptor keyRecDesc = new RecordDescriptor(keyRecDescSers);
 
-        spec.connect(new OneToOneConnectorDescriptor(spec), ordScanner, 0, insertOpA, 0);
+		ConstantTupleSourceOperatorDescriptor keyProviderOp = new ConstantTupleSourceOperatorDescriptor(
+				spec, keyRecDesc, tb.getFieldEndOffsets(), tb.getByteArray(),
+				tb.getSize());
+		PartitionConstraintHelper.addAbsoluteLocationConstraint(spec,
+				keyProviderOp, NC1_ID);
 
-        spec.connect(new OneToOneConnectorDescriptor(spec), insertOpA, 0, insertOpB, 0);
+		int[] secondaryLowKeyFields = { 0 };
+		int[] secondaryHighKeyFields = { 1 };
 
-        spec.connect(new OneToOneConnectorDescriptor(spec), insertOpB, 0, insertOpC, 0);
+		// search secondary index
+		BTreeSearchOperatorDescriptor secondaryBtreeSearchOp = new BTreeSearchOperatorDescriptor(
+				spec, secondaryRecDesc, storageManager, btreeRegistryProvider,
+				secondaryBtreeSplitProvider, secondaryInteriorFrameFactory,
+				secondaryLeafFrameFactory, secondaryTypeTraits,
+				secondaryComparatorFactories, true, secondaryLowKeyFields,
+				secondaryHighKeyFields, true, true);
+		PartitionConstraintHelper.addAbsoluteLocationConstraint(spec,
+				secondaryBtreeSearchOp, NC1_ID);
 
-        spec.connect(new OneToOneConnectorDescriptor(spec), insertOpC, 0, nullSink, 0);
+		int[] primaryLowKeyFields = { 1 }; // second field from the tuples
+		// coming from secondary index
+		int[] primaryHighKeyFields = { 1 }; // second field from the tuples
+		// coming from secondary index
 
-        spec.addRoot(nullSink);
-        runTest(spec);
+		// search primary index
+		BTreeSearchOperatorDescriptor primaryBtreeSearchOp = new BTreeSearchOperatorDescriptor(
+				spec, primaryRecDesc, storageManager, btreeRegistryProvider,
+				primaryBtreeSplitProvider, primaryInteriorFrameFactory,
+				primaryLeafFrameFactory, primaryTypeTraits,
+				primaryComparatorFactories, true, primaryLowKeyFields,
+				primaryHighKeyFields, true, true);
+		PartitionConstraintHelper.addAbsoluteLocationConstraint(spec,
+				primaryBtreeSearchOp, NC1_ID);
 
-        // scan primary index
-        System.out.println("PRINTING PRIMARY INDEX");
-        bufferCache.openFile(fileIdA);
-        IBTreeCursor scanCursorA = new RangeSearchCursor(primaryLeafFrameFactory.getFrame());
-        RangePredicate nullPredA = new RangePredicate(true, null, null, true, true, null, null);
-        BTreeOpContext opCtxA = btreeA.createOpContext(BTreeOp.BTO_SEARCH, primaryLeafFrameFactory.getFrame(),
-                primaryInteriorFrameFactory.getFrame(), null);
-        btreeA.search(scanCursorA, nullPredA, opCtxA);
-        try {
-            while (scanCursorA.hasNext()) {
-                scanCursorA.next();
-                ITupleReference frameTuple = scanCursorA.getTuple();
-                String rec = primaryCmp.printTuple(frameTuple, ordersDesc.getFields());
-                System.out.println(rec);
-            }
-        } catch (Exception e) {
-            e.printStackTrace();
-        } finally {
-            scanCursorA.close();
-        }
-        bufferCache.closeFile(fileIdA);
-        System.out.println();
+		PrinterOperatorDescriptor printer = new PrinterOperatorDescriptor(spec);
+		PartitionConstraintHelper.addAbsoluteLocationConstraint(spec, printer,
+				NC1_ID);
 
-        // scan first secondary index
-        System.out.println("PRINTING FIRST SECONDARY INDEX");
-        bufferCache.openFile(fileIdB);
-        IBTreeCursor scanCursorB = new RangeSearchCursor(secondaryLeafFrameFactory.getFrame());
-        RangePredicate nullPredB = new RangePredicate(true, null, null, true, true, null, null);
-        BTreeOpContext opCtxB = btreeB.createOpContext(BTreeOp.BTO_SEARCH, secondaryLeafFrameFactory.getFrame(),
-                secondaryInteriorFrameFactory.getFrame(), null);
-        btreeB.search(scanCursorB, nullPredB, opCtxB);
-        try {
-            while (scanCursorB.hasNext()) {
-                scanCursorB.next();
-                ITupleReference frameTuple = scanCursorB.getTuple();
-                String rec = secondaryCmp.printTuple(frameTuple, ordersDesc.getFields());
-                System.out.println(rec);
-            }
-        } catch (Exception e) {
-            e.printStackTrace();
-        } finally {
-            scanCursorB.close();
-        }
-        bufferCache.closeFile(fileIdB);
-        System.out.println();
+		spec.connect(new OneToOneConnectorDescriptor(spec), keyProviderOp, 0,
+				secondaryBtreeSearchOp, 0);
+		spec.connect(new OneToOneConnectorDescriptor(spec),
+				secondaryBtreeSearchOp, 0, primaryBtreeSearchOp, 0);
+		spec.connect(new OneToOneConnectorDescriptor(spec),
+				primaryBtreeSearchOp, 0, printer, 0);
 
-        // scan second secondary index
-        System.out.println("PRINTING SECOND SECONDARY INDEX");
-        bufferCache.openFile(fileIdC);
-        IBTreeCursor scanCursorC = new RangeSearchCursor(secondaryLeafFrameFactory.getFrame());
-        RangePredicate nullPredC = new RangePredicate(true, null, null, true, true, null, null);
-        BTreeOpContext opCtxC = btreeC.createOpContext(BTreeOp.BTO_SEARCH, secondaryLeafFrameFactory.getFrame(),
-                secondaryInteriorFrameFactory.getFrame(), null);
-        btreeC.search(scanCursorC, nullPredC, opCtxC);
-        try {
-            while (scanCursorC.hasNext()) {
-                scanCursorC.next();
-                ITupleReference frameTuple = scanCursorC.getTuple();
-                String rec = secondaryCmp.printTuple(frameTuple, ordersDesc.getFields());
-                System.out.println(rec);
-            }
-        } catch (Exception e) {
-            e.printStackTrace();
-        } finally {
-            scanCursorC.close();
-        }
-        bufferCache.closeFile(fileIdC);
-        System.out.println();
-    }
-}
+		spec.addRoot(printer);
+		runTest(spec);
+	}
+
+	@Test
+	public void insertPipelineTest() throws Exception {
+
+		JobSpecification spec = new JobSpecification();
+
+		FileSplit[] ordersSplits = new FileSplit[] { new FileSplit(NC1_ID,
+				new FileReference(new File("data/tpch0.001/orders-part2.tbl"))) };
+		IFileSplitProvider ordersSplitProvider = new ConstantFileSplitProvider(
+				ordersSplits);
+		RecordDescriptor ordersDesc = new RecordDescriptor(
+				new ISerializerDeserializer[] {
+						UTF8StringSerializerDeserializer.INSTANCE,
+						UTF8StringSerializerDeserializer.INSTANCE,
+						UTF8StringSerializerDeserializer.INSTANCE,
+						UTF8StringSerializerDeserializer.INSTANCE,
+						UTF8StringSerializerDeserializer.INSTANCE,
+						UTF8StringSerializerDeserializer.INSTANCE,
+						UTF8StringSerializerDeserializer.INSTANCE,
+						UTF8StringSerializerDeserializer.INSTANCE,
+						UTF8StringSerializerDeserializer.INSTANCE });
+
+		FileScanOperatorDescriptor ordScanner = new FileScanOperatorDescriptor(
+				spec, ordersSplitProvider, new DelimitedDataTupleParserFactory(
+						new IValueParserFactory[] {
+								UTF8StringParserFactory.INSTANCE,
+								UTF8StringParserFactory.INSTANCE,
+								UTF8StringParserFactory.INSTANCE,
+								UTF8StringParserFactory.INSTANCE,
+								UTF8StringParserFactory.INSTANCE,
+								UTF8StringParserFactory.INSTANCE,
+								UTF8StringParserFactory.INSTANCE,
+								UTF8StringParserFactory.INSTANCE,
+								UTF8StringParserFactory.INSTANCE }, '|'),
+				ordersDesc);
+		PartitionConstraintHelper.addAbsoluteLocationConstraint(spec,
+				ordScanner, NC1_ID);
+
+		// insert into primary index
+		int[] primaryFieldPermutation = { 0, 1, 2, 4, 5, 7 };
+		BTreeInsertUpdateDeleteOperatorDescriptor primaryBtreeInsertOp = new BTreeInsertUpdateDeleteOperatorDescriptor(
+				spec, ordersDesc, storageManager, btreeRegistryProvider,
+				primaryBtreeSplitProvider, primaryInteriorFrameFactory,
+				primaryLeafFrameFactory, primaryTypeTraits,
+				primaryComparatorFactories, primaryFieldPermutation,
+				BTreeOp.BTO_INSERT);
+		PartitionConstraintHelper.addAbsoluteLocationConstraint(spec,
+				primaryBtreeInsertOp, NC1_ID);
+
+		// first secondary index
+		int[] fieldPermutationB = { 4, 0 };
+		BTreeInsertUpdateDeleteOperatorDescriptor secondaryInsertOp = new BTreeInsertUpdateDeleteOperatorDescriptor(
+				spec, ordersDesc, storageManager, btreeRegistryProvider,
+				secondaryBtreeSplitProvider, secondaryInteriorFrameFactory,
+				secondaryLeafFrameFactory, secondaryTypeTraits,
+				secondaryComparatorFactories, fieldPermutationB,
+				BTreeOp.BTO_INSERT);
+		PartitionConstraintHelper.addAbsoluteLocationConstraint(spec,
+				secondaryInsertOp, NC1_ID);
+
+		NullSinkOperatorDescriptor nullSink = new NullSinkOperatorDescriptor(
+				spec);
+		PartitionConstraintHelper.addAbsoluteLocationConstraint(spec, nullSink,
+				NC1_ID);
+
+		spec.connect(new OneToOneConnectorDescriptor(spec), ordScanner, 0,
+				primaryBtreeInsertOp, 0);
+
+		spec.connect(new OneToOneConnectorDescriptor(spec),
+				primaryBtreeInsertOp, 0, secondaryInsertOp, 0);
+
+		spec.connect(new OneToOneConnectorDescriptor(spec), secondaryInsertOp,
+				0, nullSink, 0);
+
+		spec.addRoot(nullSink);
+		runTest(spec);
+	}
+
+	@Test
+	public void searchUpdatedSecondaryIndexTest() throws Exception {
+		JobSpecification spec = new JobSpecification();
+
+		// build tuple containing search keys (only use the first key as search
+		// key)
+		ArrayTupleBuilder tb = new ArrayTupleBuilder(secondaryKeyFieldCount);
+		DataOutput dos = tb.getDataOutput();
+
+		tb.reset();
+		// low key
+		UTF8StringSerializerDeserializer.INSTANCE.serialize("1998-07-21", dos);
+		tb.addFieldEndOffset();
+		// high key
+		UTF8StringSerializerDeserializer.INSTANCE.serialize("2000-10-18", dos);
+		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());
+		PartitionConstraintHelper.addAbsoluteLocationConstraint(spec,
+				keyProviderOp, NC1_ID);
+
+		int[] secondaryLowKeyFields = { 0 };
+		int[] secondaryHighKeyFields = { 1 };
+
+		// search secondary index
+		BTreeSearchOperatorDescriptor secondaryBtreeSearchOp = new BTreeSearchOperatorDescriptor(
+				spec, secondaryRecDesc, storageManager, btreeRegistryProvider,
+				secondaryBtreeSplitProvider, secondaryInteriorFrameFactory,
+				secondaryLeafFrameFactory, secondaryTypeTraits,
+				secondaryComparatorFactories, true, secondaryLowKeyFields,
+				secondaryHighKeyFields, true, true);
+		PartitionConstraintHelper.addAbsoluteLocationConstraint(spec,
+				secondaryBtreeSearchOp, NC1_ID);
+
+		// second field from the tuples coming from secondary index
+		int[] primaryLowKeyFields = { 1 };
+		// second field from the tuples coming from secondary index
+		int[] primaryHighKeyFields = { 1 };
+
+		// search primary index
+		BTreeSearchOperatorDescriptor primaryBtreeSearchOp = new BTreeSearchOperatorDescriptor(
+				spec, primaryRecDesc, storageManager, btreeRegistryProvider,
+				primaryBtreeSplitProvider, primaryInteriorFrameFactory,
+				primaryLeafFrameFactory, primaryTypeTraits,
+				primaryComparatorFactories, true, primaryLowKeyFields,
+				primaryHighKeyFields, true, true);
+		PartitionConstraintHelper.addAbsoluteLocationConstraint(spec,
+				primaryBtreeSearchOp, NC1_ID);
+
+		PrinterOperatorDescriptor printer = new PrinterOperatorDescriptor(spec);
+		PartitionConstraintHelper.addAbsoluteLocationConstraint(spec, printer,
+				NC1_ID);
+
+		spec.connect(new OneToOneConnectorDescriptor(spec), keyProviderOp, 0,
+				secondaryBtreeSearchOp, 0);
+		spec.connect(new OneToOneConnectorDescriptor(spec),
+				secondaryBtreeSearchOp, 0, primaryBtreeSearchOp, 0);
+		spec.connect(new OneToOneConnectorDescriptor(spec),
+				primaryBtreeSearchOp, 0, printer, 0);
+
+		spec.addRoot(printer);
+		runTest(spec);
+	}
+
+	@AfterClass
+	public static void cleanup() throws Exception {
+		File primary = new File(primaryFileName);
+		primary.deleteOnExit();
+
+		File secondary = new File(secondaryFileName);
+		secondary.deleteOnExit();
+	}
+}
\ No newline at end of file