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 d10d622..5008c26 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
@@ -160,7 +160,7 @@
         int btreeFileId = 0; // TODO: this relies on the way FileMappingProvider assigns ids (in sequence starting from 0)
         BTree btree = btreeRegistryProvider.getBTreeRegistry().get(btreeFileId);
         IBTreeCursor scanCursor = new RangeSearchCursor(leafFrameFactory.getFrame());
-        RangePredicate nullPred = new RangePredicate(true, null, null, true, true, null);
+        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 {
@@ -423,7 +423,7 @@
         // scan primary index         
         System.out.println("PRINTING PRIMARY INDEX");
         IBTreeCursor scanCursorA = new RangeSearchCursor(primaryLeafFrameFactory.getFrame());
-        RangePredicate nullPredA = new RangePredicate(true, null, null, true, true, null);
+        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 {
@@ -443,7 +443,7 @@
         // scan first secondary index
         System.out.println("PRINTING FIRST SECONDARY INDEX");
         IBTreeCursor scanCursorB = new RangeSearchCursor(secondaryLeafFrameFactory.getFrame());
-        RangePredicate nullPredB = new RangePredicate(true, null, null, true, true, null);
+        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 {
@@ -463,7 +463,7 @@
         // scan second secondary index
         System.out.println("PRINTING SECOND SECONDARY INDEX");
         IBTreeCursor scanCursorC = new RangeSearchCursor(secondaryLeafFrameFactory.getFrame());
-        RangePredicate nullPredC = new RangePredicate(true, null, null, true, true, null);
+        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 {
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IBTreeLeafFrame.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IBTreeLeafFrame.java
index c2dd1f4..5fb549c 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IBTreeLeafFrame.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IBTreeLeafFrame.java
@@ -15,6 +15,11 @@
 
 package edu.uci.ics.hyracks.storage.am.btree.api;
 
+import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
+import edu.uci.ics.hyracks.storage.am.btree.impls.FindTupleMode;
+import edu.uci.ics.hyracks.storage.am.btree.impls.FindTupleNoExactMatchPolicy;
+import edu.uci.ics.hyracks.storage.am.btree.impls.MultiComparator;
+
 public interface IBTreeLeafFrame extends IBTreeFrame {	
 	public void setNextLeaf(int nextPage);
 	public int getNextLeaf();
@@ -23,4 +28,6 @@
 	public int getPrevLeaf();
 	
 	public IBTreeTupleReference createTupleReference();
+	
+	public int findTupleIndex(ITupleReference searchKey, IBTreeTupleReference pageTuple, MultiComparator cmp, FindTupleMode ftm, FindTupleNoExactMatchPolicy ftp);
 }
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IPrefixSlotManager.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IPrefixSlotManager.java
index 3f74077..d8459a4 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IPrefixSlotManager.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IPrefixSlotManager.java
@@ -17,6 +17,8 @@
 
 import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
 import edu.uci.ics.hyracks.storage.am.btree.frames.FieldPrefixNSMLeafFrame;
+import edu.uci.ics.hyracks.storage.am.btree.impls.FindTupleMode;
+import edu.uci.ics.hyracks.storage.am.btree.impls.FindTupleNoExactMatchPolicy;
 import edu.uci.ics.hyracks.storage.am.btree.impls.MultiComparator;
 
 // a slot consists of two fields:
@@ -46,7 +48,7 @@
 	public int decodeSecondSlotField(int slot);		
 	public int encodeSlotFields(int firstField, int secondField);
 	
-	public int findSlot(ITupleReference tuple, IBTreeTupleReference frameTuple, IBTreeTupleReference framePrefixTuple, MultiComparator multiCmp, boolean exact);
+	public int findSlot(ITupleReference searchKey, IBTreeTupleReference frameTuple, IBTreeTupleReference framePrefixTuple, MultiComparator multiCmp, FindTupleMode mode, FindTupleNoExactMatchPolicy matchPolicy);
 	public int insertSlot(int slot, int tupleOff);
 	
 	// returns prefix slot number, returns TUPLE_UNCOMPRESSED if none found
@@ -58,13 +60,13 @@
 	public int getPrefixSlotStartOff();
 	public int getPrefixSlotEndOff();
 	
-	public int getTupleSlotOff(int slotNum);
-	public int getPrefixSlotOff(int slotNum);	
+	public int getTupleSlotOff(int tupleIndex);
+	public int getPrefixSlotOff(int tupleIndex);
 		
 	public int getSlotSize();		
 	
 	public void setSlot(int offset, int value);
 		
 	// functions for testing
-	public void setPrefixSlot(int slotNum, int slot);
+	public void setPrefixSlot(int tupleIndex, int slot);
 }
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/ISlotManager.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/ISlotManager.java
index 14cf579..cbb2d86 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/ISlotManager.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/ISlotManager.java
@@ -16,13 +16,14 @@
 package edu.uci.ics.hyracks.storage.am.btree.api;
 
 import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
-import edu.uci.ics.hyracks.storage.am.btree.impls.FindSlotMode;
+import edu.uci.ics.hyracks.storage.am.btree.impls.FindTupleMode;
+import edu.uci.ics.hyracks.storage.am.btree.impls.FindTupleNoExactMatchPolicy;
 import edu.uci.ics.hyracks.storage.am.btree.impls.MultiComparator;
 
 public interface ISlotManager {
 	public void setFrame(IBTreeFrame frame);
 	
-	public int findTupleIndex(ITupleReference tuple, IBTreeTupleReference pageTuple, MultiComparator multiCmp, FindSlotMode mode);
+	public int findTupleIndex(ITupleReference searchKey, IBTreeTupleReference frameTuple, MultiComparator multiCmp, FindTupleMode mode, FindTupleNoExactMatchPolicy matchPolicy);
 	public int insertSlot(int tupleIndex, int tupleOff);
 	
 	public int getSlotStartOff();
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeSearchOperatorNodePushable.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeSearchOperatorNodePushable.java
index e97ae7d..052d7e1 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeSearchOperatorNodePushable.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/BTreeSearchOperatorNodePushable.java
@@ -53,7 +53,8 @@
     private boolean lowKeyInclusive;
     private boolean highKeyInclusive;
     private RangePredicate rangePred;
-    private MultiComparator searchCmp;
+    private MultiComparator lowKeySearchCmp;
+    private MultiComparator highKeySearchCmp;
     private IBTreeCursor cursor;    
     private IBTreeLeafFrame cursorFrame;    
     private BTreeOpContext opCtx;
@@ -93,15 +94,30 @@
         
         // 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];
-        }
-        searchCmp = new MultiComparator(btree.getMultiComparator().getTypeTraits(), searchComparators);
+        int lowKeySearchFields = btree.getMultiComparator().getComparators().length;
+        int highKeySearchFields = btree.getMultiComparator().getComparators().length;
+        if(lowKey != null) lowKeySearchFields = lowKey.getFieldCount();
+        if(highKey != null) highKeySearchFields = highKey.getFieldCount();
         
-        rangePred = new RangePredicate(isForward, null, null, lowKeyInclusive, highKeyInclusive, searchCmp);
+        IBinaryComparator[] lowKeySearchComparators = new IBinaryComparator[lowKeySearchFields];
+        for (int i = 0; i < lowKeySearchFields; i++) {
+        	lowKeySearchComparators[i] = btree.getMultiComparator().getComparators()[i];
+        }
+        lowKeySearchCmp = new MultiComparator(btree.getMultiComparator().getTypeTraits(), lowKeySearchComparators);
+        
+        if(lowKeySearchFields == highKeySearchFields) {        
+            highKeySearchCmp = lowKeySearchCmp;            
+        }
+        else {
+        	IBinaryComparator[] highKeySearchComparators = new IBinaryComparator[highKeySearchFields];
+            for (int i = 0; i < highKeySearchFields; i++) {
+            	highKeySearchComparators[i] = btree.getMultiComparator().getComparators()[i];
+            }
+            highKeySearchCmp = new MultiComparator(btree.getMultiComparator().getTypeTraits(), highKeySearchComparators);
+        	
+        }
+        
+        rangePred = new RangePredicate(isForward, null, null, lowKeyInclusive, highKeyInclusive, lowKeySearchCmp, highKeySearchCmp);
                 
         accessor = new FrameTupleAccessor(btreeOpHelper.getHyracksContext(), recDesc);
         
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/FieldPrefixNSMLeafFrame.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/FieldPrefixNSMLeafFrame.java
index 7642638..65327ae 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/FieldPrefixNSMLeafFrame.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/FieldPrefixNSMLeafFrame.java
@@ -38,6 +38,8 @@
 import edu.uci.ics.hyracks.storage.am.btree.impls.FieldPrefixPrefixTupleReference;
 import edu.uci.ics.hyracks.storage.am.btree.impls.FieldPrefixSlotManager;
 import edu.uci.ics.hyracks.storage.am.btree.impls.FieldPrefixTupleReference;
+import edu.uci.ics.hyracks.storage.am.btree.impls.FindTupleMode;
+import edu.uci.ics.hyracks.storage.am.btree.impls.FindTupleNoExactMatchPolicy;
 import edu.uci.ics.hyracks.storage.am.btree.impls.MultiComparator;
 import edu.uci.ics.hyracks.storage.am.btree.impls.SlotOffTupleOff;
 import edu.uci.ics.hyracks.storage.am.btree.impls.SpaceStatus;
@@ -164,7 +166,7 @@
     
     @Override
     public void delete(ITupleReference tuple, MultiComparator cmp, boolean exactDelete) throws Exception {        
-        int slot = slotManager.findSlot(tuple, frameTuple, framePrefixTuple, cmp, true);
+        int slot = slotManager.findSlot(tuple, frameTuple, framePrefixTuple, cmp, FindTupleMode.FTM_EXACT, FindTupleNoExactMatchPolicy.FTP_HIGHER_KEY);
         int tupleIndex = slotManager.decodeSecondSlotField(slot);
         if(tupleIndex == FieldPrefixSlotManager.GREATEST_SLOT) {
             throw new BTreeException("Key to be deleted does not exist.");   
@@ -268,7 +270,7 @@
     
     @Override
     public void insert(ITupleReference tuple, MultiComparator cmp) throws Exception {    	
-    	int slot = slotManager.findSlot(tuple, frameTuple, framePrefixTuple, cmp, false);        
+    	int slot = slotManager.findSlot(tuple, frameTuple, framePrefixTuple, cmp, FindTupleMode.FTM_INCLUSIVE, FindTupleNoExactMatchPolicy.FTP_HIGHER_KEY);        
         
     	slot = slotManager.insertSlot(slot, buf.getInt(freeSpaceOff));                
         
@@ -425,7 +427,7 @@
     	frameTuple.setFieldCount(cmp.getFieldCount());
     	
     	// before doing anything check if key already exists
-		int slot = slotManager.findSlot(tuple, frameTuple, framePrefixTuple, cmp, true);
+		int slot = slotManager.findSlot(tuple, frameTuple, framePrefixTuple, cmp, FindTupleMode.FTM_EXACT, FindTupleNoExactMatchPolicy.FTP_HIGHER_KEY);
 		int tupleSlotNum = slotManager.decodeSecondSlotField(slot);		
 		if(tupleSlotNum != FieldPrefixSlotManager.GREATEST_SLOT) {				
 			frameTuple.resetByTupleIndex(this, tupleSlotNum);
@@ -608,4 +610,13 @@
 	public IBTreeTupleReference createTupleReference() {
 		return new FieldPrefixTupleReference(tupleWriter.createTupleReference());
 	}
+	
+	@Override
+	public int findTupleIndex(ITupleReference searchKey, IBTreeTupleReference pageTuple, MultiComparator cmp, FindTupleMode ftm, FindTupleNoExactMatchPolicy ftp) {
+		int slot = slotManager.findSlot(searchKey, pageTuple, framePrefixTuple, cmp, ftm, ftp);
+		int tupleIndex = slotManager.decodeSecondSlotField(slot);
+		if(tupleIndex == FieldPrefixSlotManager.GREATEST_SLOT) return -1;
+		else return tupleIndex;		
+	}
+		
 }
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMFrame.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMFrame.java
index ec578b8..01d24ce0 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMFrame.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMFrame.java
@@ -30,7 +30,8 @@
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeTupleWriter;
 import edu.uci.ics.hyracks.storage.am.btree.api.ISlotManager;
 import edu.uci.ics.hyracks.storage.am.btree.impls.BTreeException;
-import edu.uci.ics.hyracks.storage.am.btree.impls.FindSlotMode;
+import edu.uci.ics.hyracks.storage.am.btree.impls.FindTupleMode;
+import edu.uci.ics.hyracks.storage.am.btree.impls.FindTupleNoExactMatchPolicy;
 import edu.uci.ics.hyracks.storage.am.btree.impls.MultiComparator;
 import edu.uci.ics.hyracks.storage.am.btree.impls.OrderedSlotManager;
 import edu.uci.ics.hyracks.storage.am.btree.impls.SlotOffTupleOff;
@@ -157,7 +158,7 @@
 	@Override
 	public void delete(ITupleReference tuple, MultiComparator cmp, boolean exactDelete) throws Exception {		
 		frameTuple.setFieldCount(cmp.getFieldCount());
-		int tupleIndex = slotManager.findTupleIndex(tuple, frameTuple, cmp, FindSlotMode.FSM_EXACT);
+		int tupleIndex = slotManager.findTupleIndex(tuple, frameTuple, cmp, FindTupleMode.FTM_EXACT, FindTupleNoExactMatchPolicy.FTP_HIGHER_KEY);
 		int slotOff = slotManager.getSlotOff(tupleIndex);
 		if(tupleIndex < 0) {
 			throw new BTreeException("Key to be deleted does not exist.");
@@ -211,7 +212,7 @@
 	@Override
 	public void insert(ITupleReference tuple, MultiComparator cmp) throws Exception {
 		frameTuple.setFieldCount(cmp.getFieldCount());
-		int tupleIndex = slotManager.findTupleIndex(tuple, frameTuple, cmp, FindSlotMode.FSM_INCLUSIVE);
+		int tupleIndex = slotManager.findTupleIndex(tuple, frameTuple, cmp, FindTupleMode.FTM_INCLUSIVE, FindTupleNoExactMatchPolicy.FTP_HIGHER_KEY);
 		slotManager.insertSlot(tupleIndex, buf.getInt(freeSpaceOff));				
 		int bytesWritten = tupleWriter.writeTuple(tuple, buf, buf.getInt(freeSpaceOff));
 		
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMInteriorFrame.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMInteriorFrame.java
index 6d50461..b34a3df 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMInteriorFrame.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMInteriorFrame.java
@@ -30,7 +30,8 @@
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeTupleReference;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeTupleWriter;
 import edu.uci.ics.hyracks.storage.am.btree.impls.BTreeException;
-import edu.uci.ics.hyracks.storage.am.btree.impls.FindSlotMode;
+import edu.uci.ics.hyracks.storage.am.btree.impls.FindTupleMode;
+import edu.uci.ics.hyracks.storage.am.btree.impls.FindTupleNoExactMatchPolicy;
 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.SlotOffTupleOff;
@@ -73,7 +74,7 @@
 	@Override
 	public void insert(ITupleReference tuple, MultiComparator cmp) throws Exception {		
 		frameTuple.setFieldCount(cmp.getKeyFieldCount());		
-		int tupleIndex = slotManager.findTupleIndex(tuple, frameTuple, cmp, FindSlotMode.FSM_INCLUSIVE);
+		int tupleIndex = slotManager.findTupleIndex(tuple, frameTuple, cmp, FindTupleMode.FTM_INCLUSIVE, FindTupleNoExactMatchPolicy.FTP_HIGHER_KEY);
 		int slotOff = slotManager.getSlotOff(tupleIndex);
 		boolean isDuplicate = true;				
 		
@@ -132,7 +133,7 @@
 	public int split(IBTreeFrame rightFrame, ITupleReference tuple, MultiComparator cmp, SplitKey splitKey) throws Exception {		
 		// before doing anything check if key already exists
 		frameTuple.setFieldCount(cmp.getKeyFieldCount());
-		int tupleIndex = slotManager.findTupleIndex(tuple, frameTuple, cmp, FindSlotMode.FSM_EXACT);
+		int tupleIndex = slotManager.findTupleIndex(tuple, frameTuple, cmp, FindTupleMode.FTM_EXACT, FindTupleNoExactMatchPolicy.FTP_HIGHER_KEY);
 		int slotOff = slotManager.getSlotOff(tupleIndex);
 		if(tupleIndex >= 0) {			
 			frameTuple.resetByOffset(buf, slotManager.getTupleOff(slotOff));			
@@ -240,25 +241,28 @@
 		
 		// check for trivial cases where no low key or high key exists (e.g. during an index scan)
 		ITupleReference tuple = null;
-		FindSlotMode fsm = null;
+		FindTupleMode fsm = null;
+		MultiComparator targetCmp = null;
 		if(pred.isForward()) {						
 			tuple = pred.getLowKey();
 			if(tuple == null) {
 				return getLeftmostChildPageId(srcCmp);
 			}
-			if(pred.isLowKeyInclusive()) fsm = FindSlotMode.FSM_INCLUSIVE;
-			else fsm = FindSlotMode.FSM_EXCLUSIVE;
+			if(pred.isLowKeyInclusive()) fsm = FindTupleMode.FTM_INCLUSIVE;
+			else fsm = FindTupleMode.FTM_EXCLUSIVE;
+			targetCmp = pred.getLowKeyComparator();	
 		}
 		else {
 			tuple = pred.getHighKey();
 			if(tuple == null) {
 				return getRightmostChildPageId(srcCmp);				
 			}
-			fsm = FindSlotMode.FSM_INCLUSIVE;		
+			if(pred.isHighKeyInclusive()) fsm = FindTupleMode.FTM_EXCLUSIVE;
+			else fsm = FindTupleMode.FTM_INCLUSIVE;
+			targetCmp = pred.getHighKeyComparator();
 		}
-		
-		MultiComparator targetCmp = pred.getComparator();		
-		int tupleIndex = slotManager.findTupleIndex(tuple, frameTuple, targetCmp, fsm);
+						
+		int tupleIndex = slotManager.findTupleIndex(tuple, frameTuple, targetCmp, fsm, FindTupleNoExactMatchPolicy.FTP_HIGHER_KEY);				
 		int slotOff = slotManager.getSlotOff(tupleIndex);
 		if(tupleIndex < 0) {			
 			return buf.getInt(rightLeafOff);
@@ -275,10 +279,10 @@
 					frameTuple.resetByOffset(buf, cmpTupleOff);
 					if(targetCmp.compare(cmpFrameTuple, frameTuple) != 0) break;					
 					slotOff += slotManager.getSlotSize();	
-				}
+				}				
 				slotOff -= slotManager.getSlotSize();
 			}
-			else {
+			else {				
 				int minSlotOff = slotManager.getSlotEndOff() - slotManager.getSlotSize();
 				slotOff -= slotManager.getSlotSize();
 				while(slotOff > minSlotOff) {
@@ -286,7 +290,7 @@
 					frameTuple.resetByOffset(buf, cmpTupleOff);
 					if(targetCmp.compare(cmpFrameTuple, frameTuple) != 0) break;
 					slotOff -= slotManager.getSlotSize();										
-				}
+				}												
 				slotOff += slotManager.getSlotSize();
 			}
 			
@@ -299,7 +303,7 @@
 	@Override
 	public void delete(ITupleReference tuple, MultiComparator cmp, boolean exactDelete) throws Exception {
 		frameTuple.setFieldCount(cmp.getKeyFieldCount());
-		int tupleIndex = slotManager.findTupleIndex(tuple, frameTuple, cmp, FindSlotMode.FSM_INCLUSIVE);
+		int tupleIndex = slotManager.findTupleIndex(tuple, frameTuple, cmp, FindTupleMode.FTM_INCLUSIVE, FindTupleNoExactMatchPolicy.FTP_HIGHER_KEY);
 		int slotOff = slotManager.getSlotOff(tupleIndex);
 		int tupleOff;
 		int keySize;
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMLeafFrame.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMLeafFrame.java
index bcf6ef8..2dd6bc0 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMLeafFrame.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMLeafFrame.java
@@ -23,7 +23,8 @@
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeTupleReference;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeTupleWriter;
 import edu.uci.ics.hyracks.storage.am.btree.impls.BTreeException;
-import edu.uci.ics.hyracks.storage.am.btree.impls.FindSlotMode;
+import edu.uci.ics.hyracks.storage.am.btree.impls.FindTupleMode;
+import edu.uci.ics.hyracks.storage.am.btree.impls.FindTupleNoExactMatchPolicy;
 import edu.uci.ics.hyracks.storage.am.btree.impls.MultiComparator;
 import edu.uci.ics.hyracks.storage.am.btree.impls.SplitKey;
 
@@ -65,7 +66,7 @@
 	@Override
 	public void insert(ITupleReference tuple, MultiComparator cmp) throws Exception {		
 		frameTuple.setFieldCount(cmp.getFieldCount());
-		int tupleIndex = slotManager.findTupleIndex(tuple, frameTuple, cmp, FindSlotMode.FSM_INCLUSIVE);
+		int tupleIndex = slotManager.findTupleIndex(tuple, frameTuple, cmp, FindTupleMode.FTM_INCLUSIVE, FindTupleNoExactMatchPolicy.FTP_HIGHER_KEY);
 		int slotOff = slotManager.getSlotOff(tupleIndex);
 		boolean isDuplicate = true;
 		
@@ -106,7 +107,7 @@
 		frameTuple.setFieldCount(cmp.getFieldCount());
 		
 		// before doing anything check if key already exists
-		int tupleIndex = slotManager.findTupleIndex(tuple, frameTuple, cmp, FindSlotMode.FSM_EXACT);
+		int tupleIndex = slotManager.findTupleIndex(tuple, frameTuple, cmp, FindTupleMode.FTM_EXACT, FindTupleNoExactMatchPolicy.FTP_HIGHER_KEY);
 		if (tupleIndex >= 0) {						
 			frameTuple.resetByTupleIndex(this, tupleIndex);		
 			if (cmp.compare(tuple, frameTuple) == 0) {
@@ -173,4 +174,9 @@
 	public IBTreeTupleReference createTupleReference() {
 		return tupleWriter.createTupleReference();
 	}
+	
+	@Override
+	public int findTupleIndex(ITupleReference searchKey, IBTreeTupleReference pageTuple, MultiComparator cmp, FindTupleMode ftm, FindTupleNoExactMatchPolicy ftp) {		
+		return slotManager.findTupleIndex(searchKey, pageTuple, cmp, ftm, ftp);
+	}
 }
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/BTree.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/BTree.java
index de0122d..fc73571 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/BTree.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/BTree.java
@@ -99,7 +99,7 @@
         this.leafFrameFactory = leafFrameFactory;       
         this.cmp = cmp;
         this.treeLatch = new ReentrantReadWriteLock(true);
-        this.diskOrderScanPredicate = new RangePredicate(true, null, null, true, true, cmp);             
+        this.diskOrderScanPredicate = new RangePredicate(true, null, null, true, true, cmp, cmp);             
     }
     
     public void create(int fileId, IBTreeLeafFrame leafFrame, IBTreeMetaDataFrame metaFrame) throws Exception {
@@ -407,9 +407,12 @@
     public void search(IBTreeCursor cursor, RangePredicate pred, BTreeOpContext ctx) throws Exception {        
     	ctx.reset();
     	ctx.pred = pred;
-        ctx.cursor = cursor;        
-        if (ctx.pred.getComparator() == null)
-            ctx.pred.setComparator(cmp); // simple index scan
+        ctx.cursor = cursor;       
+        // simple index scan
+        if (ctx.pred.getLowKeyComparator() == null)
+            ctx.pred.setLowKeyComparator(cmp);
+        if (ctx.pred.getHighKeyComparator() == null)
+            ctx.pred.setHighKeyComparator(cmp);
         
         boolean repeatOp = true;
         // we use this loop to deal with possibly multiple operation restarts
@@ -520,7 +523,8 @@
 
     public void insert(ITupleReference tuple, BTreeOpContext ctx) throws Exception {       
     	ctx.reset();
-    	ctx.pred.setComparator(cmp);
+    	ctx.pred.setLowKeyComparator(cmp);
+    	ctx.pred.setHighKeyComparator(cmp);
         ctx.pred.setLowKey(tuple, true);
         ctx.pred.setHighKey(tuple, true);
         ctx.splitKey.reset();
@@ -753,7 +757,8 @@
     
     public void delete(ITupleReference tuple, BTreeOpContext ctx) throws Exception {       
     	ctx.reset();
-    	ctx.pred.setComparator(cmp);
+    	ctx.pred.setLowKeyComparator(cmp);
+    	ctx.pred.setHighKeyComparator(cmp);
         ctx.pred.setLowKey(tuple, true);
         ctx.pred.setHighKey(tuple, true);        
         ctx.splitKey.reset();
@@ -1005,7 +1010,7 @@
                     // the descent
                     boolean repeatOp = true;
                     while (repeatOp && ctx.opRestarts < MAX_RESTARTS) {
-                        int childPageId = ctx.interiorFrame.getChildPageId(ctx.pred, cmp);                                                
+                        int childPageId = ctx.interiorFrame.getChildPageId(ctx.pred, cmp);
                         performOp(childPageId, node, ctx);
 
                         if (!ctx.pageLsns.isEmpty() && ctx.pageLsns.getLast() == RESTART_OP) {
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/BTreeOpContext.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/BTreeOpContext.java
index 390831d..f15ded3 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/BTreeOpContext.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/BTreeOpContext.java
@@ -44,7 +44,7 @@
 		if(op != BTreeOp.BTO_SEARCH) {			
 			smPages = new IntArrayList(treeHeightHint, treeHeightHint);
 			freePages = new IntArrayList(treeHeightHint, treeHeightHint);
-			pred = new RangePredicate(true, null, null, true, true, null);
+			pred = new RangePredicate(true, null, null, true, true, null, null);
 			splitKey = new SplitKey(leafFrame.getTupleWriter().createTupleReference());
 		}
 		else {			
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/DiskOrderScanCursor.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/DiskOrderScanCursor.java
index 0784692..8be8b9c 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/DiskOrderScanCursor.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/DiskOrderScanCursor.java
@@ -112,8 +112,8 @@
         tupleIndex = 0;
         frame.setPage(page);
         RangePredicate pred = (RangePredicate)searchPred;
-		MultiComparator cmp = pred.getComparator();
-		frameTuple.setFieldCount(cmp.getFieldCount());
+		MultiComparator lowKeyCmp = pred.getLowKeyComparator();
+		frameTuple.setFieldCount(lowKeyCmp.getFieldCount());
         boolean leafExists = positionToNextLeaf(false);
         if(!leafExists) {
             throw new Exception("Failed to open disk-order scan cursor for B-tree. Traget B-tree has no leaves.");
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixSlotManager.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixSlotManager.java
index 9b172ba..44dde87 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixSlotManager.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FieldPrefixSlotManager.java
@@ -64,9 +64,10 @@
 	    return FieldPrefixSlotManager.TUPLE_UNCOMPRESSED;
 	}
 	
-	public int findSlot(ITupleReference tuple, IBTreeTupleReference frameTuple, IBTreeTupleReference framePrefixTuple, MultiComparator multiCmp, boolean exact) {				
+	@Override
+	public int findSlot(ITupleReference searchKey, IBTreeTupleReference frameTuple, IBTreeTupleReference framePrefixTuple, MultiComparator multiCmp, FindTupleMode mode, FindTupleNoExactMatchPolicy matchPolicy) {				
 		if(frame.getTupleCount() <= 0) encodeSlotFields(TUPLE_UNCOMPRESSED, GREATEST_SLOT);
-								
+		
 		frameTuple.setFieldCount(multiCmp.getFieldCount());
 		
 	    int prefixMid;
@@ -82,7 +83,7 @@
 	    while(prefixBegin <= prefixEnd) {
 	    	prefixMid = (prefixBegin + prefixEnd) / 2;	    			    		
 	    	framePrefixTuple.resetByTupleIndex(frame, prefixMid);	    		    	
-	    	int cmp = multiCmp.fieldRangeCompare(tuple, framePrefixTuple, 0, framePrefixTuple.getFieldCount());
+	    	int cmp = multiCmp.fieldRangeCompare(searchKey, framePrefixTuple, 0, framePrefixTuple.getFieldCount());
 	    	if(cmp < 0) {	    			
 	    		prefixEnd = prefixMid - 1;
 	    		tuplePrefixSlotNumLbound = prefixMid - 1;	    				    		
@@ -92,9 +93,16 @@
 	    		tuplePrefixSlotNumUbound = prefixMid + 1;
 	    	}
 	    	else {
-	    		tuplePrefixSlotNumLbound = prefixMid;
-	    		tuplePrefixSlotNumUbound = prefixMid;
-	    		prefixMatch = prefixMid;	    			
+	    		if(mode == FindTupleMode.FTM_EXCLUSIVE) {
+        			if(matchPolicy == FindTupleNoExactMatchPolicy.FTP_HIGHER_KEY) prefixBegin = prefixMid + 1;
+        			else prefixEnd = prefixMid - 1;
+        		}
+        		else {        			
+        			tuplePrefixSlotNumLbound = prefixMid;
+    	    		tuplePrefixSlotNumUbound = prefixMid;
+    	    		prefixMatch = prefixMid;
+        		}
+	    			    			
 	    		break;
 	    	}
 	    }
@@ -117,32 +125,47 @@
             int cmp = 0;            
             if(prefixSlotNum == TUPLE_UNCOMPRESSED) {            	            	
             	frameTuple.resetByTupleIndex(frame, tupleMid);
-            	cmp = multiCmp.compare(tuple, frameTuple);      
+            	cmp = multiCmp.compare(searchKey, frameTuple);      
             }
             else {             	
             	if(prefixSlotNum < tuplePrefixSlotNumLbound) cmp = 1;
             	else if(prefixSlotNum > tuplePrefixSlotNumUbound) cmp = -1;
             	else {
             		frameTuple.resetByTupleIndex(frame, tupleMid);
-            		cmp = multiCmp.compare(tuple, frameTuple);            	            	            	
+            		cmp = multiCmp.compare(searchKey, frameTuple);            	            	            	
             	}
             }    
             
             if(cmp < 0) tupleEnd = tupleMid - 1;
             else if(cmp > 0) tupleBegin = tupleMid + 1;
-            else return encodeSlotFields(prefixMatch, tupleMid);
+            else {
+            	if(mode == FindTupleMode.FTM_EXCLUSIVE) {
+        			if(matchPolicy == FindTupleNoExactMatchPolicy.FTP_HIGHER_KEY) tupleBegin = tupleMid + 1;
+        			else tupleEnd = tupleMid - 1;
+        		}
+        		else {
+        			return encodeSlotFields(prefixMatch, tupleMid);
+        		}            	
+            }
         }
 	    
 	    //System.out.println("RECS: " + recBegin + " " + recMid + " " + recEnd);
 	    
-        if(exact) return encodeSlotFields(prefixMatch, GREATEST_SLOT);                       
-        if(tupleBegin > (frame.getTupleCount() - 1)) return encodeSlotFields(prefixMatch, GREATEST_SLOT);
-        
+	    if(mode == FindTupleMode.FTM_EXACT) return encodeSlotFields(prefixMatch, GREATEST_SLOT);
+	    
         // do final comparison to determine whether the search key is greater than all keys or in between some existing keys
-        frameTuple.resetByTupleIndex(frame, tupleBegin);
-        int cmp = multiCmp.compare(tuple, (ITupleReference)frameTuple);        
-        if(cmp < 0) return encodeSlotFields(prefixMatch, tupleBegin);
-        else return encodeSlotFields(prefixMatch, GREATEST_SLOT);
+	    if(matchPolicy == FindTupleNoExactMatchPolicy.FTP_HIGHER_KEY) {
+        	if(tupleBegin > frame.getTupleCount() - 1) return encodeSlotFields(prefixMatch, GREATEST_SLOT);
+        	frameTuple.resetByTupleIndex(frame, tupleBegin);
+        	if(multiCmp.compare(searchKey, frameTuple) < 0) return encodeSlotFields(prefixMatch, tupleBegin);
+            else return encodeSlotFields(prefixMatch, GREATEST_SLOT);
+        }
+        else {
+        	if(tupleEnd < 0) return encodeSlotFields(prefixMatch, GREATEST_SLOT);
+        	frameTuple.resetByTupleIndex(frame, tupleEnd);
+        	if(multiCmp.compare(searchKey, frameTuple) > 0) return encodeSlotFields(prefixMatch, tupleEnd);
+            else return encodeSlotFields(prefixMatch, GREATEST_SLOT);
+        }	            
 	}	
 	
 	public int getPrefixSlotStartOff() {		
@@ -194,15 +217,15 @@
 		this.buf = frame.getBuffer();
 	}
 	
-	public int getPrefixSlotOff(int slotNum) {
-		return getPrefixSlotStartOff() - slotNum * slotSize;
+	public int getPrefixSlotOff(int tupleIndex) {
+		return getPrefixSlotStartOff() - tupleIndex * slotSize;
 	}
 	
-	public int getTupleSlotOff(int slotNum) {
-		return getTupleSlotStartOff() - slotNum * slotSize;
+	public int getTupleSlotOff(int tupleIndex) {
+		return getTupleSlotStartOff() - tupleIndex * slotSize;
 	}
 	
-	public void setPrefixSlot(int slotNum, int slot) {
-		buf.putInt(getPrefixSlotOff(slotNum), slot);
+	public void setPrefixSlot(int tupleIndex, int slot) {
+		buf.putInt(getPrefixSlotOff(tupleIndex), slot);
 	}
 }
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FindSlotMode.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FindSlotMode.java
deleted file mode 100644
index c1d878f..0000000
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FindSlotMode.java
+++ /dev/null
@@ -1,7 +0,0 @@
-package edu.uci.ics.hyracks.storage.am.btree.impls;
-
-public enum FindSlotMode {
-	FSM_INCLUSIVE,
-	FSM_EXCLUSIVE,
-	FSM_EXACT
-}
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FindTupleMode.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FindTupleMode.java
new file mode 100644
index 0000000..79dc343
--- /dev/null
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FindTupleMode.java
@@ -0,0 +1,7 @@
+package edu.uci.ics.hyracks.storage.am.btree.impls;
+
+public enum FindTupleMode {
+	FTM_INCLUSIVE,
+	FTM_EXCLUSIVE,
+	FTM_EXACT
+}
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FindTupleNoExactMatchPolicy.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FindTupleNoExactMatchPolicy.java
new file mode 100644
index 0000000..45f2765
--- /dev/null
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/FindTupleNoExactMatchPolicy.java
@@ -0,0 +1,6 @@
+package edu.uci.ics.hyracks.storage.am.btree.impls;
+
+public enum FindTupleNoExactMatchPolicy {
+	FTP_LOWER_KEY,
+	FTP_HIGHER_KEY
+}
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/OrderedSlotManager.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/OrderedSlotManager.java
index bbbd965..92c91d3 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/OrderedSlotManager.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/OrderedSlotManager.java
@@ -26,7 +26,7 @@
 	private IBTreeFrame frame;
 	
 	@Override
-	public int findTupleIndex(ITupleReference tuple, IBTreeTupleReference pageTuple, MultiComparator multiCmp, FindSlotMode mode) {
+	public int findTupleIndex(ITupleReference searchKey, IBTreeTupleReference frameTuple, MultiComparator multiCmp, FindTupleMode mode, FindTupleNoExactMatchPolicy matchPolicy) {
 		if(frame.getTupleCount() <= 0) return -1;
 				
 		int mid;
@@ -35,9 +35,9 @@
 		
         while(begin <= end) {
             mid = (begin + end) / 2;        	
-        	pageTuple.resetByTupleIndex(frame, mid);
+        	frameTuple.resetByTupleIndex(frame, mid);
         	
-        	int cmp = multiCmp.compare(tuple, pageTuple);
+        	int cmp = multiCmp.compare(searchKey, frameTuple);
         	if(cmp < 0) {
         		end = mid - 1;
         	}
@@ -45,8 +45,9 @@
         		begin = mid + 1;
         	}
         	else {
-        		if(mode == FindSlotMode.FSM_EXCLUSIVE) {
-        			begin = mid + 1;
+        		if(mode == FindTupleMode.FTM_EXCLUSIVE) {
+        			if(matchPolicy == FindTupleNoExactMatchPolicy.FTP_HIGHER_KEY) begin = mid + 1;
+        			else end = mid - 1;
         		}
         		else {        			
         			return mid;
@@ -54,14 +55,20 @@
         	}
         }
                         
-        if(mode == FindSlotMode.FSM_EXACT) return -1;             
-        if(begin > frame.getTupleCount() - 1) return -1;
-                
-        pageTuple.resetByTupleIndex(frame, begin);
-        if(multiCmp.compare(tuple, pageTuple)  < 0)
-        	return begin;
-        else
-        	return -1;
+        if(mode == FindTupleMode.FTM_EXACT) return -1;            
+        
+        if(matchPolicy == FindTupleNoExactMatchPolicy.FTP_HIGHER_KEY) {
+        	if(begin > frame.getTupleCount() - 1) return -1;
+        	frameTuple.resetByTupleIndex(frame, begin);
+        	if(multiCmp.compare(searchKey, frameTuple) < 0) return begin;
+            else return -1;
+        }
+        else {
+        	if(end < 0) return -1;
+        	frameTuple.resetByTupleIndex(frame, end);
+        	if(multiCmp.compare(searchKey, frameTuple) > 0) return end;
+            else return -1;
+        }               
 	}
 	
 	@Override
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/RangePredicate.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/RangePredicate.java
index f2419e0..5b6dea8 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/RangePredicate.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/RangePredicate.java
@@ -27,27 +27,37 @@
 	protected ITupleReference highKey = null;
 	protected boolean lowKeyInclusive = true;
 	protected boolean highKeyInclusive = true;
-	protected MultiComparator cmp;
+	protected MultiComparator lowKeyCmp;
+	protected MultiComparator highKeyCmp;
 	
 	public RangePredicate() {
 	}
 	
 	public RangePredicate(boolean isForward, ITupleReference lowKey, ITupleReference highKey, 
-			boolean lowKeyInclusive, boolean highKeyInclusive, MultiComparator cmp) {
+			boolean lowKeyInclusive, boolean highKeyInclusive, MultiComparator lowKeyCmp, MultiComparator highKeyCmp) {
 		this.isForward = isForward;
 		this.lowKey = lowKey;
 		this.highKey = highKey;
 		this.lowKeyInclusive = lowKeyInclusive;
 		this.highKeyInclusive = highKeyInclusive;
-		this.cmp = cmp;
+		this.lowKeyCmp = lowKeyCmp;
+		this.highKeyCmp = highKeyCmp;
 	}
 	
-	public MultiComparator getComparator() {
-		return cmp;
+	public MultiComparator getLowKeyComparator() {
+		return lowKeyCmp;
 	}
 	
-	public void setComparator(MultiComparator cmp) {
-		this.cmp = cmp;
+	public MultiComparator getHighKeyComparator() {
+		return highKeyCmp;
+	}
+	
+	public void setLowKeyComparator(MultiComparator lowKeyCmp) {
+		this.lowKeyCmp = lowKeyCmp;
+	}
+	
+	public void setHighKeyComparator(MultiComparator highKeyCmp) {
+		this.highKeyCmp = highKeyCmp;
 	}
 	
 	public boolean isForward() {
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/RangeSearchCursor.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/RangeSearchCursor.java
index 9970114..1299f04 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/RangeSearchCursor.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/RangeSearchCursor.java
@@ -15,6 +15,7 @@
 
 package edu.uci.ics.hyracks.storage.am.btree.impls;
 
+import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
 import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeCursor;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeLeafFrame;
@@ -25,19 +26,33 @@
 import edu.uci.ics.hyracks.storage.common.file.FileInfo;
 
 public class RangeSearchCursor implements IBTreeCursor {
-
-	private ISearchPredicate searchPred = null;	
-	private int tupleIndex = 0;
+	
 	private int fileId = -1;
 	private ICachedPage page = null;
 	private IBTreeLeafFrame frame = null;
 	private IBufferCache bufferCache = null;
 	
+	private int tupleIndex = 0;
+	private int stopTupleIndex;
+	private int tupleIndexInc = 0;
+	
+	private FindTupleMode lowKeyFtm;
+	private FindTupleMode highKeyFtm;
+	
+	private FindTupleNoExactMatchPolicy lowKeyFtp;
+	private FindTupleNoExactMatchPolicy highKeyFtp;		
+	
 	private IBTreeTupleReference frameTuple;
-			
+					
+	private RangePredicate pred;
+	private MultiComparator lowKeyCmp;
+	private MultiComparator highKeyCmp;
+	private ITupleReference lowKey;
+	private ITupleReference highKey;
+	
 	public RangeSearchCursor(IBTreeLeafFrame frame) {
 		this.frame = frame;
-		this.frameTuple = frame.createTupleReference();
+		this.frameTuple = frame.createTupleReference();		
 	}
 	
 	@Override
@@ -49,88 +64,107 @@
 		}
 	}
 	
-	public ITupleReference getTuple() {
-		return frameTuple;
+	public ITupleReference getTuple() {		
+		return frameTuple;		
 	}
 	
 	@Override
 	public ICachedPage getPage() {
 		return page;
-	}
+	}	
+	
+	private void fetchNextLeafPage(int nextLeafPage) throws HyracksDataException {
+		ICachedPage nextLeaf = bufferCache.pin(FileInfo.getDiskPageId(fileId, nextLeafPage), false);
+		nextLeaf.acquireReadLatch();
+
+		page.releaseReadLatch();
+		bufferCache.unpin(page);
 		
+		page = nextLeaf;
+		frame.setPage(page);
+	}
+	
 	@Override
 	public boolean hasNext() throws Exception {
-		if(tupleIndex >= frame.getTupleCount()) {
-			int nextLeafPage = -1;
-			if(searchPred.isForward()) {
-				nextLeafPage = frame.getNextLeaf();
-			}
-			else {
-				nextLeafPage = frame.getPrevLeaf();
-			}
-						
-			if(nextLeafPage >= 0) {			
-				ICachedPage nextLeaf = bufferCache.pin(FileInfo.getDiskPageId(fileId, nextLeafPage), false);
-				nextLeaf.acquireReadLatch();
-								
-				page.releaseReadLatch();
-				bufferCache.unpin(page);
 				
-				page = nextLeaf;
-				frame.setPage(page);
-				
-				tupleIndex = 0;
+		if(pred.isForward()) {
+			if(tupleIndex >= frame.getTupleCount()) {				
+				int nextLeafPage = frame.getNextLeaf();				
+				if(nextLeafPage >= 0) {								
+					fetchNextLeafPage(nextLeafPage);					
+					tupleIndex = 0;
+										
+					stopTupleIndex = getHighKeyIndex();
+					if(stopTupleIndex < 0) return false;		
+				}
+				else {
+					return false;
+				}
 			}
-			else {
-				return false;
-			}
-		}
-		
-		// in any case compare current key
-		RangePredicate pred = (RangePredicate)searchPred;
-		MultiComparator cmp = pred.getComparator();
-		if(searchPred.isForward()) {
-			ITupleReference highKey = pred.getHighKey();		
-			frameTuple.resetByTupleIndex(frame, tupleIndex);
-									
-			if(highKey == null) return true;
 			
-			if(pred.isHighKeyInclusive()) {
-				if(cmp.compare(highKey, frameTuple) < 0) {
-					return false;
-				}				
+			frameTuple.resetByTupleIndex(frame, tupleIndex);
+			if(highKey == null || tupleIndex <= stopTupleIndex) {				
+				return true;									
 			}
-			else {
-				if(cmp.compare(highKey, frameTuple) <= 0) {
-					return false;
-				}				
-			}
-			return true;
+			else return false;
 		}
 		else {
-			ITupleReference lowKey = pred.getLowKey();						
-			frameTuple.resetByTupleIndex(frame, frame.getTupleCount() - tupleIndex - 1);
-			if(lowKey == null) return true;
+			if(tupleIndex < 0) {				
+				int nextLeafPage = frame.getPrevLeaf();				
+				if(nextLeafPage >= 0) {								
+					fetchNextLeafPage(nextLeafPage);
+					tupleIndex = frame.getTupleCount() - 1;
+					
+					stopTupleIndex = getLowKeyIndex();										
+					if(stopTupleIndex >= frame.getTupleCount()) return false;										
+				}
+				else {
+					return false;
+				}
+			}	
 			
-			if(pred.isLowKeyInclusive()) {
-				if(cmp.compare(lowKey, frameTuple) > 0) {
-					return false;
-				}
+			frameTuple.resetByTupleIndex(frame, tupleIndex);
+			if(lowKey == null || tupleIndex >= stopTupleIndex) {				
+				return true;
 			}
-			else {
-				if(cmp.compare(lowKey, frameTuple) >= 0) {
-					return false;
-				}
-			}						
-			return true;
+			else return false;
 		}		
 	}
 
 	@Override
 	public void next() throws Exception {		
-		tupleIndex++;
+		tupleIndex += tupleIndexInc;				
 	}
 	
+	private int getLowKeyIndex() {
+		int index;
+		if(lowKey == null) index = 0;
+		else {
+			index = frame.findTupleIndex(lowKey, frameTuple, lowKeyCmp, lowKeyFtm, lowKeyFtp);
+			if(pred.lowKeyInclusive) {															
+				index++;
+			}
+			else {
+				if(index < 0) index = frame.getTupleCount();
+			}
+		}
+		return index;
+	}	
+	
+	private int getHighKeyIndex() {
+		int index;
+		if(highKey == null) index = frame.getTupleCount() - 1;
+		else {				
+			index = frame.findTupleIndex(highKey, frameTuple, highKeyCmp, highKeyFtm, highKeyFtp);
+			if(pred.highKeyInclusive) {										
+				if(index < 0) index = frame.getTupleCount() - 1;
+				else index--;
+			}
+		}
+		return index;
+	}
+
+	
 	@Override
 	public void open(ICachedPage page, ISearchPredicate searchPred) throws Exception {		
 		// in case open is called multiple times without closing
@@ -138,65 +172,54 @@
 			this.page.releaseReadLatch();
 			bufferCache.unpin(this.page);
 		}
-						
-		this.searchPred = searchPred;
+		
 		this.page = page;
 		frame.setPage(page);					
 		
-		// position tupleIndex to the first appropriate key
-		// TODO: can be done more efficiently with binary search but this needs some thinking/refactoring
-		RangePredicate pred = (RangePredicate)searchPred;
-		MultiComparator cmp = pred.getComparator();
-		frameTuple.setFieldCount(cmp.getFieldCount());
-		if(searchPred.isForward()) {
-			ITupleReference lowKey = pred.getLowKey();
-			
-			frameTuple.resetByTupleIndex(frame, tupleIndex);
-			if(lowKey == null) return; // null means -infinity
-			
-			if(pred.isLowKeyInclusive()) {
-				while(cmp.compare(lowKey, frameTuple) > 0) {
-					tupleIndex++;
-					if(tupleIndex >= frame.getTupleCount()) break;
-					frameTuple.resetByTupleIndex(frame, tupleIndex);
-				}				
-			}
-			else {
-				while(cmp.compare(lowKey, frameTuple) >= 0) {
-					tupleIndex++;
-					if(tupleIndex >= frame.getTupleCount()) break;
-					frameTuple.resetByTupleIndex(frame, tupleIndex);
-				}
-			}
+		pred = (RangePredicate)searchPred;
+		lowKeyCmp = pred.getLowKeyComparator();
+		highKeyCmp = pred.getHighKeyComparator();
+		
+		lowKey = pred.getLowKey();
+		highKey = pred.getHighKey();
+		
+		// field count must be identical for lowKeyCmp and highKeyCmp (key count may be different)
+		frameTuple.setFieldCount(lowKeyCmp.getFieldCount());
+		
+		// init
+		lowKeyFtm = FindTupleMode.FTM_EXCLUSIVE;
+		if(pred.lowKeyInclusive) {
+			lowKeyFtp = FindTupleNoExactMatchPolicy.FTP_LOWER_KEY;
 		}
 		else {
-			ITupleReference highKey = pred.getHighKey();
-			
-			frameTuple.resetByTupleIndex(frame, frame.getTupleCount() - tupleIndex - 1);
-			if(highKey == null) return; // null means +infinity
-			    
-			if(pred.isHighKeyInclusive()) {				
-				while(cmp.compare(highKey, frameTuple) < 0) {					
-					tupleIndex++;
-					if(tupleIndex >= frame.getTupleCount()) break;
-					frameTuple.resetByTupleIndex(frame, frame.getTupleCount() - tupleIndex - 1);					
-				}				
-			}
-			else {
-				while(cmp.compare(highKey, frameTuple) <= 0) {				
-					tupleIndex++;
-					if(tupleIndex >= frame.getTupleCount()) break;
-					frameTuple.resetByTupleIndex(frame, frame.getTupleCount() - tupleIndex - 1);
-				}
-			}
+			lowKeyFtp = FindTupleNoExactMatchPolicy.FTP_HIGHER_KEY;
 		}
+		
+		highKeyFtm = FindTupleMode.FTM_EXCLUSIVE;
+		if(pred.highKeyInclusive) {
+			highKeyFtp = FindTupleNoExactMatchPolicy.FTP_HIGHER_KEY;
+		}
+		else {
+			highKeyFtp = FindTupleNoExactMatchPolicy.FTP_LOWER_KEY;
+		}
+		
+		if(pred.isForward()) {						
+			tupleIndex = getLowKeyIndex();
+			stopTupleIndex = getHighKeyIndex();
+			tupleIndexInc = 1;
+		}
+		else {			
+			tupleIndex = getHighKeyIndex();
+			stopTupleIndex = getLowKeyIndex();
+			tupleIndexInc = -1;
+		}		
 	}
 	
 	@Override
 	public void reset() {
 		tupleIndex = 0;
 		page = null;	
-		searchPred = null;
+		pred = null;
 	}
 
     @Override
@@ -208,4 +231,4 @@
     public void setFileId(int fileId) {
         this.fileId = fileId;
     }	
-}
+}
\ No newline at end of file
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/tuples/TypeAwareTupleReference.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/tuples/TypeAwareTupleReference.java
index 1012800..89325de 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/tuples/TypeAwareTupleReference.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/tuples/TypeAwareTupleReference.java
@@ -86,7 +86,7 @@
 	@Override
 	public int getFieldLength(int fIdx) {
 		if(fIdx == 0) {
-			return decodedFieldSlots[0];			
+			return decodedFieldSlots[0];		
 		}
 		else {
 			return decodedFieldSlots[fIdx] - decodedFieldSlots[fIdx-1];
diff --git a/hyracks-storage-am-btree/src/test/java/edu/uci/ics/hyracks/storage/am/btree/BTreeTest.java b/hyracks-storage-am-btree/src/test/java/edu/uci/ics/hyracks/storage/am/btree/BTreeTest.java
index 78b6d4d78..0f083d6 100644
--- a/hyracks-storage-am-btree/src/test/java/edu/uci/ics/hyracks/storage/am/btree/BTreeTest.java
+++ b/hyracks-storage-am-btree/src/test/java/edu/uci/ics/hyracks/storage/am/btree/BTreeTest.java
@@ -219,7 +219,7 @@
         
         print("ORDERED SCAN:\n");
         IBTreeCursor scanCursor = new RangeSearchCursor(leafFrame);
-        RangePredicate nullPred = new RangePredicate(true, null, null, true, true, null);
+        RangePredicate nullPred = new RangePredicate(true, null, null, true, true, null, null);
         BTreeOpContext searchOpCtx = btree.createOpContext(BTreeOp.BTO_SEARCH, leafFrame, interiorFrame, null);
         btree.search(scanCursor, nullPred, searchOpCtx);
         try {
@@ -294,7 +294,7 @@
         searchCmps[0] = IntegerBinaryComparatorFactory.INSTANCE.createBinaryComparator();
         MultiComparator searchCmp = new MultiComparator(typeTraits, searchCmps);
         
-        RangePredicate rangePred = new RangePredicate(true, lowKey, highKey, true, true, searchCmp);
+        RangePredicate rangePred = new RangePredicate(true, lowKey, highKey, true, true, searchCmp, searchCmp);
         btree.search(rangeCursor, rangePred, searchOpCtx);
         
         try {
@@ -425,7 +425,7 @@
         // try a simple index scan
         print("ORDERED SCAN:\n");        
         IBTreeCursor scanCursor = new RangeSearchCursor(leafFrame);
-        RangePredicate nullPred = new RangePredicate(true, null, null, true, true, null);
+        RangePredicate nullPred = new RangePredicate(true, null, null, true, true, null, null);
         BTreeOpContext searchOpCtx = btree.createOpContext(BTreeOp.BTO_SEARCH, leafFrame, interiorFrame, null);
         btree.search(scanCursor, nullPred, searchOpCtx);
         
@@ -482,7 +482,7 @@
         searchCmps[0] = IntegerBinaryComparatorFactory.INSTANCE.createBinaryComparator();       
         MultiComparator searchCmp = new MultiComparator(typeTraits, searchCmps); // use only a single comparator for searching
         
-        RangePredicate rangePred = new RangePredicate(true, lowKey, highKey, true, true, searchCmp);
+        RangePredicate rangePred = new RangePredicate(true, lowKey, highKey, true, true, searchCmp, searchCmp);
         btree.search(rangeCursor, rangePred, searchOpCtx);
         
         try {
@@ -604,7 +604,7 @@
     	// ordered scan
         print("ORDERED SCAN:\n");        
         IBTreeCursor scanCursor = new RangeSearchCursor(leafFrame);
-        RangePredicate nullPred = new RangePredicate(true, null, null, true, true, null);
+        RangePredicate nullPred = new RangePredicate(true, null, null, true, true, null, null);
         BTreeOpContext searchOpCtx = btree.createOpContext(BTreeOp.BTO_SEARCH, leafFrame, interiorFrame, null);
         btree.search(scanCursor, nullPred, searchOpCtx);
         
@@ -661,7 +661,7 @@
         searchCmps[0] = UTF8StringBinaryComparatorFactory.INSTANCE.createBinaryComparator();
         MultiComparator searchCmp = new MultiComparator(typeTraits, searchCmps);
         
-        RangePredicate rangePred = new RangePredicate(true, lowKey, highKey, true, true, searchCmp);
+        RangePredicate rangePred = new RangePredicate(true, lowKey, highKey, true, true, searchCmp, searchCmp);
         btree.search(rangeCursor, rangePred, searchOpCtx);
 
         try {
@@ -988,7 +988,7 @@
         MultiComparator searchCmp = new MultiComparator(typeTraits, searchCmps);
         
         // TODO: check when searching backwards
-        RangePredicate rangePred = new RangePredicate(true, lowKey, highKey, true, true, searchCmp);
+        RangePredicate rangePred = new RangePredicate(true, lowKey, highKey, true, true, searchCmp, searchCmp);
         BTreeOpContext searchOpCtx = btree.createOpContext(BTreeOp.BTO_SEARCH, leafFrame, interiorFrame, null);
         btree.search(rangeCursor, rangePred, searchOpCtx);
         
@@ -1150,7 +1150,7 @@
 
         print("ORDERED SCAN:\n");
         IBTreeCursor scanCursor = new RangeSearchCursor(leafFrame);
-        RangePredicate nullPred = new RangePredicate(true, null, null, true, true, null);
+        RangePredicate nullPred = new RangePredicate(true, null, null, true, true, null, null);
         BTreeOpContext searchOpCtx = btree.createOpContext(BTreeOp.BTO_SEARCH, leafFrame, interiorFrame, null);
         btree.search(scanCursor, nullPred, searchOpCtx);
 
@@ -1213,7 +1213,7 @@
                 
         //print("INDEX RANGE SEARCH ON: " + cmp.printKey(lowKey, 0) + " " + cmp.printKey(highKey, 0) + "\n");                
         
-        RangePredicate rangePred = new RangePredicate(true, lowKey, highKey, true, true, searchCmp);
+        RangePredicate rangePred = new RangePredicate(true, lowKey, highKey, true, true, searchCmp, searchCmp);
         btree.search(rangeCursor, rangePred, searchOpCtx);
         
         try {
diff --git a/hyracks-storage-am-btree/src/test/java/edu/uci/ics/hyracks/storage/am/btree/RangeSearchCursorTest.java b/hyracks-storage-am-btree/src/test/java/edu/uci/ics/hyracks/storage/am/btree/RangeSearchCursorTest.java
new file mode 100644
index 0000000..9bfaa6b
--- /dev/null
+++ b/hyracks-storage-am-btree/src/test/java/edu/uci/ics/hyracks/storage/am/btree/RangeSearchCursorTest.java
@@ -0,0 +1,543 @@
+package edu.uci.ics.hyracks.storage.am.btree;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInput;
+import java.io.DataInputStream;
+import java.io.DataOutput;
+import java.io.File;
+import java.io.RandomAccessFile;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Random;
+import java.util.TreeSet;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import edu.uci.ics.hyracks.api.comm.IFrameTupleAccessor;
+import edu.uci.ics.hyracks.api.context.IHyracksContext;
+import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparator;
+import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
+import edu.uci.ics.hyracks.api.dataflow.value.ITypeTrait;
+import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
+import edu.uci.ics.hyracks.api.dataflow.value.TypeTrait;
+import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
+import edu.uci.ics.hyracks.control.nc.runtime.RootHyracksContext;
+import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
+import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAccessor;
+import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAppender;
+import edu.uci.ics.hyracks.dataflow.common.data.accessors.FrameTupleReference;
+import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
+import edu.uci.ics.hyracks.dataflow.common.data.comparators.IntegerBinaryComparatorFactory;
+import edu.uci.ics.hyracks.dataflow.common.data.marshalling.IntegerSerializerDeserializer;
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeCursor;
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeInteriorFrame;
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeInteriorFrameFactory;
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeLeafFrame;
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeLeafFrameFactory;
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeMetaDataFrame;
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeMetaDataFrameFactory;
+import edu.uci.ics.hyracks.storage.am.btree.frames.FieldPrefixNSMLeafFrameFactory;
+import edu.uci.ics.hyracks.storage.am.btree.frames.MetaDataFrameFactory;
+import edu.uci.ics.hyracks.storage.am.btree.frames.NSMInteriorFrameFactory;
+import edu.uci.ics.hyracks.storage.am.btree.frames.NSMLeafFrameFactory;
+import edu.uci.ics.hyracks.storage.am.btree.impls.BTree;
+import edu.uci.ics.hyracks.storage.am.btree.impls.BTreeException;
+import edu.uci.ics.hyracks.storage.am.btree.impls.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.buffercache.BufferCache;
+import edu.uci.ics.hyracks.storage.common.buffercache.ClockPageReplacementStrategy;
+import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
+import edu.uci.ics.hyracks.storage.common.buffercache.ICacheMemoryAllocator;
+import edu.uci.ics.hyracks.storage.common.buffercache.IPageReplacementStrategy;
+import edu.uci.ics.hyracks.storage.common.file.FileInfo;
+import edu.uci.ics.hyracks.storage.common.file.FileManager;
+
+public class RangeSearchCursorTest {
+	private static final int PAGE_SIZE = 256;
+    private static final int NUM_PAGES = 10;
+    private static final int HYRACKS_FRAME_SIZE = 128;
+    
+    private String tmpDir = System.getProperty("java.io.tmpdir");
+    
+    private void print(String str) {
+    	System.out.print(str);
+    }       
+    
+    public class BufferAllocator implements ICacheMemoryAllocator {
+        @Override
+        public ByteBuffer[] allocate(int pageSize, int numPages) {
+            ByteBuffer[] buffers = new ByteBuffer[numPages];
+            for (int i = 0; i < numPages; ++i) {
+                buffers[i] = ByteBuffer.allocate(pageSize);
+            }
+            return buffers;
+        }
+    }
+            
+    // declare fields
+    int fieldCount = 2;
+    ITypeTrait[] typeTraits = new ITypeTrait[fieldCount];
+        
+    TypeAwareTupleWriterFactory tupleWriterFactory = new TypeAwareTupleWriterFactory(typeTraits);        
+    IBTreeLeafFrameFactory leafFrameFactory = new NSMLeafFrameFactory(tupleWriterFactory);
+    IBTreeInteriorFrameFactory interiorFrameFactory = new NSMInteriorFrameFactory(tupleWriterFactory);
+    IBTreeMetaDataFrameFactory metaFrameFactory = new MetaDataFrameFactory();
+    
+    IBTreeLeafFrame leafFrame = leafFrameFactory.getFrame();
+    IBTreeInteriorFrame interiorFrame = interiorFrameFactory.getFrame();
+    IBTreeMetaDataFrame metaFrame = metaFrameFactory.getFrame();                
+    
+    IHyracksContext ctx = new RootHyracksContext(HYRACKS_FRAME_SIZE);        
+    ByteBuffer frame = ctx.getResourceManager().allocateFrame();
+	FrameTupleAppender appender = new FrameTupleAppender(ctx);	
+	
+	ISerializerDeserializer[] recDescSers = { IntegerSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE};
+	RecordDescriptor recDesc = new RecordDescriptor(recDescSers);
+	IFrameTupleAccessor accessor = new FrameTupleAccessor(ctx, recDesc);	
+	FrameTupleReference tuple = new FrameTupleReference();
+    
+	Random rnd = new Random(50);    
+	
+    @Before
+    public void setUp() {
+    	typeTraits[0] = new TypeTrait(4);
+        typeTraits[1] = new TypeTrait(4);
+        accessor.reset(frame);
+    }       
+    
+    @Test
+    public void uniqueIndexTest() throws Exception {
+    	    	
+    	System.out.println("TESTING RANGE SEARCH CURSOR ON UNIQUE INDEX");
+    	
+        FileManager fileManager = new FileManager();
+        ICacheMemoryAllocator allocator = new BufferAllocator();
+        IPageReplacementStrategy prs = new ClockPageReplacementStrategy();
+        IBufferCache bufferCache = new BufferCache(allocator, prs, fileManager, PAGE_SIZE, NUM_PAGES);
+        
+        File f = new File(tmpDir + "/" + "btreetest.bin");
+        RandomAccessFile raf = new RandomAccessFile(f, "rw");
+        int fileId = 0;
+        FileInfo fi = new FileInfo(fileId, raf);
+        fileManager.registerFile(fi);
+                
+        // declare keys
+        int keyFieldCount = 1;
+        IBinaryComparator[] cmps = new IBinaryComparator[keyFieldCount];
+        cmps[0] = IntegerBinaryComparatorFactory.INSTANCE.createBinaryComparator();    
+                       
+        MultiComparator cmp = new MultiComparator(typeTraits, cmps);
+                
+        BTree btree = new BTree(bufferCache, interiorFrameFactory, leafFrameFactory, cmp);
+        btree.create(fileId, leafFrame, metaFrame);
+        btree.open(fileId);
+                
+        ArrayTupleBuilder tb = new ArrayTupleBuilder(cmp.getFieldCount());
+    	DataOutput dos = tb.getDataOutput();
+		
+		BTreeOpContext insertOpCtx = btree.createOpContext(BTreeOp.BTO_INSERT, leafFrame, interiorFrame, metaFrame);
+		
+		// generate keys
+		int numKeys = 50;
+		int maxKey = 1000;
+		TreeSet<Integer> uniqueKeys = new TreeSet<Integer>(); 
+		ArrayList<Integer> keys = new ArrayList<Integer>();			
+		while(uniqueKeys.size() < numKeys) {
+			int key = rnd.nextInt() % maxKey;
+			uniqueKeys.add(key);
+		}
+		for(Integer i : uniqueKeys) {
+			keys.add(i);
+		}		
+				
+		// insert keys into btree
+        for (int i = 0; i < keys.size(); i++) {        	        	
+        	
+        	tb.reset();
+        	IntegerSerializerDeserializer.INSTANCE.serialize(keys.get(i).intValue(), dos);
+        	tb.addFieldEndOffset();
+        	IntegerSerializerDeserializer.INSTANCE.serialize(i, dos);
+        	tb.addFieldEndOffset();        	
+        	        	
+        	appender.reset(frame, true);
+        	appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+        	
+        	tuple.reset(accessor, 0);
+        	        	
+            try {                                
+                btree.insert(tuple, insertOpCtx);
+            } catch (BTreeException e) {            	
+            } catch (Exception e) {
+            	e.printStackTrace();
+            }           
+        }        
+        
+        //btree.printTree(leafFrame, interiorFrame, recDescSers);
+        
+        int minSearchKey = -100;
+        int maxSearchKey = 100;
+                
+        //System.out.println("STARTING SEARCH TESTS");
+        
+        // forward searches
+        performSearches(keys, btree, leafFrame, interiorFrame, minSearchKey, maxSearchKey, true, true, true, false);        
+        performSearches(keys, btree, leafFrame, interiorFrame, minSearchKey, maxSearchKey, true, false, true, false);        
+        performSearches(keys, btree, leafFrame, interiorFrame, minSearchKey, maxSearchKey, true, true, false, false);        
+        performSearches(keys, btree, leafFrame, interiorFrame, minSearchKey, maxSearchKey, true, true, true, false);        
+        
+        // backward searches
+        performSearches(keys, btree, leafFrame, interiorFrame, minSearchKey, maxSearchKey, false, true, true, false);
+        performSearches(keys, btree, leafFrame, interiorFrame, minSearchKey, maxSearchKey, false, false, true, false);
+        performSearches(keys, btree, leafFrame, interiorFrame, minSearchKey, maxSearchKey, false, true, false, false);
+        performSearches(keys, btree, leafFrame, interiorFrame, minSearchKey, maxSearchKey, false, true, true, false);
+        
+        btree.close();
+
+        bufferCache.close();
+        fileManager.close();
+    }    
+    
+    @Test
+    public void nonUniqueIndexTest() throws Exception {
+    	    	
+    	System.out.println("TESTING RANGE SEARCH CURSOR ON NONUNIQUE INDEX");
+    	
+        FileManager fileManager = new FileManager();
+        ICacheMemoryAllocator allocator = new BufferAllocator();
+        IPageReplacementStrategy prs = new ClockPageReplacementStrategy();
+        IBufferCache bufferCache = new BufferCache(allocator, prs, fileManager, PAGE_SIZE, NUM_PAGES);
+        
+        File f = new File(tmpDir + "/" + "btreetest.bin");
+        RandomAccessFile raf = new RandomAccessFile(f, "rw");
+        int fileId = 0;
+        FileInfo fi = new FileInfo(fileId, raf);
+        fileManager.registerFile(fi);
+                
+        // declare keys
+        int keyFieldCount = 2;
+        IBinaryComparator[] cmps = new IBinaryComparator[keyFieldCount];
+        cmps[0] = IntegerBinaryComparatorFactory.INSTANCE.createBinaryComparator();
+        cmps[1] = IntegerBinaryComparatorFactory.INSTANCE.createBinaryComparator();
+                    
+        MultiComparator cmp = new MultiComparator(typeTraits, cmps);
+                
+        BTree btree = new BTree(bufferCache, interiorFrameFactory, leafFrameFactory, cmp);
+        btree.create(fileId, leafFrame, metaFrame);
+        btree.open(fileId);
+                
+        ArrayTupleBuilder tb = new ArrayTupleBuilder(cmp.getFieldCount());
+    	DataOutput dos = tb.getDataOutput();
+		
+		BTreeOpContext insertOpCtx = btree.createOpContext(BTreeOp.BTO_INSERT, leafFrame, interiorFrame, metaFrame);
+		
+		// generate keys
+		int numKeys = 50;
+		int maxKey = 10;		
+		ArrayList<Integer> keys = new ArrayList<Integer>();			
+		for(int i = 0; i < numKeys; i++) {
+			int k = rnd.nextInt() % maxKey;
+			keys.add(k);
+		}
+		Collections.sort(keys);
+		
+		// insert keys into btree
+        for (int i = 0; i < keys.size(); i++) {        	        	
+        	
+        	tb.reset();
+        	IntegerSerializerDeserializer.INSTANCE.serialize(keys.get(i).intValue(), dos);
+        	tb.addFieldEndOffset();
+        	IntegerSerializerDeserializer.INSTANCE.serialize(i, dos);
+        	tb.addFieldEndOffset();        	
+        	        	
+        	appender.reset(frame, true);
+        	appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+        	
+        	tuple.reset(accessor, 0);
+        	        	
+            try {                                
+                btree.insert(tuple, insertOpCtx);
+            } catch (BTreeException e) {            	
+            } catch (Exception e) {
+            	e.printStackTrace();
+            }           
+        }        
+        
+        //btree.printTree(leafFrame, interiorFrame, recDescSers);
+        
+        int minSearchKey = -100;
+        int maxSearchKey = 100;
+                
+        //System.out.println("STARTING SEARCH TESTS");
+        
+        // forward searches
+        performSearches(keys, btree, leafFrame, interiorFrame, minSearchKey, maxSearchKey, true, true, true, false);        
+        performSearches(keys, btree, leafFrame, interiorFrame, minSearchKey, maxSearchKey, true, false, true, false);        
+        performSearches(keys, btree, leafFrame, interiorFrame, minSearchKey, maxSearchKey, true, true, false, false);        
+        performSearches(keys, btree, leafFrame, interiorFrame, minSearchKey, maxSearchKey, true, true, true, false);        
+        
+        // backward searches
+        performSearches(keys, btree, leafFrame, interiorFrame, minSearchKey, maxSearchKey, false, true, true, false);
+        performSearches(keys, btree, leafFrame, interiorFrame, minSearchKey, maxSearchKey, false, false, true, false);
+        performSearches(keys, btree, leafFrame, interiorFrame, minSearchKey, maxSearchKey, false, true, false, false);
+        performSearches(keys, btree, leafFrame, interiorFrame, minSearchKey, maxSearchKey, false, true, true, false);
+        
+        btree.close();
+
+        bufferCache.close();
+        fileManager.close();
+    }    
+    
+    @Test
+    public void nonUniqueFieldPrefixIndexTest() throws Exception {
+    	    	
+    	System.out.println("TESTING RANGE SEARCH CURSOR ON NONUNIQUE FIELD-PREFIX COMPRESSED INDEX");
+    	
+    	IBTreeLeafFrameFactory leafFrameFactory = new FieldPrefixNSMLeafFrameFactory(tupleWriterFactory);
+    	IBTreeLeafFrame leafFrame = leafFrameFactory.getFrame();
+    	
+        FileManager fileManager = new FileManager();
+        ICacheMemoryAllocator allocator = new BufferAllocator();
+        IPageReplacementStrategy prs = new ClockPageReplacementStrategy();
+        IBufferCache bufferCache = new BufferCache(allocator, prs, fileManager, PAGE_SIZE, NUM_PAGES);
+        
+        File f = new File(tmpDir + "/" + "btreetest.bin");
+        RandomAccessFile raf = new RandomAccessFile(f, "rw");
+        int fileId = 0;
+        FileInfo fi = new FileInfo(fileId, raf);
+        fileManager.registerFile(fi);
+                
+        // declare keys
+        int keyFieldCount = 2;
+        IBinaryComparator[] cmps = new IBinaryComparator[keyFieldCount];
+        cmps[0] = IntegerBinaryComparatorFactory.INSTANCE.createBinaryComparator();
+        cmps[1] = IntegerBinaryComparatorFactory.INSTANCE.createBinaryComparator();
+                    
+        MultiComparator cmp = new MultiComparator(typeTraits, cmps);
+                
+        BTree btree = new BTree(bufferCache, interiorFrameFactory, leafFrameFactory, cmp);
+        btree.create(fileId, leafFrame, metaFrame);
+        btree.open(fileId);        
+        
+        ArrayTupleBuilder tb = new ArrayTupleBuilder(cmp.getFieldCount());
+    	DataOutput dos = tb.getDataOutput();
+		
+		BTreeOpContext insertOpCtx = btree.createOpContext(BTreeOp.BTO_INSERT, leafFrame, interiorFrame, metaFrame);
+		
+		// generate keys
+		int numKeys = 50;
+		int maxKey = 10;		
+		ArrayList<Integer> keys = new ArrayList<Integer>();			
+		for(int i = 0; i < numKeys; i++) {
+			int k = rnd.nextInt() % maxKey;
+			keys.add(k);
+		}
+		Collections.sort(keys);
+		
+		// insert keys into btree
+        for (int i = 0; i < keys.size(); i++) {        	        	
+        	
+        	tb.reset();
+        	IntegerSerializerDeserializer.INSTANCE.serialize(keys.get(i).intValue(), dos);
+        	tb.addFieldEndOffset();
+        	IntegerSerializerDeserializer.INSTANCE.serialize(i, dos);
+        	tb.addFieldEndOffset();        	
+        	        	
+        	appender.reset(frame, true);
+        	appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+        	
+        	tuple.reset(accessor, 0);
+        	        	
+            try {                                
+                btree.insert(tuple, insertOpCtx);
+            } catch (BTreeException e) {            	
+            } catch (Exception e) {
+            	e.printStackTrace();
+            }           
+        }        
+        
+        //btree.printTree(leafFrame, interiorFrame, recDescSers);
+        
+        int minSearchKey = -100;
+        int maxSearchKey = 100;
+                
+        //System.out.println("STARTING SEARCH TESTS");
+        
+        // forward searches
+        performSearches(keys, btree, leafFrame, interiorFrame, minSearchKey, maxSearchKey, true, true, true, false);        
+        performSearches(keys, btree, leafFrame, interiorFrame, minSearchKey, maxSearchKey, true, false, true, false);        
+        performSearches(keys, btree, leafFrame, interiorFrame, minSearchKey, maxSearchKey, true, true, false, false);        
+        performSearches(keys, btree, leafFrame, interiorFrame, minSearchKey, maxSearchKey, true, true, true, false);        
+        
+        // backward searches
+        performSearches(keys, btree, leafFrame, interiorFrame, minSearchKey, maxSearchKey, false, true, true, false);
+        performSearches(keys, btree, leafFrame, interiorFrame, minSearchKey, maxSearchKey, false, false, true, false);
+        performSearches(keys, btree, leafFrame, interiorFrame, minSearchKey, maxSearchKey, false, true, false, false);
+        performSearches(keys, btree, leafFrame, interiorFrame, minSearchKey, maxSearchKey, false, true, true, false);
+        
+        btree.close();
+
+        bufferCache.close();
+        fileManager.close();
+    }    
+    
+    public RangePredicate createRangePredicate(int lk, int hk, boolean isForward, boolean lowKeyInclusive, boolean highKeyInclusive, MultiComparator cmp, ITypeTrait[] typeTraits) throws HyracksDataException {        
+    	// build low and high keys
+    	ArrayTupleBuilder ktb = new ArrayTupleBuilder(1);
+		DataOutput kdos = ktb.getDataOutput();
+		
+		ISerializerDeserializer[] keyDescSers = { IntegerSerializerDeserializer.INSTANCE };
+		RecordDescriptor keyDesc = new RecordDescriptor(keyDescSers);
+		IFrameTupleAccessor keyAccessor = new FrameTupleAccessor(ctx, keyDesc);
+		keyAccessor.reset(frame);
+		
+		appender.reset(frame, true);
+		
+		// build and append low key
+		ktb.reset();
+    	IntegerSerializerDeserializer.INSTANCE.serialize(lk, kdos);
+    	ktb.addFieldEndOffset();    	
+    	appender.append(ktb.getFieldEndOffsets(), ktb.getByteArray(), 0, ktb.getSize());
+    	
+    	// build and append high key
+    	ktb.reset();
+    	IntegerSerializerDeserializer.INSTANCE.serialize(hk, kdos); 	
+    	ktb.addFieldEndOffset();
+    	appender.append(ktb.getFieldEndOffsets(), ktb.getByteArray(), 0, ktb.getSize());
+        
+    	// create tuplereferences for search keys
+    	FrameTupleReference lowKey = new FrameTupleReference();
+    	lowKey.reset(keyAccessor, 0);    			
+    	
+		FrameTupleReference highKey = new FrameTupleReference();
+		highKey.reset(keyAccessor, 1);
+    			
+        IBinaryComparator[] searchCmps = new IBinaryComparator[1];
+        searchCmps[0] = IntegerBinaryComparatorFactory.INSTANCE.createBinaryComparator();
+        MultiComparator searchCmp = new MultiComparator(typeTraits, searchCmps);
+        
+        RangePredicate rangePred = new RangePredicate(isForward, lowKey, highKey, lowKeyInclusive, highKeyInclusive, searchCmp, searchCmp);
+        return rangePred;
+    }
+    
+    public void getExpectedResults(ArrayList<Integer> expectedResults, ArrayList<Integer> keys, int lk, int hk, boolean isForward, boolean lowKeyInclusive, boolean highKeyInclusive) {    	
+    	
+    	// special cases
+    	if(lk == hk && (!lowKeyInclusive || !highKeyInclusive)) return;    	
+    	if(lk > hk) return;
+    	
+    	if(isForward) {
+    		for(int i = 0; i < keys.size(); i++) {
+    			if( (lk == keys.get(i) && lowKeyInclusive) || (hk == keys.get(i) && highKeyInclusive) ) {
+    				expectedResults.add(keys.get(i));
+    				continue;
+    			}
+    			
+    			if(lk < keys.get(i) && hk > keys.get(i)) {
+    				expectedResults.add(keys.get(i));
+    				continue;
+    			}	
+    		}
+    	}
+    	else {
+    		for(int i = keys.size() - 1; i >= 0; i--) {    			
+    			if( (lk == keys.get(i) && lowKeyInclusive) || (hk == keys.get(i) && highKeyInclusive) ) {
+    				expectedResults.add(keys.get(i));
+    				continue;
+    			}
+    			
+    			if(lk < keys.get(i) && hk > keys.get(i)) {
+    				expectedResults.add(keys.get(i));
+    				continue;
+    			}	
+    		}
+    	}    	
+    }
+        
+    public boolean performSearches(ArrayList<Integer> keys, BTree btree, IBTreeLeafFrame leafFrame, IBTreeInteriorFrame interiorFrame, int minKey, int maxKey, boolean isForward, boolean lowKeyInclusive, boolean highKeyInclusive, boolean printExpectedResults) throws Exception {
+    	
+    	ArrayList<Integer> results = new ArrayList<Integer>();
+    	ArrayList<Integer> expectedResults = new ArrayList<Integer>();
+    	    	
+    	for(int i = minKey; i < maxKey; i++) {
+    		for(int j = minKey; j < maxKey; j++) {
+    			    			   
+    			//if(i != -100 || j != 1) continue;
+    			
+    			results.clear();
+    			expectedResults.clear();
+    			    			
+    			int lowKey = i;
+    			int highKey = j;
+
+    			IBTreeCursor rangeCursor = new RangeSearchCursor(leafFrame);        
+    			RangePredicate rangePred = createRangePredicate(lowKey, highKey, isForward, lowKeyInclusive, highKeyInclusive, btree.getMultiComparator(), btree.getMultiComparator().getTypeTraits());
+    			BTreeOpContext searchOpCtx = btree.createOpContext(BTreeOp.BTO_SEARCH, leafFrame, interiorFrame, null);
+    			btree.search(rangeCursor, rangePred, searchOpCtx);
+
+    			try {           
+    				while (rangeCursor.hasNext()) {    					
+    					rangeCursor.next();
+    					ITupleReference frameTuple = rangeCursor.getTuple();
+    					ByteArrayInputStream inStream = new ByteArrayInputStream(frameTuple.getFieldData(0), frameTuple.getFieldStart(0), frameTuple.getFieldLength(0));
+    					DataInput dataIn = new DataInputStream(inStream);
+    					Integer res = IntegerSerializerDeserializer.INSTANCE.deserialize(dataIn);
+    					results.add(res);                              
+    				}            
+    			} catch (Exception e) {
+    				e.printStackTrace();
+    			} finally {
+    				rangeCursor.close();
+    			}  
+    			
+    			getExpectedResults(expectedResults, keys, lowKey, highKey, isForward, lowKeyInclusive, highKeyInclusive);    			
+    			
+    			if(printExpectedResults) {    				
+    				if(expectedResults.size() > 0) {
+    					char l, u;
+    					
+    					if(lowKeyInclusive) l = '[';
+    					else l = '(';
+    					
+    					if(highKeyInclusive) u = ']';
+    					else u = ')';
+    					
+    					System.out.println("RANGE: " + l + " " + lowKey + " , " + highKey + " " + u);
+    					for(Integer r : expectedResults)
+    						System.out.print(r + " ");
+    					System.out.println();
+    				}
+    			}
+    			
+    			if(results.size() == expectedResults.size()) {
+    				for(int k = 0; k < results.size(); k++) {
+    					if(!results.get(k).equals(expectedResults.get(k))) {
+    						System.out.println("DIFFERENT RESULTS AT: i=" + i + " j=" + j + " k=" + k);
+    						System.out.println(results.get(k) + " " + expectedResults.get(k));
+    						return false;
+    					}
+    				}
+    			}
+    			else {
+    				System.out.println("UNEQUAL NUMBER OF RESULTS AT: i=" + i + " j=" + j);
+    				System.out.println("RESULTS: " + results.size());
+    				System.out.println("EXPECTED RESULTS: " + expectedResults.size());
+    				return false;
+    			}    			
+    		}   
+    	}
+    	
+    	return true;
+    }
+    
+    @After
+    public void tearDown() {
+
+    }      
+}
diff --git a/hyracks-storage-am-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/invertedindex/impls/SimpleConjunctiveSearcher.java b/hyracks-storage-am-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/invertedindex/impls/SimpleConjunctiveSearcher.java
index bb13d94..b6e3547 100644
--- a/hyracks-storage-am-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/invertedindex/impls/SimpleConjunctiveSearcher.java
+++ b/hyracks-storage-am-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/invertedindex/impls/SimpleConjunctiveSearcher.java
@@ -54,7 +54,7 @@
 	private final IBTreeInteriorFrame interiorFrame;
 	private final IBTreeCursor btreeCursor;
 	private final FrameTupleReference searchKey = new FrameTupleReference();
-	private final RangePredicate pred = new RangePredicate(true, null, null, true, true, null);	
+	private final RangePredicate pred = new RangePredicate(true, null, null, true, true, null, null);	
 	
 	private final IBinaryTokenizer queryTokenizer;
 	
@@ -86,7 +86,8 @@
 		}
 		
 		MultiComparator searchCmp = new MultiComparator(btreeCmp.getTypeTraits(), keyCmps);
-		pred.setComparator(searchCmp);
+		pred.setLowKeyComparator(searchCmp);
+		pred.setHighKeyComparator(searchCmp);
 		pred.setLowKey(searchKey, true);
 		pred.setHighKey(searchKey, true);
 				
@@ -139,7 +140,7 @@
 			// append first inverted list to temporary results
 			searchKey.reset(queryTokenAccessor, 0);
 			btree.search(btreeCursor, pred, opCtx);
-			while(btreeCursor.hasNext()) {
+			while(btreeCursor.hasNext()) {				
 				btreeCursor.next();
 				maxResultBufIdx = appendTupleToNewResults(btreeCursor, maxResultBufIdx);				
 			}
diff --git a/hyracks-storage-am-invertedindex/src/test/java/edu/uci/ics/hyracks/storage/am/invertedindex/searchers/SimpleConjunctiveSearcherTest.java b/hyracks-storage-am-invertedindex/src/test/java/edu/uci/ics/hyracks/storage/am/invertedindex/searchers/SimpleConjunctiveSearcherTest.java
index 788e01b..3a5c903 100644
--- a/hyracks-storage-am-invertedindex/src/test/java/edu/uci/ics/hyracks/storage/am/invertedindex/searchers/SimpleConjunctiveSearcherTest.java
+++ b/hyracks-storage-am-invertedindex/src/test/java/edu/uci/ics/hyracks/storage/am/invertedindex/searchers/SimpleConjunctiveSearcherTest.java
@@ -35,9 +35,9 @@
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeLeafFrameFactory;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeMetaDataFrame;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeMetaDataFrameFactory;
-import edu.uci.ics.hyracks.storage.am.btree.frames.FieldPrefixNSMLeafFrameFactory;
 import edu.uci.ics.hyracks.storage.am.btree.frames.MetaDataFrameFactory;
 import edu.uci.ics.hyracks.storage.am.btree.frames.NSMInteriorFrameFactory;
+import edu.uci.ics.hyracks.storage.am.btree.frames.NSMLeafFrameFactory;
 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;
@@ -58,14 +58,15 @@
 public class SimpleConjunctiveSearcherTest {
 	
 	// testing params
-	private static final int PAGE_SIZE = 256;
-    private static final int NUM_PAGES = 10;
-    private static final int HYRACKS_FRAME_SIZE = 256;
+	//private static final int PAGE_SIZE = 256;
+    //private static final int NUM_PAGES = 10;
+    //private static final int HYRACKS_FRAME_SIZE = 256;
 
 	// realistic params
-	//private static final int PAGE_SIZE = 32768;
-    //private static final int NUM_PAGES = 100;
-    //private static final int HYRACKS_FRAME_SIZE = 32768;
+	//private static final int PAGE_SIZE = 65536;
+	private static final int PAGE_SIZE = 32768;
+    private static final int NUM_PAGES = 10;
+    private static final int HYRACKS_FRAME_SIZE = 32768;
     
 	private String tmpDir = System.getProperty("java.io.tmpdir");
 	
@@ -82,7 +83,7 @@
     
 	@Test	
 	public void test01() throws Exception { 
-		    	
+		
     	FileManager fileManager = new FileManager();
     	ICacheMemoryAllocator allocator = new BufferAllocator();
     	IPageReplacementStrategy prs = new ClockPageReplacementStrategy();
@@ -110,8 +111,8 @@
     	    	
     	TypeAwareTupleWriterFactory tupleWriterFactory = new TypeAwareTupleWriterFactory(typeTraits);
     	//SimpleTupleWriterFactory tupleWriterFactory = new SimpleTupleWriterFactory();
-        //IBTreeLeafFrameFactory leafFrameFactory = new NSMLeafFrameFactory(tupleWriterFactory);
-    	IBTreeLeafFrameFactory leafFrameFactory = new FieldPrefixNSMLeafFrameFactory(tupleWriterFactory);
+        IBTreeLeafFrameFactory leafFrameFactory = new NSMLeafFrameFactory(tupleWriterFactory);
+    	//IBTreeLeafFrameFactory leafFrameFactory = new FieldPrefixNSMLeafFrameFactory(tupleWriterFactory);
         IBTreeInteriorFrameFactory interiorFrameFactory = new NSMInteriorFrameFactory(tupleWriterFactory);
         IBTreeMetaDataFrameFactory metaFrameFactory = new MetaDataFrameFactory();
         
@@ -146,7 +147,7 @@
 		tokens.add("science");		
 		tokens.add("major");				
 		
-		int maxId = 1000;
+		int maxId = 1000000;
 		int addProb = 0;
 		int addProbStep = 2;
 		
@@ -177,6 +178,9 @@
     		}
     	}     
     	
+    	int numPages = btree.getMaxPage(metaFrame);
+    	System.out.println("NUMPAGES: " + numPages);
+    	
     	// build query as tuple reference
     	ISerializerDeserializer[] querySerde = { UTF8StringSerializerDeserializer.INSTANCE };
 		RecordDescriptor queryRecDesc = new RecordDescriptor(querySerde);
@@ -218,7 +222,7 @@
     	long timeEnd = System.currentTimeMillis();    	
     	System.out.println("SEARCH TIME: " + (timeEnd - timeStart) + "ms");
     	    	    	
-    	System.out.println("INTERSECTION RESULTS");
+    	//System.out.println("INTERSECTION RESULTS");
     	IInvertedIndexResultCursor resultCursor = searcher.getResultCursor();
     	while(resultCursor.hasNext()) {
     		resultCursor.next();
@@ -229,9 +233,9 @@
 					ByteArrayInputStream inStream = new ByteArrayInputStream(resultTuple.getFieldData(j), resultTuple.getFieldStart(j), resultTuple.getFieldLength(j));
 					DataInput dataIn = new DataInputStream(inStream);
 					Object o = resultSerde[j].deserialize(dataIn);		
-					System.out.print(o + " ");
+					//System.out.print(o + " ");
 				}
-				System.out.println();
+				//System.out.println();
 			}
     	}
     	    	    	
@@ -242,14 +246,15 @@
 		
     	// ordered scan        
         IBTreeCursor scanCursor = new RangeSearchCursor(leafFrame);
-        RangePredicate nullPred = new RangePredicate(true, null, null, null);
-        btree.search(scanCursor, nullPred, leafFrame, interiorFrame);
+        RangePredicate nullPred = new RangePredicate(true, null, null, true, true, null);
+        BTreeOpContext searchOpCtx = btree.createOpContext(BTreeOp.BTO_SEARCH, leafFrame, interiorFrame, metaFrame);
+        btree.search(scanCursor, nullPred, searchOpCtx);
         
         try {
             while (scanCursor.hasNext()) {
                 scanCursor.next();
                 ITupleReference frameTuple = scanCursor.getTuple();                
-                String rec = cmp.printTuple(frameTuple, recDescSers);
+                String rec = cmp.printTuple(frameTuple, btreeSerde);
                 System.out.println(rec);
             }
         } catch (Exception e) {
@@ -258,6 +263,7 @@
             scanCursor.close();
         }
         */
+        
                 
         btree.close();