diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/frames/BTreeFieldPrefixNSMLeafFrame.java b/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/frames/BTreeFieldPrefixNSMLeafFrame.java
index 9624158..eb87c57 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/frames/BTreeFieldPrefixNSMLeafFrame.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/frames/BTreeFieldPrefixNSMLeafFrame.java
@@ -790,4 +790,14 @@
         throw new IllegalStateException("nyi");
     }
 
+    @Override
+    public ITupleReference getLeftmostTuple() throws HyracksDataException {
+        throw new UnsupportedOperationException("Not implemented");
+    }
+
+    @Override
+    public ITupleReference getRightmostTuple() throws HyracksDataException {
+        throw new UnsupportedOperationException("Not implemented");
+    }
+
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/BTree.java b/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/BTree.java
index 63c54c3..26564e0 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/BTree.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/BTree.java
@@ -838,8 +838,8 @@
      * for now, we are reusing it while it is an inner class !!!!
      */
     public class BTreeAccessor implements ITreeIndexAccessor {
-        private BTree btree;
-        private BTreeOpContext ctx;
+        protected BTree btree;
+        protected BTreeOpContext ctx;
 
         public BTreeAccessor(BTree btree, IModificationOperationCallback modificationCalback,
                 ISearchOperationCallback searchCallback) {
@@ -896,6 +896,10 @@
             return new BTreeRangeSearchCursor(leafFrame, exclusive);
         }
 
+        public BTreeRangeSearchCursor createPointCursor(boolean exclusive) {
+            return createSearchCursor(exclusive);
+        }
+
         @Override
         public void search(IIndexCursor cursor, ISearchPredicate searchPred) throws HyracksDataException {
             ctx.setOperation(IndexOperation.SEARCH);
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/BTreeRangeSearchCursor.java b/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/BTreeRangeSearchCursor.java
index 5ce9e1a..a75c9f7 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/BTreeRangeSearchCursor.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/BTreeRangeSearchCursor.java
@@ -40,37 +40,37 @@
 
 public class BTreeRangeSearchCursor implements ITreeIndexCursor {
 
-    private final IBTreeLeafFrame frame;
-    private final ITreeIndexTupleReference frameTuple;
-    private final boolean exclusiveLatchNodes;
-    private boolean isPageDirty;
+    protected final IBTreeLeafFrame frame;
+    protected final ITreeIndexTupleReference frameTuple;
+    protected final boolean exclusiveLatchNodes;
+    protected boolean isPageDirty;
 
-    private IBufferCache bufferCache = null;
-    private int fileId = -1;
+    protected IBufferCache bufferCache = null;
+    protected int fileId = -1;
 
-    private ICachedPage page = null;
-    private int pageId = -1; // This is used by the LSMRTree flush operation
+    protected ICachedPage page = null;
+    protected int pageId = -1; // This is used by the LSMRTree flush operation
 
-    private int tupleIndex = 0;
-    private int stopTupleIndex;
+    protected int tupleIndex = 0;
+    protected int stopTupleIndex;
 
-    private final RangePredicate reusablePredicate;
-    private final ArrayTupleReference reconciliationTuple;
-    private IIndexAccessor accessor;
-    private ISearchOperationCallback searchCb;
-    private MultiComparator originalKeyCmp;
-    private ArrayTupleBuilder tupleBuilder;
+    protected final RangePredicate reusablePredicate;
+    protected final ArrayTupleReference reconciliationTuple;
+    protected IIndexAccessor accessor;
+    protected ISearchOperationCallback searchCb;
+    protected MultiComparator originalKeyCmp;
+    protected ArrayTupleBuilder tupleBuilder;
 
-    private FindTupleMode lowKeyFtm;
-    private FindTupleMode highKeyFtm;
-    private FindTupleNoExactMatchPolicy lowKeyFtp;
-    private FindTupleNoExactMatchPolicy highKeyFtp;
+    protected FindTupleMode lowKeyFtm;
+    protected FindTupleMode highKeyFtm;
+    protected FindTupleNoExactMatchPolicy lowKeyFtp;
+    protected FindTupleNoExactMatchPolicy highKeyFtp;
 
-    private RangePredicate pred;
-    private MultiComparator lowKeyCmp;
-    private MultiComparator highKeyCmp;
+    protected RangePredicate pred;
+    protected MultiComparator lowKeyCmp;
+    protected MultiComparator highKeyCmp;
     protected ITupleReference lowKey;
-    private ITupleReference highKey;
+    protected ITupleReference highKey;
 
     public BTreeRangeSearchCursor(IBTreeLeafFrame frame, boolean exclusiveLatchNodes) {
         this.frame = frame;
@@ -83,12 +83,7 @@
     @Override
     public void destroy() throws HyracksDataException {
         if (page != null) {
-            if (exclusiveLatchNodes) {
-                page.releaseWriteLatch(isPageDirty);
-            } else {
-                page.releaseReadLatch();
-            }
-            bufferCache.unpin(page);
+            releasePage();
         }
 
         tupleIndex = 0;
@@ -120,18 +115,10 @@
         return pageId;
     }
 
-    private void fetchNextLeafPage(int nextLeafPage) throws HyracksDataException {
+    protected void fetchNextLeafPage(int nextLeafPage) throws HyracksDataException {
         do {
-            ICachedPage nextLeaf = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, nextLeafPage), false);
-            if (exclusiveLatchNodes) {
-                nextLeaf.acquireWriteLatch();
-                page.releaseWriteLatch(isPageDirty);
-            } else {
-                nextLeaf.acquireReadLatch();
-                page.releaseReadLatch();
-            }
-            bufferCache.unpin(page);
-
+            ICachedPage nextLeaf = acquirePage(nextLeafPage);
+            releasePage();
             page = nextLeaf;
             isPageDirty = false;
             frame.setPage(page);
@@ -173,13 +160,7 @@
                 TupleUtils.copyTuple(tupleBuilder, frameTuple, originalKeyCmp.getKeyFieldCount());
                 reconciliationTuple.reset(tupleBuilder.getFieldEndOffsets(), tupleBuilder.getByteArray());
 
-                // unlatch/unpin
-                if (exclusiveLatchNodes) {
-                    page.releaseWriteLatch(isPageDirty);
-                } else {
-                    page.releaseReadLatch();
-                }
-                bufferCache.unpin(page);
+                releasePage();
                 page = null;
                 isPageDirty = false;
 
@@ -210,7 +191,7 @@
         tupleIndex++;
     }
 
-    private int getLowKeyIndex() throws HyracksDataException {
+    protected int getLowKeyIndex() throws HyracksDataException {
         if (lowKey == null) {
             return 0;
         }
@@ -227,7 +208,7 @@
         return index;
     }
 
-    private int getHighKeyIndex() throws HyracksDataException {
+    protected int getHighKeyIndex() throws HyracksDataException {
         if (highKey == null) {
             return frame.getTupleCount() - 1;
         }
@@ -248,12 +229,7 @@
     public void open(ICursorInitialState initialState, ISearchPredicate searchPred) throws HyracksDataException {
         // in case open is called multiple times without closing
         if (page != null) {
-            if (exclusiveLatchNodes) {
-                page.releaseWriteLatch(isPageDirty);
-            } else {
-                page.releaseReadLatch();
-            }
-            bufferCache.unpin(page);
+            resetBeforeOpen();
         }
         accessor = ((BTreeCursorInitialState) initialState).getAccessor();
         searchCb = initialState.getSearchOperationCallback();
@@ -291,6 +267,10 @@
         stopTupleIndex = getHighKeyIndex();
     }
 
+    protected void resetBeforeOpen() throws HyracksDataException {
+        releasePage();
+    }
+
     @Override
     public void close() throws HyracksDataException {
         destroy();
@@ -310,4 +290,23 @@
     public boolean isExclusiveLatchNodes() {
         return exclusiveLatchNodes;
     }
+
+    protected void releasePage() throws HyracksDataException {
+        if (exclusiveLatchNodes) {
+            page.releaseWriteLatch(isPageDirty);
+        } else {
+            page.releaseReadLatch();
+        }
+        bufferCache.unpin(page);
+    }
+
+    protected ICachedPage acquirePage(int pageId) throws HyracksDataException {
+        ICachedPage nextPage = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, pageId), false);
+        if (exclusiveLatchNodes) {
+            nextPage.acquireWriteLatch();
+        } else {
+            nextPage.acquireReadLatch();
+        }
+        return nextPage;
+    }
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/DiskBTree.java b/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/DiskBTree.java
new file mode 100644
index 0000000..32fa9df
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/DiskBTree.java
@@ -0,0 +1,289 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hyracks.storage.am.btree.impls;
+
+import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.api.io.FileReference;
+import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
+import org.apache.hyracks.storage.am.btree.api.IBTreeFrame;
+import org.apache.hyracks.storage.am.btree.api.IBTreeLeafFrame;
+import org.apache.hyracks.storage.am.common.api.IPageManager;
+import org.apache.hyracks.storage.am.common.api.ITreeIndexCursor;
+import org.apache.hyracks.storage.am.common.api.ITreeIndexFrame;
+import org.apache.hyracks.storage.am.common.api.ITreeIndexFrameFactory;
+import org.apache.hyracks.storage.am.common.impls.TreeIndexDiskOrderScanCursor;
+import org.apache.hyracks.storage.am.common.ophelpers.IndexOperation;
+import org.apache.hyracks.storage.common.IIndexAccessParameters;
+import org.apache.hyracks.storage.common.IIndexCursor;
+import org.apache.hyracks.storage.common.IModificationOperationCallback;
+import org.apache.hyracks.storage.common.ISearchOperationCallback;
+import org.apache.hyracks.storage.common.ISearchPredicate;
+import org.apache.hyracks.storage.common.MultiComparator;
+import org.apache.hyracks.storage.common.buffercache.IBufferCache;
+import org.apache.hyracks.storage.common.buffercache.ICachedPage;
+import org.apache.hyracks.storage.common.file.BufferedFileHandle;
+
+public class DiskBTree extends BTree {
+
+    public DiskBTree(IBufferCache bufferCache, IPageManager freePageManager,
+            ITreeIndexFrameFactory interiorFrameFactory, ITreeIndexFrameFactory leafFrameFactory,
+            IBinaryComparatorFactory[] cmpFactories, int fieldCount, FileReference file) {
+        super(bufferCache, freePageManager, interiorFrameFactory, leafFrameFactory, cmpFactories, fieldCount, file);
+    }
+
+    private void diskOrderScan(ITreeIndexCursor icursor, BTreeOpContext ctx) throws HyracksDataException {
+        TreeIndexDiskOrderScanCursor cursor = (TreeIndexDiskOrderScanCursor) icursor;
+        ctx.reset();
+        RangePredicate diskOrderScanPred = new RangePredicate(null, null, true, true, ctx.getCmp(), ctx.getCmp());
+        int maxPageId = freePageManager.getMaxPageId(ctx.getMetaFrame());
+        int currentPageId = bulkloadLeafStart;
+        ICachedPage page = bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), currentPageId), false);
+        try {
+            cursor.setBufferCache(bufferCache);
+            cursor.setFileId(getFileId());
+            cursor.setCurrentPageId(currentPageId);
+            cursor.setMaxPageId(maxPageId);
+            ctx.getCursorInitialState().setPage(page);
+            ctx.getCursorInitialState().setSearchOperationCallback(ctx.getSearchCallback());
+            ctx.getCursorInitialState().setOriginialKeyComparator(ctx.getCmp());
+            cursor.open(ctx.getCursorInitialState(), diskOrderScanPred);
+        } catch (Exception e) {
+            bufferCache.unpin(page);
+            throw HyracksDataException.create(e);
+        }
+    }
+
+    private void search(ITreeIndexCursor cursor, ISearchPredicate searchPred, BTreeOpContext ctx)
+            throws HyracksDataException {
+        ctx.reset();
+        ctx.setPred((RangePredicate) searchPred);
+        ctx.setCursor(cursor);
+        // simple index scan
+        if (ctx.getPred().getLowKeyComparator() == null) {
+            ctx.getPred().setLowKeyComparator(ctx.getCmp());
+        }
+        if (ctx.getPred().getHighKeyComparator() == null) {
+            ctx.getPred().setHighKeyComparator(ctx.getCmp());
+        }
+        cursor.setBufferCache(bufferCache);
+        cursor.setFileId(getFileId());
+
+        DiskBTreeRangeSearchCursor diskCursor = (DiskBTreeRangeSearchCursor) cursor;
+
+        if (diskCursor.numSearchPages() == 0) {
+            // we have to search from root to leaf
+            ICachedPage rootNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), rootPage), false);
+            diskCursor.addSearchPage(rootPage);
+            searchDown(rootNode, rootPage, ctx, diskCursor);
+        } else {
+            // we first check whether the leaf page matches because page may be shifted during cursor.hasNext
+            if (ctx.getLeafFrame().getPage() != diskCursor.getPage()) {
+                ctx.getLeafFrame().setPage(diskCursor.getPage());
+                ctx.getCursorInitialState().setPage(diskCursor.getPage());
+                ctx.getCursorInitialState().setPageId(diskCursor.getPageId());
+            }
+
+            if (fitInPage(ctx.getPred().getLowKey(), ctx.getPred().getLowKeyComparator(), ctx.getLeafFrame())) {
+                // the input still falls into the previous search leaf
+                diskCursor.open(ctx.getCursorInitialState(), searchPred);
+            } else {
+                // unpin the previous leaf page
+                bufferCache.unpin(ctx.getLeafFrame().getPage());
+                diskCursor.removeLastSearchPage();
+
+                ICachedPage page = searchUp(ctx, diskCursor);
+                int pageId = diskCursor.getLastSearchPage();
+
+                searchDown(page, pageId, ctx, diskCursor);
+            }
+        }
+    }
+
+    private ICachedPage searchUp(BTreeOpContext ctx, DiskBTreeRangeSearchCursor cursor) throws HyracksDataException {
+        int index = cursor.numSearchPages() - 1;
+        // no need to check root page
+        for (; index >= 0; index--) {
+            int pageId = cursor.getLastSearchPage();
+            ICachedPage page = bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), pageId), false);
+            ctx.getInteriorFrame().setPage(page);
+            if (index == 0 || fitInPage(ctx.getPred().getLowKey(), ctx.getPred().getLowKeyComparator(),
+                    ctx.getInteriorFrame())) {
+                // we've found the right page
+                return page;
+            } else {
+                // unpin the current page
+                bufferCache.unpin(page);
+                cursor.removeLastSearchPage();
+            }
+        }
+
+        // if no page is available (which is the case for single-level BTree)
+        // we simply return the root page
+        ICachedPage page = bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), rootPage), false);
+        cursor.addSearchPage(rootPage);
+        return page;
+    }
+
+    private boolean fitInPage(ITupleReference key, MultiComparator comparator, IBTreeFrame frame)
+            throws HyracksDataException {
+        ITupleReference rightmostTuple = frame.getRightmostTuple();
+        int cmp = comparator.compare(key, rightmostTuple);
+        if (cmp > 0) {
+            return false;
+        }
+        ITupleReference leftmostTuple = frame.getLeftmostTuple();
+        return comparator.compare(key, leftmostTuple) >= 0;
+    }
+
+    private void searchDown(ICachedPage page, int pageId, BTreeOpContext ctx, DiskBTreeRangeSearchCursor cursor)
+            throws HyracksDataException {
+        ICachedPage currentPage = page;
+        ctx.getInteriorFrame().setPage(currentPage);
+
+        try {
+            int childPageId = pageId;
+            while (!ctx.getInteriorFrame().isLeaf()) {
+                // walk down the tree until we find the leaf
+                childPageId = ctx.getInteriorFrame().getChildPageId(ctx.getPred());
+
+                // save the child page tuple index
+                cursor.addSearchPage(childPageId);
+                bufferCache.unpin(currentPage);
+                currentPage = bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), childPageId), false);
+                ctx.getInteriorFrame().setPage(currentPage);
+            }
+
+            ctx.getCursorInitialState().setSearchOperationCallback(ctx.getSearchCallback());
+            ctx.getCursorInitialState().setOriginialKeyComparator(ctx.getCmp());
+            ctx.getCursorInitialState().setPage(currentPage);
+            ctx.getCursorInitialState().setPageId(childPageId);
+            ctx.getLeafFrame().setPage(currentPage);
+            cursor.open(ctx.getCursorInitialState(), ctx.getPred());
+        } catch (HyracksDataException e) {
+            if (!ctx.isExceptionHandled() && currentPage != null) {
+                bufferCache.unpin(currentPage);
+                ctx.setExceptionHandled(true);
+            }
+            throw e;
+        } catch (Exception e) {
+            if (currentPage != null) {
+                bufferCache.unpin(currentPage);
+            }
+            HyracksDataException wrappedException = HyracksDataException.create(e);
+            ctx.setExceptionHandled(true);
+            throw wrappedException;
+        }
+    }
+
+    @Override
+    public BTreeAccessor createAccessor(IIndexAccessParameters iap) {
+        return new DiskBTreeAccessor(this, iap.getModificationCallback(), iap.getSearchOperationCallback());
+    }
+
+    @Override
+    public DiskBTreeAccessor createAccessor(IModificationOperationCallback modificationCallback,
+            ISearchOperationCallback searchCallback, int[] logTupleFields) {
+        return new DiskBTreeAccessor(this, modificationCallback, searchCallback, logTupleFields);
+    }
+
+    public class DiskBTreeAccessor extends BTreeAccessor {
+
+        public DiskBTreeAccessor(DiskBTree btree, IModificationOperationCallback modificationCalback,
+                ISearchOperationCallback searchCallback) {
+            super(btree, modificationCalback, searchCallback);
+        }
+
+        public DiskBTreeAccessor(DiskBTree btree, IModificationOperationCallback modificationCalback,
+                ISearchOperationCallback searchCallback, int[] logTupleFields) {
+            super(btree, modificationCalback, searchCallback, logTupleFields);
+        }
+
+        @Override
+        public void insert(ITupleReference tuple) throws HyracksDataException {
+            throw new UnsupportedOperationException("Insert is not supported by DiskBTree. ");
+        }
+
+        @Override
+        public void update(ITupleReference tuple) throws HyracksDataException {
+            throw new UnsupportedOperationException("Update is not supported by DiskBTree. ");
+        }
+
+        @Override
+        public void delete(ITupleReference tuple) throws HyracksDataException {
+            throw new UnsupportedOperationException("Delete is not supported by DiskBTree. ");
+        }
+
+        @Override
+        public void upsert(ITupleReference tuple) throws HyracksDataException {
+            throw new UnsupportedOperationException("Upsert is not supported by DiskBTree. ");
+        }
+
+        @Override
+        public DiskBTreeRangeSearchCursor createSearchCursor(boolean exclusive) {
+            IBTreeLeafFrame leafFrame = (IBTreeLeafFrame) btree.getLeafFrameFactory().createFrame();
+            return new DiskBTreeRangeSearchCursor(leafFrame, exclusive);
+        }
+
+        @Override
+        public BTreeRangeSearchCursor createPointCursor(boolean exclusive) {
+            IBTreeLeafFrame leafFrame = (IBTreeLeafFrame) btree.getLeafFrameFactory().createFrame();
+            return new DiskBTreePointSearchCursor(leafFrame, exclusive);
+        }
+
+        @Override
+        public void search(IIndexCursor cursor, ISearchPredicate searchPred) throws HyracksDataException {
+            ctx.setOperation(IndexOperation.SEARCH);
+            ((DiskBTree) btree).search((ITreeIndexCursor) cursor, searchPred, ctx);
+        }
+
+        @Override
+        public ITreeIndexCursor createDiskOrderScanCursor() {
+            IBTreeLeafFrame leafFrame = (IBTreeLeafFrame) btree.getLeafFrameFactory().createFrame();
+            return new DiskBTreeDiskScanCursor(leafFrame);
+        }
+
+        @Override
+        public void diskOrderScan(ITreeIndexCursor cursor) throws HyracksDataException {
+            ctx.setOperation(IndexOperation.DISKORDERSCAN);
+            ((DiskBTree) btree).diskOrderScan(cursor, ctx);
+        }
+    }
+
+    private class DiskBTreeDiskScanCursor extends TreeIndexDiskOrderScanCursor {
+
+        public DiskBTreeDiskScanCursor(ITreeIndexFrame frame) {
+            super(frame);
+        }
+
+        @Override
+        protected void releasePage() throws HyracksDataException {
+            bufferCache.unpin(page);
+        }
+
+        @Override
+        protected ICachedPage acquireNextPage() throws HyracksDataException {
+            ICachedPage nextPage = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, currentPageId), false);
+            return nextPage;
+        }
+
+    }
+
+}
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/DiskBTreePointSearchCursor.java b/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/DiskBTreePointSearchCursor.java
new file mode 100644
index 0000000..5839d0e
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/DiskBTreePointSearchCursor.java
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hyracks.storage.am.btree.impls;
+
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.storage.am.btree.api.IBTreeLeafFrame;
+import org.apache.hyracks.storage.am.common.ophelpers.FindTupleMode;
+import org.apache.hyracks.storage.am.common.ophelpers.FindTupleNoExactMatchPolicy;
+import org.apache.hyracks.storage.common.ICursorInitialState;
+import org.apache.hyracks.storage.common.ISearchPredicate;
+
+public class DiskBTreePointSearchCursor extends DiskBTreeRangeSearchCursor {
+
+    private boolean nextHasBeenCalled;
+
+    public DiskBTreePointSearchCursor(IBTreeLeafFrame frame, boolean exclusiveLatchNodes) {
+        super(frame, exclusiveLatchNodes);
+    }
+
+    @Override
+    public boolean hasNext() throws HyracksDataException {
+        return tupleIndex >= 0 && !nextHasBeenCalled;
+    }
+
+    @Override
+    public void next() throws HyracksDataException {
+        nextHasBeenCalled = true;
+    }
+
+    @Override
+    public void open(ICursorInitialState initialState, ISearchPredicate searchPred) throws HyracksDataException {
+        // in case open is called multiple times without closing
+        if (page != null) {
+            resetBeforeOpen();
+        }
+        accessor = ((BTreeCursorInitialState) initialState).getAccessor();
+        searchCb = initialState.getSearchOperationCallback();
+        originalKeyCmp = initialState.getOriginalKeyComparator();
+        pageId = ((BTreeCursorInitialState) initialState).getPageId();
+        page = initialState.getPage();
+        isPageDirty = false;
+        frame.setPage(page);
+
+        pred = (RangePredicate) searchPred;
+        lowKeyCmp = pred.getLowKeyComparator();
+        lowKey = pred.getLowKey();
+
+        reusablePredicate.setLowKeyComparator(originalKeyCmp);
+
+        lowKeyFtm = FindTupleMode.EXACT;
+        lowKeyFtp = FindTupleNoExactMatchPolicy.NONE;
+
+        nextHasBeenCalled = false;
+
+        // only get the low key position
+        tupleIndex = getLowKeyIndex();
+    }
+
+}
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/DiskBTreeRangeSearchCursor.java b/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/DiskBTreeRangeSearchCursor.java
new file mode 100644
index 0000000..7d4ee0d
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/DiskBTreeRangeSearchCursor.java
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hyracks.storage.am.btree.impls;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.storage.am.btree.api.IBTreeLeafFrame;
+import org.apache.hyracks.storage.common.buffercache.ICachedPage;
+import org.apache.hyracks.storage.common.file.BufferedFileHandle;
+
+public class DiskBTreeRangeSearchCursor extends BTreeRangeSearchCursor {
+
+    // keep track of the pages (root -> leaf) we've searched
+    protected final List<Integer> searchPages = new ArrayList<>(5);
+
+    public DiskBTreeRangeSearchCursor(IBTreeLeafFrame frame, boolean exclusiveLatchNodes) {
+        super(frame, exclusiveLatchNodes);
+    }
+
+    @Override
+    public boolean hasNext() throws HyracksDataException {
+        int nextLeafPage;
+        if (tupleIndex >= frame.getTupleCount()) {
+            nextLeafPage = frame.getNextLeaf();
+            if (nextLeafPage >= 0) {
+                fetchNextLeafPage(nextLeafPage);
+                tupleIndex = 0;
+                // update page ids and positions
+                searchPages.set(searchPages.size() - 1, nextLeafPage);
+                stopTupleIndex = getHighKeyIndex();
+                if (stopTupleIndex < 0) {
+                    return false;
+                }
+            } else {
+                return false;
+            }
+        }
+        if (tupleIndex > stopTupleIndex) {
+            return false;
+        }
+
+        frameTuple.resetByTupleIndex(frame, tupleIndex);
+        return true;
+    }
+
+    @Override
+    protected void resetBeforeOpen() throws HyracksDataException {
+        // do nothing
+        // we allow a disk btree range cursor be stateful, that is, the next search can be based on the previous search
+    }
+
+    public int numSearchPages() {
+        return searchPages.size();
+    }
+
+    public void addSearchPage(int page) {
+        searchPages.add(page);
+    }
+
+    public int getLastSearchPage() {
+        return searchPages.get(searchPages.size() - 1);
+    }
+
+    public int removeLastSearchPage() {
+        return searchPages.remove(searchPages.size() - 1);
+    }
+
+    public ICachedPage getPage() {
+        return page;
+    }
+
+    @Override
+    protected void releasePage() throws HyracksDataException {
+        bufferCache.unpin(page);
+    }
+
+    @Override
+    protected ICachedPage acquirePage(int pageId) throws HyracksDataException {
+        return bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, pageId), false);
+    }
+
+    @Override
+    public void close() throws HyracksDataException {
+        super.close();
+        searchPages.clear();
+    }
+}
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/util/BTreeUtils.java b/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/util/BTreeUtils.java
index 7f992f9..0c06bc3 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/util/BTreeUtils.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/util/BTreeUtils.java
@@ -29,6 +29,7 @@
 import org.apache.hyracks.storage.am.btree.frames.BTreeNSMInteriorFrameFactory;
 import org.apache.hyracks.storage.am.btree.frames.BTreeNSMLeafFrameFactory;
 import org.apache.hyracks.storage.am.btree.impls.BTree;
+import org.apache.hyracks.storage.am.btree.impls.DiskBTree;
 import org.apache.hyracks.storage.am.btree.tuples.BTreeTypeAwareTupleWriterFactory;
 import org.apache.hyracks.storage.am.common.api.IPageManager;
 import org.apache.hyracks.storage.am.common.api.ITreeIndexFrameFactory;
@@ -51,6 +52,17 @@
                 typeTraits.length, file);
     }
 
+    public static DiskBTree createDiskBTree(IBufferCache bufferCache, ITypeTraits[] typeTraits,
+            IBinaryComparatorFactory[] cmpFactories, BTreeLeafFrameType leafType, FileReference file,
+            IPageManager freePageManager, boolean updateAware) throws HyracksDataException {
+        BTreeTypeAwareTupleWriterFactory tupleWriterFactory =
+                new BTreeTypeAwareTupleWriterFactory(typeTraits, updateAware);
+        ITreeIndexFrameFactory leafFrameFactory = getLeafFrameFactory(tupleWriterFactory, leafType);
+        ITreeIndexFrameFactory interiorFrameFactory = new BTreeNSMInteriorFrameFactory(tupleWriterFactory);
+        return new DiskBTree(bufferCache, freePageManager, interiorFrameFactory, leafFrameFactory, cmpFactories,
+                typeTraits.length, file);
+    }
+
     public static BTree createBTree(IBufferCache bufferCache, IPageManager freePageManager, ITypeTraits[] typeTraits,
             IBinaryComparatorFactory[] cmpFactories, BTreeLeafFrameType leafType, FileReference file,
             boolean updateAware) throws HyracksDataException {
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/ITreeIndexFrame.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/ITreeIndexFrame.java
index 8dc30c8..7ac49c6 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/ITreeIndexFrame.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/ITreeIndexFrame.java
@@ -118,4 +118,8 @@
     public ITreeIndexTupleReference createTupleReference();
 
     public void setMultiComparator(MultiComparator cmp);
+
+    public ITupleReference getLeftmostTuple() throws HyracksDataException;
+
+    public ITupleReference getRightmostTuple() throws HyracksDataException;
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/frames/TreeIndexNSMFrame.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/frames/TreeIndexNSMFrame.java
index 5f001e0..6106358 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/frames/TreeIndexNSMFrame.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/frames/TreeIndexNSMFrame.java
@@ -335,4 +335,23 @@
         return buf.capacity() - getFreeSpaceOff() - (getTupleCount() * slotManager.getSlotSize());
     }
 
+    public ITupleReference getLeftmostTuple() {
+        int tupleCount = getTupleCount();
+        if (tupleCount == 0) {
+            return null;
+        } else {
+            frameTuple.resetByTupleIndex(this, 0);
+            return frameTuple;
+        }
+    }
+
+    public ITupleReference getRightmostTuple() {
+        int tupleCount = getTupleCount();
+        if (tupleCount == 0) {
+            return null;
+        } else {
+            frameTuple.resetByTupleIndex(this, tupleCount - 1);
+            return frameTuple;
+        }
+    }
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/impls/TreeIndexDiskOrderScanCursor.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/impls/TreeIndexDiskOrderScanCursor.java
index 6bc5be2..3cbcf70 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/impls/TreeIndexDiskOrderScanCursor.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/impls/TreeIndexDiskOrderScanCursor.java
@@ -32,12 +32,12 @@
 
 public class TreeIndexDiskOrderScanCursor implements ITreeIndexCursor {
 
-    private int tupleIndex = 0;
-    private int fileId = -1;
-    private int currentPageId = -1;
-    private int maxPageId = -1;
-    private ICachedPage page = null;
-    private IBufferCache bufferCache = null;
+    protected int tupleIndex = 0;
+    protected int fileId = -1;
+    protected int currentPageId = -1;
+    protected int maxPageId = -1;
+    protected ICachedPage page = null;
+    protected IBufferCache bufferCache = null;
 
     private final ITreeIndexFrame frame;
     private final ITreeIndexTupleReference frameTuple;
@@ -49,8 +49,7 @@
 
     @Override
     public void destroy() throws HyracksDataException {
-        page.releaseReadLatch();
-        bufferCache.unpin(page);
+        releasePage();
         page = null;
     }
 
@@ -75,12 +74,9 @@
                 break;
             }
 
-            page.releaseReadLatch();
-            bufferCache.unpin(page);
+            releasePage();
 
-            ICachedPage nextPage = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, currentPageId), false);
-            nextPage.acquireReadLatch();
-
+            ICachedPage nextPage = acquireNextPage();
             page = nextPage;
             frame.setPage(page);
             tupleIndex = 0;
@@ -120,8 +116,7 @@
     public void open(ICursorInitialState initialState, ISearchPredicate searchPred) throws HyracksDataException {
         // in case open is called multiple times without closing
         if (page != null) {
-            page.releaseReadLatch();
-            bufferCache.unpin(page);
+            releasePage();
         }
         page = initialState.getPage();
         tupleIndex = 0;
@@ -159,4 +154,15 @@
     public boolean isExclusiveLatchNodes() {
         return false;
     }
+
+    protected void releasePage() throws HyracksDataException {
+        page.releaseReadLatch();
+        bufferCache.unpin(page);
+    }
+
+    protected ICachedPage acquireNextPage() throws HyracksDataException {
+        ICachedPage nextPage = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, currentPageId), false);
+        nextPage.acquireReadLatch();
+        return nextPage;
+    }
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeDiskComponent.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeDiskComponent.java
index d759167..0fc79eb 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeDiskComponent.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeDiskComponent.java
@@ -22,26 +22,27 @@
 import java.util.Set;
 
 import org.apache.hyracks.storage.am.btree.impls.BTree;
+import org.apache.hyracks.storage.am.btree.impls.DiskBTree;
 import org.apache.hyracks.storage.am.common.api.IMetadataPageManager;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentFilter;
 import org.apache.hyracks.storage.am.lsm.common.impls.AbstractLSMDiskComponent;
 import org.apache.hyracks.storage.am.lsm.common.impls.AbstractLSMIndex;
 
 public class LSMBTreeDiskComponent extends AbstractLSMDiskComponent {
-    protected final BTree btree;
+    protected final DiskBTree btree;
 
-    public LSMBTreeDiskComponent(AbstractLSMIndex lsmIndex, BTree btree, ILSMComponentFilter filter) {
+    public LSMBTreeDiskComponent(AbstractLSMIndex lsmIndex, DiskBTree btree, ILSMComponentFilter filter) {
         super(lsmIndex, getMetadataPageManager(btree), filter);
         this.btree = btree;
     }
 
     @Override
-    public BTree getIndex() {
+    public DiskBTree getIndex() {
         return btree;
     }
 
     @Override
-    public BTree getMetadataHolder() {
+    public DiskBTree getMetadataHolder() {
         return btree;
     }
 
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeDiskComponentFactory.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeDiskComponentFactory.java
index 0e54f53..3b332cd 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeDiskComponentFactory.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeDiskComponentFactory.java
@@ -20,7 +20,7 @@
 package org.apache.hyracks.storage.am.lsm.btree.impls;
 
 import org.apache.hyracks.api.exceptions.HyracksDataException;
-import org.apache.hyracks.storage.am.btree.impls.BTree;
+import org.apache.hyracks.storage.am.btree.impls.DiskBTree;
 import org.apache.hyracks.storage.am.lsm.common.api.IComponentFilterHelper;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMDiskComponentFactory;
 import org.apache.hyracks.storage.am.lsm.common.impls.AbstractLSMIndex;
@@ -28,10 +28,10 @@
 import org.apache.hyracks.storage.am.lsm.common.impls.TreeIndexFactory;
 
 public class LSMBTreeDiskComponentFactory implements ILSMDiskComponentFactory {
-    protected final TreeIndexFactory<BTree> btreeFactory;
+    protected final TreeIndexFactory<DiskBTree> btreeFactory;
     protected final IComponentFilterHelper filterHelper;
 
-    public LSMBTreeDiskComponentFactory(TreeIndexFactory<BTree> btreeFactory, IComponentFilterHelper filterHelper) {
+    public LSMBTreeDiskComponentFactory(TreeIndexFactory<DiskBTree> btreeFactory, IComponentFilterHelper filterHelper) {
         this.btreeFactory = btreeFactory;
         this.filterHelper = filterHelper;
     }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeDiskComponentScanCursor.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeDiskComponentScanCursor.java
index efaf555..a789ddd 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeDiskComponentScanCursor.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeDiskComponentScanCursor.java
@@ -26,10 +26,8 @@
 import org.apache.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
 import org.apache.hyracks.dataflow.common.comm.io.ArrayTupleReference;
 import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
-import org.apache.hyracks.storage.am.btree.api.IBTreeLeafFrame;
 import org.apache.hyracks.storage.am.btree.impls.BTree;
 import org.apache.hyracks.storage.am.btree.impls.BTree.BTreeAccessor;
-import org.apache.hyracks.storage.am.btree.impls.BTreeRangeSearchCursor;
 import org.apache.hyracks.storage.am.common.impls.NoOpIndexAccessParameters;
 import org.apache.hyracks.storage.am.common.tuples.PermutingTupleReference;
 import org.apache.hyracks.storage.am.lsm.btree.tuples.LSMBTreeTupleReference;
@@ -74,11 +72,9 @@
         btreeAccessors = new BTreeAccessor[numBTrees];
         for (int i = 0; i < numBTrees; i++) {
             ILSMComponent component = operationalComponents.get(i);
-            IBTreeLeafFrame leafFrame = (IBTreeLeafFrame) lsmInitialState.getLeafFrameFactory().createFrame();
-            rangeCursors[i] = new BTreeRangeSearchCursor(leafFrame, false);
             BTree btree = (BTree) component.getIndex();
-
             btreeAccessors[i] = btree.createAccessor(NoOpIndexAccessParameters.INSTANCE);
+            rangeCursors[i] = btreeAccessors[i].createSearchCursor(false);
             btreeAccessors[i].search(rangeCursors[i], searchPred);
         }
 
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreePointSearchCursor.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreePointSearchCursor.java
index 0284ee1..31e0356 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreePointSearchCursor.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreePointSearchCursor.java
@@ -24,11 +24,10 @@
 import org.apache.hyracks.api.exceptions.HyracksDataException;
 import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
 import org.apache.hyracks.storage.am.bloomfilter.impls.BloomFilter;
-import org.apache.hyracks.storage.am.btree.api.IBTreeLeafFrame;
 import org.apache.hyracks.storage.am.btree.impls.BTree;
 import org.apache.hyracks.storage.am.btree.impls.BTree.BTreeAccessor;
-import org.apache.hyracks.storage.am.btree.impls.BTreeRangeSearchCursor;
 import org.apache.hyracks.storage.am.btree.impls.RangePredicate;
+import org.apache.hyracks.storage.am.common.api.ITreeIndexCursor;
 import org.apache.hyracks.storage.am.common.impls.NoOpIndexAccessParameters;
 import org.apache.hyracks.storage.am.common.impls.NoOpOperationCallback;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponent;
@@ -44,7 +43,7 @@
 
 public class LSMBTreePointSearchCursor implements IIndexCursor {
 
-    private BTreeRangeSearchCursor[] rangeCursors;
+    private ITreeIndexCursor[] btreeCursors;
     private final ILSMIndexOperationContext opCtx;
     private ISearchOperationCallback searchCallback;
     private RangePredicate predicate;
@@ -77,21 +76,21 @@
             if (bloomFilters[i] != null && !bloomFilters[i].contains(predicate.getLowKey(), hashes)) {
                 continue;
             }
-            btreeAccessors[i].search(rangeCursors[i], predicate);
-            if (rangeCursors[i].hasNext()) {
-                rangeCursors[i].next();
+            btreeAccessors[i].search(btreeCursors[i], predicate);
+            if (btreeCursors[i].hasNext()) {
+                btreeCursors[i].next();
                 // We use the predicate's to lock the key instead of the tuple that we get from cursor
                 // to avoid copying the tuple when we do the "unlatch dance".
                 if (reconciled || searchCallback.proceed(predicate.getLowKey())) {
                     // if proceed is successful, then there's no need for doing the "unlatch dance"
-                    if (((ILSMTreeTupleReference) rangeCursors[i].getTuple()).isAntimatter()) {
+                    if (((ILSMTreeTupleReference) btreeCursors[i].getTuple()).isAntimatter()) {
                         if (reconciled) {
                             searchCallback.cancel(predicate.getLowKey());
                         }
-                        rangeCursors[i].destroy();
+                        btreeCursors[i].destroy();
                         return false;
                     } else {
-                        frameTuple = rangeCursors[i].getTuple();
+                        frameTuple = btreeCursors[i].getTuple();
                         foundTuple = true;
                         foundIn = i;
                         return true;
@@ -99,20 +98,20 @@
                 }
                 if (i == 0 && includeMutableComponent) {
                     // unlatch/unpin
-                    rangeCursors[i].close();
+                    btreeCursors[i].close();
                     searchCallback.reconcile(predicate.getLowKey());
                     reconciled = true;
 
                     // retraverse
-                    btreeAccessors[0].search(rangeCursors[i], predicate);
-                    if (rangeCursors[i].hasNext()) {
-                        rangeCursors[i].next();
-                        if (((ILSMTreeTupleReference) rangeCursors[i].getTuple()).isAntimatter()) {
+                    btreeAccessors[0].search(btreeCursors[i], predicate);
+                    if (btreeCursors[i].hasNext()) {
+                        btreeCursors[i].next();
+                        if (((ILSMTreeTupleReference) btreeCursors[i].getTuple()).isAntimatter()) {
                             searchCallback.cancel(predicate.getLowKey());
-                            rangeCursors[i].destroy();
+                            btreeCursors[i].destroy();
                             return false;
                         } else {
-                            frameTuple = rangeCursors[i].getTuple();
+                            frameTuple = btreeCursors[i].getTuple();
                             foundTuple = true;
                             searchCallback.complete(predicate.getLowKey());
                             foundIn = i;
@@ -120,10 +119,10 @@
                         }
                     } else {
                         searchCallback.cancel(predicate.getLowKey());
-                        rangeCursors[i].destroy();
+                        btreeCursors[i].destroy();
                     }
                 } else {
-                    frameTuple = rangeCursors[i].getTuple();
+                    frameTuple = btreeCursors[i].getTuple();
                     searchCallback.reconcile(frameTuple);
                     searchCallback.complete(frameTuple);
                     foundTuple = true;
@@ -131,7 +130,7 @@
                     return true;
                 }
             } else {
-                rangeCursors[i].destroy();
+                btreeCursors[i].destroy();
             }
         }
         return false;
@@ -140,9 +139,9 @@
     @Override
     public void close() throws HyracksDataException {
         try {
-            if (rangeCursors != null) {
-                for (int i = 0; i < rangeCursors.length; ++i) {
-                    rangeCursors[i].close();
+            if (btreeCursors != null) {
+                for (int i = 0; i < numBTrees; ++i) {
+                    btreeCursors[i].close();
                 }
             }
             nextHasBeenCalled = false;
@@ -162,9 +161,9 @@
         searchCallback = lsmInitialState.getSearchOperationCallback();
         predicate = (RangePredicate) lsmInitialState.getSearchPredicate();
         numBTrees = operationalComponents.size();
-        if (rangeCursors == null || rangeCursors.length < numBTrees) {
+        if (btreeCursors == null || btreeCursors.length != numBTrees) {
             // object creation: should be relatively low
-            rangeCursors = new BTreeRangeSearchCursor[numBTrees];
+            btreeCursors = new ITreeIndexCursor[numBTrees];
             btreeAccessors = new BTreeAccessor[numBTrees];
             bloomFilters = new BloomFilter[numBTrees];
         }
@@ -172,37 +171,21 @@
 
         for (int i = 0; i < numBTrees; i++) {
             ILSMComponent component = operationalComponents.get(i);
-            BTree btree;
+            BTree btree = (BTree) component.getIndex();
             if (component.getType() == LSMComponentType.MEMORY) {
                 includeMutableComponent = true;
-                if (rangeCursors[i] == null) {
-                    // create a new one
-                    IBTreeLeafFrame leafFrame = (IBTreeLeafFrame) lsmInitialState.getLeafFrameFactory().createFrame();
-                    rangeCursors[i] = new BTreeRangeSearchCursor(leafFrame, false);
-                } else {
-                    // reset
-                    rangeCursors[i].close();
-                }
-                btree = ((LSMBTreeMemoryComponent) component).getIndex();
-                // no bloom filter for in-memory BTree
                 bloomFilters[i] = null;
             } else {
-                if (rangeCursors[i] != null) {
-                    // can re-use cursor
-                    rangeCursors[i].close();
-                } else {
-                    // create new cursor <should be relatively rare>
-                    IBTreeLeafFrame leafFrame = (IBTreeLeafFrame) lsmInitialState.getLeafFrameFactory().createFrame();
-                    rangeCursors[i] = new BTreeRangeSearchCursor(leafFrame, false);
-                }
-                btree = ((LSMBTreeWithBloomFilterDiskComponent) component).getIndex();
                 bloomFilters[i] = ((LSMBTreeWithBloomFilterDiskComponent) component).getBloomFilter();
             }
+
             if (btreeAccessors[i] == null) {
                 btreeAccessors[i] = btree.createAccessor(NoOpIndexAccessParameters.INSTANCE);
+                btreeCursors[i] = btreeAccessors[i].createPointCursor(false);
             } else {
                 // re-use
                 btreeAccessors[i].reset(btree, NoOpOperationCallback.INSTANCE, NoOpOperationCallback.INSTANCE);
+                btreeCursors[i].close();
             }
         }
         nextHasBeenCalled = false;
@@ -219,7 +202,7 @@
         if (lsmHarness != null) {
             try {
                 closeCursors();
-                rangeCursors = null;
+                btreeCursors = null;
             } finally {
                 lsmHarness.endSearch(opCtx);
             }
@@ -253,10 +236,10 @@
     }
 
     private void closeCursors() throws HyracksDataException {
-        if (rangeCursors != null) {
+        if (btreeCursors != null) {
             for (int i = 0; i < numBTrees; ++i) {
-                if (rangeCursors[i] != null) {
-                    rangeCursors[i].destroy();
+                if (btreeCursors[i] != null) {
+                    btreeCursors[i].destroy();
                 }
             }
         }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeRangeSearchCursor.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeRangeSearchCursor.java
index 36ca87f..1e401a2 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeRangeSearchCursor.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeRangeSearchCursor.java
@@ -26,10 +26,8 @@
 import org.apache.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
 import org.apache.hyracks.dataflow.common.comm.io.ArrayTupleReference;
 import org.apache.hyracks.dataflow.common.utils.TupleUtils;
-import org.apache.hyracks.storage.am.btree.api.IBTreeLeafFrame;
 import org.apache.hyracks.storage.am.btree.impls.BTree;
 import org.apache.hyracks.storage.am.btree.impls.BTree.BTreeAccessor;
-import org.apache.hyracks.storage.am.btree.impls.BTreeRangeSearchCursor;
 import org.apache.hyracks.storage.am.btree.impls.RangePredicate;
 import org.apache.hyracks.storage.am.common.impls.NoOpIndexAccessParameters;
 import org.apache.hyracks.storage.am.common.impls.NoOpOperationCallback;
@@ -338,30 +336,25 @@
             rangeCursors = new IIndexCursor[numBTrees];
             btreeAccessors = new BTreeAccessor[numBTrees];
         }
+
         for (int i = 0; i < numBTrees; i++) {
             ILSMComponent component = operationalComponents.get(i);
             BTree btree;
-            if (rangeCursors[i] == null) {
-                // create, should be relatively rare
-                IBTreeLeafFrame leafFrame = (IBTreeLeafFrame) lsmInitialState.getLeafFrameFactory().createFrame();
-                rangeCursors[i] = new BTreeRangeSearchCursor(leafFrame, false);
-            } else {
-                // re-use
-                rangeCursors[i].close();
-            }
-
             if (component.getType() == LSMComponentType.MEMORY) {
                 includeMutableComponent = true;
             }
             btree = (BTree) component.getIndex();
             if (btreeAccessors[i] == null) {
                 btreeAccessors[i] = btree.createAccessor(NoOpIndexAccessParameters.INSTANCE);
+                rangeCursors[i] = btreeAccessors[i].createSearchCursor(false);
             } else {
                 // re-use
                 btreeAccessors[i].reset(btree, NoOpOperationCallback.INSTANCE, NoOpOperationCallback.INSTANCE);
+                rangeCursors[i].close();
             }
             btreeAccessors[i].search(rangeCursors[i], searchPred);
         }
+
         setPriorityQueueComparator();
         initPriorityQueue();
         canCallProceed = true;
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeWithBloomFilterDiskComponentFactory.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeWithBloomFilterDiskComponentFactory.java
index 09e83cc..eaf9cd5 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeWithBloomFilterDiskComponentFactory.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeWithBloomFilterDiskComponentFactory.java
@@ -20,7 +20,7 @@
 
 import org.apache.hyracks.api.exceptions.HyracksDataException;
 import org.apache.hyracks.storage.am.bloomfilter.impls.BloomFilterFactory;
-import org.apache.hyracks.storage.am.btree.impls.BTree;
+import org.apache.hyracks.storage.am.btree.impls.DiskBTree;
 import org.apache.hyracks.storage.am.lsm.common.api.IComponentFilterHelper;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMDiskComponentFactory;
 import org.apache.hyracks.storage.am.lsm.common.impls.AbstractLSMIndex;
@@ -28,11 +28,11 @@
 import org.apache.hyracks.storage.am.lsm.common.impls.TreeIndexFactory;
 
 public class LSMBTreeWithBloomFilterDiskComponentFactory implements ILSMDiskComponentFactory {
-    protected final TreeIndexFactory<BTree> btreeFactory;
+    protected final TreeIndexFactory<DiskBTree> btreeFactory;
     protected final IComponentFilterHelper filterHelper;
     protected final BloomFilterFactory bloomFilterFactory;
 
-    public LSMBTreeWithBloomFilterDiskComponentFactory(TreeIndexFactory<BTree> btreeFactory,
+    public LSMBTreeWithBloomFilterDiskComponentFactory(TreeIndexFactory<DiskBTree> btreeFactory,
             BloomFilterFactory bloomFilterFactory, IComponentFilterHelper filterHelper) {
         this.btreeFactory = btreeFactory;
         this.filterHelper = filterHelper;
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/utils/LSMBTreeUtil.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/utils/LSMBTreeUtil.java
index 08e5af0..4706faa 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/utils/LSMBTreeUtil.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/utils/LSMBTreeUtil.java
@@ -30,6 +30,7 @@
 import org.apache.hyracks.storage.am.btree.frames.BTreeNSMInteriorFrameFactory;
 import org.apache.hyracks.storage.am.btree.frames.BTreeNSMLeafFrameFactory;
 import org.apache.hyracks.storage.am.btree.impls.BTree;
+import org.apache.hyracks.storage.am.btree.impls.DiskBTree;
 import org.apache.hyracks.storage.am.btree.tuples.BTreeTypeAwareTupleWriterFactory;
 import org.apache.hyracks.storage.am.common.api.IMetadataPageManagerFactory;
 import org.apache.hyracks.storage.am.common.api.ITreeIndexFrameFactory;
@@ -54,6 +55,7 @@
 import org.apache.hyracks.storage.am.lsm.common.frames.LSMComponentFilterFrameFactory;
 import org.apache.hyracks.storage.am.lsm.common.impls.BTreeFactory;
 import org.apache.hyracks.storage.am.lsm.common.impls.ComponentFilterHelper;
+import org.apache.hyracks.storage.am.lsm.common.impls.DiskBTreeFactory;
 import org.apache.hyracks.storage.am.lsm.common.impls.LSMComponentFilterManager;
 import org.apache.hyracks.storage.am.lsm.common.impls.TreeIndexFactory;
 import org.apache.hyracks.storage.common.buffercache.IBufferCache;
@@ -87,10 +89,11 @@
         ITreeIndexFrameFactory interiorFrameFactory = new BTreeNSMInteriorFrameFactory(insertTupleWriterFactory);
         ITreeIndexFrameFactory bulkLoadLeafFrameFactory = new BTreeNSMLeafFrameFactory(bulkLoadTupleWriterFactory);
 
-        TreeIndexFactory<BTree> diskBTreeFactory = new BTreeFactory(ioManager, diskBufferCache, freePageManagerFactory,
-                interiorFrameFactory, copyTupleLeafFrameFactory, cmpFactories, typeTraits.length);
-        TreeIndexFactory<BTree> bulkLoadBTreeFactory =
-                new BTreeFactory(ioManager, diskBufferCache, freePageManagerFactory, interiorFrameFactory,
+        TreeIndexFactory<DiskBTree> diskBTreeFactory =
+                new DiskBTreeFactory(ioManager, diskBufferCache, freePageManagerFactory, interiorFrameFactory,
+                        copyTupleLeafFrameFactory, cmpFactories, typeTraits.length);
+        TreeIndexFactory<DiskBTree> bulkLoadBTreeFactory =
+                new DiskBTreeFactory(ioManager, diskBufferCache, freePageManagerFactory, interiorFrameFactory,
                         bulkLoadLeafFrameFactory, cmpFactories, typeTraits.length);
 
         ComponentFilterHelper filterHelper = null;
@@ -151,16 +154,17 @@
         ITreeIndexFrameFactory transactionLeafFrameFactory =
                 new BTreeNSMLeafFrameFactory(transactionTupleWriterFactory);
 
-        TreeIndexFactory<BTree> diskBTreeFactory = new BTreeFactory(ioManager, diskBufferCache, freePageManagerFactory,
-                interiorFrameFactory, copyTupleLeafFrameFactory, cmpFactories, typeTraits.length);
-        TreeIndexFactory<BTree> bulkLoadBTreeFactory = new BTreeFactory(ioManager, diskBufferCache,
+        TreeIndexFactory<DiskBTree> diskBTreeFactory =
+                new DiskBTreeFactory(ioManager, diskBufferCache, freePageManagerFactory, interiorFrameFactory,
+                        copyTupleLeafFrameFactory, cmpFactories, typeTraits.length);
+        TreeIndexFactory<DiskBTree> bulkLoadBTreeFactory = new DiskBTreeFactory(ioManager, diskBufferCache,
                 freePageManagerFactory, interiorFrameFactory, insertLeafFrameFactory, cmpFactories, typeTraits.length);
 
         BloomFilterFactory bloomFilterFactory = new BloomFilterFactory(diskBufferCache, bloomFilterKeyFields);
 
         // This is the component factory for transactions
-        TreeIndexFactory<BTree> transactionBTreeFactory =
-                new BTreeFactory(ioManager, diskBufferCache, freePageManagerFactory, interiorFrameFactory,
+        TreeIndexFactory<DiskBTree> transactionBTreeFactory =
+                new DiskBTreeFactory(ioManager, diskBufferCache, freePageManagerFactory, interiorFrameFactory,
                         transactionLeafFrameFactory, cmpFactories, typeTraits.length);
         //TODO remove BloomFilter from external dataset's secondary LSMBTree index
         ILSMIndexFileManager fileNameManager = new LSMBTreeFileManager(ioManager, file, diskBTreeFactory, true);
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/DiskBTreeFactory.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/DiskBTreeFactory.java
new file mode 100644
index 0000000..ba863ac
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/DiskBTreeFactory.java
@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hyracks.storage.am.lsm.common.impls;
+
+import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
+import org.apache.hyracks.api.io.FileReference;
+import org.apache.hyracks.api.io.IIOManager;
+import org.apache.hyracks.storage.am.btree.impls.DiskBTree;
+import org.apache.hyracks.storage.am.common.api.IPageManagerFactory;
+import org.apache.hyracks.storage.am.common.api.ITreeIndexFrameFactory;
+import org.apache.hyracks.storage.common.buffercache.IBufferCache;
+
+public class DiskBTreeFactory extends TreeIndexFactory<DiskBTree> {
+
+    public DiskBTreeFactory(IIOManager ioManager, IBufferCache bufferCache, IPageManagerFactory freePageManagerFactory,
+            ITreeIndexFrameFactory interiorFrameFactory, ITreeIndexFrameFactory leafFrameFactory,
+            IBinaryComparatorFactory[] cmpFactories, int fieldCount) {
+        super(ioManager, bufferCache, freePageManagerFactory, interiorFrameFactory, leafFrameFactory, cmpFactories,
+                fieldCount);
+    }
+
+    @Override
+    public DiskBTree createIndexInstance(FileReference file) {
+        return new DiskBTree(bufferCache, freePageManagerFactory.createPageManager(bufferCache), interiorFrameFactory,
+                leafFrameFactory, cmpFactories, fieldCount, file);
+    }
+
+}
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/ondisk/OnDiskInvertedIndex.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/ondisk/OnDiskInvertedIndex.java
index 8db298d..4942eda 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/ondisk/OnDiskInvertedIndex.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/ondisk/OnDiskInvertedIndex.java
@@ -37,6 +37,7 @@
 import org.apache.hyracks.dataflow.common.utils.TupleUtils;
 import org.apache.hyracks.storage.am.btree.frames.BTreeLeafFrameType;
 import org.apache.hyracks.storage.am.btree.impls.BTree;
+import org.apache.hyracks.storage.am.btree.impls.DiskBTree;
 import org.apache.hyracks.storage.am.btree.impls.RangePredicate;
 import org.apache.hyracks.storage.am.btree.util.BTreeUtils;
 import org.apache.hyracks.storage.am.common.api.IIndexOperationContext;
@@ -90,7 +91,7 @@
         btreeValueTypeTraits[3] = IntegerPointable.TYPE_TRAITS;
     }
 
-    protected BTree btree;
+    protected DiskBTree btree;
     protected int rootPageId = 0;
     protected IBufferCache bufferCache;
     protected int fileId = -1;
@@ -117,7 +118,7 @@
         this.invListCmpFactories = invListCmpFactories;
         this.tokenTypeTraits = tokenTypeTraits;
         this.tokenCmpFactories = tokenCmpFactories;
-        this.btree = BTreeUtils.createBTree(bufferCache, getBTreeTypeTraits(tokenTypeTraits), tokenCmpFactories,
+        this.btree = BTreeUtils.createDiskBTree(bufferCache, getBTreeTypeTraits(tokenTypeTraits), tokenCmpFactories,
                 BTreeLeafFrameType.REGULAR_NSM, btreeFile, pageManagerFactory.createPageManager(bufferCache), false);
         this.numTokenFields = btree.getComparatorFactories().length;
         this.numInvListKeys = invListCmpFactories.length;
@@ -240,7 +241,7 @@
         private final boolean verifyInput;
         private final MultiComparator allCmp;
 
-        private IFIFOPageQueue queue;
+        private final IFIFOPageQueue queue;
 
         public OnDiskInvertedIndexBulkLoader(float btreeFillFactor, boolean verifyInput, long numElementsHint,
                 boolean checkIfEmptyIndex, int startPageId) throws HyracksDataException {
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-btree-test/src/test/java/org/apache/hyracks/storage/am/btree/BTreeSearchCursorTest.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-btree-test/src/test/java/org/apache/hyracks/storage/am/btree/BTreeSearchCursorTest.java
index 6c234b7..68a3984 100644
--- a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-btree-test/src/test/java/org/apache/hyracks/storage/am/btree/BTreeSearchCursorTest.java
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-btree-test/src/test/java/org/apache/hyracks/storage/am/btree/BTreeSearchCursorTest.java
@@ -24,6 +24,7 @@
 import java.io.DataInputStream;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.List;
 import java.util.Random;
 import java.util.TreeSet;
 
@@ -43,19 +44,19 @@
 import org.apache.hyracks.storage.am.btree.frames.BTreeNSMInteriorFrameFactory;
 import org.apache.hyracks.storage.am.btree.frames.BTreeNSMLeafFrameFactory;
 import org.apache.hyracks.storage.am.btree.impls.BTree;
-import org.apache.hyracks.storage.am.btree.impls.BTreeRangeSearchCursor;
+import org.apache.hyracks.storage.am.btree.impls.BTree.BTreeAccessor;
 import org.apache.hyracks.storage.am.btree.impls.RangePredicate;
 import org.apache.hyracks.storage.am.btree.tuples.BTreeTypeAwareTupleWriterFactory;
 import org.apache.hyracks.storage.am.btree.util.AbstractBTreeTest;
 import org.apache.hyracks.storage.am.common.TestOperationCallback;
 import org.apache.hyracks.storage.am.common.api.IMetadataPageManager;
 import org.apache.hyracks.storage.am.common.api.ITreeIndexAccessor;
-import org.apache.hyracks.storage.am.common.api.ITreeIndexCursor;
 import org.apache.hyracks.storage.am.common.api.ITreeIndexFrameFactory;
 import org.apache.hyracks.storage.am.common.api.ITreeIndexMetadataFrameFactory;
 import org.apache.hyracks.storage.am.common.frames.LIFOMetaDataFrameFactory;
 import org.apache.hyracks.storage.am.common.freepage.LinkedMetaDataPageManager;
 import org.apache.hyracks.storage.am.common.impls.IndexAccessParameters;
+import org.apache.hyracks.storage.common.IIndexCursor;
 import org.apache.hyracks.storage.common.MultiComparator;
 import org.apache.hyracks.storage.common.buffercache.IBufferCache;
 import org.junit.Assert;
@@ -63,12 +64,12 @@
 import org.junit.Test;
 
 public class BTreeSearchCursorTest extends AbstractBTreeTest {
-    private final int fieldCount = 2;
-    private final ITypeTraits[] typeTraits = new ITypeTraits[fieldCount];
-    private final BTreeTypeAwareTupleWriterFactory tupleWriterFactory =
+    protected final int fieldCount = 2;
+    protected final ITypeTraits[] typeTraits = new ITypeTraits[fieldCount];
+    protected final BTreeTypeAwareTupleWriterFactory tupleWriterFactory =
             new BTreeTypeAwareTupleWriterFactory(typeTraits, false);
-    private final ITreeIndexMetadataFrameFactory metaFrameFactory = new LIFOMetaDataFrameFactory();
-    private final Random rnd = new Random(50);
+    protected final ITreeIndexMetadataFrameFactory metaFrameFactory = new LIFOMetaDataFrameFactory();
+    protected final Random rnd = new Random(50);
 
     @Override
     @Before
@@ -104,13 +105,6 @@
         btree.create();
         btree.activate();
 
-        ArrayTupleBuilder tupleBuilder = new ArrayTupleBuilder(fieldCount);
-        ArrayTupleReference tuple = new ArrayTupleReference();
-
-        IndexAccessParameters actx =
-                new IndexAccessParameters(TestOperationCallback.INSTANCE, TestOperationCallback.INSTANCE);
-        ITreeIndexAccessor indexAccessor = btree.createAccessor(actx);
-
         // generate keys
         int numKeys = 50;
         int maxKey = 1000;
@@ -124,18 +118,7 @@
             keys.add(i);
         }
 
-        // insert keys into btree
-        for (int i = 0; i < keys.size(); i++) {
-
-            TupleUtils.createIntegerTuple(tupleBuilder, tuple, keys.get(i), i);
-            tuple.reset(tupleBuilder.getFieldEndOffsets(), tupleBuilder.getByteArray());
-
-            try {
-                indexAccessor.insert(tuple);
-            } catch (Exception e) {
-                e.printStackTrace();
-            }
-        }
+        insertBTree(keys, btree);
 
         int minSearchKey = -100;
         int maxSearchKey = 100;
@@ -181,13 +164,6 @@
         btree.create();
         btree.activate();
 
-        ArrayTupleBuilder tupleBuilder = new ArrayTupleBuilder(fieldCount);
-        ArrayTupleReference tuple = new ArrayTupleReference();
-
-        IndexAccessParameters actx =
-                new IndexAccessParameters(TestOperationCallback.INSTANCE, TestOperationCallback.INSTANCE);
-        ITreeIndexAccessor indexAccessor = btree.createAccessor(actx);
-
         // generate keys
         int numKeys = 50;
         int maxKey = 10;
@@ -198,18 +174,7 @@
         }
         Collections.sort(keys);
 
-        // insert keys into btree
-        for (int i = 0; i < keys.size(); i++) {
-
-            TupleUtils.createIntegerTuple(tupleBuilder, tuple, keys.get(i), i);
-            tuple.reset(tupleBuilder.getFieldEndOffsets(), tupleBuilder.getByteArray());
-
-            try {
-                indexAccessor.insert(tuple);
-            } catch (Exception e) {
-                e.printStackTrace();
-            }
-        }
+        insertBTree(keys, btree);
 
         int minSearchKey = -100;
         int maxSearchKey = 100;
@@ -254,14 +219,6 @@
                 fieldCount, harness.getFileReference());
         btree.create();
         btree.activate();
-
-        ArrayTupleBuilder tupleBuilder = new ArrayTupleBuilder(fieldCount);
-        ArrayTupleReference tuple = new ArrayTupleReference();
-
-        IndexAccessParameters actx =
-                new IndexAccessParameters(TestOperationCallback.INSTANCE, TestOperationCallback.INSTANCE);
-        ITreeIndexAccessor indexAccessor = btree.createAccessor(actx);
-
         // generate keys
         int numKeys = 50;
         int maxKey = 10;
@@ -272,18 +229,7 @@
         }
         Collections.sort(keys);
 
-        // insert keys into btree
-        for (int i = 0; i < keys.size(); i++) {
-
-            TupleUtils.createIntegerTuple(tupleBuilder, tuple, keys.get(i), i);
-            tuple.reset(tupleBuilder.getFieldEndOffsets(), tupleBuilder.getByteArray());
-
-            try {
-                indexAccessor.insert(tuple);
-            } catch (Exception e) {
-                e.printStackTrace();
-            }
-        }
+        insertBTree(keys, btree);
 
         int minSearchKey = -100;
         int maxSearchKey = 100;
@@ -304,7 +250,6 @@
 
     public RangePredicate createRangePredicate(int lk, int hk, boolean lowKeyInclusive, boolean highKeyInclusive)
             throws HyracksDataException {
-
         // create tuplereferences for search keys
         ITupleReference lowKey = TupleUtils.createIntegerTuple(false, lk);
         ITupleReference highKey = TupleUtils.createIntegerTuple(false, hk);
@@ -351,18 +296,17 @@
 
         for (int i = minKey; i < maxKey; i++) {
             for (int j = minKey; j < maxKey; j++) {
-
                 results.clear();
                 expectedResults.clear();
 
                 int lowKey = i;
                 int highKey = j;
 
-                ITreeIndexCursor rangeCursor = new BTreeRangeSearchCursor(leafFrame, false);
                 RangePredicate rangePred = createRangePredicate(lowKey, highKey, lowKeyInclusive, highKeyInclusive);
                 IndexAccessParameters actx =
                         new IndexAccessParameters(TestOperationCallback.INSTANCE, TestOperationCallback.INSTANCE);
                 ITreeIndexAccessor indexAccessor = btree.createAccessor(actx);
+                IIndexCursor rangeCursor = indexAccessor.createSearchCursor(false);
                 indexAccessor.search(rangeCursor, rangePred);
 
                 try {
@@ -435,4 +379,19 @@
 
         return true;
     }
+
+    protected void insertBTree(List<Integer> keys, BTree btree) throws HyracksDataException {
+        IndexAccessParameters actx =
+                new IndexAccessParameters(TestOperationCallback.INSTANCE, TestOperationCallback.INSTANCE);
+        BTreeAccessor accessor = btree.createAccessor(actx);
+        ArrayTupleBuilder tupleBuilder = new ArrayTupleBuilder(fieldCount);
+        ArrayTupleReference tuple = new ArrayTupleReference();
+
+        // insert keys into btree
+        for (int i = 0; i < keys.size(); i++) {
+            TupleUtils.createIntegerTuple(tupleBuilder, tuple, keys.get(i), i);
+            tuple.reset(tupleBuilder.getFieldEndOffsets(), tupleBuilder.getByteArray());
+            accessor.insert(tuple);
+        }
+    }
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-btree-test/src/test/java/org/apache/hyracks/storage/am/btree/DiskBTreeSearchCursorTest.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-btree-test/src/test/java/org/apache/hyracks/storage/am/btree/DiskBTreeSearchCursorTest.java
new file mode 100644
index 0000000..c7f0425
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-btree-test/src/test/java/org/apache/hyracks/storage/am/btree/DiskBTreeSearchCursorTest.java
@@ -0,0 +1,203 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hyracks.storage.am.btree;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInput;
+import java.io.DataInputStream;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.TreeSet;
+
+import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.data.std.accessors.PointableBinaryComparatorFactory;
+import org.apache.hyracks.data.std.primitive.IntegerPointable;
+import org.apache.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
+import org.apache.hyracks.dataflow.common.comm.io.ArrayTupleReference;
+import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
+import org.apache.hyracks.dataflow.common.data.marshalling.IntegerSerializerDeserializer;
+import org.apache.hyracks.dataflow.common.utils.TupleUtils;
+import org.apache.hyracks.storage.am.btree.api.IBTreeInteriorFrame;
+import org.apache.hyracks.storage.am.btree.api.IBTreeLeafFrame;
+import org.apache.hyracks.storage.am.btree.frames.BTreeNSMInteriorFrameFactory;
+import org.apache.hyracks.storage.am.btree.frames.BTreeNSMLeafFrameFactory;
+import org.apache.hyracks.storage.am.btree.impls.BTree;
+import org.apache.hyracks.storage.am.btree.impls.BTree.BTreeAccessor;
+import org.apache.hyracks.storage.am.btree.impls.DiskBTree;
+import org.apache.hyracks.storage.am.btree.impls.RangePredicate;
+import org.apache.hyracks.storage.am.common.TestOperationCallback;
+import org.apache.hyracks.storage.am.common.api.IMetadataPageManager;
+import org.apache.hyracks.storage.am.common.api.ITreeIndexFrameFactory;
+import org.apache.hyracks.storage.am.common.freepage.LinkedMetaDataPageManager;
+import org.apache.hyracks.storage.am.common.impls.IndexAccessParameters;
+import org.apache.hyracks.storage.common.IIndexBulkLoader;
+import org.apache.hyracks.storage.common.IIndexCursor;
+import org.apache.hyracks.storage.common.buffercache.IBufferCache;
+import org.junit.Assert;
+import org.junit.FixMethodOrder;
+import org.junit.Test;
+import org.junit.runners.MethodSorters;
+
+@FixMethodOrder(MethodSorters.NAME_ASCENDING)
+public class DiskBTreeSearchCursorTest extends BTreeSearchCursorTest {
+
+    @Test
+    public void batchPointLookupSingleLevelTest() throws Exception {
+        if (LOGGER.isInfoEnabled()) {
+            LOGGER.info("TESTING POINT LOOKUP CURSOR ON UNIQUE 1 LEVEL INDEX");
+        }
+        // only 4 keys
+        batchPointLookupTest(4, 4, -8, 8);
+    }
+
+    @Test
+    public void batchPointLookupMultiLevelTest() throws Exception {
+        if (LOGGER.isInfoEnabled()) {
+            LOGGER.info("TESTING POINT LOOKUP CURSOR ON UNIQUE MULTI-LEVEL INDEX");
+        }
+        // only 10000 keys
+        batchPointLookupTest(10000, 20000, -1000, 1000);
+    }
+
+    @Test
+    public void batchPointLookupMultiLevelOutRangeTest() throws Exception {
+        if (LOGGER.isInfoEnabled()) {
+            LOGGER.info("TESTING POINT LOOKUP CURSOR ON UNIQUE 1 LEVEL INDEX");
+        }
+        // only 100 keys
+        batchPointLookupTest(100, 200, -1000, 1000);
+    }
+
+    private void batchPointLookupTest(int numKeys, int maxKey, int minSearchKey, int maxSearchKey) throws Exception {
+
+        IBufferCache bufferCache = harness.getBufferCache();
+
+        // declare keys
+        int keyFieldCount = 1;
+        IBinaryComparatorFactory[] cmpFactories = new IBinaryComparatorFactory[keyFieldCount];
+        cmpFactories[0] = PointableBinaryComparatorFactory.of(IntegerPointable.FACTORY);
+
+        ITreeIndexFrameFactory leafFrameFactory = new BTreeNSMLeafFrameFactory(tupleWriterFactory);
+        ITreeIndexFrameFactory interiorFrameFactory = new BTreeNSMInteriorFrameFactory(tupleWriterFactory);
+
+        IBTreeLeafFrame leafFrame = (IBTreeLeafFrame) leafFrameFactory.createFrame();
+        IBTreeInteriorFrame interiorFrame = (IBTreeInteriorFrame) interiorFrameFactory.createFrame();
+
+        IMetadataPageManager freePageManager = new LinkedMetaDataPageManager(bufferCache, metaFrameFactory);
+
+        DiskBTree btree = new DiskBTree(bufferCache, freePageManager, interiorFrameFactory, leafFrameFactory,
+                cmpFactories, fieldCount, harness.getFileReference());
+        btree.create();
+        btree.activate();
+
+        TreeSet<Integer> uniqueKeys = new TreeSet<>();
+        ArrayList<Integer> keys = new ArrayList<>();
+        while (uniqueKeys.size() < numKeys) {
+            int key = rnd.nextInt() % maxKey;
+            uniqueKeys.add(key);
+        }
+        for (Integer i : uniqueKeys) {
+            keys.add(i);
+        }
+
+        insertBTree(keys, btree);
+
+        // forward searches
+        Assert.assertTrue(performBatchLookups(keys, btree, leafFrame, interiorFrame, minSearchKey, maxSearchKey));
+
+        btree.deactivate();
+        btree.destroy();
+    }
+
+    private boolean performBatchLookups(ArrayList<Integer> keys, BTree btree, IBTreeLeafFrame leafFrame,
+            IBTreeInteriorFrame interiorFrame, int minKey, int maxKey) throws Exception {
+
+        ArrayList<Integer> results = new ArrayList<>();
+        ArrayList<Integer> expectedResults = new ArrayList<>();
+        BTreeAccessor indexAccessor = btree.createAccessor(
+                new IndexAccessParameters(TestOperationCallback.INSTANCE, TestOperationCallback.INSTANCE));
+        IIndexCursor pointCursor = indexAccessor.createPointCursor(false);
+        for (int i = minKey; i < maxKey; i++) {
+            results.clear();
+            expectedResults.clear();
+
+            int lowKey = i;
+            int highKey = i;
+            RangePredicate rangePred = createRangePredicate(lowKey, highKey, true, true);
+            indexAccessor.search(pointCursor, rangePred);
+            try {
+                while (pointCursor.hasNext()) {
+                    pointCursor.next();
+                    ITupleReference frameTuple = pointCursor.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();
+            }
+
+            getExpectedResults(expectedResults, keys, lowKey, highKey, true, true);
+
+            if (results.size() == expectedResults.size()) {
+                for (int k = 0; k < results.size(); k++) {
+                    if (!results.get(k).equals(expectedResults.get(k))) {
+                        if (LOGGER.isInfoEnabled()) {
+                            LOGGER.info("DIFFERENT RESULTS AT: i=" + i + " k=" + k);
+                            LOGGER.info(results.get(k) + " " + expectedResults.get(k));
+                        }
+                        return false;
+                    }
+                }
+            } else {
+                if (LOGGER.isInfoEnabled()) {
+                    LOGGER.info("UNEQUAL NUMBER OF RESULTS AT: i=" + i);
+                    LOGGER.info("RESULTS: " + results.size());
+                    LOGGER.info("EXPECTED RESULTS: " + expectedResults.size());
+                }
+                return false;
+            }
+        }
+
+        pointCursor.close();
+
+        return true;
+    }
+
+    @Override
+    protected void insertBTree(List<Integer> keys, BTree btree) throws HyracksDataException {
+        ArrayTupleBuilder tupleBuilder = new ArrayTupleBuilder(fieldCount);
+        ArrayTupleReference tuple = new ArrayTupleReference();
+
+        IIndexBulkLoader bulkloader = btree.createBulkLoader(1, true, 0, true);
+        // insert keys into btree
+        for (int i = 0; i < keys.size(); i++) {
+            TupleUtils.createIntegerTuple(tupleBuilder, tuple, keys.get(i), i);
+            tuple.reset(tupleBuilder.getFieldEndOffsets(), tupleBuilder.getByteArray());
+
+            bulkloader.add(tuple);
+        }
+        bulkloader.end();
+    }
+
+}
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/TestLsmBtreeUtil.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/TestLsmBtreeUtil.java
index 4000c1d..d614aff 100644
--- a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/TestLsmBtreeUtil.java
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/TestLsmBtreeUtil.java
@@ -28,7 +28,7 @@
 import org.apache.hyracks.storage.am.bloomfilter.impls.BloomFilterFactory;
 import org.apache.hyracks.storage.am.btree.frames.BTreeNSMInteriorFrameFactory;
 import org.apache.hyracks.storage.am.btree.frames.BTreeNSMLeafFrameFactory;
-import org.apache.hyracks.storage.am.btree.impls.BTree;
+import org.apache.hyracks.storage.am.btree.impls.DiskBTree;
 import org.apache.hyracks.storage.am.common.api.IMetadataPageManagerFactory;
 import org.apache.hyracks.storage.am.common.api.ITreeIndexFrameFactory;
 import org.apache.hyracks.storage.am.common.tuples.TypeAwareTupleWriterFactory;
@@ -46,8 +46,8 @@
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMOperationTracker;
 import org.apache.hyracks.storage.am.lsm.common.api.IVirtualBufferCache;
 import org.apache.hyracks.storage.am.lsm.common.frames.LSMComponentFilterFrameFactory;
-import org.apache.hyracks.storage.am.lsm.common.impls.BTreeFactory;
 import org.apache.hyracks.storage.am.lsm.common.impls.ComponentFilterHelper;
+import org.apache.hyracks.storage.am.lsm.common.impls.DiskBTreeFactory;
 import org.apache.hyracks.storage.am.lsm.common.impls.LSMComponentFilterManager;
 import org.apache.hyracks.storage.am.lsm.common.impls.TreeIndexFactory;
 import org.apache.hyracks.storage.common.buffercache.IBufferCache;
@@ -77,9 +77,10 @@
         ITreeIndexFrameFactory deleteLeafFrameFactory = new BTreeNSMLeafFrameFactory(deleteTupleWriterFactory);
         ITreeIndexFrameFactory interiorFrameFactory = new BTreeNSMInteriorFrameFactory(insertTupleWriterFactory);
 
-        TreeIndexFactory<BTree> diskBTreeFactory = new BTreeFactory(ioManager, diskBufferCache, freePageManagerFactory,
-                interiorFrameFactory, copyTupleLeafFrameFactory, cmpFactories, typeTraits.length);
-        TreeIndexFactory<BTree> bulkLoadBTreeFactory = new BTreeFactory(ioManager, diskBufferCache,
+        TreeIndexFactory<DiskBTree> diskBTreeFactory =
+                new DiskBTreeFactory(ioManager, diskBufferCache, freePageManagerFactory, interiorFrameFactory,
+                        copyTupleLeafFrameFactory, cmpFactories, typeTraits.length);
+        TreeIndexFactory<DiskBTree> bulkLoadBTreeFactory = new DiskBTreeFactory(ioManager, diskBufferCache,
                 freePageManagerFactory, interiorFrameFactory, insertLeafFrameFactory, cmpFactories, typeTraits.length);
 
         ComponentFilterHelper filterHelper = null;