diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IBTreeFrame.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IBTreeFrame.java
index a24c4d7..affc8ce 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IBTreeFrame.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/api/IBTreeFrame.java
@@ -1,3 +1,18 @@
+/*
+ * Copyright 2009-2010 by The Regents of the University of California
+ * Licensed 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 from
+ * 
+ *     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 edu.uci.ics.hyracks.storage.am.btree.api;
 
 import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
diff --git a/hyracks-storage-am-lsmtree-btree/pom.xml b/hyracks-storage-am-lsmtree-btree/pom.xml
index 4175e72..062beb7 100644
--- a/hyracks-storage-am-lsmtree-btree/pom.xml
+++ b/hyracks-storage-am-lsmtree-btree/pom.xml
@@ -37,6 +37,13 @@
   		<version>0.2.0-SNAPSHOT</version>
   		<type>jar</type>
   		<scope>compile</scope>
+  	</dependency> 
+  	<dependency>
+  		<groupId>edu.uci.ics.hyracks</groupId>
+  		<artifactId>hyracks-storage-am-lsmtree-common</artifactId>
+  		<version>0.2.0-SNAPSHOT</version>
+  		<type>jar</type>
+  		<scope>compile</scope>
   	</dependency> 	  		
   	<dependency>
   		<groupId>junit</groupId>
diff --git a/hyracks-storage-am-lsmtree-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsmtree/freepage/InMemoryBufferCache.java b/hyracks-storage-am-lsmtree-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsmtree/freepage/InMemoryBufferCache.java
deleted file mode 100644
index c523440..0000000
--- a/hyracks-storage-am-lsmtree-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsmtree/freepage/InMemoryBufferCache.java
+++ /dev/null
@@ -1,146 +0,0 @@
-package edu.uci.ics.hyracks.storage.am.lsmtree.freepage;
-
-import java.nio.ByteBuffer;
-import java.util.concurrent.locks.ReadWriteLock;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
-
-import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
-import edu.uci.ics.hyracks.api.io.FileReference;
-import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCacheInternal;
-import edu.uci.ics.hyracks.storage.common.buffercache.ICacheMemoryAllocator;
-import edu.uci.ics.hyracks.storage.common.buffercache.ICachedPage;
-import edu.uci.ics.hyracks.storage.common.buffercache.ICachedPageInternal;
-import edu.uci.ics.hyracks.storage.common.file.BufferedFileHandle;
-
-public class InMemoryBufferCache implements IBufferCacheInternal {
-	
-	private final int pageSize;
-	private final int numPages;
-    private final CachedPage[] cachedPages;
-
-	//Constructor
-	public InMemoryBufferCache(ICacheMemoryAllocator allocator, int pageSize, int numPages){
-		
-        this.pageSize = pageSize;
-        this.numPages = numPages;
-		ByteBuffer[] buffers = allocator.allocate(this.pageSize, this.numPages);
-		cachedPages = new CachedPage[buffers.length];
-        for (int i = 0; i < buffers.length; ++i) {
-            cachedPages[i] = new CachedPage(i, buffers[i]);
-        }
-	}
-	
-	@Override
-	public void createFile(FileReference fileRef) throws HyracksDataException {
-		// Do nothing
-	}
-
-	@Override
-	public void openFile(int fileId) throws HyracksDataException {
-		// Do nothing		
-	}
-
-	@Override
-	public void closeFile(int fileId) throws HyracksDataException {
-		// Do nothing
-	}
-
-	@Override
-	public void deleteFile(int fileId) throws HyracksDataException {
-		// Do nothing
-	}
-
-	@Override
-	public ICachedPage tryPin(long dpid) throws HyracksDataException {
-		// Just call pin!
-		return null;
-	}
-
-	@Override
-	public ICachedPage pin(long dpid, boolean newPage){
-		return cachedPages[BufferedFileHandle.getPageId(dpid)];
-	}
-
-	@Override
-	public void unpin(ICachedPage page) throws HyracksDataException {
-		//Do Nothing
-	}
-
-	@Override
-	public int getPageSize() {
-		return pageSize;
-	}
-
-	@Override
-	public int getNumPages() {
-		return numPages;
-	}
-
-	@Override
-	public void close() {
-		// Do nothing	
-	}
-
-	@Override
-	public ICachedPageInternal getPage(int cpid) {
-		return cachedPages[cpid];
-	}
-
-    private class CachedPage implements ICachedPageInternal {
-        private final int cpid;
-        private final ByteBuffer buffer;
-        private final ReadWriteLock latch;
-
-        public CachedPage(int cpid, ByteBuffer buffer) {
-            this.cpid = cpid;
-            this.buffer = buffer;
-            latch = new ReentrantReadWriteLock(true);
-        }
-
-        @Override
-        public ByteBuffer getBuffer() {
-            return buffer;
-        }
-
-        @Override
-        public Object getReplacementStrategyObject() {
-        	//Do nothing
-            return null;
-        }
-
-        @Override
-        public boolean pinIfGoodVictim() {
-        	//Do nothing
-        	return false;
-        }
-
-        @Override
-        public int getCachedPageId() {
-            return cpid;
-        }
-
-        @Override
-        public void acquireReadLatch() {
-            latch.readLock().lock();
-        }
-
-        private void acquireWriteLatch(boolean markDirty) {
-            latch.writeLock().lock();
-        }
-
-        @Override
-        public void acquireWriteLatch() {
-            acquireWriteLatch(true);
-        }
-
-        @Override
-        public void releaseReadLatch() {
-            latch.readLock().unlock();
-        }
-
-        @Override
-        public void releaseWriteLatch() {
-            latch.writeLock().unlock();
-        }
-    }
-}
diff --git a/hyracks-storage-am-lsmtree-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsmtree/freepage/InMemoryBufferCacheFactory.java b/hyracks-storage-am-lsmtree-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsmtree/freepage/InMemoryBufferCacheFactory.java
deleted file mode 100644
index 07e1c15..0000000
--- a/hyracks-storage-am-lsmtree-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsmtree/freepage/InMemoryBufferCacheFactory.java
+++ /dev/null
@@ -1,26 +0,0 @@
-package edu.uci.ics.hyracks.storage.am.lsmtree.freepage;
-
-import edu.uci.ics.hyracks.storage.common.buffercache.HeapBufferAllocator;
-import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
-import edu.uci.ics.hyracks.storage.common.buffercache.ICacheMemoryAllocator;
-
-public class InMemoryBufferCacheFactory {
-	
-	private IBufferCache bufferCache;
-    private final int pageSize;
-    private final int numPages;
-	
-    public InMemoryBufferCacheFactory(int pageSize, int numPages) {
-    	this.pageSize = pageSize;
-    	this.numPages = numPages;
-        bufferCache = null;
-    }
-    
-    public synchronized IBufferCache createInMemoryBufferCache() {
-        if (bufferCache == null) {
-            ICacheMemoryAllocator allocator = new HeapBufferAllocator();
-            bufferCache = new InMemoryBufferCache(allocator, pageSize, numPages);
-        }
-        return bufferCache;
-    }
-}
diff --git a/hyracks-storage-am-lsmtree-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsmtree/freepage/InMemoryBufferCacheTest.java b/hyracks-storage-am-lsmtree-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsmtree/freepage/InMemoryBufferCacheTest.java
index 418ed5c..70e3ac6 100644
--- a/hyracks-storage-am-lsmtree-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsmtree/freepage/InMemoryBufferCacheTest.java
+++ b/hyracks-storage-am-lsmtree-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsmtree/freepage/InMemoryBufferCacheTest.java
@@ -2,8 +2,10 @@
 
 import static org.junit.Assert.fail;
 
-import org.junit.*;
+import org.junit.Test;
 
+import edu.uci.ics.hyracks.storage.am.lsmtree.common.impls.InMemoryBufferCache;
+import edu.uci.ics.hyracks.storage.common.buffercache.HeapBufferAllocator;
 import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
 import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCacheInternal;
 import edu.uci.ics.hyracks.storage.common.buffercache.ICachedPage;
@@ -16,9 +18,7 @@
     //TEST InMemoryBufferCache.pin()
     @Test
     public void InMemoryBufferCacheTest01() throws Exception {
-    	
-        InMemoryBufferCacheFactory InMemBufferCacheFactory = new InMemoryBufferCacheFactory(PAGE_SIZE, NUM_PAGES);
-        IBufferCache memBufferCache = InMemBufferCacheFactory.createInMemoryBufferCache();
+        IBufferCache memBufferCache = new InMemoryBufferCache(new HeapBufferAllocator(), PAGE_SIZE, NUM_PAGES);
 
         try {
         	ICachedPage memCachedPage = memBufferCache.pin(10, true);
@@ -43,8 +43,7 @@
     @Test
     public void InMemoryBufferCacheTest02() throws Exception {
     	
-        InMemoryBufferCacheFactory InMemBufferCacheFactory = new InMemoryBufferCacheFactory(PAGE_SIZE, NUM_PAGES);
-        IBufferCache memBufferCache = InMemBufferCacheFactory.createInMemoryBufferCache();
+        IBufferCache memBufferCache = new InMemoryBufferCache(new HeapBufferAllocator(), PAGE_SIZE, NUM_PAGES);
 
         try {
         	ICachedPage memCachedPage = memBufferCache.pin(500, true);
@@ -68,8 +67,7 @@
     @Test
     public void InMemoryBufferCacheTest03() throws Exception {
     	
-        InMemoryBufferCacheFactory InMemBufferCacheFactory = new InMemoryBufferCacheFactory(PAGE_SIZE, NUM_PAGES);
-        IBufferCache memBufferCache = InMemBufferCacheFactory.createInMemoryBufferCache();
+        IBufferCache memBufferCache = new InMemoryBufferCache(new HeapBufferAllocator(), PAGE_SIZE, NUM_PAGES);
 
         try {
         	ICachedPage memCachedPage = ((IBufferCacheInternal) memBufferCache).getPage(20);
@@ -94,8 +92,7 @@
     @Test
     public void InMemoryBufferCacheTest04() throws Exception {
     	
-        InMemoryBufferCacheFactory InMemBufferCacheFactory = new InMemoryBufferCacheFactory(PAGE_SIZE, NUM_PAGES);
-        IBufferCache memBufferCache = InMemBufferCacheFactory.createInMemoryBufferCache();
+        IBufferCache memBufferCache = new InMemoryBufferCache(new HeapBufferAllocator(), PAGE_SIZE, NUM_PAGES);
 
         try {
         	ICachedPage memCachedPage = ((IBufferCacheInternal) memBufferCache).getPage(1000);
diff --git a/hyracks-storage-am-lsmtree-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsmtree/perf/InMemoryBTreeRunner.java b/hyracks-storage-am-lsmtree-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsmtree/perf/InMemoryBTreeRunner.java
index e411856..030301b 100644
--- a/hyracks-storage-am-lsmtree-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsmtree/perf/InMemoryBTreeRunner.java
+++ b/hyracks-storage-am-lsmtree-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsmtree/perf/InMemoryBTreeRunner.java
@@ -17,9 +17,9 @@
 import edu.uci.ics.hyracks.storage.am.common.frames.LIFOMetaDataFrameFactory;
 import edu.uci.ics.hyracks.storage.am.common.ophelpers.MultiComparator;
 import edu.uci.ics.hyracks.storage.am.common.tuples.TypeAwareTupleWriterFactory;
+import edu.uci.ics.hyracks.storage.am.lsmtree.common.impls.InMemoryBufferCache;
 import edu.uci.ics.hyracks.storage.am.lsmtree.datagen.DataGenThread;
 import edu.uci.ics.hyracks.storage.am.lsmtree.datagen.TupleBatch;
-import edu.uci.ics.hyracks.storage.am.lsmtree.freepage.InMemoryBufferCache;
 import edu.uci.ics.hyracks.storage.am.lsmtree.freepage.InMemoryFreePageManager;
 import edu.uci.ics.hyracks.storage.common.buffercache.HeapBufferAllocator;
 import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
diff --git a/hyracks-storage-am-lsmtree-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsmtree/perf/LSMTreeRunner.java b/hyracks-storage-am-lsmtree-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsmtree/perf/LSMTreeRunner.java
index b9175ad..957e044 100644
--- a/hyracks-storage-am-lsmtree-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsmtree/perf/LSMTreeRunner.java
+++ b/hyracks-storage-am-lsmtree-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsmtree/perf/LSMTreeRunner.java
@@ -13,10 +13,11 @@
 import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexAccessor;
 import edu.uci.ics.hyracks.storage.am.common.api.TreeIndexException;
 import edu.uci.ics.hyracks.storage.am.common.ophelpers.MultiComparator;
+import edu.uci.ics.hyracks.storage.am.lsmtree.common.impls.InMemoryBufferCache;
 import edu.uci.ics.hyracks.storage.am.lsmtree.datagen.DataGenThread;
 import edu.uci.ics.hyracks.storage.am.lsmtree.datagen.TupleBatch;
-import edu.uci.ics.hyracks.storage.am.lsmtree.freepage.InMemoryBufferCacheFactory;
 import edu.uci.ics.hyracks.storage.am.lsmtree.impls.LSMTree;
+import edu.uci.ics.hyracks.storage.common.buffercache.HeapBufferAllocator;
 import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
 import edu.uci.ics.hyracks.storage.common.file.IFileMapManager;
 import edu.uci.ics.hyracks.storage.common.file.IFileMapProvider;
@@ -44,7 +45,7 @@
     private final int onDiskPageSize;
     private final int onDiskNumPages;
     
-    public LSMTreeRunner(int numBatches, int inMemPageSize, int inMeNumPages, int onDiskPageSize, int onDiskNumPages, ITypeTraits[] typeTraits, MultiComparator cmp) throws HyracksDataException, BTreeException {
+    public LSMTreeRunner(int numBatches, int inMemPageSize, int inMemNumPages, int onDiskPageSize, int onDiskNumPages, ITypeTraits[] typeTraits, MultiComparator cmp) throws HyracksDataException, BTreeException {
         this.numBatches = numBatches;
         
         this.onDiskPageSize = onDiskPageSize;
@@ -62,10 +63,7 @@
         lsmtreeFileId = fmp.lookupFileId(file);
         bufferCache.openFile(lsmtreeFileId);
         
-        
-        // In Memory
-		InMemoryBufferCacheFactory InMemBufferCacheFactory = new InMemoryBufferCacheFactory(inMemPageSize, inMeNumPages);
-		memBufferCache = InMemBufferCacheFactory.createInMemoryBufferCache();
+        IBufferCache memBufferCache = new InMemoryBufferCache(new HeapBufferAllocator(), inMemPageSize, inMemNumPages);
 		
         lsmtree = LSMTreeUtils.createLSMTree(memBufferCache, bufferCache, lsmtreeFileId, typeTraits, cmp.getComparators(), BTreeLeafFrameType.REGULAR_NSM, (IFileMapManager)fmp);
     }
diff --git a/hyracks-storage-am-lsmtree-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsmtree/tuples/LSMTypeAwareTupleReference.java b/hyracks-storage-am-lsmtree-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsmtree/tuples/LSMTypeAwareTupleReference.java
index a5a3ee4..6dec2c6 100644
--- a/hyracks-storage-am-lsmtree-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsmtree/tuples/LSMTypeAwareTupleReference.java
+++ b/hyracks-storage-am-lsmtree-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsmtree/tuples/LSMTypeAwareTupleReference.java
@@ -11,18 +11,18 @@
 
 	@Override
 	protected int getNullFlagsBytes() {
-		//+1.0 is for insert/delete tuple checking
+		// +1.0 is for insert/delete tuple checking.
 		return (int) Math.ceil((fieldCount + 1.0) / 8.0);
     }
 
 	@Override
 	public boolean isDelete() {
-		byte[] temp = buf.array();
+		// TODO: Alex. Rewrite this to be more efficient...
+	    byte[] temp = buf.array();
 		byte firstByte = temp[tupleStartOff];
 		final byte mask = (byte) (1 << 7);
 		final byte compare = (byte) (1 << 7);
-		
-		//check the first bit is 0 or 1
+		// Check the first bit is 0 or 1.
 		if((byte)(firstByte & mask) == compare) {
 			return true;
 		}
diff --git a/hyracks-storage-am-lsmtree-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsmtree/common/impls/InMemoryBufferCache.java b/hyracks-storage-am-lsmtree-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsmtree/common/impls/InMemoryBufferCache.java
index 25891dc..d793b17 100644
--- a/hyracks-storage-am-lsmtree-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsmtree/common/impls/InMemoryBufferCache.java
+++ b/hyracks-storage-am-lsmtree-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsmtree/common/impls/InMemoryBufferCache.java
@@ -1,6 +1,23 @@
+/*
+ * Copyright 2009-2010 by The Regents of the University of California
+ * Licensed 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 from
+ * 
+ *     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 edu.uci.ics.hyracks.storage.am.lsmtree.common.impls;
 
 import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.concurrent.locks.ReadWriteLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 
@@ -13,77 +30,95 @@
 import edu.uci.ics.hyracks.storage.common.file.BufferedFileHandle;
 
 public class InMemoryBufferCache implements IBufferCacheInternal {
-	
-	private final int pageSize;
-	private final int numPages;
-    protected final CachedPage[] cachedPages;
-
-	//Constructor
+    protected final ICacheMemoryAllocator allocator;
+    protected final int pageSize;
+    protected final CachedPage[] pages;
+    protected final List<CachedPage> overflowPages = new ArrayList<CachedPage>();
+    
 	public InMemoryBufferCache(ICacheMemoryAllocator allocator, int pageSize, int numPages){
-		
-        this.pageSize = pageSize;
-        this.numPages = numPages;
-		ByteBuffer[] buffers = allocator.allocate(this.pageSize, this.numPages);
-		cachedPages = new CachedPage[buffers.length];
+        this.allocator = allocator;
+	    this.pageSize = pageSize;
+		ByteBuffer[] buffers = allocator.allocate(pageSize, numPages);
+		pages = new CachedPage[buffers.length];
         for (int i = 0; i < buffers.length; ++i) {
-            cachedPages[i] = new CachedPage(i, buffers[i]);
+            pages[i] = new CachedPage(i, buffers[i]);
         }
 	}
+
+	@Override
+	public ICachedPage pin(long dpid, boolean newPage) {
+	    int pageId = BufferedFileHandle.getPageId(dpid);
+	    if (pageId < pages.length) {
+	        // Common case: Return regular page.
+	        return pages[pageId];
+	    } else {
+	        // Rare case: Return overflow page, possibly expanding overflow array.
+	        synchronized(overflowPages) {
+	            int numNewPages = pageId - pages.length - overflowPages.size() + 1;          
+	            if (numNewPages > 0) {
+	                ByteBuffer[] buffers = allocator.allocate(pageSize, numNewPages);
+	                for (int i = 0; i < numNewPages; i++) {
+	                    CachedPage overflowPage = new CachedPage(pages.length + overflowPages.size(), buffers[i]);
+	                    overflowPages.add(overflowPage);
+	                }
+	            }
+	            return overflowPages.get(pageId - pages.length);
+	        }
+	    }
+	}
+
+	@Override
+    public ICachedPage tryPin(long dpid) throws HyracksDataException {
+        return pin(dpid, false);
+    }
 	
 	@Override
-	public void createFile(FileReference fileRef) throws HyracksDataException {
-		// Do nothing
-	}
+    public int getPageSize() {
+        return pageSize;
+    }
 
+    @Override
+    public int getNumPages() {
+        return pages.length;
+    }
+
+    @Override
+    public ICachedPageInternal getPage(int cpid) {
+        return pages[cpid];
+    }
+    
+    public int getNumOverflowPages() {
+        return overflowPages.size();
+    }
+	
 	@Override
-	public void openFile(int fileId) throws HyracksDataException {
-		// Do nothing		
-	}
+    public void createFile(FileReference fileRef) throws HyracksDataException {
+        // Do nothing.
+    }
 
-	@Override
-	public void closeFile(int fileId) throws HyracksDataException {
-		// Do nothing
-	}
+    @Override
+    public void openFile(int fileId) throws HyracksDataException {
+        // Do nothing.
+    }
 
-	@Override
-	public void deleteFile(int fileId) throws HyracksDataException {
-		// Do nothing
-	}
+    @Override
+    public void closeFile(int fileId) throws HyracksDataException {
+        // Do nothing.
+    }
 
-	@Override
-	public ICachedPage tryPin(long dpid) throws HyracksDataException {
-		// Just call pin!
-		return null;
-	}
-
-	@Override
-	public ICachedPage pin(long dpid, boolean newPage){
-		return cachedPages[BufferedFileHandle.getPageId(dpid)];
-	}
-
+    @Override
+    public void deleteFile(int fileId) throws HyracksDataException {
+        // Do nothing.
+    }
+	
 	@Override
 	public void unpin(ICachedPage page) throws HyracksDataException {
-		//Do Nothing
-	}
-
-	@Override
-	public int getPageSize() {
-		return pageSize;
-	}
-
-	@Override
-	public int getNumPages() {
-		return numPages;
+		// Do Nothing.
 	}
 
 	@Override
 	public void close() {
-		// Do nothing	
-	}
-
-	@Override
-	public ICachedPageInternal getPage(int cpid) {
-		return cachedPages[cpid];
+		// Do nothing.
 	}
 
     private class CachedPage implements ICachedPageInternal {
@@ -104,13 +139,13 @@
 
         @Override
         public Object getReplacementStrategyObject() {
-        	//Do nothing
+        	// Do nothing.
             return null;
         }
 
         @Override
         public boolean pinIfGoodVictim() {
-        	//Do nothing
+        	// Do nothing.
         	return false;
         }
 
@@ -124,13 +159,9 @@
             latch.readLock().lock();
         }
 
-        private void acquireWriteLatch(boolean markDirty) {
-            latch.writeLock().lock();
-        }
-
         @Override
         public void acquireWriteLatch() {
-            acquireWriteLatch(true);
+            latch.writeLock().lock();
         }
 
         @Override
diff --git a/hyracks-storage-am-lsmtree-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsmtree/common/impls/InMemoryBufferCacheFactory.java b/hyracks-storage-am-lsmtree-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsmtree/common/impls/InMemoryBufferCacheFactory.java
deleted file mode 100644
index fca8dc4..0000000
--- a/hyracks-storage-am-lsmtree-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsmtree/common/impls/InMemoryBufferCacheFactory.java
+++ /dev/null
@@ -1,26 +0,0 @@
-package edu.uci.ics.hyracks.storage.am.lsmtree.common.impls;
-
-import edu.uci.ics.hyracks.storage.common.buffercache.HeapBufferAllocator;
-import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
-import edu.uci.ics.hyracks.storage.common.buffercache.ICacheMemoryAllocator;
-
-public class InMemoryBufferCacheFactory {
-	
-	private IBufferCache bufferCache;
-    private final int pageSize;
-    private final int numPages;
-	
-    public InMemoryBufferCacheFactory(int pageSize, int numPages) {
-    	this.pageSize = pageSize;
-    	this.numPages = numPages;
-        bufferCache = null;
-    }
-    
-    public synchronized IBufferCache createInMemoryBufferCache() {
-        if (bufferCache == null) {
-            ICacheMemoryAllocator allocator = new HeapBufferAllocator();
-            bufferCache = new InMemoryBufferCache(allocator, pageSize, numPages);
-        }
-        return bufferCache;
-    }
-}
diff --git a/hyracks-storage-am-lsmtree-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsmtree/rtree/impls/LSMRTreeInMemoryBufferCache.java b/hyracks-storage-am-lsmtree-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsmtree/rtree/impls/LSMRTreeInMemoryBufferCache.java
index 85c73ae..3e8d235 100644
--- a/hyracks-storage-am-lsmtree-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsmtree/rtree/impls/LSMRTreeInMemoryBufferCache.java
+++ b/hyracks-storage-am-lsmtree-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsmtree/rtree/impls/LSMRTreeInMemoryBufferCache.java
@@ -17,9 +17,9 @@
         int fileId = BufferedFileHandle.getFileId(dpid);
 
         if (pageId == 0 || pageId == 1) {
-            return cachedPages[pageId + 2 * fileId];
+            return pages[pageId + 2 * fileId];
         } else {
-            return cachedPages[pageId];
+            return pages[pageId];
         }
     }
 
diff --git a/hyracks-tests/hyracks-storage-am-lsmtree-btree-test/pom.xml b/hyracks-tests/hyracks-storage-am-lsmtree-btree-test/pom.xml
new file mode 100644
index 0000000..428971e
--- /dev/null
+++ b/hyracks-tests/hyracks-storage-am-lsmtree-btree-test/pom.xml
@@ -0,0 +1,55 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+  <groupId>edu.uci.ics.hyracks</groupId>
+  <artifactId>hyracks-storage-am-lsmtree-btree-test</artifactId>
+  <version>0.2.0-SNAPSHOT</version>
+
+  <parent>
+    <groupId>edu.uci.ics.hyracks</groupId>
+    <artifactId>hyracks-tests</artifactId>
+    <version>0.2.0-SNAPSHOT</version>
+  </parent>
+
+  <build>
+    <plugins>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-compiler-plugin</artifactId>
+        <version>2.0.2</version>
+        <configuration>
+          <source>1.6</source>
+          <target>1.6</target>
+        </configuration>
+      </plugin>
+    </plugins>
+  </build>
+  <dependencies>
+  	<dependency>
+  		<groupId>junit</groupId>
+  		<artifactId>junit</artifactId>
+  		<version>4.8.1</version>
+  		<type>jar</type>
+  		<scope>test</scope>
+  	</dependency>
+  	<dependency>
+  		<groupId>edu.uci.ics.hyracks</groupId>
+  		<artifactId>hyracks-control-nc</artifactId>
+  		<version>0.2.0-SNAPSHOT</version>
+  		<scope>compile</scope>
+  	</dependency>
+  	<dependency>
+  		<groupId>edu.uci.ics.hyracks</groupId>
+  		<artifactId>hyracks-storage-am-lsmtree-btree</artifactId>
+  		<version>0.2.0-SNAPSHOT</version>
+  		<type>jar</type>
+  		<scope>compile</scope>
+  	</dependency>
+  	<dependency>
+  		<groupId>edu.uci.ics.hyracks</groupId>
+  		<artifactId>hyracks-test-support</artifactId>
+  		<version>0.2.0-SNAPSHOT</version>
+  		<type>jar</type>
+  		<scope>test</scope>
+  	</dependency>
+  </dependencies>
+</project>
diff --git a/hyracks-tests/hyracks-storage-am-lsmtree-btree-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsmtree/btree/AbstractLSMTreeTest.java b/hyracks-tests/hyracks-storage-am-lsmtree-btree-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsmtree/btree/AbstractLSMTreeTest.java
new file mode 100644
index 0000000..53ce238
--- /dev/null
+++ b/hyracks-tests/hyracks-storage-am-lsmtree-btree-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsmtree/btree/AbstractLSMTreeTest.java
@@ -0,0 +1,76 @@
+package edu.uci.ics.hyracks.storage.am.lsmtree.btree;
+
+import java.io.File;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.Random;
+import java.util.logging.Logger;
+
+import org.junit.After;
+import org.junit.Before;
+
+import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
+import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
+import edu.uci.ics.hyracks.api.io.FileReference;
+import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
+import edu.uci.ics.hyracks.storage.common.file.IFileMapProvider;
+import edu.uci.ics.hyracks.test.support.TestStorageManagerComponentHolder;
+import edu.uci.ics.hyracks.test.support.TestUtils;
+
+public abstract class AbstractLSMTreeTest {
+    protected static final Logger LOGGER = Logger.getLogger(AbstractLSMTreeTest.class.getName());
+    public static final long RANDOM_SEED = 50;
+    
+    private static final int PAGE_SIZE = 256;
+    private static final int NUM_PAGES = 10;
+    private static final int MAX_OPEN_FILES = 10;
+    private static final int HYRACKS_FRAME_SIZE = 128;
+        
+    protected IHyracksTaskContext ctx; 
+    protected IBufferCache bufferCache;
+    protected int btreeFileId;
+    
+    protected final Random rnd = new Random();
+    protected final static SimpleDateFormat simpleDateFormat = new SimpleDateFormat("ddMMyy-hhmmssSS");
+    protected final static String tmpDir = System.getProperty("java.io.tmpdir");
+    protected final static String sep = System.getProperty("file.separator");
+    protected String fileName;    
+    
+    @Before
+    public void setUp() throws HyracksDataException {
+        fileName = tmpDir + sep + simpleDateFormat.format(new Date());
+        ctx = TestUtils.create(getHyracksFrameSize());
+        TestStorageManagerComponentHolder.init(getPageSize(), getNumPages(), getMaxOpenFiles());
+        bufferCache = TestStorageManagerComponentHolder.getBufferCache(ctx);
+        IFileMapProvider fmp = TestStorageManagerComponentHolder.getFileMapProvider(ctx);
+        FileReference file = new FileReference(new File(fileName)); 
+        bufferCache.createFile(file);
+        btreeFileId = fmp.lookupFileId(file);
+        bufferCache.openFile(btreeFileId);
+        rnd.setSeed(RANDOM_SEED);
+    }
+    
+    @After
+    public void tearDown() throws HyracksDataException {
+        bufferCache.closeFile(btreeFileId);
+        bufferCache.close();
+        File f = new File(fileName);
+        f.deleteOnExit();
+    }
+    
+    public int getPageSize() {
+        return PAGE_SIZE;
+    }
+    
+    public int getNumPages() {
+        return NUM_PAGES;
+    }
+    
+    public int getHyracksFrameSize() {
+        return HYRACKS_FRAME_SIZE;
+    }
+    
+    public int getMaxOpenFiles() {
+        return MAX_OPEN_FILES;
+    }
+}
diff --git a/hyracks-tests/hyracks-storage-am-lsmtree-btree-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsmtree/btree/LSMTreeDeleteTest.java b/hyracks-tests/hyracks-storage-am-lsmtree-btree-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsmtree/btree/LSMTreeDeleteTest.java
new file mode 100644
index 0000000..16cab5c
--- /dev/null
+++ b/hyracks-tests/hyracks-storage-am-lsmtree-btree-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsmtree/btree/LSMTreeDeleteTest.java
@@ -0,0 +1,1102 @@
+package edu.uci.ics.hyracks.storage.am.lsmtree.btree;
+
+import static org.junit.Assert.fail;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInput;
+import java.io.DataInputStream;
+import java.io.DataOutput;
+import java.io.File;
+import java.nio.ByteBuffer;
+
+import org.junit.Test;
+
+import edu.uci.ics.hyracks.api.comm.IFrameTupleAccessor;
+import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
+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.ITypeTraits;
+import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
+import edu.uci.ics.hyracks.api.io.FileReference;
+import edu.uci.ics.hyracks.data.std.accessors.PointableBinaryComparatorFactory;
+import edu.uci.ics.hyracks.data.std.primitive.IntegerPointable;
+import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
+import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleReference;
+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.marshalling.IntegerSerializerDeserializer;
+import edu.uci.ics.hyracks.storage.am.btree.frames.BTreeNSMInteriorFrameFactory;
+import edu.uci.ics.hyracks.storage.am.btree.frames.BTreeNSMLeafFrameFactory;
+import edu.uci.ics.hyracks.storage.am.btree.impls.RangePredicate;
+import edu.uci.ics.hyracks.storage.am.common.api.IFreePageManager;
+import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexAccessor;
+import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexCursor;
+import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexFrameFactory;
+import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexMetaDataFrameFactory;
+import edu.uci.ics.hyracks.storage.am.common.api.TreeIndexException;
+import edu.uci.ics.hyracks.storage.am.common.frames.LIFOMetaDataFrameFactory;
+import edu.uci.ics.hyracks.storage.am.common.ophelpers.MultiComparator;
+import edu.uci.ics.hyracks.storage.am.lsmtree.freepage.FreePageManagerFactory;
+import edu.uci.ics.hyracks.storage.am.lsmtree.freepage.InMemoryBufferCacheFactory;
+import edu.uci.ics.hyracks.storage.am.lsmtree.freepage.InMemoryFreePageManager;
+import edu.uci.ics.hyracks.storage.am.lsmtree.impls.BTreeFactory;
+import edu.uci.ics.hyracks.storage.am.lsmtree.impls.LSMTree;
+import edu.uci.ics.hyracks.storage.am.lsmtree.impls.LSMTreeRangeSearchCursor;
+import edu.uci.ics.hyracks.storage.am.lsmtree.tuples.LSMTypeAwareTupleWriterFactory;
+import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
+import edu.uci.ics.hyracks.storage.common.file.IFileMapManager;
+import edu.uci.ics.hyracks.storage.common.file.IFileMapProvider;
+import edu.uci.ics.hyracks.test.support.TestStorageManagerComponentHolder;
+import edu.uci.ics.hyracks.test.support.TestUtils;
+
+public class LSMTreeDeleteTest extends AbstractLSMTreeTest {
+
+    private static final int PAGE_SIZE = 256;
+    private static final int NUM_PAGES = 100;
+    private static final int MAX_OPEN_FILES = 100;
+    private static final int HYRACKS_FRAME_SIZE = 128;
+    private IHyracksTaskContext ctx = TestUtils.create(HYRACKS_FRAME_SIZE);
+
+    // BASIC DELETE TEST
+    // create a fix-length lsm tree, and do 100 deletes. That is insert 100
+    // delete nodes into the in-memory tree.
+    @Test
+    public void Test1() throws Exception {
+        // in disk
+        TestStorageManagerComponentHolder.init(PAGE_SIZE, NUM_PAGES, MAX_OPEN_FILES);
+        IBufferCache bufferCache = TestStorageManagerComponentHolder.getBufferCache(ctx);
+        IFileMapProvider fmp = TestStorageManagerComponentHolder.getFileMapProvider(ctx);
+        FileReference file = new FileReference(new File(fileName));
+        bufferCache.createFile(file);
+        int fileId = fmp.lookupFileId(file);
+        bufferCache.openFile(fileId);
+
+        // in memory
+        InMemoryBufferCacheFactory InMemBufferCacheFactory = new InMemoryBufferCacheFactory(PAGE_SIZE, NUM_PAGES);
+        IBufferCache memBufferCache = InMemBufferCacheFactory.createInMemoryBufferCache();
+
+        // declare fields
+        int fieldCount = 2;
+        ITypeTraits[] typeTraits = new ITypeTraits[fieldCount];
+        typeTraits[0] = IntegerPointable.TYPE_TRAITS;
+        typeTraits[1] = IntegerPointable.TYPE_TRAITS;
+
+        // declare keys
+        int keyFieldCount = 1;
+        IBinaryComparator[] cmps = new IBinaryComparator[keyFieldCount];
+        cmps[0] = PointableBinaryComparatorFactory.of(IntegerPointable.FACTORY).createBinaryComparator();
+
+        MultiComparator cmp = new MultiComparator(cmps);
+
+        LSMTypeAwareTupleWriterFactory insertTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(typeTraits, false);
+        LSMTypeAwareTupleWriterFactory deleteTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(typeTraits, true);
+
+        ITreeIndexFrameFactory insertLeafFrameFactory = new BTreeNSMLeafFrameFactory(insertTupleWriterFactory);
+        ITreeIndexFrameFactory deleteLeafFrameFactory = new BTreeNSMLeafFrameFactory(deleteTupleWriterFactory);
+        ITreeIndexFrameFactory interiorFrameFactory = new BTreeNSMInteriorFrameFactory(insertTupleWriterFactory);
+        ITreeIndexMetaDataFrameFactory metaFrameFactory = new LIFOMetaDataFrameFactory();
+
+        IFreePageManager memFreePageManager = new InMemoryFreePageManager(100, metaFrameFactory);
+
+        FreePageManagerFactory freePageManagerFactory = new FreePageManagerFactory(bufferCache, metaFrameFactory);
+        BTreeFactory bTreeFactory = new BTreeFactory(bufferCache, freePageManagerFactory, cmp, fieldCount,
+                interiorFrameFactory, insertLeafFrameFactory);
+
+        LSMTree lsmtree = new LSMTree(memBufferCache, bufferCache, fieldCount, cmp, memFreePageManager,
+                interiorFrameFactory, insertLeafFrameFactory, deleteLeafFrameFactory, bTreeFactory,
+                (IFileMapManager) fmp);
+
+        lsmtree.create(fileId);
+        lsmtree.open(fileId);
+
+        ByteBuffer frame = ctx.allocateFrame();
+        FrameTupleAppender appender = new FrameTupleAppender(ctx.getFrameSize());
+
+        ArrayTupleBuilder tb = new ArrayTupleBuilder(fieldCount);
+        DataOutput dos = tb.getDataOutput();
+
+        ISerializerDeserializer[] recDescSers = { IntegerSerializerDeserializer.INSTANCE,
+                IntegerSerializerDeserializer.INSTANCE };
+        RecordDescriptor recDesc = new RecordDescriptor(recDescSers);
+
+        IFrameTupleAccessor accessor = new FrameTupleAccessor(ctx.getFrameSize(), recDesc);
+        accessor.reset(frame);
+
+        FrameTupleReference tuple = new FrameTupleReference();
+
+        ITreeIndexAccessor lsmTreeAccessor = lsmtree.createAccessor();
+
+        int resultSize = 50;
+        int[][] resultArray = new int[resultSize][3];
+
+        for (int i = 0; i < resultSize; i++) {
+            resultArray[i][0] = i;
+            resultArray[i][1] = i + 1;
+            resultArray[i][2] = 1;
+        }
+
+        // delete
+        for (int i = 0; i < resultSize; i++) {
+
+            int f0 = resultArray[i][0];
+            int f1 = resultArray[i][1];
+
+            tb.reset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+            tb.addFieldEndOffset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+            tb.addFieldEndOffset();
+
+            appender.reset(frame, true);
+            appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+
+            tuple.reset(accessor, 0);
+
+            ArrayTupleReference t = new ArrayTupleReference();
+            t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+
+            try {
+                lsmTreeAccessor.delete(t);
+            } catch (TreeIndexException e) {
+                System.out.println("test01:" + e);
+                e.printStackTrace();
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+
+        // scan
+        ITreeIndexCursor scanCursor = new LSMTreeRangeSearchCursor();
+        RangePredicate nullPred = new RangePredicate(true, null, null, true, true, null, null);
+        lsmTreeAccessor.search(scanCursor, nullPred);
+
+        try {
+            int scanTupleIndex = 0;
+            int arrayIndex = 0;
+            Object o = null;
+            while (scanCursor.hasNext()) {
+                scanCursor.next();
+                ITupleReference frameTuple = scanCursor.getTuple();
+                int numPrintFields = Math.min(frameTuple.getFieldCount(), recDescSers.length);
+
+                for (int i = 0; i < numPrintFields; i++) {
+                    ByteArrayInputStream inStream = new ByteArrayInputStream(frameTuple.getFieldData(i),
+                            frameTuple.getFieldStart(i), frameTuple.getFieldLength(i));
+                    DataInput dataIn = new DataInputStream(inStream);
+                    o = recDescSers[i].deserialize(dataIn);
+
+                }
+                while (resultArray[arrayIndex][2] != 0) {
+                    arrayIndex++;
+                }
+                if (Integer.parseInt(o.toString()) != resultArray[arrayIndex][1]) {
+                    fail("Input value and Output value doesn't match on the " + scanTupleIndex + " tuple\n");
+                }
+                scanTupleIndex++;
+                arrayIndex++;
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+        } finally {
+            scanCursor.close();
+        }
+
+        lsmtree.close();
+        bufferCache.closeFile(fileId);
+        memBufferCache.close();
+    }
+
+    // INSERT-DELETE TEST
+    // create a fix-length lsm tree. First, do 100 insertions,
+    // and then do 50 deletions which has the same 50 keys which are part of the
+    // insertions.
+    @Test
+    public void Test2() throws Exception {
+        // in disk
+        TestStorageManagerComponentHolder.init(PAGE_SIZE, NUM_PAGES, MAX_OPEN_FILES);
+        IBufferCache bufferCache = TestStorageManagerComponentHolder.getBufferCache(ctx);
+        IFileMapProvider fmp = TestStorageManagerComponentHolder.getFileMapProvider(ctx);
+        FileReference file = new FileReference(new File(fileName));
+        bufferCache.createFile(file);
+        int fileId = fmp.lookupFileId(file);
+        bufferCache.openFile(fileId);
+
+        // in memory
+        InMemoryBufferCacheFactory InMemBufferCacheFactory = new InMemoryBufferCacheFactory(PAGE_SIZE, NUM_PAGES);
+        IBufferCache memBufferCache = InMemBufferCacheFactory.createInMemoryBufferCache();
+
+        // declare fields
+        int fieldCount = 2;
+        ITypeTraits[] typeTraits = new ITypeTraits[fieldCount];
+        typeTraits[0] = IntegerPointable.TYPE_TRAITS;
+        typeTraits[1] = IntegerPointable.TYPE_TRAITS;
+
+        // declare keys
+        int keyFieldCount = 1;
+        IBinaryComparator[] cmps = new IBinaryComparator[keyFieldCount];
+        cmps[0] = PointableBinaryComparatorFactory.of(IntegerPointable.FACTORY).createBinaryComparator();
+
+        MultiComparator cmp = new MultiComparator(cmps);
+
+        LSMTypeAwareTupleWriterFactory insertTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(typeTraits, false);
+        LSMTypeAwareTupleWriterFactory deleteTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(typeTraits, true);
+
+        ITreeIndexFrameFactory insertLeafFrameFactory = new BTreeNSMLeafFrameFactory(insertTupleWriterFactory);
+        ITreeIndexFrameFactory deleteLeafFrameFactory = new BTreeNSMLeafFrameFactory(deleteTupleWriterFactory);
+        ITreeIndexFrameFactory interiorFrameFactory = new BTreeNSMInteriorFrameFactory(insertTupleWriterFactory);
+        ITreeIndexMetaDataFrameFactory metaFrameFactory = new LIFOMetaDataFrameFactory();
+
+        IFreePageManager memFreePageManager = new InMemoryFreePageManager(100, metaFrameFactory);
+
+        FreePageManagerFactory freePageManagerFactory = new FreePageManagerFactory(bufferCache, metaFrameFactory);
+        BTreeFactory bTreeFactory = new BTreeFactory(bufferCache, freePageManagerFactory, cmp, fieldCount,
+                interiorFrameFactory, insertLeafFrameFactory);
+
+        LSMTree lsmtree = new LSMTree(memBufferCache, bufferCache, fieldCount, cmp, memFreePageManager,
+                interiorFrameFactory, insertLeafFrameFactory, deleteLeafFrameFactory, bTreeFactory,
+                (IFileMapManager) fmp);
+
+        lsmtree.create(fileId);
+        lsmtree.open(fileId);
+
+        ByteBuffer frame = ctx.allocateFrame();
+        FrameTupleAppender appender = new FrameTupleAppender(ctx.getFrameSize());
+
+        ArrayTupleBuilder tb = new ArrayTupleBuilder(fieldCount);
+        DataOutput dos = tb.getDataOutput();
+
+        ISerializerDeserializer[] recDescSers = { IntegerSerializerDeserializer.INSTANCE,
+                IntegerSerializerDeserializer.INSTANCE };
+        RecordDescriptor recDesc = new RecordDescriptor(recDescSers);
+
+        IFrameTupleAccessor accessor = new FrameTupleAccessor(ctx.getFrameSize(), recDesc);
+        accessor.reset(frame);
+
+        FrameTupleReference tuple = new FrameTupleReference();
+
+        int resultSize = 100;
+        int deleteStartPosition = 50;
+        int[][] resultArray = new int[resultSize][3];
+
+        for (int i = 0; i < resultSize; i++) {
+            resultArray[i][0] = i;
+            resultArray[i][1] = i + 1;
+            resultArray[i][2] = 0;
+        }
+
+        // insert
+        ITreeIndexAccessor lsmTreeAccessor = lsmtree.createAccessor();
+        for (int i = 0; i < resultSize; i++) {
+
+            int f0 = resultArray[i][0];
+            int f1 = resultArray[i][1];
+
+            tb.reset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+            tb.addFieldEndOffset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+            tb.addFieldEndOffset();
+
+            appender.reset(frame, true);
+            appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+
+            tuple.reset(accessor, 0);
+
+            ArrayTupleReference t = new ArrayTupleReference();
+            t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+
+            try {
+                lsmTreeAccessor.insert(t);
+            } catch (TreeIndexException e) {
+                System.out.println("test02:" + e);
+                e.printStackTrace();
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+
+        // delete
+        for (int i = deleteStartPosition; i < resultSize; i++) {
+
+            int f0 = resultArray[i][0];
+            int f1 = ++resultArray[i][1];
+            resultArray[i][2] = 1;
+
+            tb.reset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+            tb.addFieldEndOffset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+            tb.addFieldEndOffset();
+
+            appender.reset(frame, true);
+            appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+
+            tuple.reset(accessor, 0);
+
+            ArrayTupleReference t = new ArrayTupleReference();
+            t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+
+            try {
+                lsmTreeAccessor.delete(t);
+            } catch (TreeIndexException e) {
+                System.out.println("test02:" + e);
+                e.printStackTrace();
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+
+        // scan
+        ITreeIndexCursor scanCursor = new LSMTreeRangeSearchCursor();
+        RangePredicate nullPred = new RangePredicate(true, null, null, true, true, null, null);
+        lsmTreeAccessor.search(scanCursor, nullPred);
+
+        try {
+            int scanTupleIndex = 0;
+            int arrayIndex = 0;
+            Object o = null;
+            while (scanCursor.hasNext()) {
+                scanCursor.next();
+                ITupleReference frameTuple = scanCursor.getTuple();
+                int numPrintFields = Math.min(frameTuple.getFieldCount(), recDescSers.length);
+
+                for (int i = 0; i < numPrintFields; i++) {
+                    ByteArrayInputStream inStream = new ByteArrayInputStream(frameTuple.getFieldData(i),
+                            frameTuple.getFieldStart(i), frameTuple.getFieldLength(i));
+                    DataInput dataIn = new DataInputStream(inStream);
+                    o = recDescSers[i].deserialize(dataIn);
+
+                }
+                while (resultArray[arrayIndex][2] != 0) {
+                    arrayIndex++;
+                }
+                if (Integer.parseInt(o.toString()) != resultArray[arrayIndex][1]) {
+                    fail("Input value and Output value doesn't match on the " + scanTupleIndex + " tuple\n");
+                }
+
+                scanTupleIndex++;
+                arrayIndex++;
+
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+        } finally {
+            scanCursor.close();
+        }
+
+        lsmtree.close();
+        bufferCache.closeFile(fileId);
+        memBufferCache.close();
+    }
+
+    // DELETE->INSERT TEST
+    // create a fix-length lsm tree. First, do 100 deletions,
+    // and then do 50 insertions which has the same 50 keys which are part of
+    // the deletions.
+    @Test
+    public void Test3() throws Exception {
+        System.out.println("TEST3");
+        // in disk
+        TestStorageManagerComponentHolder.init(PAGE_SIZE, NUM_PAGES, MAX_OPEN_FILES);
+        IBufferCache bufferCache = TestStorageManagerComponentHolder.getBufferCache(ctx);
+        IFileMapProvider fmp = TestStorageManagerComponentHolder.getFileMapProvider(ctx);
+        FileReference file = new FileReference(new File(fileName));
+        bufferCache.createFile(file);
+        int fileId = fmp.lookupFileId(file);
+        bufferCache.openFile(fileId);
+
+        // in mem
+        InMemoryBufferCacheFactory InMemBufferCacheFactory = new InMemoryBufferCacheFactory(PAGE_SIZE, NUM_PAGES);
+        IBufferCache memBufferCache = InMemBufferCacheFactory.createInMemoryBufferCache();
+
+        // declare fields
+        int fieldCount = 2;
+        ITypeTraits[] typeTraits = new ITypeTraits[fieldCount];
+        typeTraits[0] = IntegerPointable.TYPE_TRAITS;
+        typeTraits[1] = IntegerPointable.TYPE_TRAITS;
+
+        // declare keys
+        int keyFieldCount = 1;
+        IBinaryComparator[] cmps = new IBinaryComparator[keyFieldCount];
+        cmps[0] = PointableBinaryComparatorFactory.of(IntegerPointable.FACTORY).createBinaryComparator();
+
+        MultiComparator cmp = new MultiComparator(cmps);
+
+        LSMTypeAwareTupleWriterFactory insertTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(typeTraits, false);
+        LSMTypeAwareTupleWriterFactory deleteTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(typeTraits, true);
+
+        ITreeIndexFrameFactory insertLeafFrameFactory = new BTreeNSMLeafFrameFactory(insertTupleWriterFactory);
+        ITreeIndexFrameFactory deleteLeafFrameFactory = new BTreeNSMLeafFrameFactory(deleteTupleWriterFactory);
+        // change
+        ITreeIndexFrameFactory interiorFrameFactory = new BTreeNSMInteriorFrameFactory(insertTupleWriterFactory);
+        ITreeIndexMetaDataFrameFactory metaFrameFactory = new LIFOMetaDataFrameFactory();
+
+        IFreePageManager memFreePageManager = new InMemoryFreePageManager(100, metaFrameFactory);
+
+        FreePageManagerFactory freePageManagerFactory = new FreePageManagerFactory(bufferCache, metaFrameFactory);
+        BTreeFactory bTreeFactory = new BTreeFactory(bufferCache, freePageManagerFactory, cmp, fieldCount,
+                interiorFrameFactory, insertLeafFrameFactory);
+
+        LSMTree lsmtree = new LSMTree(memBufferCache, bufferCache, fieldCount, cmp, memFreePageManager,
+                interiorFrameFactory, insertLeafFrameFactory, deleteLeafFrameFactory, bTreeFactory,
+                (IFileMapManager) fmp);
+
+        lsmtree.create(fileId);
+        lsmtree.open(fileId);
+
+        ByteBuffer frame = ctx.allocateFrame();
+        FrameTupleAppender appender = new FrameTupleAppender(ctx.getFrameSize());
+
+        ArrayTupleBuilder tb = new ArrayTupleBuilder(fieldCount);
+        DataOutput dos = tb.getDataOutput();
+
+        ISerializerDeserializer[] recDescSers = { IntegerSerializerDeserializer.INSTANCE,
+                IntegerSerializerDeserializer.INSTANCE };
+        RecordDescriptor recDesc = new RecordDescriptor(recDescSers);
+
+        IFrameTupleAccessor accessor = new FrameTupleAccessor(ctx.getFrameSize(), recDesc);
+        accessor.reset(frame);
+
+        FrameTupleReference tuple = new FrameTupleReference();
+
+        int resultSize = 100;
+        int insertStartPosition = 50;
+        int[][] resultArray = new int[resultSize][3];
+
+        for (int i = 0; i < resultSize; i++) {
+            resultArray[i][0] = i;
+            resultArray[i][1] = i + 1;
+            resultArray[i][2] = 1;
+        }
+
+        // delete
+        ITreeIndexAccessor lsmTreeAccessor = lsmtree.createAccessor();
+        for (int i = 0; i < resultSize; i++) {
+
+            int f0 = resultArray[i][0];
+            int f1 = resultArray[i][1];
+
+            tb.reset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+            tb.addFieldEndOffset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+            tb.addFieldEndOffset();
+
+            appender.reset(frame, true);
+            appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+
+            tuple.reset(accessor, 0);
+
+            ArrayTupleReference t = new ArrayTupleReference();
+            t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+
+            try {
+                lsmTreeAccessor.delete(t);
+            } catch (TreeIndexException e) {
+                System.out.println("test03:" + e);
+                e.printStackTrace();
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+
+        // insert
+        for (int i = insertStartPosition; i < resultSize; i++) {
+
+            int f0 = resultArray[i][0];
+            int f1 = ++resultArray[i][1];
+            resultArray[i][2] = 0;
+
+            tb.reset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+            tb.addFieldEndOffset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+            tb.addFieldEndOffset();
+
+            appender.reset(frame, true);
+            appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+
+            tuple.reset(accessor, 0);
+
+            ArrayTupleReference t = new ArrayTupleReference();
+            t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+
+            try {
+                lsmTreeAccessor.insert(t);
+            } catch (TreeIndexException e) {
+                System.out.println("test03:" + e);
+                e.printStackTrace();
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+
+        // scan
+        ITreeIndexCursor scanCursor = new LSMTreeRangeSearchCursor();
+        RangePredicate nullPred = new RangePredicate(true, null, null, true, true, null, null);
+        lsmTreeAccessor.search(scanCursor, nullPred);
+
+        try {
+            int scanTupleIndex = 0;
+            int arrayIndex = 0;
+            Object o = null;
+            while (scanCursor.hasNext()) {
+                scanCursor.next();
+                ITupleReference frameTuple = scanCursor.getTuple();
+                int numPrintFields = Math.min(frameTuple.getFieldCount(), recDescSers.length);
+
+                for (int i = 0; i < numPrintFields; i++) {
+                    ByteArrayInputStream inStream = new ByteArrayInputStream(frameTuple.getFieldData(i),
+                            frameTuple.getFieldStart(i), frameTuple.getFieldLength(i));
+                    DataInput dataIn = new DataInputStream(inStream);
+                    o = recDescSers[i].deserialize(dataIn);
+                }
+                while (resultArray[arrayIndex][2] != 0) {
+                    arrayIndex++;
+                }
+                if (Integer.parseInt(o.toString()) != resultArray[arrayIndex][1]) {
+                    fail("Input value and Output value doesn't match on the " + scanTupleIndex + " tuple\n");
+                }
+
+                scanTupleIndex++;
+                arrayIndex++;
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+        } finally {
+            scanCursor.close();
+        }
+
+        lsmtree.close();
+        bufferCache.closeFile(fileId);
+        memBufferCache.close();
+    }
+
+    // TEST DELETION and PageAllocationException
+    // create a fix-length lsm tree. First, do 811 deletions,
+    // the page will be run out on the 810th deletions, if there is any
+    // exception returns, the test case fails.
+    @Test
+    public void Test4() throws Exception {
+        // in disk
+        TestStorageManagerComponentHolder.init(PAGE_SIZE, NUM_PAGES, MAX_OPEN_FILES);
+        IBufferCache bufferCache = TestStorageManagerComponentHolder.getBufferCache(ctx);
+        IFileMapProvider fmp = TestStorageManagerComponentHolder.getFileMapProvider(ctx);
+        FileReference file = new FileReference(new File(fileName));
+        bufferCache.createFile(file);
+        int fileId = fmp.lookupFileId(file);
+        bufferCache.openFile(fileId);
+
+        // in memory
+        InMemoryBufferCacheFactory InMemBufferCacheFactory = new InMemoryBufferCacheFactory(PAGE_SIZE, NUM_PAGES);
+        IBufferCache memBufferCache = InMemBufferCacheFactory.createInMemoryBufferCache();
+
+        // declare fields
+        int fieldCount = 2;
+        ITypeTraits[] typeTraits = new ITypeTraits[fieldCount];
+        typeTraits[0] = IntegerPointable.TYPE_TRAITS;
+        typeTraits[1] = IntegerPointable.TYPE_TRAITS;
+
+        // declare keys
+        int keyFieldCount = 1;
+        IBinaryComparator[] cmps = new IBinaryComparator[keyFieldCount];
+        cmps[0] = PointableBinaryComparatorFactory.of(IntegerPointable.FACTORY).createBinaryComparator();
+
+        MultiComparator cmp = new MultiComparator(cmps);
+
+        LSMTypeAwareTupleWriterFactory insertTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(typeTraits, false);
+        LSMTypeAwareTupleWriterFactory deleteTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(typeTraits, true);
+
+        ITreeIndexFrameFactory insertLeafFrameFactory = new BTreeNSMLeafFrameFactory(insertTupleWriterFactory);
+        ITreeIndexFrameFactory deleteLeafFrameFactory = new BTreeNSMLeafFrameFactory(deleteTupleWriterFactory);
+        ITreeIndexFrameFactory interiorFrameFactory = new BTreeNSMInteriorFrameFactory(insertTupleWriterFactory);
+        ITreeIndexMetaDataFrameFactory metaFrameFactory = new LIFOMetaDataFrameFactory();
+
+        IFreePageManager memFreePageManager = new InMemoryFreePageManager(100, metaFrameFactory);
+
+        FreePageManagerFactory freePageManagerFactory = new FreePageManagerFactory(bufferCache, metaFrameFactory);
+        BTreeFactory bTreeFactory = new BTreeFactory(bufferCache, freePageManagerFactory, cmp, fieldCount,
+                interiorFrameFactory, insertLeafFrameFactory);
+
+        // For the Flush Mechanism
+        LSMTree lsmtree = new LSMTree(memBufferCache, bufferCache, fieldCount, cmp, memFreePageManager,
+                interiorFrameFactory, insertLeafFrameFactory, deleteLeafFrameFactory, bTreeFactory,
+                (IFileMapManager) fmp);
+
+        lsmtree.create(fileId);
+        lsmtree.open(fileId);
+
+        ByteBuffer frame = ctx.allocateFrame();
+        FrameTupleAppender appender = new FrameTupleAppender(ctx.getFrameSize());
+
+        ArrayTupleBuilder tb = new ArrayTupleBuilder(fieldCount);
+        DataOutput dos = tb.getDataOutput();
+
+        ISerializerDeserializer[] recDescSers = { IntegerSerializerDeserializer.INSTANCE,
+                IntegerSerializerDeserializer.INSTANCE };
+        RecordDescriptor recDesc = new RecordDescriptor(recDescSers);
+
+        IFrameTupleAccessor accessor = new FrameTupleAccessor(ctx.getFrameSize(), recDesc);
+        accessor.reset(frame);
+
+        FrameTupleReference tuple = new FrameTupleReference();
+
+        int resultSize = 811;
+        int[][] resultArray = new int[resultSize][2];
+
+        for (int i = 0; i < resultSize; i++) {
+            resultArray[i][0] = i;
+            resultArray[i][1] = i + 1;
+        }
+
+        // delete
+        ITreeIndexAccessor lsmTreeAccessor = lsmtree.createAccessor();
+        for (int i = 0; i < resultSize; i++) {
+
+            int f0 = resultArray[i][0];
+            int f1 = resultArray[i][1];
+
+            tb.reset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+            tb.addFieldEndOffset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+            tb.addFieldEndOffset();
+
+            appender.reset(frame, true);
+            appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+
+            tuple.reset(accessor, 0);
+
+            ArrayTupleReference t = new ArrayTupleReference();
+            t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+
+            try {
+                lsmTreeAccessor.delete(t);
+            } catch (TreeIndexException e) {
+                System.out.println("test04:" + e);
+                e.printStackTrace();
+                fail("test04: Catch TreeIndexException" + e);
+            } catch (Exception e) {
+                e.printStackTrace();
+                fail("test04: Catch Other Exceptions" + e);
+            }
+        }
+    }
+
+    // DELETE -> DELETE
+    // create a fix-length lsm tree. First, do 100 deletions,
+    // and then do 50 deletions which has the same 50 keys which are part of the
+    // first deletions.
+    @Test
+    public void Test5() throws Exception {
+        // in disk
+        TestStorageManagerComponentHolder.init(PAGE_SIZE, NUM_PAGES, MAX_OPEN_FILES);
+        IBufferCache bufferCache = TestStorageManagerComponentHolder.getBufferCache(ctx);
+        IFileMapProvider fmp = TestStorageManagerComponentHolder.getFileMapProvider(ctx);
+        FileReference file = new FileReference(new File(fileName));
+        bufferCache.createFile(file);
+        int fileId = fmp.lookupFileId(file);
+        bufferCache.openFile(fileId);
+
+        // in memory
+        InMemoryBufferCacheFactory InMemBufferCacheFactory = new InMemoryBufferCacheFactory(PAGE_SIZE, NUM_PAGES);
+        IBufferCache memBufferCache = InMemBufferCacheFactory.createInMemoryBufferCache();
+
+        // declare fields
+        int fieldCount = 2;
+        ITypeTraits[] typeTraits = new ITypeTraits[fieldCount];
+        typeTraits[0] = IntegerPointable.TYPE_TRAITS;
+        typeTraits[1] = IntegerPointable.TYPE_TRAITS;
+
+        // declare keys
+        int keyFieldCount = 1;
+        IBinaryComparator[] cmps = new IBinaryComparator[keyFieldCount];
+        cmps[0] = PointableBinaryComparatorFactory.of(IntegerPointable.FACTORY).createBinaryComparator();
+
+        MultiComparator cmp = new MultiComparator(cmps);
+
+        LSMTypeAwareTupleWriterFactory insertTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(typeTraits, false);
+        LSMTypeAwareTupleWriterFactory deleteTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(typeTraits, true);
+
+        ITreeIndexFrameFactory insertLeafFrameFactory = new BTreeNSMLeafFrameFactory(insertTupleWriterFactory);
+        ITreeIndexFrameFactory deleteLeafFrameFactory = new BTreeNSMLeafFrameFactory(deleteTupleWriterFactory);
+        ITreeIndexFrameFactory interiorFrameFactory = new BTreeNSMInteriorFrameFactory(insertTupleWriterFactory);
+        ITreeIndexMetaDataFrameFactory metaFrameFactory = new LIFOMetaDataFrameFactory();
+
+        IFreePageManager memFreePageManager = new InMemoryFreePageManager(100, metaFrameFactory);
+
+        FreePageManagerFactory freePageManagerFactory = new FreePageManagerFactory(bufferCache, metaFrameFactory);
+        BTreeFactory bTreeFactory = new BTreeFactory(bufferCache, freePageManagerFactory, cmp, fieldCount,
+                interiorFrameFactory, insertLeafFrameFactory);
+
+        LSMTree lsmtree = new LSMTree(memBufferCache, bufferCache, fieldCount, cmp, memFreePageManager,
+                interiorFrameFactory, insertLeafFrameFactory, deleteLeafFrameFactory, bTreeFactory,
+                (IFileMapManager) fmp);
+
+        lsmtree.create(fileId);
+        lsmtree.open(fileId);
+
+        ByteBuffer frame = ctx.allocateFrame();
+        FrameTupleAppender appender = new FrameTupleAppender(ctx.getFrameSize());
+
+        ArrayTupleBuilder tb = new ArrayTupleBuilder(fieldCount);
+        DataOutput dos = tb.getDataOutput();
+
+        ISerializerDeserializer[] recDescSers = { IntegerSerializerDeserializer.INSTANCE,
+                IntegerSerializerDeserializer.INSTANCE };
+        RecordDescriptor recDesc = new RecordDescriptor(recDescSers);
+
+        IFrameTupleAccessor accessor = new FrameTupleAccessor(ctx.getFrameSize(), recDesc);
+        accessor.reset(frame);
+
+        FrameTupleReference tuple = new FrameTupleReference();
+
+        int resultSize = 100;
+        int insertStartPosition = 50;
+        int[][] resultArray = new int[resultSize][3];
+
+        for (int i = 0; i < resultSize; i++) {
+            resultArray[i][0] = i;
+            resultArray[i][1] = i + 1;
+            resultArray[i][2] = 1;
+        }
+
+        // First deletion part
+        ITreeIndexAccessor lsmTreeAccessor = lsmtree.createAccessor();
+        for (int i = 0; i < resultSize; i++) {
+
+            int f0 = resultArray[i][0];
+            int f1 = resultArray[i][1];
+
+            tb.reset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+            tb.addFieldEndOffset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+            tb.addFieldEndOffset();
+
+            appender.reset(frame, true);
+            appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+
+            tuple.reset(accessor, 0);
+
+            ArrayTupleReference t = new ArrayTupleReference();
+            t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+
+            try {
+                lsmTreeAccessor.delete(t);
+            } catch (TreeIndexException e) {
+                System.out.println("test05:" + e);
+                e.printStackTrace();
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+
+        // Second delete part
+        for (int i = insertStartPosition; i < resultSize; i++) {
+
+            int f0 = resultArray[i][0];
+            int f1 = ++resultArray[i][1];
+            resultArray[i][2] = 1;
+
+            tb.reset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+            tb.addFieldEndOffset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+            tb.addFieldEndOffset();
+
+            appender.reset(frame, true);
+            appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+
+            tuple.reset(accessor, 0);
+
+            ArrayTupleReference t = new ArrayTupleReference();
+            t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+
+            try {
+                lsmTreeAccessor.insert(t);
+            } catch (TreeIndexException e) {
+                System.out.println("test05:" + e);
+                e.printStackTrace();
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+
+        // scan
+        ITreeIndexCursor scanCursor = new LSMTreeRangeSearchCursor();
+        RangePredicate nullPred = new RangePredicate(true, null, null, true, true, null, null);
+        lsmTreeAccessor.search(scanCursor, nullPred);
+
+        try {
+            int scanTupleIndex = 0;
+            int arrayIndex = 0;
+            Object o = null;
+            while (scanCursor.hasNext()) {
+                scanCursor.next();
+                ITupleReference frameTuple = scanCursor.getTuple();
+                int numPrintFields = Math.min(frameTuple.getFieldCount(), recDescSers.length);
+
+                for (int i = 0; i < numPrintFields; i++) {
+                    ByteArrayInputStream inStream = new ByteArrayInputStream(frameTuple.getFieldData(i),
+                            frameTuple.getFieldStart(i), frameTuple.getFieldLength(i));
+                    DataInput dataIn = new DataInputStream(inStream);
+                    o = recDescSers[i].deserialize(dataIn);
+
+                }
+                while (resultArray[arrayIndex][2] != 0) {
+                    arrayIndex++;
+                }
+                if (Integer.parseInt(o.toString()) != resultArray[arrayIndex][1]) {
+                    fail("Input value and Output value doesn't match on the " + scanTupleIndex + " tuple\n");
+                }
+
+                scanTupleIndex++;
+                arrayIndex++;
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+        } finally {
+            scanCursor.close();
+        }
+
+        lsmtree.close();
+        bufferCache.closeFile(fileId);
+        memBufferCache.close();
+    }
+
+    // INSERT -> DELETE -> INSERT
+    // create a fix-length lsm tree. Do the insertion, deletion and insertion.
+    // the final result will be
+    // | 0~9 | 10~19 | 20~39 | 40~59 | 60~79 | 80~99 |
+    // | f1=10 | f1=9 | f1=8 | f1=7 | f1=6 | f1=5 |
+    // | Insert| Insert| Delete| Delete| Insert| Insert|
+    @Test
+    public void Test6() throws Exception {
+
+        // in disk
+        TestStorageManagerComponentHolder.init(PAGE_SIZE, NUM_PAGES, MAX_OPEN_FILES);
+        IBufferCache bufferCache = TestStorageManagerComponentHolder.getBufferCache(ctx);
+        IFileMapProvider fmp = TestStorageManagerComponentHolder.getFileMapProvider(ctx);
+        FileReference file = new FileReference(new File(fileName));
+        bufferCache.createFile(file);
+        int fileId = fmp.lookupFileId(file);
+        bufferCache.openFile(fileId);
+
+        // in mem
+        InMemoryBufferCacheFactory InMemBufferCacheFactory = new InMemoryBufferCacheFactory(PAGE_SIZE, NUM_PAGES);
+        IBufferCache memBufferCache = InMemBufferCacheFactory.createInMemoryBufferCache();
+
+        // declare fields
+        int fieldCount = 2;
+        ITypeTraits[] typeTraits = new ITypeTraits[fieldCount];
+        typeTraits[0] = IntegerPointable.TYPE_TRAITS;
+        typeTraits[1] = IntegerPointable.TYPE_TRAITS;
+
+        // declare keys
+        int keyFieldCount = 1;
+        IBinaryComparator[] cmps = new IBinaryComparator[keyFieldCount];
+        cmps[0] = PointableBinaryComparatorFactory.of(IntegerPointable.FACTORY).createBinaryComparator();
+
+        MultiComparator cmp = new MultiComparator(cmps);
+
+        LSMTypeAwareTupleWriterFactory insertTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(typeTraits, false);
+        LSMTypeAwareTupleWriterFactory deleteTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(typeTraits, true);
+
+        ITreeIndexFrameFactory insertLeafFrameFactory = new BTreeNSMLeafFrameFactory(insertTupleWriterFactory);
+        ITreeIndexFrameFactory deleteLeafFrameFactory = new BTreeNSMLeafFrameFactory(deleteTupleWriterFactory);
+        ITreeIndexFrameFactory interiorFrameFactory = new BTreeNSMInteriorFrameFactory(insertTupleWriterFactory);
+        ITreeIndexMetaDataFrameFactory metaFrameFactory = new LIFOMetaDataFrameFactory();
+
+        IFreePageManager memFreePageManager = new InMemoryFreePageManager(100, metaFrameFactory);
+
+        FreePageManagerFactory freePageManagerFactory = new FreePageManagerFactory(bufferCache, metaFrameFactory);
+        BTreeFactory bTreeFactory = new BTreeFactory(bufferCache, freePageManagerFactory, cmp, fieldCount,
+                interiorFrameFactory, insertLeafFrameFactory);
+
+        LSMTree lsmtree = new LSMTree(memBufferCache, bufferCache, fieldCount, cmp, memFreePageManager,
+                interiorFrameFactory, insertLeafFrameFactory, deleteLeafFrameFactory, bTreeFactory,
+                (IFileMapManager) fmp);
+
+        lsmtree.create(fileId);
+        lsmtree.open(fileId);
+
+        ByteBuffer frame = ctx.allocateFrame();
+        FrameTupleAppender appender = new FrameTupleAppender(ctx.getFrameSize());
+
+        ArrayTupleBuilder tb = new ArrayTupleBuilder(fieldCount);
+        DataOutput dos = tb.getDataOutput();
+
+        ISerializerDeserializer[] recDescSers = { IntegerSerializerDeserializer.INSTANCE,
+                IntegerSerializerDeserializer.INSTANCE };
+        RecordDescriptor recDesc = new RecordDescriptor(recDescSers);
+
+        IFrameTupleAccessor accessor = new FrameTupleAccessor(ctx.getFrameSize(), recDesc);
+        accessor.reset(frame);
+
+        FrameTupleReference tuple = new FrameTupleReference();
+
+        ITreeIndexAccessor lsmTreeAccessor = lsmtree.createAccessor();
+
+        int resultSize = 180;
+        int[][] resultArray = new int[resultSize][3];
+
+        // insert
+        for (int i = 0; i < 180; i++) {
+            int f0 = i % 100;
+            int f1;
+            if (i >= 100) {
+                f1 = 6;
+            } else {
+                f1 = 5;
+            }
+
+            resultArray[f0][0] = f0;
+            resultArray[f0][1] = f1;
+            resultArray[f0][2] = 0;
+
+            tb.reset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+            tb.addFieldEndOffset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+            tb.addFieldEndOffset();
+
+            appender.reset(frame, true);
+            appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+
+            tuple.reset(accessor, 0);
+
+            ArrayTupleReference t = new ArrayTupleReference();
+            t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+
+            try {
+                lsmTreeAccessor.insert(t);
+            } catch (TreeIndexException e) {
+                System.out.println("test06:" + e);
+                e.printStackTrace();
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+
+        // delete
+        for (int i = 0; i < 100; i++) {
+            int f0 = i % 60;
+            int f1;
+            if (i >= 60) {
+                f1 = 8;
+            } else {
+                f1 = 7;
+            }
+
+            resultArray[f0][0] = f0;
+            resultArray[f0][1] = f1;
+            resultArray[f0][2] = 1;
+
+            tb.reset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+            tb.addFieldEndOffset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+            tb.addFieldEndOffset();
+
+            appender.reset(frame, true);
+            appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+
+            tuple.reset(accessor, 0);
+
+            ArrayTupleReference t = new ArrayTupleReference();
+            t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+
+            try {
+                lsmTreeAccessor.delete(t);
+            } catch (TreeIndexException e) {
+                System.out.println("test06:" + e);
+                e.printStackTrace();
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+
+        }
+
+        // reinsert
+        for (int i = 0; i < 30; i++) {
+            int f0 = i % 20;
+            int f1;
+            if (i >= 20) {
+                f1 = 10;
+            } else {
+                f1 = 9;
+            }
+
+            resultArray[f0][0] = f0;
+            resultArray[f0][1] = f1;
+            resultArray[f0][2] = 0;
+
+            tb.reset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+            tb.addFieldEndOffset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+            tb.addFieldEndOffset();
+
+            appender.reset(frame, true);
+            appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+
+            tuple.reset(accessor, 0);
+
+            ArrayTupleReference t = new ArrayTupleReference();
+            t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+
+            try {
+                lsmTreeAccessor.insert(t);
+            } catch (TreeIndexException e) {
+                System.out.println("test06:" + e);
+                e.printStackTrace();
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+
+        // scan
+        ITreeIndexCursor scanCursor = new LSMTreeRangeSearchCursor();
+        RangePredicate nullPred = new RangePredicate(true, null, null, true, true, null, null);
+        lsmTreeAccessor.search(scanCursor, nullPred);
+
+        try {
+            int scanTupleIndex = 0;
+            int arrayIndex = 0;
+            Object o = null;
+            while (scanCursor.hasNext()) {
+                scanCursor.next();
+                ITupleReference frameTuple = scanCursor.getTuple();
+                int numPrintFields = Math.min(frameTuple.getFieldCount(), recDescSers.length);
+
+                for (int i = 0; i < numPrintFields; i++) {
+                    ByteArrayInputStream inStream = new ByteArrayInputStream(frameTuple.getFieldData(i),
+                            frameTuple.getFieldStart(i), frameTuple.getFieldLength(i));
+                    DataInput dataIn = new DataInputStream(inStream);
+                    o = recDescSers[i].deserialize(dataIn);
+                }
+                while (resultArray[arrayIndex][2] != 0) {
+                    arrayIndex++;
+                }
+                if (Integer.parseInt(o.toString()) != resultArray[arrayIndex][1]) {
+                    fail("Input value and Output value doesn't match on the " + scanTupleIndex + " tuple\n");
+                }
+
+                scanTupleIndex++;
+                arrayIndex++;
+            }
+
+        } catch (Exception e) {
+            e.printStackTrace();
+        } finally {
+            scanCursor.close();
+        }
+
+        lsmtree.close();
+        bufferCache.closeFile(fileId);
+        memBufferCache.close();
+    }
+}
diff --git a/hyracks-tests/hyracks-storage-am-lsmtree-btree-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsmtree/btree/LSMTreeFlushTest.java b/hyracks-tests/hyracks-storage-am-lsmtree-btree-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsmtree/btree/LSMTreeFlushTest.java
new file mode 100644
index 0000000..079fc23
--- /dev/null
+++ b/hyracks-tests/hyracks-storage-am-lsmtree-btree-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsmtree/btree/LSMTreeFlushTest.java
@@ -0,0 +1,755 @@
+package edu.uci.ics.hyracks.storage.am.lsmtree.btree;
+
+import java.io.DataOutput;
+import java.io.File;
+import java.nio.ByteBuffer;
+
+import org.junit.Test;
+
+import edu.uci.ics.hyracks.api.comm.IFrameTupleAccessor;
+import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
+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.ITypeTraits;
+import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
+import edu.uci.ics.hyracks.api.io.FileReference;
+import edu.uci.ics.hyracks.data.std.accessors.PointableBinaryComparatorFactory;
+import edu.uci.ics.hyracks.data.std.primitive.IntegerPointable;
+import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
+import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleReference;
+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.marshalling.IntegerSerializerDeserializer;
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeLeafFrame;
+import edu.uci.ics.hyracks.storage.am.btree.frames.BTreeNSMInteriorFrameFactory;
+import edu.uci.ics.hyracks.storage.am.btree.frames.BTreeNSMLeafFrameFactory;
+import edu.uci.ics.hyracks.storage.am.common.api.IFreePageManager;
+import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexAccessor;
+import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexFrameFactory;
+import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexMetaDataFrameFactory;
+import edu.uci.ics.hyracks.storage.am.common.api.TreeIndexException;
+import edu.uci.ics.hyracks.storage.am.common.frames.LIFOMetaDataFrameFactory;
+import edu.uci.ics.hyracks.storage.am.common.freepage.LinkedListFreePageManager;
+import edu.uci.ics.hyracks.storage.am.common.ophelpers.MultiComparator;
+import edu.uci.ics.hyracks.storage.am.lsmtree.freepage.FreePageManagerFactory;
+import edu.uci.ics.hyracks.storage.am.lsmtree.freepage.InMemoryBufferCacheFactory;
+import edu.uci.ics.hyracks.storage.am.lsmtree.freepage.InMemoryFreePageManager;
+import edu.uci.ics.hyracks.storage.am.lsmtree.impls.BTreeFactory;
+import edu.uci.ics.hyracks.storage.am.lsmtree.impls.LSMTree;
+import edu.uci.ics.hyracks.storage.am.lsmtree.tuples.LSMEntireTupleWriterFactory;
+import edu.uci.ics.hyracks.storage.am.lsmtree.tuples.LSMTypeAwareTupleWriterFactory;
+import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
+import edu.uci.ics.hyracks.storage.common.file.IFileMapManager;
+import edu.uci.ics.hyracks.storage.common.file.IFileMapProvider;
+import edu.uci.ics.hyracks.test.support.TestStorageManagerComponentHolder;
+import edu.uci.ics.hyracks.test.support.TestUtils;
+
+public class LSMTreeFlushTest extends AbstractLSMTreeTest {
+    private static final int PAGE_SIZE = 256;
+    private static final int NUM_PAGES = 100;
+    private static final int MAX_OPEN_FILES = 10000;
+    private static final int HYRACKS_FRAME_SIZE = 128;
+    private IHyracksTaskContext ctx = TestUtils.create(HYRACKS_FRAME_SIZE);
+
+    // BASIC TEST
+    // @Test
+    // public void Test1() throws Exception {
+    // System.out.printf("TEST1 START\n");
+    // //in disk
+    // TestStorageManagerComponentHolder.init(PAGE_SIZE, NUM_PAGES,
+    // MAX_OPEN_FILES);
+    // IBufferCache bufferCache =
+    // TestStorageManagerComponentHolder.getBufferCache(ctx);
+    // IFileMapProvider fmp =
+    // TestStorageManagerComponentHolder.getFileMapProvider(ctx);
+    // FileReference file = new FileReference(new File(fileName));
+    // bufferCache.createFile(file);
+    // int fileId = fmp.lookupFileId(file);
+    // bufferCache.openFile(fileId);
+    //
+    // //in memory
+    // InMemoryBufferCacheFactory InMemBufferCacheFactory = new
+    // InMemoryBufferCacheFactory(PAGE_SIZE, NUM_PAGES);
+    // IBufferCache memBufferCache =
+    // InMemBufferCacheFactory.createInMemoryBufferCache();
+    //
+    // // declare fields
+    // int fieldCount = 2;
+    // ITypeTrait[] typeTraits = new ITypeTrait[fieldCount];
+    // typeTraits[0] = new TypeTrait(4);
+    // typeTraits[1] = new TypeTrait(4);
+    //
+    // // declare keys
+    // int keyFieldCount = 1;
+    // IBinaryComparatorFactory[] cmpFactories = new
+    // IBinaryComparatorFactory[keyFieldCount];
+    // cmpFactories[0] = IntegerBinaryComparatorFactory.INSTANCE;
+    //
+    // MultiComparator cmp = BTreeUtils.createMultiComparator(cmpFactories);
+    //
+    // LSMTypeAwareTupleWriterFactory insertTupleWriterFactory = new
+    // LSMTypeAwareTupleWriterFactory(typeTraits, false);
+    // LSMTypeAwareTupleWriterFactory deleteTupleWriterFactory = new
+    // LSMTypeAwareTupleWriterFactory(typeTraits, true);
+    //
+    // ITreeIndexFrameFactory insertLeafFrameFactory = new
+    // BTreeNSMLeafFrameFactory(insertTupleWriterFactory);
+    // ITreeIndexFrameFactory deleteLeafFrameFactory = new
+    // BTreeNSMLeafFrameFactory(deleteTupleWriterFactory);
+    // ITreeIndexFrameFactory interiorFrameFactory = new
+    // BTreeNSMInteriorFrameFactory(insertTupleWriterFactory);
+    // ITreeIndexMetaDataFrameFactory metaFrameFactory = new
+    // LIFOMetaDataFrameFactory();
+    //
+    // IBTreeLeafFrame insertLeafFrame = (IBTreeLeafFrame)
+    // insertLeafFrameFactory.createFrame();
+    //
+    // IFreePageManager freePageManager = new
+    // LinkedListFreePageManager(bufferCache, fileId, 0, metaFrameFactory);
+    // IFreePageManager memFreePageManager = new InMemoryFreePageManager(100,
+    // metaFrameFactory);
+    //
+    // // For the Flush Mechanism
+    // LSMEntireTupleWriterFactory flushTupleWriterFactory = new
+    // LSMEntireTupleWriterFactory(typeTraits);
+    // ITreeIndexFrameFactory flushLeafFrameFactory = new
+    // BTreeNSMLeafFrameFactory(flushTupleWriterFactory);
+    // FreePageManagerFactory freePageManagerFactory = new
+    // FreePageManagerFactory(bufferCache, metaFrameFactory);
+    // BTreeFactory bTreeFactory = new BTreeFactory(bufferCache,
+    // freePageManagerFactory, cmp, fieldCount, interiorFrameFactory,
+    // flushLeafFrameFactory);
+    //
+    //
+    //
+    // // LSMTree lsmtree = new LSMTree(3, 100, 2, memBufferCache, bufferCache,
+    // fieldCount, cmp, memFreePageManager,
+    // // freePageManager, interiorFrameFactory, insertLeafFrameFactory,
+    // deleteLeafFrameFactory, bTreeFactory, flushLeafFrameFactory,
+    // (IFileMapManager)fmp);
+    // //
+    // LSMTree lsmtree = LSMTreeUtils.createLSMTree(memBufferCache, bufferCache,
+    // fileId, typeTraits, cmp.getComparators(), BTreeLeafFrameType.REGULAR_NSM,
+    // (IFileMapManager)fmp);
+    // lsmtree.create(fileId);
+    // lsmtree.open(fileId);
+    //
+    // ByteBuffer frame = ctx.allocateFrame();
+    // FrameTupleAppender appender = new FrameTupleAppender(ctx.getFrameSize());
+    //
+    // ArrayTupleBuilder tb = new ArrayTupleBuilder(fieldCount);
+    // DataOutput dos = tb.getDataOutput();
+    //
+    // ISerializerDeserializer[] recDescSers = {
+    // IntegerSerializerDeserializer.INSTANCE,
+    // IntegerSerializerDeserializer.INSTANCE };
+    // RecordDescriptor recDesc = new RecordDescriptor(recDescSers);
+    //
+    // IFrameTupleAccessor accessor = new FrameTupleAccessor(ctx.getFrameSize(),
+    // recDesc);
+    // accessor.reset(frame);
+    //
+    // FrameTupleReference tuple = new FrameTupleReference();
+    //
+    // int resultSize = 100;
+    // int[][] resultArray = new int[resultSize][2];
+    //
+    //
+    // //insert 100 tuples
+    // for (int i = 0; i < resultSize; i++){
+    // resultArray[i][0] = i;
+    // resultArray[i][1] = 1;
+    // }
+    //
+    //
+    // LSMTreeOpContext insertOpCtx = lsmtree.createOpContext(IndexOp.INSERT);
+    // for (int i = 0; i < resultSize; i++) {
+    //
+    // int f0 = resultArray[i][0];
+    // int f1 = resultArray[i][1];
+    //
+    // tb.reset();
+    // IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+    // tb.addFieldEndOffset();
+    // IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+    // tb.addFieldEndOffset();
+    //
+    // appender.reset(frame, true);
+    // appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0,
+    // tb.getSize());
+    //
+    // tuple.reset(accessor, 0);
+    //
+    // ArrayTupleReference t = new ArrayTupleReference();
+    // t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+    //
+    // try {
+    // lsmtree.insert(t, insertOpCtx);
+    // } catch (TreeIndexException e) {
+    // System.out.println("test01:" + e);
+    // e.printStackTrace();
+    // } catch (Exception e) {
+    // e.printStackTrace();
+    // }
+    // }
+    // // Delete the first 50 keys in the in-memory tree
+    // insertOpCtx = lsmtree.createOpContext(IndexOp.DELETE);
+    // for (int i = 0; i < 50; i++){
+    // resultArray[i][0] = i;
+    // resultArray[i][1] = 1;
+    // }
+    //
+    // for (int i = 0; i < 50; i++) {
+    //
+    // int f0 = resultArray[i][0];
+    // int f1 = resultArray[i][1];
+    //
+    // tb.reset();
+    // IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+    // tb.addFieldEndOffset();
+    // IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+    // tb.addFieldEndOffset();
+    //
+    // appender.reset(frame, true);
+    // appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0,
+    // tb.getSize());
+    //
+    // tuple.reset(accessor, 0);
+    //
+    // ArrayTupleReference t = new ArrayTupleReference();
+    // t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+    //
+    // try {
+    // lsmtree.delete(t, insertOpCtx);
+    // } catch (TreeIndexException e) {
+    // System.out.println("test01:" + e);
+    // e.printStackTrace();
+    // } catch (Exception e) {
+    // e.printStackTrace();
+    // }
+    // }
+    //
+    //
+    // //Flush the tree into the first in Disk tree
+    // lsmtree.flushInMemoryBtree();
+    //
+    // //insert 50 delete nodes
+    // insertOpCtx = lsmtree.createOpContext(IndexOp.DELETE);
+    // for (int i = 0; i < 50; i++){
+    // resultArray[i][0] = i;
+    // resultArray[i][1] = 2;
+    // }
+    //
+    // for (int i = 0; i < 50; i++) {
+    //
+    // int f0 = resultArray[i][0];
+    // int f1 = resultArray[i][1];
+    //
+    // tb.reset();
+    // IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+    // tb.addFieldEndOffset();
+    // IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+    // tb.addFieldEndOffset();
+    //
+    // appender.reset(frame, true);
+    // appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0,
+    // tb.getSize());
+    //
+    // tuple.reset(accessor, 0);
+    //
+    // ArrayTupleReference t = new ArrayTupleReference();
+    // t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+    //
+    // try {
+    // lsmtree.delete(t, insertOpCtx);
+    // } catch (TreeIndexException e) {
+    // System.out.println("test01:" + e);
+    // e.printStackTrace();
+    // } catch (Exception e) {
+    // e.printStackTrace();
+    // }
+    // }
+    //
+    // // insert 25 nodes
+    // insertOpCtx = lsmtree.createOpContext(IndexOp.INSERT);
+    // for (int i = 0; i < resultSize; i++){
+    // resultArray[i][0] = i;
+    // resultArray[i][1] = 2;
+    // }
+    // for (int i = 0; i < 25; i++) {
+    //
+    // int f0 = resultArray[i][0];
+    // int f1 = resultArray[i][1];
+    //
+    // tb.reset();
+    // IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+    // tb.addFieldEndOffset();
+    // IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+    // tb.addFieldEndOffset();
+    //
+    // appender.reset(frame, true);
+    // appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0,
+    // tb.getSize());
+    //
+    // tuple.reset(accessor, 0);
+    //
+    // ArrayTupleReference t = new ArrayTupleReference();
+    // t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+    //
+    // try {
+    // lsmtree.insert(t, insertOpCtx);
+    // } catch (TreeIndexException e) {
+    // System.out.println("test01:" + e);
+    // e.printStackTrace();
+    // } catch (Exception e) {
+    // e.printStackTrace();
+    // }
+    // }
+    //
+    // //Flush the tree into the fist in Disk tree, which have fieldId as "1"
+    // lsmtree.flushInMemoryBtree();
+    //
+    // //Print out the first in Disk Btree
+    // System.out.println("LSMTreeFlushTest: start print the first tree");
+    // lsmtree.scanDiskTree(0);
+    // //Print out the second in Disk Btree
+    // System.out.println("LSMTreeFlushTest: start print the second tree");
+    // lsmtree.scanDiskTree(1);
+    //
+    //
+    // lsmtree.close();
+    // bufferCache.closeFile(fileId);
+    // memBufferCache.close();
+    //
+    // System.out.printf("End of TEST1()\n");
+    //
+    // }
+    // TEST auto Flush
+    @Test
+    public void Test2() throws Exception {
+        System.out.printf("TEST2 START\n");
+        // in disk
+        TestStorageManagerComponentHolder.init(PAGE_SIZE, NUM_PAGES, MAX_OPEN_FILES);
+        IBufferCache bufferCache = TestStorageManagerComponentHolder.getBufferCache(ctx);
+        IFileMapProvider fmp = TestStorageManagerComponentHolder.getFileMapProvider(ctx);
+        FileReference file = new FileReference(new File(fileName));
+        bufferCache.createFile(file);
+        int fileId = fmp.lookupFileId(file);
+        bufferCache.openFile(fileId);
+
+        // in memory
+        InMemoryBufferCacheFactory InMemBufferCacheFactory = new InMemoryBufferCacheFactory(PAGE_SIZE, NUM_PAGES);
+        IBufferCache memBufferCache = InMemBufferCacheFactory.createInMemoryBufferCache();
+
+        // declare fields
+        int fieldCount = 2;
+        ITypeTraits[] typeTraits = new ITypeTraits[fieldCount];
+        typeTraits[0] = IntegerPointable.TYPE_TRAITS;
+        typeTraits[1] = IntegerPointable.TYPE_TRAITS;
+
+        // declare keys
+        int keyFieldCount = 1;
+        IBinaryComparator[] cmps = new IBinaryComparator[keyFieldCount];
+        cmps[0] = PointableBinaryComparatorFactory.of(IntegerPointable.FACTORY).createBinaryComparator();
+
+        MultiComparator cmp = new MultiComparator(cmps);
+
+        LSMTypeAwareTupleWriterFactory insertTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(typeTraits, false);
+        LSMTypeAwareTupleWriterFactory deleteTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(typeTraits, true);
+
+        ITreeIndexFrameFactory insertLeafFrameFactory = new BTreeNSMLeafFrameFactory(insertTupleWriterFactory);
+        ITreeIndexFrameFactory deleteLeafFrameFactory = new BTreeNSMLeafFrameFactory(deleteTupleWriterFactory);
+        ITreeIndexFrameFactory interiorFrameFactory = new BTreeNSMInteriorFrameFactory(insertTupleWriterFactory);
+        ITreeIndexMetaDataFrameFactory metaFrameFactory = new LIFOMetaDataFrameFactory();
+
+        IBTreeLeafFrame insertLeafFrame = (IBTreeLeafFrame) insertLeafFrameFactory.createFrame();
+
+        IFreePageManager freePageManager = new LinkedListFreePageManager(bufferCache, fileId, 0, metaFrameFactory);
+        IFreePageManager memFreePageManager = new InMemoryFreePageManager(NUM_PAGES, metaFrameFactory);
+
+        // For the Flush Mechanism
+        LSMEntireTupleWriterFactory flushTupleWriterFactory = new LSMEntireTupleWriterFactory(typeTraits);
+        ITreeIndexFrameFactory flushLeafFrameFactory = new BTreeNSMLeafFrameFactory(flushTupleWriterFactory);
+        FreePageManagerFactory freePageManagerFactory = new FreePageManagerFactory(bufferCache, metaFrameFactory);
+        BTreeFactory bTreeFactory = new BTreeFactory(bufferCache, freePageManagerFactory, cmp, fieldCount,
+                interiorFrameFactory, flushLeafFrameFactory);
+
+        LSMTree lsmtree = new LSMTree(memBufferCache, bufferCache, fieldCount, cmp, memFreePageManager,
+                interiorFrameFactory, insertLeafFrameFactory, deleteLeafFrameFactory, bTreeFactory,
+                (IFileMapManager) fmp);
+
+        lsmtree.create(fileId);
+        lsmtree.open(fileId);
+
+        ByteBuffer frame = ctx.allocateFrame();
+        FrameTupleAppender appender = new FrameTupleAppender(ctx.getFrameSize());
+
+        ArrayTupleBuilder tb = new ArrayTupleBuilder(fieldCount);
+        DataOutput dos = tb.getDataOutput();
+
+        ISerializerDeserializer[] recDescSers = { IntegerSerializerDeserializer.INSTANCE,
+                IntegerSerializerDeserializer.INSTANCE };
+        RecordDescriptor recDesc = new RecordDescriptor(recDescSers);
+
+        IFrameTupleAccessor accessor = new FrameTupleAccessor(ctx.getFrameSize(), recDesc);
+        accessor.reset(frame);
+
+        FrameTupleReference tuple = new FrameTupleReference();
+
+        int resultSize = 820;
+        int[][] resultArray = new int[resultSize][2];
+
+        // insert 820 tuples
+        for (int i = 0; i < resultSize; i++) {
+            resultArray[i][0] = i;
+            resultArray[i][1] = i;
+        }
+
+        ITreeIndexAccessor lsmTreeAccessor = lsmtree.createAccessor();
+        for (int i = 0; i < resultSize; i++) {
+
+            int f0 = resultArray[i][0];
+            int f1 = resultArray[i][1];
+
+            tb.reset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+            tb.addFieldEndOffset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+            tb.addFieldEndOffset();
+
+            appender.reset(frame, true);
+            appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+
+            tuple.reset(accessor, 0);
+
+            ArrayTupleReference t = new ArrayTupleReference();
+            t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+
+            try {
+                lsmTreeAccessor.insert(t);
+            } catch (TreeIndexException e) {
+                System.out.println("test02:" + e);
+                e.printStackTrace();
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+
+        // Print out the third in Disk Btree
+        System.out.println("LSMTreeFlushTest: start print the first tree");
+        // lsmtree.scanDiskTree(2);
+        // Print out the second in Disk Btree
+        System.out.println("LSMTreeFlushTest: start print the first tree");
+        // lsmtree.scanDiskTree(1);
+        // Print out the first in Disk Btree
+        System.out.println("LSMTreeFlushTest: start print the first tree");
+        lsmtree.scanDiskTree(0);
+
+        lsmtree.close();
+        bufferCache.closeFile(fileId);
+        memBufferCache.close();
+
+        System.out.printf("End of TEST2()\n");
+
+    }
+
+    // @Test
+    // public void Test3() throws Exception {
+    // System.out.printf("TEST3 START\n");
+    // //in disk
+    // TestStorageManagerComponentHolder.init(PAGE_SIZE, NUM_PAGES,
+    // MAX_OPEN_FILES);
+    // IBufferCache bufferCache =
+    // TestStorageManagerComponentHolder.getBufferCache(ctx);
+    // IFileMapProvider fmp =
+    // TestStorageManagerComponentHolder.getFileMapProvider(ctx);
+    // FileReference file = new FileReference(new File(fileName));
+    // bufferCache.createFile(file);
+    // int fileId = fmp.lookupFileId(file);
+    // bufferCache.openFile(fileId);
+    //
+    // //in memory
+    // InMemoryBufferCacheFactory InMemBufferCacheFactory = new
+    // InMemoryBufferCacheFactory(PAGE_SIZE, NUM_PAGES);
+    // IBufferCache memBufferCache =
+    // InMemBufferCacheFactory.createInMemoryBufferCache();
+    //
+    // // declare fields
+    // int fieldCount = 2;
+    // ITypeTrait[] typeTraits = new ITypeTrait[fieldCount];
+    // typeTraits[0] = new TypeTrait(4);
+    // typeTraits[1] = new TypeTrait(4);
+    //
+    // // declare keys
+    // int keyFieldCount = 1;
+    // IBinaryComparatorFactory[] cmpFactories = new
+    // IBinaryComparatorFactory[keyFieldCount];
+    // cmpFactories[0] = IntegerBinaryComparatorFactory.INSTANCE;
+    //
+    // MultiComparator cmp = BTreeUtils.createMultiComparator(cmpFactories);
+    //
+    // LSMTypeAwareTupleWriterFactory insertTupleWriterFactory = new
+    // LSMTypeAwareTupleWriterFactory(typeTraits, false);
+    // LSMTypeAwareTupleWriterFactory deleteTupleWriterFactory = new
+    // LSMTypeAwareTupleWriterFactory(typeTraits, true);
+    //
+    // ITreeIndexFrameFactory insertLeafFrameFactory = new
+    // BTreeNSMLeafFrameFactory(insertTupleWriterFactory);
+    // ITreeIndexFrameFactory deleteLeafFrameFactory = new
+    // BTreeNSMLeafFrameFactory(deleteTupleWriterFactory);
+    // ITreeIndexFrameFactory interiorFrameFactory = new
+    // BTreeNSMInteriorFrameFactory(insertTupleWriterFactory);
+    // ITreeIndexMetaDataFrameFactory metaFrameFactory = new
+    // LIFOMetaDataFrameFactory();
+    //
+    // IBTreeLeafFrame insertLeafFrame = (IBTreeLeafFrame)
+    // insertLeafFrameFactory.createFrame();
+    //
+    // IFreePageManager freePageManager = new
+    // LinkedListFreePageManager(bufferCache, fileId, 0, metaFrameFactory);
+    // IFreePageManager memFreePageManager = new InMemoryFreePageManager(30,
+    // metaFrameFactory);
+    //
+    // // For the Flush Mechanism
+    // LSMEntireTupleWriterFactory flushTupleWriterFactory = new
+    // LSMEntireTupleWriterFactory(typeTraits);
+    // ITreeIndexFrameFactory flushLeafFrameFactory = new
+    // BTreeNSMLeafFrameFactory(flushTupleWriterFactory);
+    // FreePageManagerFactory freePageManagerFactory = new
+    // FreePageManagerFactory(bufferCache, metaFrameFactory);
+    // BTreeFactory bTreeFactory = new BTreeFactory(bufferCache,
+    // freePageManagerFactory, cmp, fieldCount, interiorFrameFactory,
+    // flushLeafFrameFactory);
+    //
+    //
+    //
+    // LSMTree lsmtree = new LSMTree(memBufferCache, bufferCache, fieldCount,
+    // cmp, memFreePageManager, interiorFrameFactory, insertLeafFrameFactory,
+    // deleteLeafFrameFactory, bTreeFactory, (IFileMapManager)fmp);
+    //
+    // lsmtree.create(fileId);
+    // lsmtree.open(fileId);
+    //
+    // ByteBuffer frame = ctx.allocateFrame();
+    // FrameTupleAppender appender = new FrameTupleAppender(ctx.getFrameSize());
+    //
+    // ArrayTupleBuilder tb = new ArrayTupleBuilder(fieldCount);
+    // DataOutput dos = tb.getDataOutput();
+    //
+    // ISerializerDeserializer[] recDescSers = {
+    // IntegerSerializerDeserializer.INSTANCE,
+    // IntegerSerializerDeserializer.INSTANCE };
+    // RecordDescriptor recDesc = new RecordDescriptor(recDescSers);
+    //
+    // IFrameTupleAccessor accessor = new FrameTupleAccessor(ctx.getFrameSize(),
+    // recDesc);
+    // accessor.reset(frame);
+    //
+    // FrameTupleReference tuple = new FrameTupleReference();
+    //
+    // int resultSize = 500;
+    // int[][] resultArray = new int[resultSize][2];
+    //
+    //
+    // //insert 250 tuples
+    // System.out.printf("Start for 1st Insert\n");
+    // LSMTreeOpContext insertOpCtx = lsmtree.createOpContext(IndexOp.INSERT);
+    // for (int i = 0; i < 252; i++){
+    // resultArray[i][0] = i;
+    // resultArray[i][1] = i;
+    // }
+    // for (int i = 0; i < 252; i++) {
+    //
+    // int f0 = resultArray[i][0];
+    // int f1 = resultArray[i][1];
+    //
+    // tb.reset();
+    // IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+    // tb.addFieldEndOffset();
+    // IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+    // tb.addFieldEndOffset();
+    //
+    // appender.reset(frame, true);
+    // appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0,
+    // tb.getSize());
+    //
+    // tuple.reset(accessor, 0);
+    //
+    // ArrayTupleReference t = new ArrayTupleReference();
+    // t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+    //
+    // try {
+    // lsmtree.insert(t, insertOpCtx);
+    // } catch (TreeIndexException e) {
+    // System.out.println("test03:" + e);
+    // e.printStackTrace();
+    // } catch (Exception e) {
+    // e.printStackTrace();
+    // }
+    // }
+    // System.out.printf("Start for 2nd Insert\n");
+    // //delete 126~251. Deletion of 251 cause the flush
+    // insertOpCtx.reset(IndexOp.DELETE);
+    // // LSMTreeOpContext insertOpCtx =
+    // lsmtree.createOpContext(IndexOp.DELETE);
+    // for (int i = 125; i < 253; i++){
+    // resultArray[i][0] = i;
+    // resultArray[i][1] = i;
+    // }
+    // for (int i = 126; i < 253; i++) {
+    // int f0 = resultArray[i][0];
+    // int f1 = resultArray[i][1];
+    //
+    // tb.reset();
+    // IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+    // tb.addFieldEndOffset();
+    // IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+    // tb.addFieldEndOffset();
+    //
+    // appender.reset(frame, true);
+    // appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0,
+    // tb.getSize());
+    //
+    // tuple.reset(accessor, 0);
+    //
+    // ArrayTupleReference t = new ArrayTupleReference();
+    // t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+    //
+    // try {
+    // lsmtree.delete(t, insertOpCtx);
+    // } catch (TreeIndexException e) {
+    // System.out.println("test03:" + e);
+    // e.printStackTrace();
+    // } catch (Exception e) {
+    // e.printStackTrace();
+    // }
+    // }
+    // //delete 0~250
+    // insertOpCtx = lsmtree.createOpContext(IndexOp.INSERT);
+    // for (int i = 130; i > 0; i--){
+    // resultArray[i][0] = i;
+    // resultArray[i][1] = i;
+    // }
+    // for (int i = 130; i > 0; i--) {
+    //
+    // int f0 = resultArray[i][0];
+    // int f1 = resultArray[i][1];
+    //
+    // tb.reset();
+    // IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+    // tb.addFieldEndOffset();
+    // IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+    // tb.addFieldEndOffset();
+    //
+    // appender.reset(frame, true);
+    // appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0,
+    // tb.getSize());
+    //
+    // tuple.reset(accessor, 0);
+    //
+    // ArrayTupleReference t = new ArrayTupleReference();
+    // t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+    //
+    // try {
+    // lsmtree.insert(t, insertOpCtx);
+    // } catch (TreeIndexException e) {
+    // System.out.println("test03:" + e);
+    // e.printStackTrace();
+    // } catch (Exception e) {
+    // e.printStackTrace();
+    // }
+    // }
+    //
+    // //
+    // //
+    // //
+    // // //Print out the second in Disk Btree
+    // // System.out.println("LSMTreeFlushTest: start print the second tree");
+    // // lsmtree.scanDiskTree(1);
+    // // //Print out the first in Disk Btree
+    // // System.out.println("LSMTreeFlushTest: start print the first tree");
+    // // lsmtree.scanDiskTree(0);
+    // //
+    // // //Print out the In-memory Tree
+    // //
+    // System.out.println("LSMTreeFlushTest: start print the In-memory tree");
+    // // lsmtree.scanInMemoryTree();
+    // // //TODO: scan whole tree
+    //
+    // LOGGER.info("RANGE SEARCH:");
+    //
+    // BTreeOpContext searchOpCtx = lsmtree.createOpContext(IndexOp.SEARCH);
+    // ITreeIndexCursor rangeCursor = new LSMTreeRangeSearchCursor();
+    //
+    // // build low and high keys
+    // ArrayTupleBuilder ktb = new ArrayTupleBuilder(cmp.getKeyFieldCount());
+    // DataOutput kdos = ktb.getDataOutput();
+    //
+    // ISerializerDeserializer[] keyDescSers = {
+    // IntegerSerializerDeserializer.INSTANCE };
+    // RecordDescriptor keyDesc = new RecordDescriptor(keyDescSers);
+    // IFrameTupleAccessor keyAccessor = new
+    // FrameTupleAccessor(ctx.getFrameSize(), keyDesc);
+    // keyAccessor.reset(frame);
+    //
+    // appender.reset(frame, true);
+    //
+    // // build and append low key
+    // ktb.reset();
+    // IntegerSerializerDeserializer.INSTANCE.serialize(-1, kdos);
+    // ktb.addFieldEndOffset();
+    // appender.append(ktb.getFieldEndOffsets(), ktb.getByteArray(), 0,
+    // ktb.getSize());
+    //
+    // // build and append high key
+    // ktb.reset();
+    // IntegerSerializerDeserializer.INSTANCE.serialize(300, 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(searchCmps);
+    //
+    // RangePredicate rangePred = new RangePredicate(true, lowKey, highKey,
+    // true, true, searchCmp, searchCmp);
+    // lsmtree.search(rangeCursor, rangePred, searchOpCtx);
+    //
+    // try {
+    // while (rangeCursor.hasNext()) {
+    // rangeCursor.next();
+    // ITupleReference frameTuple = rangeCursor.getTuple();
+    // String rec = TupleUtils.printTuple(frameTuple, recDescSers);
+    // if(((LSMTypeAwareTupleReference)frameTuple).isDelete()) {
+    // System.out.println("del " + rec);
+    // }
+    // else {
+    // System.out.println("ins " + rec);
+    // }
+    // // System.out.println("------------------");
+    // }
+    // } catch (Exception e) {
+    // e.printStackTrace();
+    // } finally {
+    // rangeCursor.close();
+    // }
+    //
+    // lsmtree.close();
+    // bufferCache.closeFile(fileId);
+    // memBufferCache.close();
+    //
+    // System.out.printf("End of TEST3()\n");
+    //
+    // }
+
+}
\ No newline at end of file
diff --git a/hyracks-tests/hyracks-storage-am-lsmtree-btree-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsmtree/btree/LSMTreeMergeTest.java b/hyracks-tests/hyracks-storage-am-lsmtree-btree-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsmtree/btree/LSMTreeMergeTest.java
new file mode 100644
index 0000000..03924f0
--- /dev/null
+++ b/hyracks-tests/hyracks-storage-am-lsmtree-btree-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsmtree/btree/LSMTreeMergeTest.java
@@ -0,0 +1,378 @@
+package edu.uci.ics.hyracks.storage.am.lsmtree.btree;
+
+import java.io.DataOutput;
+import java.io.File;
+import java.nio.ByteBuffer;
+
+import org.junit.Test;
+
+import edu.uci.ics.hyracks.api.comm.IFrameTupleAccessor;
+import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
+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.ITypeTraits;
+import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
+import edu.uci.ics.hyracks.api.io.FileReference;
+import edu.uci.ics.hyracks.data.std.accessors.PointableBinaryComparatorFactory;
+import edu.uci.ics.hyracks.data.std.primitive.IntegerPointable;
+import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
+import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleReference;
+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.marshalling.IntegerSerializerDeserializer;
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeLeafFrame;
+import edu.uci.ics.hyracks.storage.am.btree.frames.BTreeNSMInteriorFrameFactory;
+import edu.uci.ics.hyracks.storage.am.btree.frames.BTreeNSMLeafFrameFactory;
+import edu.uci.ics.hyracks.storage.am.common.api.IFreePageManager;
+import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexAccessor;
+import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexFrameFactory;
+import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexMetaDataFrameFactory;
+import edu.uci.ics.hyracks.storage.am.common.api.TreeIndexException;
+import edu.uci.ics.hyracks.storage.am.common.frames.LIFOMetaDataFrameFactory;
+import edu.uci.ics.hyracks.storage.am.common.freepage.LinkedListFreePageManager;
+import edu.uci.ics.hyracks.storage.am.common.ophelpers.MultiComparator;
+import edu.uci.ics.hyracks.storage.am.lsmtree.freepage.FreePageManagerFactory;
+import edu.uci.ics.hyracks.storage.am.lsmtree.freepage.InMemoryBufferCacheFactory;
+import edu.uci.ics.hyracks.storage.am.lsmtree.freepage.InMemoryFreePageManager;
+import edu.uci.ics.hyracks.storage.am.lsmtree.impls.BTreeFactory;
+import edu.uci.ics.hyracks.storage.am.lsmtree.impls.LSMTree;
+import edu.uci.ics.hyracks.storage.am.lsmtree.tuples.LSMEntireTupleWriterFactory;
+import edu.uci.ics.hyracks.storage.am.lsmtree.tuples.LSMTypeAwareTupleWriterFactory;
+import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
+import edu.uci.ics.hyracks.storage.common.file.IFileMapManager;
+import edu.uci.ics.hyracks.storage.common.file.IFileMapProvider;
+import edu.uci.ics.hyracks.test.support.TestStorageManagerComponentHolder;
+import edu.uci.ics.hyracks.test.support.TestUtils;
+
+public class LSMTreeMergeTest extends AbstractLSMTreeTest {
+    private static final int PAGE_SIZE = 256;
+    private static final int NUM_PAGES = 30;
+    private static final int MAX_OPEN_FILES = 100;
+    private static final int HYRACKS_FRAME_SIZE = 128;
+    private IHyracksTaskContext ctx = TestUtils.create(HYRACKS_FRAME_SIZE);
+
+    @Test
+    public void Test1() throws Exception {
+        System.out.printf("TEST1 START\n");
+        // in disk
+        TestStorageManagerComponentHolder.init(PAGE_SIZE, NUM_PAGES, MAX_OPEN_FILES);
+        IBufferCache bufferCache = TestStorageManagerComponentHolder.getBufferCache(ctx);
+        IFileMapProvider fmp = TestStorageManagerComponentHolder.getFileMapProvider(ctx);
+        FileReference file = new FileReference(new File(fileName));
+        bufferCache.createFile(file);
+        int fileId = fmp.lookupFileId(file);
+        bufferCache.openFile(fileId);
+
+        // in memory
+        InMemoryBufferCacheFactory InMemBufferCacheFactory = new InMemoryBufferCacheFactory(PAGE_SIZE, NUM_PAGES);
+        IBufferCache memBufferCache = InMemBufferCacheFactory.createInMemoryBufferCache();
+
+        // declare fields
+        int fieldCount = 2;
+        ITypeTraits[] typeTraits = new ITypeTraits[fieldCount];
+        typeTraits[0] = IntegerPointable.TYPE_TRAITS;
+        typeTraits[1] = IntegerPointable.TYPE_TRAITS;
+
+        // declare keys
+        int keyFieldCount = 1;
+        IBinaryComparator[] cmps = new IBinaryComparator[keyFieldCount];
+        cmps[0] = PointableBinaryComparatorFactory.of(IntegerPointable.FACTORY).createBinaryComparator();
+
+        MultiComparator cmp = new MultiComparator(cmps);
+
+        LSMTypeAwareTupleWriterFactory insertTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(typeTraits, false);
+        LSMTypeAwareTupleWriterFactory deleteTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(typeTraits, true);
+
+        ITreeIndexFrameFactory insertLeafFrameFactory = new BTreeNSMLeafFrameFactory(insertTupleWriterFactory);
+        ITreeIndexFrameFactory deleteLeafFrameFactory = new BTreeNSMLeafFrameFactory(deleteTupleWriterFactory);
+        ITreeIndexFrameFactory interiorFrameFactory = new BTreeNSMInteriorFrameFactory(insertTupleWriterFactory);
+        ITreeIndexMetaDataFrameFactory metaFrameFactory = new LIFOMetaDataFrameFactory();
+
+        IBTreeLeafFrame insertLeafFrame = (IBTreeLeafFrame) insertLeafFrameFactory.createFrame();
+
+        IFreePageManager freePageManager = new LinkedListFreePageManager(bufferCache, fileId, 0, metaFrameFactory);
+        IFreePageManager memFreePageManager = new InMemoryFreePageManager(NUM_PAGES, metaFrameFactory);
+
+        // For the Flush Mechanism
+        LSMEntireTupleWriterFactory flushTupleWriterFactory = new LSMEntireTupleWriterFactory(typeTraits);
+        ITreeIndexFrameFactory flushLeafFrameFactory = new BTreeNSMLeafFrameFactory(flushTupleWriterFactory);
+        FreePageManagerFactory freePageManagerFactory = new FreePageManagerFactory(bufferCache, metaFrameFactory);
+        BTreeFactory bTreeFactory = new BTreeFactory(bufferCache, freePageManagerFactory, cmp, fieldCount,
+                interiorFrameFactory, flushLeafFrameFactory);
+
+        LSMTree lsmtree = new LSMTree(memBufferCache, bufferCache, fieldCount, cmp, memFreePageManager,
+                interiorFrameFactory, insertLeafFrameFactory, deleteLeafFrameFactory, bTreeFactory,
+                (IFileMapManager) fmp);
+
+        // LSMTree lsmtree = LSMTreeUtils.createLSMTree(10, 30, 2,
+        // memBufferCache, bufferCache, fileId, typeTraits,
+        // cmp.getComparators(), BTreeLeafFrameType.REGULAR_NSM,
+        // (IFileMapManager)fmp);
+
+        lsmtree.create(fileId);
+        lsmtree.open(fileId);
+
+        ByteBuffer frame = ctx.allocateFrame();
+        FrameTupleAppender appender = new FrameTupleAppender(ctx.getFrameSize());
+
+        ArrayTupleBuilder tb = new ArrayTupleBuilder(fieldCount);
+        DataOutput dos = tb.getDataOutput();
+
+        ISerializerDeserializer[] recDescSers = { IntegerSerializerDeserializer.INSTANCE,
+                IntegerSerializerDeserializer.INSTANCE };
+        RecordDescriptor recDesc = new RecordDescriptor(recDescSers);
+
+        IFrameTupleAccessor accessor = new FrameTupleAccessor(ctx.getFrameSize(), recDesc);
+        accessor.reset(frame);
+
+        FrameTupleReference tuple = new FrameTupleReference();
+
+        int resultSize = 100000;
+        int[][] resultArray = new int[resultSize][2];
+
+        // insert 0~250 tuples
+        System.out.printf("Start for 1st Insert\n");
+        ITreeIndexAccessor lsmTreeAccessor = lsmtree.createAccessor();
+        for (int i = 0; i < 251; i++) {
+            resultArray[i][0] = i;
+            resultArray[i][1] = i;
+        }
+        for (int i = 0; i < 251; i++) {
+
+            int f0 = resultArray[i][0];
+            int f1 = resultArray[i][1];
+
+            tb.reset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+            tb.addFieldEndOffset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+            tb.addFieldEndOffset();
+
+            appender.reset(frame, true);
+            appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+
+            tuple.reset(accessor, 0);
+
+            ArrayTupleReference t = new ArrayTupleReference();
+            t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+
+            try {
+                lsmTreeAccessor.insert(t);
+            } catch (TreeIndexException e) {
+                System.out.println("test03:" + e);
+                e.printStackTrace();
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+        System.out.printf("Start for 2nd Insert\n");
+        // delete 126~250.
+        for (int i = 126; i < 251; i++) {
+            resultArray[i][0] = i;
+            resultArray[i][1] = i;
+        }
+        for (int i = 126; i < 251; i++) {
+            int f0 = resultArray[i][0];
+            int f1 = resultArray[i][1];
+
+            tb.reset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+            tb.addFieldEndOffset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+            tb.addFieldEndOffset();
+
+            appender.reset(frame, true);
+            appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+
+            tuple.reset(accessor, 0);
+
+            ArrayTupleReference t = new ArrayTupleReference();
+            t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+
+            try {
+                lsmTreeAccessor.delete(t);
+            } catch (TreeIndexException e) {
+                System.out.println("test03:" + e);
+                e.printStackTrace();
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+        // insert 251~1
+        for (int i = 251; i > 0; i--) {
+            resultArray[i][0] = i;
+            resultArray[i][1] = i;
+        }
+        for (int i = 251; i > 0; i--) {
+
+            int f0 = resultArray[i][0];
+            int f1 = resultArray[i][1];
+
+            tb.reset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+            tb.addFieldEndOffset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+            tb.addFieldEndOffset();
+
+            appender.reset(frame, true);
+            appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+
+            tuple.reset(accessor, 0);
+
+            ArrayTupleReference t = new ArrayTupleReference();
+            t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+
+            try {
+                lsmTreeAccessor.insert(t);
+            } catch (TreeIndexException e) {
+                System.out.println("test03:" + e);
+                e.printStackTrace();
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+        // delete 100~0
+        for (int i = 100; i >= 0; i--) {
+            resultArray[i][0] = i;
+            resultArray[i][1] = i;
+        }
+        for (int i = 100; i >= 0; i--) {
+
+            int f0 = resultArray[i][0];
+            int f1 = resultArray[i][1];
+
+            tb.reset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+            tb.addFieldEndOffset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+            tb.addFieldEndOffset();
+
+            appender.reset(frame, true);
+            appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+
+            tuple.reset(accessor, 0);
+
+            ArrayTupleReference t = new ArrayTupleReference();
+            t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+
+            try {
+                lsmTreeAccessor.delete(t);
+            } catch (TreeIndexException e) {
+                System.out.println("test03:" + e);
+                e.printStackTrace();
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+        // insert 1~50
+        for (int i = 1; i < 51; i++) {
+            resultArray[i][0] = i;
+            resultArray[i][1] = i;
+        }
+        for (int i = 1; i < 51; i++) {
+
+            int f0 = resultArray[i][0];
+            int f1 = resultArray[i][1];
+
+            tb.reset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+            tb.addFieldEndOffset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+            tb.addFieldEndOffset();
+
+            appender.reset(frame, true);
+            appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+
+            tuple.reset(accessor, 0);
+
+            ArrayTupleReference t = new ArrayTupleReference();
+            t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+
+            try {
+                lsmTreeAccessor.insert(t);
+            } catch (TreeIndexException e) {
+                System.out.println("test03:" + e);
+                e.printStackTrace();
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+
+        lsmtree.merge();
+
+        // Output should be:
+        // In memory tree = 0->del, 1~50 ins
+        // MergedTree = 0->ins, 1~100->del, 101~251->ins
+        // Whole search = 1~50,101~251
+
+        // System.out.println("LSMTreeFlushTest: start print the first tree");
+        // lsmtree.scanDiskTree(1);
+        //
+        // Print out the first in Disk Btree
+        // System.out.println("LSMTreeFlushTest: start print the first tree");
+        // lsmtree.scanDiskTree(0);
+        // Print out the In-memory Tree
+        System.out.println("LSMTreeFlushTest: start print the In-memory tree");
+        lsmtree.scanInMemoryTree();
+        // TODO: scan whole tree
+        /*
+         * System.out.println("Range SEARCH:");
+         * 
+         * BTreeOpContext searchOpCtx = lsmtree.createOpContext(IndexOp.SEARCH);
+         * ITreeIndexCursor rangeCursor = new LSMTreeRangeSearchCursor();
+         * 
+         * // build low and high keys ArrayTupleBuilder ktb = new
+         * ArrayTupleBuilder(cmp.getKeyFieldCount()); DataOutput kdos =
+         * ktb.getDataOutput();
+         * 
+         * ISerializerDeserializer[] keyDescSers = {
+         * IntegerSerializerDeserializer.INSTANCE }; RecordDescriptor keyDesc =
+         * new RecordDescriptor(keyDescSers); IFrameTupleAccessor keyAccessor =
+         * new FrameTupleAccessor( ctx.getFrameSize(), keyDesc);
+         * keyAccessor.reset(frame);
+         * 
+         * appender.reset(frame, true);
+         * 
+         * // build and append low key ktb.reset();
+         * IntegerSerializerDeserializer.INSTANCE.serialize(-1, kdos);
+         * ktb.addFieldEndOffset(); appender.append(ktb.getFieldEndOffsets(),
+         * ktb.getByteArray(), 0, ktb.getSize());
+         * 
+         * // build and append high key ktb.reset();
+         * IntegerSerializerDeserializer.INSTANCE.serialize(300, 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(searchCmps);
+         * 
+         * RangePredicate rangePred = new RangePredicate(true, lowKey, highKey,
+         * true, true, searchCmp, searchCmp); lsmtree.search(rangeCursor,
+         * rangePred, searchOpCtx);
+         * 
+         * try { while (rangeCursor.hasNext()) { rangeCursor.next();
+         * ITupleReference frameTuple = rangeCursor.getTuple(); String rec =
+         * TupleUtils.printTuple(frameTuple, recDescSers); if
+         * (((LSMTypeAwareTupleReference) frameTuple).isDelete()) {
+         * System.out.println("del " + rec); } else { System.out.println("ins "
+         * + rec); } // System.out.println("------------------"); } } catch
+         * (Exception e) { e.printStackTrace(); } finally { rangeCursor.close();
+         * }
+         */
+        lsmtree.close();
+        bufferCache.closeFile(fileId);
+        memBufferCache.close();
+
+        System.out.printf("End of TEST1()\n");
+
+    }
+}
diff --git a/hyracks-tests/hyracks-storage-am-lsmtree-btree-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsmtree/btree/LSMTreeSearchTest.java b/hyracks-tests/hyracks-storage-am-lsmtree-btree-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsmtree/btree/LSMTreeSearchTest.java
new file mode 100644
index 0000000..9b48303
--- /dev/null
+++ b/hyracks-tests/hyracks-storage-am-lsmtree-btree-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsmtree/btree/LSMTreeSearchTest.java
@@ -0,0 +1,419 @@
+package edu.uci.ics.hyracks.storage.am.lsmtree.btree;
+
+import java.io.DataOutput;
+import java.io.File;
+import java.nio.ByteBuffer;
+import java.util.Date;
+import java.util.Random;
+
+import org.junit.Test;
+
+import edu.uci.ics.hyracks.api.comm.IFrameTupleAccessor;
+import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
+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.ITypeTraits;
+import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
+import edu.uci.ics.hyracks.api.io.FileReference;
+import edu.uci.ics.hyracks.data.std.accessors.PointableBinaryComparatorFactory;
+import edu.uci.ics.hyracks.data.std.primitive.IntegerPointable;
+import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
+import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleReference;
+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.marshalling.IntegerSerializerDeserializer;
+import edu.uci.ics.hyracks.dataflow.common.util.TupleUtils;
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeInteriorFrame;
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeLeafFrame;
+import edu.uci.ics.hyracks.storage.am.btree.frames.BTreeNSMInteriorFrameFactory;
+import edu.uci.ics.hyracks.storage.am.btree.frames.BTreeNSMLeafFrameFactory;
+import edu.uci.ics.hyracks.storage.am.btree.impls.BTree;
+import edu.uci.ics.hyracks.storage.am.btree.impls.RangePredicate;
+import edu.uci.ics.hyracks.storage.am.common.api.IFreePageManager;
+import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexAccessor;
+import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexCursor;
+import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexFrameFactory;
+import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexMetaDataFrame;
+import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexMetaDataFrameFactory;
+import edu.uci.ics.hyracks.storage.am.common.api.TreeIndexException;
+import edu.uci.ics.hyracks.storage.am.common.frames.LIFOMetaDataFrameFactory;
+import edu.uci.ics.hyracks.storage.am.common.freepage.LinkedListFreePageManager;
+import edu.uci.ics.hyracks.storage.am.common.ophelpers.MultiComparator;
+import edu.uci.ics.hyracks.storage.am.common.tuples.TypeAwareTupleWriterFactory;
+import edu.uci.ics.hyracks.storage.am.lsmtree.freepage.FreePageManagerFactory;
+import edu.uci.ics.hyracks.storage.am.lsmtree.freepage.InMemoryBufferCacheFactory;
+import edu.uci.ics.hyracks.storage.am.lsmtree.freepage.InMemoryFreePageManager;
+import edu.uci.ics.hyracks.storage.am.lsmtree.impls.BTreeFactory;
+import edu.uci.ics.hyracks.storage.am.lsmtree.impls.InDiskTreeInfo;
+import edu.uci.ics.hyracks.storage.am.lsmtree.impls.LSMTree;
+import edu.uci.ics.hyracks.storage.am.lsmtree.impls.LSMTreeRangeSearchCursor;
+import edu.uci.ics.hyracks.storage.am.lsmtree.tuples.LSMEntireTupleWriterFactory;
+import edu.uci.ics.hyracks.storage.am.lsmtree.tuples.LSMTypeAwareTupleReference;
+import edu.uci.ics.hyracks.storage.am.lsmtree.tuples.LSMTypeAwareTupleWriterFactory;
+import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
+import edu.uci.ics.hyracks.storage.common.file.IFileMapManager;
+import edu.uci.ics.hyracks.storage.common.file.IFileMapProvider;
+import edu.uci.ics.hyracks.test.support.TestStorageManagerComponentHolder;
+import edu.uci.ics.hyracks.test.support.TestUtils;
+
+// TODO: needs a big cleanup phase.
+public class LSMTreeSearchTest extends AbstractLSMTreeTest {
+
+    private static final int PAGE_SIZE = 256;
+    private static final int NUM_PAGES = 10;
+    private static final int MAX_OPEN_FILES = 100;
+    private static final int HYRACKS_FRAME_SIZE = 128;
+    private IHyracksTaskContext ctx = TestUtils.create(HYRACKS_FRAME_SIZE);
+
+    @Test
+    public void Test1() throws Exception {
+        // in disk
+        TestStorageManagerComponentHolder.init(PAGE_SIZE, NUM_PAGES, MAX_OPEN_FILES);
+        IBufferCache bufferCache = TestStorageManagerComponentHolder.getBufferCache(ctx);
+        IFileMapProvider fmp = TestStorageManagerComponentHolder.getFileMapProvider(ctx);
+        FileReference file = new FileReference(new File(fileName));
+        bufferCache.createFile(file);
+        int fileId = fmp.lookupFileId(file);
+        bufferCache.openFile(fileId);
+
+        // in memory
+        InMemoryBufferCacheFactory InMemBufferCacheFactory = new InMemoryBufferCacheFactory(PAGE_SIZE, NUM_PAGES);
+        IBufferCache memBufferCache = InMemBufferCacheFactory.createInMemoryBufferCache();
+
+        // declare fields
+        int fieldCount = 2;
+        ITypeTraits[] typeTraits = new ITypeTraits[fieldCount];
+        typeTraits[0] = IntegerPointable.TYPE_TRAITS;
+        typeTraits[1] = IntegerPointable.TYPE_TRAITS;
+
+        // declare keys
+        int keyFieldCount = 1;
+        IBinaryComparator[] cmps = new IBinaryComparator[keyFieldCount];
+        cmps[0] = PointableBinaryComparatorFactory.of(IntegerPointable.FACTORY).createBinaryComparator();
+
+        MultiComparator cmp = new MultiComparator(cmps);
+
+        LSMTypeAwareTupleWriterFactory insertTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(typeTraits, false);
+        LSMTypeAwareTupleWriterFactory deleteTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(typeTraits, true);
+
+        ITreeIndexFrameFactory insertLeafFrameFactory = new BTreeNSMLeafFrameFactory(insertTupleWriterFactory);
+        ITreeIndexFrameFactory deleteLeafFrameFactory = new BTreeNSMLeafFrameFactory(deleteTupleWriterFactory);
+        ITreeIndexFrameFactory interiorFrameFactory = new BTreeNSMInteriorFrameFactory(insertTupleWriterFactory);
+        ITreeIndexMetaDataFrameFactory metaFrameFactory = new LIFOMetaDataFrameFactory();
+
+        IBTreeLeafFrame insertLeafFrame = (IBTreeLeafFrame) insertLeafFrameFactory.createFrame();
+
+        IFreePageManager freePageManager = new LinkedListFreePageManager(bufferCache, fileId, 0, metaFrameFactory);
+        IFreePageManager memFreePageManager = new InMemoryFreePageManager(100, metaFrameFactory);
+
+        LSMEntireTupleWriterFactory flushTupleWriterFactory = new LSMEntireTupleWriterFactory(typeTraits);
+        ITreeIndexFrameFactory flushLeafFrameFactory = new BTreeNSMLeafFrameFactory(flushTupleWriterFactory);
+        FreePageManagerFactory freePageManagerFactory = new FreePageManagerFactory(bufferCache, metaFrameFactory);
+        BTreeFactory bTreeFactory = new BTreeFactory(bufferCache, freePageManagerFactory, cmp, fieldCount,
+                interiorFrameFactory, flushLeafFrameFactory);
+
+        LSMTree lsmtree = new LSMTree(memBufferCache, bufferCache, fieldCount, cmp, memFreePageManager,
+                interiorFrameFactory, insertLeafFrameFactory, deleteLeafFrameFactory, bTreeFactory,
+                (IFileMapManager) fmp);
+
+        lsmtree.create(fileId);
+        lsmtree.open(fileId);
+
+        ByteBuffer frame = ctx.allocateFrame();
+        FrameTupleAppender appender = new FrameTupleAppender(ctx.getFrameSize());
+
+        ArrayTupleBuilder tb = new ArrayTupleBuilder(fieldCount);
+        DataOutput dos = tb.getDataOutput();
+
+        ISerializerDeserializer[] recDescSers = { IntegerSerializerDeserializer.INSTANCE,
+                IntegerSerializerDeserializer.INSTANCE };
+        RecordDescriptor recDesc = new RecordDescriptor(recDescSers);
+
+        IFrameTupleAccessor accessor = new FrameTupleAccessor(ctx.getFrameSize(), recDesc);
+        accessor.reset(frame);
+
+        FrameTupleReference tuple = new FrameTupleReference();
+
+        ITreeIndexAccessor lsmTreeAccessor = lsmtree.createAccessor();
+
+        // delete
+        for (int i = 26; i < 36; i++) {
+
+            int f0 = i;
+            int f1 = -1;
+
+            tb.reset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+            tb.addFieldEndOffset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+            tb.addFieldEndOffset();
+
+            appender.reset(frame, true);
+            appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+
+            tuple.reset(accessor, 0);
+
+            ArrayTupleReference t = new ArrayTupleReference();
+            t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+
+            try {
+                lsmTreeAccessor.delete(t);
+            } catch (TreeIndexException e) {
+                System.out.println("test01:" + e);
+                e.printStackTrace();
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+
+        for (int i = 21; i < 31; i++) {
+            int f0 = i;
+            int f1 = 0;
+
+            tb.reset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+            tb.addFieldEndOffset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+            tb.addFieldEndOffset();
+
+            appender.reset(frame, true);
+            appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+
+            tuple.reset(accessor, 0);
+
+            ArrayTupleReference t = new ArrayTupleReference();
+            t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+
+            try {
+                lsmTreeAccessor.insert(t);
+            } catch (TreeIndexException e) {
+                System.out.println("test01:" + e);
+                e.printStackTrace();
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+
+        // In disk insertions 1
+
+        LOGGER.info("Start in-disk insertions");
+
+        fileName = tmpDir + sep + simpleDateFormat.format(new Date());
+        FileReference file_1 = new FileReference(new File(fileName));
+        bufferCache.createFile(file_1);
+        int fileId_1 = fmp.lookupFileId(file_1);
+        bufferCache.openFile(fileId_1);
+
+        TypeAwareTupleWriterFactory tupleWriterFactory = new TypeAwareTupleWriterFactory(typeTraits);
+        ITreeIndexFrameFactory leafFrameFactory = new BTreeNSMLeafFrameFactory(tupleWriterFactory);
+
+        IBTreeLeafFrame leafFrame = (IBTreeLeafFrame) leafFrameFactory.createFrame();
+        IBTreeInteriorFrame interiorFrame = (IBTreeInteriorFrame) interiorFrameFactory.createFrame();
+        ITreeIndexMetaDataFrame metaFrame = metaFrameFactory.createFrame();
+
+        IFreePageManager freePageManager_1 = new LinkedListFreePageManager(bufferCache, fileId_1, 0, metaFrameFactory);
+
+        BTree btree_1 = new BTree(bufferCache, fieldCount, cmp, freePageManager_1, interiorFrameFactory,
+                leafFrameFactory);
+        btree_1.create(fileId_1);
+        btree_1.open(fileId_1);
+
+        // TODO: rename this one.
+        InDiskTreeInfo info_1 = new InDiskTreeInfo(btree_1);
+        lsmtree.inDiskTreeInfoList.add(info_1);
+
+        Random rnd = new Random();
+        rnd.setSeed(50);
+
+        LOGGER.info("INSERTING INTO TREE");
+
+        // ByteBuffer
+        frame = ctx.allocateFrame();
+        // FrameTupleAppender
+        appender = new FrameTupleAppender(ctx.getFrameSize());
+        // ArrayTupleBuilder
+        tb = new ArrayTupleBuilder(fieldCount);
+        // DataOutput
+        dos = tb.getDataOutput();
+
+        recDesc = new RecordDescriptor(recDescSers);
+
+        accessor.reset(frame);
+
+        tuple = new FrameTupleReference();
+
+        ITreeIndexAccessor indexAccessor_1 = btree_1.createAccessor();
+
+        // 10000
+        for (int i = 16; i < 41; i++) {
+
+            int f0 = i;
+            int f1 = 1;
+
+            tb.reset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+            tb.addFieldEndOffset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+            tb.addFieldEndOffset();
+
+            appender.reset(frame, true);
+            appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+
+            tuple.reset(accessor, 0);
+
+            ArrayTupleReference t = new ArrayTupleReference();
+            t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+
+            if (i % 10 == 0) {
+                System.out.println("INSERTING " + i + " : " + f0 + " " + f1);
+            }
+
+            try {
+                indexAccessor_1.insert(t);
+            } catch (TreeIndexException e) {
+                e.printStackTrace();
+                System.out.println("Error: " + e);
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+        // btree_1.close();
+
+        // In disk insertions 2
+
+        LOGGER.info("Start in-disk insertions");
+
+        fileName = tmpDir + sep + simpleDateFormat.format(new Date());
+        FileReference file_2 = new FileReference(new File(fileName));
+        bufferCache.createFile(file_2);
+        int fileId_2 = fmp.lookupFileId(file_2);
+        bufferCache.openFile(fileId_2);
+
+        IFreePageManager freePageManager_2 = new LinkedListFreePageManager(bufferCache, fileId_2, 0, metaFrameFactory);
+        BTree btree_2 = new BTree(bufferCache, fieldCount, cmp, freePageManager_2, interiorFrameFactory,
+                leafFrameFactory);
+        btree_2.create(fileId_2);
+        btree_2.open(fileId_2);
+
+        InDiskTreeInfo info_2 = new InDiskTreeInfo(btree_2);
+        lsmtree.inDiskTreeInfoList.add(info_2);
+
+        LOGGER.info("INSERTING INTO TREE");
+
+        // ByteBuffer
+        frame = ctx.allocateFrame();
+        // FrameTupleAppender
+        appender = new FrameTupleAppender(ctx.getFrameSize());
+        // ArrayTupleBuilder
+        tb = new ArrayTupleBuilder(fieldCount);
+        // DataOutput
+        dos = tb.getDataOutput();
+
+        recDesc = new RecordDescriptor(recDescSers);
+
+        accessor.reset(frame);
+
+        tuple = new FrameTupleReference();
+
+        ITreeIndexAccessor indexAccessor_2 = btree_2.createAccessor();
+
+        // 10000
+        for (int i = 11; i < 51; i++) {
+
+            int f0 = i;
+            int f1 = 2;
+
+            tb.reset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+            tb.addFieldEndOffset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+            tb.addFieldEndOffset();
+
+            appender.reset(frame, true);
+            appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+
+            tuple.reset(accessor, 0);
+
+            ArrayTupleReference t = new ArrayTupleReference();
+            t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+
+            if (i % 10 == 0) {
+                System.out.println("INSERTING " + i + " : " + f0 + " " + f1);
+            }
+
+            try {
+                indexAccessor_2.insert(t);
+            } catch (TreeIndexException e) {
+                e.printStackTrace();
+                System.out.println("Error: " + e);
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+        // btree_2.close();
+
+        // range search in [-1000, 1000]
+        LOGGER.info("RANGE SEARCH:");
+
+        ITreeIndexCursor rangeCursor = new LSMTreeRangeSearchCursor();
+
+        // build low and high keys
+        ArrayTupleBuilder ktb = new ArrayTupleBuilder(cmp.getKeyFieldCount());
+        DataOutput kdos = ktb.getDataOutput();
+
+        ISerializerDeserializer[] keyDescSers = { IntegerSerializerDeserializer.INSTANCE };
+        RecordDescriptor keyDesc = new RecordDescriptor(keyDescSers);
+        IFrameTupleAccessor keyAccessor = new FrameTupleAccessor(ctx.getFrameSize(), keyDesc);
+        keyAccessor.reset(frame);
+
+        appender.reset(frame, true);
+
+        // build and append low key
+        ktb.reset();
+        IntegerSerializerDeserializer.INSTANCE.serialize(-100, kdos);
+        ktb.addFieldEndOffset();
+        appender.append(ktb.getFieldEndOffsets(), ktb.getByteArray(), 0, ktb.getSize());
+
+        // build and append high key
+        ktb.reset();
+        IntegerSerializerDeserializer.INSTANCE.serialize(100, 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 = cmps;
+        MultiComparator searchCmp = new MultiComparator(searchCmps);
+
+        RangePredicate rangePred = new RangePredicate(true, lowKey, highKey, true, true, searchCmp, searchCmp);
+        lsmTreeAccessor.search(rangeCursor, rangePred);
+
+        try {
+            while (rangeCursor.hasNext()) {
+                rangeCursor.next();
+                ITupleReference frameTuple = rangeCursor.getTuple();
+                String rec = TupleUtils.printTuple(frameTuple, recDescSers);
+                if (((LSMTypeAwareTupleReference) frameTuple).isDelete()) {
+                    System.out.println("del " + rec);
+                } else {
+                    System.out.println("ins " + rec);
+                }
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+        } finally {
+            rangeCursor.close();
+        }
+
+        lsmtree.close();
+        bufferCache.closeFile(fileId);
+        memBufferCache.close();
+    }
+}
\ No newline at end of file
diff --git a/hyracks-tests/hyracks-storage-am-lsmtree-common-test/pom.xml b/hyracks-tests/hyracks-storage-am-lsmtree-common-test/pom.xml
new file mode 100644
index 0000000..8a1c3ae
--- /dev/null
+++ b/hyracks-tests/hyracks-storage-am-lsmtree-common-test/pom.xml
@@ -0,0 +1,55 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+  <groupId>edu.uci.ics.hyracks</groupId>
+  <artifactId>hyracks-storage-am-lsmtree-common-test</artifactId>
+  <version>0.2.0-SNAPSHOT</version>
+
+  <parent>
+    <groupId>edu.uci.ics.hyracks</groupId>
+    <artifactId>hyracks-tests</artifactId>
+    <version>0.2.0-SNAPSHOT</version>
+  </parent>
+
+  <build>
+    <plugins>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-compiler-plugin</artifactId>
+        <version>2.0.2</version>
+        <configuration>
+          <source>1.6</source>
+          <target>1.6</target>
+        </configuration>
+      </plugin>
+    </plugins>
+  </build>
+  <dependencies>
+  	<dependency>
+  		<groupId>junit</groupId>
+  		<artifactId>junit</artifactId>
+  		<version>4.8.1</version>
+  		<type>jar</type>
+  		<scope>test</scope>
+  	</dependency>
+  	<dependency>
+  		<groupId>edu.uci.ics.hyracks</groupId>
+  		<artifactId>hyracks-control-nc</artifactId>
+  		<version>0.2.0-SNAPSHOT</version>
+  		<scope>compile</scope>
+  	</dependency>
+  	<dependency>
+  		<groupId>edu.uci.ics.hyracks</groupId>
+  		<artifactId>hyracks-storage-am-lsmtree-common</artifactId>
+  		<version>0.2.0-SNAPSHOT</version>
+  		<type>jar</type>
+  		<scope>compile</scope>
+  	</dependency>
+  	<dependency>
+  		<groupId>edu.uci.ics.hyracks</groupId>
+  		<artifactId>hyracks-test-support</artifactId>
+  		<version>0.2.0-SNAPSHOT</version>
+  		<type>jar</type>
+  		<scope>test</scope>
+  	</dependency>
+  </dependencies>
+</project>
diff --git a/hyracks-tests/hyracks-storage-am-lsmtree-common-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsmtree/common/AbstractLSMTreeTest.java b/hyracks-tests/hyracks-storage-am-lsmtree-common-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsmtree/common/AbstractLSMTreeTest.java
new file mode 100644
index 0000000..53ce238
--- /dev/null
+++ b/hyracks-tests/hyracks-storage-am-lsmtree-common-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsmtree/common/AbstractLSMTreeTest.java
@@ -0,0 +1,76 @@
+package edu.uci.ics.hyracks.storage.am.lsmtree.btree;
+
+import java.io.File;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.Random;
+import java.util.logging.Logger;
+
+import org.junit.After;
+import org.junit.Before;
+
+import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
+import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
+import edu.uci.ics.hyracks.api.io.FileReference;
+import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
+import edu.uci.ics.hyracks.storage.common.file.IFileMapProvider;
+import edu.uci.ics.hyracks.test.support.TestStorageManagerComponentHolder;
+import edu.uci.ics.hyracks.test.support.TestUtils;
+
+public abstract class AbstractLSMTreeTest {
+    protected static final Logger LOGGER = Logger.getLogger(AbstractLSMTreeTest.class.getName());
+    public static final long RANDOM_SEED = 50;
+    
+    private static final int PAGE_SIZE = 256;
+    private static final int NUM_PAGES = 10;
+    private static final int MAX_OPEN_FILES = 10;
+    private static final int HYRACKS_FRAME_SIZE = 128;
+        
+    protected IHyracksTaskContext ctx; 
+    protected IBufferCache bufferCache;
+    protected int btreeFileId;
+    
+    protected final Random rnd = new Random();
+    protected final static SimpleDateFormat simpleDateFormat = new SimpleDateFormat("ddMMyy-hhmmssSS");
+    protected final static String tmpDir = System.getProperty("java.io.tmpdir");
+    protected final static String sep = System.getProperty("file.separator");
+    protected String fileName;    
+    
+    @Before
+    public void setUp() throws HyracksDataException {
+        fileName = tmpDir + sep + simpleDateFormat.format(new Date());
+        ctx = TestUtils.create(getHyracksFrameSize());
+        TestStorageManagerComponentHolder.init(getPageSize(), getNumPages(), getMaxOpenFiles());
+        bufferCache = TestStorageManagerComponentHolder.getBufferCache(ctx);
+        IFileMapProvider fmp = TestStorageManagerComponentHolder.getFileMapProvider(ctx);
+        FileReference file = new FileReference(new File(fileName)); 
+        bufferCache.createFile(file);
+        btreeFileId = fmp.lookupFileId(file);
+        bufferCache.openFile(btreeFileId);
+        rnd.setSeed(RANDOM_SEED);
+    }
+    
+    @After
+    public void tearDown() throws HyracksDataException {
+        bufferCache.closeFile(btreeFileId);
+        bufferCache.close();
+        File f = new File(fileName);
+        f.deleteOnExit();
+    }
+    
+    public int getPageSize() {
+        return PAGE_SIZE;
+    }
+    
+    public int getNumPages() {
+        return NUM_PAGES;
+    }
+    
+    public int getHyracksFrameSize() {
+        return HYRACKS_FRAME_SIZE;
+    }
+    
+    public int getMaxOpenFiles() {
+        return MAX_OPEN_FILES;
+    }
+}
diff --git a/hyracks-tests/hyracks-storage-am-lsmtree-common-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsmtree/common/LSMTreeDeleteTest.java b/hyracks-tests/hyracks-storage-am-lsmtree-common-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsmtree/common/LSMTreeDeleteTest.java
new file mode 100644
index 0000000..16cab5c
--- /dev/null
+++ b/hyracks-tests/hyracks-storage-am-lsmtree-common-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsmtree/common/LSMTreeDeleteTest.java
@@ -0,0 +1,1102 @@
+package edu.uci.ics.hyracks.storage.am.lsmtree.btree;
+
+import static org.junit.Assert.fail;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInput;
+import java.io.DataInputStream;
+import java.io.DataOutput;
+import java.io.File;
+import java.nio.ByteBuffer;
+
+import org.junit.Test;
+
+import edu.uci.ics.hyracks.api.comm.IFrameTupleAccessor;
+import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
+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.ITypeTraits;
+import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
+import edu.uci.ics.hyracks.api.io.FileReference;
+import edu.uci.ics.hyracks.data.std.accessors.PointableBinaryComparatorFactory;
+import edu.uci.ics.hyracks.data.std.primitive.IntegerPointable;
+import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
+import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleReference;
+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.marshalling.IntegerSerializerDeserializer;
+import edu.uci.ics.hyracks.storage.am.btree.frames.BTreeNSMInteriorFrameFactory;
+import edu.uci.ics.hyracks.storage.am.btree.frames.BTreeNSMLeafFrameFactory;
+import edu.uci.ics.hyracks.storage.am.btree.impls.RangePredicate;
+import edu.uci.ics.hyracks.storage.am.common.api.IFreePageManager;
+import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexAccessor;
+import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexCursor;
+import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexFrameFactory;
+import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexMetaDataFrameFactory;
+import edu.uci.ics.hyracks.storage.am.common.api.TreeIndexException;
+import edu.uci.ics.hyracks.storage.am.common.frames.LIFOMetaDataFrameFactory;
+import edu.uci.ics.hyracks.storage.am.common.ophelpers.MultiComparator;
+import edu.uci.ics.hyracks.storage.am.lsmtree.freepage.FreePageManagerFactory;
+import edu.uci.ics.hyracks.storage.am.lsmtree.freepage.InMemoryBufferCacheFactory;
+import edu.uci.ics.hyracks.storage.am.lsmtree.freepage.InMemoryFreePageManager;
+import edu.uci.ics.hyracks.storage.am.lsmtree.impls.BTreeFactory;
+import edu.uci.ics.hyracks.storage.am.lsmtree.impls.LSMTree;
+import edu.uci.ics.hyracks.storage.am.lsmtree.impls.LSMTreeRangeSearchCursor;
+import edu.uci.ics.hyracks.storage.am.lsmtree.tuples.LSMTypeAwareTupleWriterFactory;
+import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
+import edu.uci.ics.hyracks.storage.common.file.IFileMapManager;
+import edu.uci.ics.hyracks.storage.common.file.IFileMapProvider;
+import edu.uci.ics.hyracks.test.support.TestStorageManagerComponentHolder;
+import edu.uci.ics.hyracks.test.support.TestUtils;
+
+public class LSMTreeDeleteTest extends AbstractLSMTreeTest {
+
+    private static final int PAGE_SIZE = 256;
+    private static final int NUM_PAGES = 100;
+    private static final int MAX_OPEN_FILES = 100;
+    private static final int HYRACKS_FRAME_SIZE = 128;
+    private IHyracksTaskContext ctx = TestUtils.create(HYRACKS_FRAME_SIZE);
+
+    // BASIC DELETE TEST
+    // create a fix-length lsm tree, and do 100 deletes. That is insert 100
+    // delete nodes into the in-memory tree.
+    @Test
+    public void Test1() throws Exception {
+        // in disk
+        TestStorageManagerComponentHolder.init(PAGE_SIZE, NUM_PAGES, MAX_OPEN_FILES);
+        IBufferCache bufferCache = TestStorageManagerComponentHolder.getBufferCache(ctx);
+        IFileMapProvider fmp = TestStorageManagerComponentHolder.getFileMapProvider(ctx);
+        FileReference file = new FileReference(new File(fileName));
+        bufferCache.createFile(file);
+        int fileId = fmp.lookupFileId(file);
+        bufferCache.openFile(fileId);
+
+        // in memory
+        InMemoryBufferCacheFactory InMemBufferCacheFactory = new InMemoryBufferCacheFactory(PAGE_SIZE, NUM_PAGES);
+        IBufferCache memBufferCache = InMemBufferCacheFactory.createInMemoryBufferCache();
+
+        // declare fields
+        int fieldCount = 2;
+        ITypeTraits[] typeTraits = new ITypeTraits[fieldCount];
+        typeTraits[0] = IntegerPointable.TYPE_TRAITS;
+        typeTraits[1] = IntegerPointable.TYPE_TRAITS;
+
+        // declare keys
+        int keyFieldCount = 1;
+        IBinaryComparator[] cmps = new IBinaryComparator[keyFieldCount];
+        cmps[0] = PointableBinaryComparatorFactory.of(IntegerPointable.FACTORY).createBinaryComparator();
+
+        MultiComparator cmp = new MultiComparator(cmps);
+
+        LSMTypeAwareTupleWriterFactory insertTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(typeTraits, false);
+        LSMTypeAwareTupleWriterFactory deleteTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(typeTraits, true);
+
+        ITreeIndexFrameFactory insertLeafFrameFactory = new BTreeNSMLeafFrameFactory(insertTupleWriterFactory);
+        ITreeIndexFrameFactory deleteLeafFrameFactory = new BTreeNSMLeafFrameFactory(deleteTupleWriterFactory);
+        ITreeIndexFrameFactory interiorFrameFactory = new BTreeNSMInteriorFrameFactory(insertTupleWriterFactory);
+        ITreeIndexMetaDataFrameFactory metaFrameFactory = new LIFOMetaDataFrameFactory();
+
+        IFreePageManager memFreePageManager = new InMemoryFreePageManager(100, metaFrameFactory);
+
+        FreePageManagerFactory freePageManagerFactory = new FreePageManagerFactory(bufferCache, metaFrameFactory);
+        BTreeFactory bTreeFactory = new BTreeFactory(bufferCache, freePageManagerFactory, cmp, fieldCount,
+                interiorFrameFactory, insertLeafFrameFactory);
+
+        LSMTree lsmtree = new LSMTree(memBufferCache, bufferCache, fieldCount, cmp, memFreePageManager,
+                interiorFrameFactory, insertLeafFrameFactory, deleteLeafFrameFactory, bTreeFactory,
+                (IFileMapManager) fmp);
+
+        lsmtree.create(fileId);
+        lsmtree.open(fileId);
+
+        ByteBuffer frame = ctx.allocateFrame();
+        FrameTupleAppender appender = new FrameTupleAppender(ctx.getFrameSize());
+
+        ArrayTupleBuilder tb = new ArrayTupleBuilder(fieldCount);
+        DataOutput dos = tb.getDataOutput();
+
+        ISerializerDeserializer[] recDescSers = { IntegerSerializerDeserializer.INSTANCE,
+                IntegerSerializerDeserializer.INSTANCE };
+        RecordDescriptor recDesc = new RecordDescriptor(recDescSers);
+
+        IFrameTupleAccessor accessor = new FrameTupleAccessor(ctx.getFrameSize(), recDesc);
+        accessor.reset(frame);
+
+        FrameTupleReference tuple = new FrameTupleReference();
+
+        ITreeIndexAccessor lsmTreeAccessor = lsmtree.createAccessor();
+
+        int resultSize = 50;
+        int[][] resultArray = new int[resultSize][3];
+
+        for (int i = 0; i < resultSize; i++) {
+            resultArray[i][0] = i;
+            resultArray[i][1] = i + 1;
+            resultArray[i][2] = 1;
+        }
+
+        // delete
+        for (int i = 0; i < resultSize; i++) {
+
+            int f0 = resultArray[i][0];
+            int f1 = resultArray[i][1];
+
+            tb.reset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+            tb.addFieldEndOffset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+            tb.addFieldEndOffset();
+
+            appender.reset(frame, true);
+            appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+
+            tuple.reset(accessor, 0);
+
+            ArrayTupleReference t = new ArrayTupleReference();
+            t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+
+            try {
+                lsmTreeAccessor.delete(t);
+            } catch (TreeIndexException e) {
+                System.out.println("test01:" + e);
+                e.printStackTrace();
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+
+        // scan
+        ITreeIndexCursor scanCursor = new LSMTreeRangeSearchCursor();
+        RangePredicate nullPred = new RangePredicate(true, null, null, true, true, null, null);
+        lsmTreeAccessor.search(scanCursor, nullPred);
+
+        try {
+            int scanTupleIndex = 0;
+            int arrayIndex = 0;
+            Object o = null;
+            while (scanCursor.hasNext()) {
+                scanCursor.next();
+                ITupleReference frameTuple = scanCursor.getTuple();
+                int numPrintFields = Math.min(frameTuple.getFieldCount(), recDescSers.length);
+
+                for (int i = 0; i < numPrintFields; i++) {
+                    ByteArrayInputStream inStream = new ByteArrayInputStream(frameTuple.getFieldData(i),
+                            frameTuple.getFieldStart(i), frameTuple.getFieldLength(i));
+                    DataInput dataIn = new DataInputStream(inStream);
+                    o = recDescSers[i].deserialize(dataIn);
+
+                }
+                while (resultArray[arrayIndex][2] != 0) {
+                    arrayIndex++;
+                }
+                if (Integer.parseInt(o.toString()) != resultArray[arrayIndex][1]) {
+                    fail("Input value and Output value doesn't match on the " + scanTupleIndex + " tuple\n");
+                }
+                scanTupleIndex++;
+                arrayIndex++;
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+        } finally {
+            scanCursor.close();
+        }
+
+        lsmtree.close();
+        bufferCache.closeFile(fileId);
+        memBufferCache.close();
+    }
+
+    // INSERT-DELETE TEST
+    // create a fix-length lsm tree. First, do 100 insertions,
+    // and then do 50 deletions which has the same 50 keys which are part of the
+    // insertions.
+    @Test
+    public void Test2() throws Exception {
+        // in disk
+        TestStorageManagerComponentHolder.init(PAGE_SIZE, NUM_PAGES, MAX_OPEN_FILES);
+        IBufferCache bufferCache = TestStorageManagerComponentHolder.getBufferCache(ctx);
+        IFileMapProvider fmp = TestStorageManagerComponentHolder.getFileMapProvider(ctx);
+        FileReference file = new FileReference(new File(fileName));
+        bufferCache.createFile(file);
+        int fileId = fmp.lookupFileId(file);
+        bufferCache.openFile(fileId);
+
+        // in memory
+        InMemoryBufferCacheFactory InMemBufferCacheFactory = new InMemoryBufferCacheFactory(PAGE_SIZE, NUM_PAGES);
+        IBufferCache memBufferCache = InMemBufferCacheFactory.createInMemoryBufferCache();
+
+        // declare fields
+        int fieldCount = 2;
+        ITypeTraits[] typeTraits = new ITypeTraits[fieldCount];
+        typeTraits[0] = IntegerPointable.TYPE_TRAITS;
+        typeTraits[1] = IntegerPointable.TYPE_TRAITS;
+
+        // declare keys
+        int keyFieldCount = 1;
+        IBinaryComparator[] cmps = new IBinaryComparator[keyFieldCount];
+        cmps[0] = PointableBinaryComparatorFactory.of(IntegerPointable.FACTORY).createBinaryComparator();
+
+        MultiComparator cmp = new MultiComparator(cmps);
+
+        LSMTypeAwareTupleWriterFactory insertTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(typeTraits, false);
+        LSMTypeAwareTupleWriterFactory deleteTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(typeTraits, true);
+
+        ITreeIndexFrameFactory insertLeafFrameFactory = new BTreeNSMLeafFrameFactory(insertTupleWriterFactory);
+        ITreeIndexFrameFactory deleteLeafFrameFactory = new BTreeNSMLeafFrameFactory(deleteTupleWriterFactory);
+        ITreeIndexFrameFactory interiorFrameFactory = new BTreeNSMInteriorFrameFactory(insertTupleWriterFactory);
+        ITreeIndexMetaDataFrameFactory metaFrameFactory = new LIFOMetaDataFrameFactory();
+
+        IFreePageManager memFreePageManager = new InMemoryFreePageManager(100, metaFrameFactory);
+
+        FreePageManagerFactory freePageManagerFactory = new FreePageManagerFactory(bufferCache, metaFrameFactory);
+        BTreeFactory bTreeFactory = new BTreeFactory(bufferCache, freePageManagerFactory, cmp, fieldCount,
+                interiorFrameFactory, insertLeafFrameFactory);
+
+        LSMTree lsmtree = new LSMTree(memBufferCache, bufferCache, fieldCount, cmp, memFreePageManager,
+                interiorFrameFactory, insertLeafFrameFactory, deleteLeafFrameFactory, bTreeFactory,
+                (IFileMapManager) fmp);
+
+        lsmtree.create(fileId);
+        lsmtree.open(fileId);
+
+        ByteBuffer frame = ctx.allocateFrame();
+        FrameTupleAppender appender = new FrameTupleAppender(ctx.getFrameSize());
+
+        ArrayTupleBuilder tb = new ArrayTupleBuilder(fieldCount);
+        DataOutput dos = tb.getDataOutput();
+
+        ISerializerDeserializer[] recDescSers = { IntegerSerializerDeserializer.INSTANCE,
+                IntegerSerializerDeserializer.INSTANCE };
+        RecordDescriptor recDesc = new RecordDescriptor(recDescSers);
+
+        IFrameTupleAccessor accessor = new FrameTupleAccessor(ctx.getFrameSize(), recDesc);
+        accessor.reset(frame);
+
+        FrameTupleReference tuple = new FrameTupleReference();
+
+        int resultSize = 100;
+        int deleteStartPosition = 50;
+        int[][] resultArray = new int[resultSize][3];
+
+        for (int i = 0; i < resultSize; i++) {
+            resultArray[i][0] = i;
+            resultArray[i][1] = i + 1;
+            resultArray[i][2] = 0;
+        }
+
+        // insert
+        ITreeIndexAccessor lsmTreeAccessor = lsmtree.createAccessor();
+        for (int i = 0; i < resultSize; i++) {
+
+            int f0 = resultArray[i][0];
+            int f1 = resultArray[i][1];
+
+            tb.reset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+            tb.addFieldEndOffset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+            tb.addFieldEndOffset();
+
+            appender.reset(frame, true);
+            appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+
+            tuple.reset(accessor, 0);
+
+            ArrayTupleReference t = new ArrayTupleReference();
+            t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+
+            try {
+                lsmTreeAccessor.insert(t);
+            } catch (TreeIndexException e) {
+                System.out.println("test02:" + e);
+                e.printStackTrace();
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+
+        // delete
+        for (int i = deleteStartPosition; i < resultSize; i++) {
+
+            int f0 = resultArray[i][0];
+            int f1 = ++resultArray[i][1];
+            resultArray[i][2] = 1;
+
+            tb.reset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+            tb.addFieldEndOffset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+            tb.addFieldEndOffset();
+
+            appender.reset(frame, true);
+            appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+
+            tuple.reset(accessor, 0);
+
+            ArrayTupleReference t = new ArrayTupleReference();
+            t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+
+            try {
+                lsmTreeAccessor.delete(t);
+            } catch (TreeIndexException e) {
+                System.out.println("test02:" + e);
+                e.printStackTrace();
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+
+        // scan
+        ITreeIndexCursor scanCursor = new LSMTreeRangeSearchCursor();
+        RangePredicate nullPred = new RangePredicate(true, null, null, true, true, null, null);
+        lsmTreeAccessor.search(scanCursor, nullPred);
+
+        try {
+            int scanTupleIndex = 0;
+            int arrayIndex = 0;
+            Object o = null;
+            while (scanCursor.hasNext()) {
+                scanCursor.next();
+                ITupleReference frameTuple = scanCursor.getTuple();
+                int numPrintFields = Math.min(frameTuple.getFieldCount(), recDescSers.length);
+
+                for (int i = 0; i < numPrintFields; i++) {
+                    ByteArrayInputStream inStream = new ByteArrayInputStream(frameTuple.getFieldData(i),
+                            frameTuple.getFieldStart(i), frameTuple.getFieldLength(i));
+                    DataInput dataIn = new DataInputStream(inStream);
+                    o = recDescSers[i].deserialize(dataIn);
+
+                }
+                while (resultArray[arrayIndex][2] != 0) {
+                    arrayIndex++;
+                }
+                if (Integer.parseInt(o.toString()) != resultArray[arrayIndex][1]) {
+                    fail("Input value and Output value doesn't match on the " + scanTupleIndex + " tuple\n");
+                }
+
+                scanTupleIndex++;
+                arrayIndex++;
+
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+        } finally {
+            scanCursor.close();
+        }
+
+        lsmtree.close();
+        bufferCache.closeFile(fileId);
+        memBufferCache.close();
+    }
+
+    // DELETE->INSERT TEST
+    // create a fix-length lsm tree. First, do 100 deletions,
+    // and then do 50 insertions which has the same 50 keys which are part of
+    // the deletions.
+    @Test
+    public void Test3() throws Exception {
+        System.out.println("TEST3");
+        // in disk
+        TestStorageManagerComponentHolder.init(PAGE_SIZE, NUM_PAGES, MAX_OPEN_FILES);
+        IBufferCache bufferCache = TestStorageManagerComponentHolder.getBufferCache(ctx);
+        IFileMapProvider fmp = TestStorageManagerComponentHolder.getFileMapProvider(ctx);
+        FileReference file = new FileReference(new File(fileName));
+        bufferCache.createFile(file);
+        int fileId = fmp.lookupFileId(file);
+        bufferCache.openFile(fileId);
+
+        // in mem
+        InMemoryBufferCacheFactory InMemBufferCacheFactory = new InMemoryBufferCacheFactory(PAGE_SIZE, NUM_PAGES);
+        IBufferCache memBufferCache = InMemBufferCacheFactory.createInMemoryBufferCache();
+
+        // declare fields
+        int fieldCount = 2;
+        ITypeTraits[] typeTraits = new ITypeTraits[fieldCount];
+        typeTraits[0] = IntegerPointable.TYPE_TRAITS;
+        typeTraits[1] = IntegerPointable.TYPE_TRAITS;
+
+        // declare keys
+        int keyFieldCount = 1;
+        IBinaryComparator[] cmps = new IBinaryComparator[keyFieldCount];
+        cmps[0] = PointableBinaryComparatorFactory.of(IntegerPointable.FACTORY).createBinaryComparator();
+
+        MultiComparator cmp = new MultiComparator(cmps);
+
+        LSMTypeAwareTupleWriterFactory insertTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(typeTraits, false);
+        LSMTypeAwareTupleWriterFactory deleteTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(typeTraits, true);
+
+        ITreeIndexFrameFactory insertLeafFrameFactory = new BTreeNSMLeafFrameFactory(insertTupleWriterFactory);
+        ITreeIndexFrameFactory deleteLeafFrameFactory = new BTreeNSMLeafFrameFactory(deleteTupleWriterFactory);
+        // change
+        ITreeIndexFrameFactory interiorFrameFactory = new BTreeNSMInteriorFrameFactory(insertTupleWriterFactory);
+        ITreeIndexMetaDataFrameFactory metaFrameFactory = new LIFOMetaDataFrameFactory();
+
+        IFreePageManager memFreePageManager = new InMemoryFreePageManager(100, metaFrameFactory);
+
+        FreePageManagerFactory freePageManagerFactory = new FreePageManagerFactory(bufferCache, metaFrameFactory);
+        BTreeFactory bTreeFactory = new BTreeFactory(bufferCache, freePageManagerFactory, cmp, fieldCount,
+                interiorFrameFactory, insertLeafFrameFactory);
+
+        LSMTree lsmtree = new LSMTree(memBufferCache, bufferCache, fieldCount, cmp, memFreePageManager,
+                interiorFrameFactory, insertLeafFrameFactory, deleteLeafFrameFactory, bTreeFactory,
+                (IFileMapManager) fmp);
+
+        lsmtree.create(fileId);
+        lsmtree.open(fileId);
+
+        ByteBuffer frame = ctx.allocateFrame();
+        FrameTupleAppender appender = new FrameTupleAppender(ctx.getFrameSize());
+
+        ArrayTupleBuilder tb = new ArrayTupleBuilder(fieldCount);
+        DataOutput dos = tb.getDataOutput();
+
+        ISerializerDeserializer[] recDescSers = { IntegerSerializerDeserializer.INSTANCE,
+                IntegerSerializerDeserializer.INSTANCE };
+        RecordDescriptor recDesc = new RecordDescriptor(recDescSers);
+
+        IFrameTupleAccessor accessor = new FrameTupleAccessor(ctx.getFrameSize(), recDesc);
+        accessor.reset(frame);
+
+        FrameTupleReference tuple = new FrameTupleReference();
+
+        int resultSize = 100;
+        int insertStartPosition = 50;
+        int[][] resultArray = new int[resultSize][3];
+
+        for (int i = 0; i < resultSize; i++) {
+            resultArray[i][0] = i;
+            resultArray[i][1] = i + 1;
+            resultArray[i][2] = 1;
+        }
+
+        // delete
+        ITreeIndexAccessor lsmTreeAccessor = lsmtree.createAccessor();
+        for (int i = 0; i < resultSize; i++) {
+
+            int f0 = resultArray[i][0];
+            int f1 = resultArray[i][1];
+
+            tb.reset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+            tb.addFieldEndOffset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+            tb.addFieldEndOffset();
+
+            appender.reset(frame, true);
+            appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+
+            tuple.reset(accessor, 0);
+
+            ArrayTupleReference t = new ArrayTupleReference();
+            t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+
+            try {
+                lsmTreeAccessor.delete(t);
+            } catch (TreeIndexException e) {
+                System.out.println("test03:" + e);
+                e.printStackTrace();
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+
+        // insert
+        for (int i = insertStartPosition; i < resultSize; i++) {
+
+            int f0 = resultArray[i][0];
+            int f1 = ++resultArray[i][1];
+            resultArray[i][2] = 0;
+
+            tb.reset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+            tb.addFieldEndOffset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+            tb.addFieldEndOffset();
+
+            appender.reset(frame, true);
+            appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+
+            tuple.reset(accessor, 0);
+
+            ArrayTupleReference t = new ArrayTupleReference();
+            t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+
+            try {
+                lsmTreeAccessor.insert(t);
+            } catch (TreeIndexException e) {
+                System.out.println("test03:" + e);
+                e.printStackTrace();
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+
+        // scan
+        ITreeIndexCursor scanCursor = new LSMTreeRangeSearchCursor();
+        RangePredicate nullPred = new RangePredicate(true, null, null, true, true, null, null);
+        lsmTreeAccessor.search(scanCursor, nullPred);
+
+        try {
+            int scanTupleIndex = 0;
+            int arrayIndex = 0;
+            Object o = null;
+            while (scanCursor.hasNext()) {
+                scanCursor.next();
+                ITupleReference frameTuple = scanCursor.getTuple();
+                int numPrintFields = Math.min(frameTuple.getFieldCount(), recDescSers.length);
+
+                for (int i = 0; i < numPrintFields; i++) {
+                    ByteArrayInputStream inStream = new ByteArrayInputStream(frameTuple.getFieldData(i),
+                            frameTuple.getFieldStart(i), frameTuple.getFieldLength(i));
+                    DataInput dataIn = new DataInputStream(inStream);
+                    o = recDescSers[i].deserialize(dataIn);
+                }
+                while (resultArray[arrayIndex][2] != 0) {
+                    arrayIndex++;
+                }
+                if (Integer.parseInt(o.toString()) != resultArray[arrayIndex][1]) {
+                    fail("Input value and Output value doesn't match on the " + scanTupleIndex + " tuple\n");
+                }
+
+                scanTupleIndex++;
+                arrayIndex++;
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+        } finally {
+            scanCursor.close();
+        }
+
+        lsmtree.close();
+        bufferCache.closeFile(fileId);
+        memBufferCache.close();
+    }
+
+    // TEST DELETION and PageAllocationException
+    // create a fix-length lsm tree. First, do 811 deletions,
+    // the page will be run out on the 810th deletions, if there is any
+    // exception returns, the test case fails.
+    @Test
+    public void Test4() throws Exception {
+        // in disk
+        TestStorageManagerComponentHolder.init(PAGE_SIZE, NUM_PAGES, MAX_OPEN_FILES);
+        IBufferCache bufferCache = TestStorageManagerComponentHolder.getBufferCache(ctx);
+        IFileMapProvider fmp = TestStorageManagerComponentHolder.getFileMapProvider(ctx);
+        FileReference file = new FileReference(new File(fileName));
+        bufferCache.createFile(file);
+        int fileId = fmp.lookupFileId(file);
+        bufferCache.openFile(fileId);
+
+        // in memory
+        InMemoryBufferCacheFactory InMemBufferCacheFactory = new InMemoryBufferCacheFactory(PAGE_SIZE, NUM_PAGES);
+        IBufferCache memBufferCache = InMemBufferCacheFactory.createInMemoryBufferCache();
+
+        // declare fields
+        int fieldCount = 2;
+        ITypeTraits[] typeTraits = new ITypeTraits[fieldCount];
+        typeTraits[0] = IntegerPointable.TYPE_TRAITS;
+        typeTraits[1] = IntegerPointable.TYPE_TRAITS;
+
+        // declare keys
+        int keyFieldCount = 1;
+        IBinaryComparator[] cmps = new IBinaryComparator[keyFieldCount];
+        cmps[0] = PointableBinaryComparatorFactory.of(IntegerPointable.FACTORY).createBinaryComparator();
+
+        MultiComparator cmp = new MultiComparator(cmps);
+
+        LSMTypeAwareTupleWriterFactory insertTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(typeTraits, false);
+        LSMTypeAwareTupleWriterFactory deleteTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(typeTraits, true);
+
+        ITreeIndexFrameFactory insertLeafFrameFactory = new BTreeNSMLeafFrameFactory(insertTupleWriterFactory);
+        ITreeIndexFrameFactory deleteLeafFrameFactory = new BTreeNSMLeafFrameFactory(deleteTupleWriterFactory);
+        ITreeIndexFrameFactory interiorFrameFactory = new BTreeNSMInteriorFrameFactory(insertTupleWriterFactory);
+        ITreeIndexMetaDataFrameFactory metaFrameFactory = new LIFOMetaDataFrameFactory();
+
+        IFreePageManager memFreePageManager = new InMemoryFreePageManager(100, metaFrameFactory);
+
+        FreePageManagerFactory freePageManagerFactory = new FreePageManagerFactory(bufferCache, metaFrameFactory);
+        BTreeFactory bTreeFactory = new BTreeFactory(bufferCache, freePageManagerFactory, cmp, fieldCount,
+                interiorFrameFactory, insertLeafFrameFactory);
+
+        // For the Flush Mechanism
+        LSMTree lsmtree = new LSMTree(memBufferCache, bufferCache, fieldCount, cmp, memFreePageManager,
+                interiorFrameFactory, insertLeafFrameFactory, deleteLeafFrameFactory, bTreeFactory,
+                (IFileMapManager) fmp);
+
+        lsmtree.create(fileId);
+        lsmtree.open(fileId);
+
+        ByteBuffer frame = ctx.allocateFrame();
+        FrameTupleAppender appender = new FrameTupleAppender(ctx.getFrameSize());
+
+        ArrayTupleBuilder tb = new ArrayTupleBuilder(fieldCount);
+        DataOutput dos = tb.getDataOutput();
+
+        ISerializerDeserializer[] recDescSers = { IntegerSerializerDeserializer.INSTANCE,
+                IntegerSerializerDeserializer.INSTANCE };
+        RecordDescriptor recDesc = new RecordDescriptor(recDescSers);
+
+        IFrameTupleAccessor accessor = new FrameTupleAccessor(ctx.getFrameSize(), recDesc);
+        accessor.reset(frame);
+
+        FrameTupleReference tuple = new FrameTupleReference();
+
+        int resultSize = 811;
+        int[][] resultArray = new int[resultSize][2];
+
+        for (int i = 0; i < resultSize; i++) {
+            resultArray[i][0] = i;
+            resultArray[i][1] = i + 1;
+        }
+
+        // delete
+        ITreeIndexAccessor lsmTreeAccessor = lsmtree.createAccessor();
+        for (int i = 0; i < resultSize; i++) {
+
+            int f0 = resultArray[i][0];
+            int f1 = resultArray[i][1];
+
+            tb.reset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+            tb.addFieldEndOffset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+            tb.addFieldEndOffset();
+
+            appender.reset(frame, true);
+            appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+
+            tuple.reset(accessor, 0);
+
+            ArrayTupleReference t = new ArrayTupleReference();
+            t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+
+            try {
+                lsmTreeAccessor.delete(t);
+            } catch (TreeIndexException e) {
+                System.out.println("test04:" + e);
+                e.printStackTrace();
+                fail("test04: Catch TreeIndexException" + e);
+            } catch (Exception e) {
+                e.printStackTrace();
+                fail("test04: Catch Other Exceptions" + e);
+            }
+        }
+    }
+
+    // DELETE -> DELETE
+    // create a fix-length lsm tree. First, do 100 deletions,
+    // and then do 50 deletions which has the same 50 keys which are part of the
+    // first deletions.
+    @Test
+    public void Test5() throws Exception {
+        // in disk
+        TestStorageManagerComponentHolder.init(PAGE_SIZE, NUM_PAGES, MAX_OPEN_FILES);
+        IBufferCache bufferCache = TestStorageManagerComponentHolder.getBufferCache(ctx);
+        IFileMapProvider fmp = TestStorageManagerComponentHolder.getFileMapProvider(ctx);
+        FileReference file = new FileReference(new File(fileName));
+        bufferCache.createFile(file);
+        int fileId = fmp.lookupFileId(file);
+        bufferCache.openFile(fileId);
+
+        // in memory
+        InMemoryBufferCacheFactory InMemBufferCacheFactory = new InMemoryBufferCacheFactory(PAGE_SIZE, NUM_PAGES);
+        IBufferCache memBufferCache = InMemBufferCacheFactory.createInMemoryBufferCache();
+
+        // declare fields
+        int fieldCount = 2;
+        ITypeTraits[] typeTraits = new ITypeTraits[fieldCount];
+        typeTraits[0] = IntegerPointable.TYPE_TRAITS;
+        typeTraits[1] = IntegerPointable.TYPE_TRAITS;
+
+        // declare keys
+        int keyFieldCount = 1;
+        IBinaryComparator[] cmps = new IBinaryComparator[keyFieldCount];
+        cmps[0] = PointableBinaryComparatorFactory.of(IntegerPointable.FACTORY).createBinaryComparator();
+
+        MultiComparator cmp = new MultiComparator(cmps);
+
+        LSMTypeAwareTupleWriterFactory insertTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(typeTraits, false);
+        LSMTypeAwareTupleWriterFactory deleteTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(typeTraits, true);
+
+        ITreeIndexFrameFactory insertLeafFrameFactory = new BTreeNSMLeafFrameFactory(insertTupleWriterFactory);
+        ITreeIndexFrameFactory deleteLeafFrameFactory = new BTreeNSMLeafFrameFactory(deleteTupleWriterFactory);
+        ITreeIndexFrameFactory interiorFrameFactory = new BTreeNSMInteriorFrameFactory(insertTupleWriterFactory);
+        ITreeIndexMetaDataFrameFactory metaFrameFactory = new LIFOMetaDataFrameFactory();
+
+        IFreePageManager memFreePageManager = new InMemoryFreePageManager(100, metaFrameFactory);
+
+        FreePageManagerFactory freePageManagerFactory = new FreePageManagerFactory(bufferCache, metaFrameFactory);
+        BTreeFactory bTreeFactory = new BTreeFactory(bufferCache, freePageManagerFactory, cmp, fieldCount,
+                interiorFrameFactory, insertLeafFrameFactory);
+
+        LSMTree lsmtree = new LSMTree(memBufferCache, bufferCache, fieldCount, cmp, memFreePageManager,
+                interiorFrameFactory, insertLeafFrameFactory, deleteLeafFrameFactory, bTreeFactory,
+                (IFileMapManager) fmp);
+
+        lsmtree.create(fileId);
+        lsmtree.open(fileId);
+
+        ByteBuffer frame = ctx.allocateFrame();
+        FrameTupleAppender appender = new FrameTupleAppender(ctx.getFrameSize());
+
+        ArrayTupleBuilder tb = new ArrayTupleBuilder(fieldCount);
+        DataOutput dos = tb.getDataOutput();
+
+        ISerializerDeserializer[] recDescSers = { IntegerSerializerDeserializer.INSTANCE,
+                IntegerSerializerDeserializer.INSTANCE };
+        RecordDescriptor recDesc = new RecordDescriptor(recDescSers);
+
+        IFrameTupleAccessor accessor = new FrameTupleAccessor(ctx.getFrameSize(), recDesc);
+        accessor.reset(frame);
+
+        FrameTupleReference tuple = new FrameTupleReference();
+
+        int resultSize = 100;
+        int insertStartPosition = 50;
+        int[][] resultArray = new int[resultSize][3];
+
+        for (int i = 0; i < resultSize; i++) {
+            resultArray[i][0] = i;
+            resultArray[i][1] = i + 1;
+            resultArray[i][2] = 1;
+        }
+
+        // First deletion part
+        ITreeIndexAccessor lsmTreeAccessor = lsmtree.createAccessor();
+        for (int i = 0; i < resultSize; i++) {
+
+            int f0 = resultArray[i][0];
+            int f1 = resultArray[i][1];
+
+            tb.reset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+            tb.addFieldEndOffset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+            tb.addFieldEndOffset();
+
+            appender.reset(frame, true);
+            appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+
+            tuple.reset(accessor, 0);
+
+            ArrayTupleReference t = new ArrayTupleReference();
+            t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+
+            try {
+                lsmTreeAccessor.delete(t);
+            } catch (TreeIndexException e) {
+                System.out.println("test05:" + e);
+                e.printStackTrace();
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+
+        // Second delete part
+        for (int i = insertStartPosition; i < resultSize; i++) {
+
+            int f0 = resultArray[i][0];
+            int f1 = ++resultArray[i][1];
+            resultArray[i][2] = 1;
+
+            tb.reset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+            tb.addFieldEndOffset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+            tb.addFieldEndOffset();
+
+            appender.reset(frame, true);
+            appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+
+            tuple.reset(accessor, 0);
+
+            ArrayTupleReference t = new ArrayTupleReference();
+            t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+
+            try {
+                lsmTreeAccessor.insert(t);
+            } catch (TreeIndexException e) {
+                System.out.println("test05:" + e);
+                e.printStackTrace();
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+
+        // scan
+        ITreeIndexCursor scanCursor = new LSMTreeRangeSearchCursor();
+        RangePredicate nullPred = new RangePredicate(true, null, null, true, true, null, null);
+        lsmTreeAccessor.search(scanCursor, nullPred);
+
+        try {
+            int scanTupleIndex = 0;
+            int arrayIndex = 0;
+            Object o = null;
+            while (scanCursor.hasNext()) {
+                scanCursor.next();
+                ITupleReference frameTuple = scanCursor.getTuple();
+                int numPrintFields = Math.min(frameTuple.getFieldCount(), recDescSers.length);
+
+                for (int i = 0; i < numPrintFields; i++) {
+                    ByteArrayInputStream inStream = new ByteArrayInputStream(frameTuple.getFieldData(i),
+                            frameTuple.getFieldStart(i), frameTuple.getFieldLength(i));
+                    DataInput dataIn = new DataInputStream(inStream);
+                    o = recDescSers[i].deserialize(dataIn);
+
+                }
+                while (resultArray[arrayIndex][2] != 0) {
+                    arrayIndex++;
+                }
+                if (Integer.parseInt(o.toString()) != resultArray[arrayIndex][1]) {
+                    fail("Input value and Output value doesn't match on the " + scanTupleIndex + " tuple\n");
+                }
+
+                scanTupleIndex++;
+                arrayIndex++;
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+        } finally {
+            scanCursor.close();
+        }
+
+        lsmtree.close();
+        bufferCache.closeFile(fileId);
+        memBufferCache.close();
+    }
+
+    // INSERT -> DELETE -> INSERT
+    // create a fix-length lsm tree. Do the insertion, deletion and insertion.
+    // the final result will be
+    // | 0~9 | 10~19 | 20~39 | 40~59 | 60~79 | 80~99 |
+    // | f1=10 | f1=9 | f1=8 | f1=7 | f1=6 | f1=5 |
+    // | Insert| Insert| Delete| Delete| Insert| Insert|
+    @Test
+    public void Test6() throws Exception {
+
+        // in disk
+        TestStorageManagerComponentHolder.init(PAGE_SIZE, NUM_PAGES, MAX_OPEN_FILES);
+        IBufferCache bufferCache = TestStorageManagerComponentHolder.getBufferCache(ctx);
+        IFileMapProvider fmp = TestStorageManagerComponentHolder.getFileMapProvider(ctx);
+        FileReference file = new FileReference(new File(fileName));
+        bufferCache.createFile(file);
+        int fileId = fmp.lookupFileId(file);
+        bufferCache.openFile(fileId);
+
+        // in mem
+        InMemoryBufferCacheFactory InMemBufferCacheFactory = new InMemoryBufferCacheFactory(PAGE_SIZE, NUM_PAGES);
+        IBufferCache memBufferCache = InMemBufferCacheFactory.createInMemoryBufferCache();
+
+        // declare fields
+        int fieldCount = 2;
+        ITypeTraits[] typeTraits = new ITypeTraits[fieldCount];
+        typeTraits[0] = IntegerPointable.TYPE_TRAITS;
+        typeTraits[1] = IntegerPointable.TYPE_TRAITS;
+
+        // declare keys
+        int keyFieldCount = 1;
+        IBinaryComparator[] cmps = new IBinaryComparator[keyFieldCount];
+        cmps[0] = PointableBinaryComparatorFactory.of(IntegerPointable.FACTORY).createBinaryComparator();
+
+        MultiComparator cmp = new MultiComparator(cmps);
+
+        LSMTypeAwareTupleWriterFactory insertTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(typeTraits, false);
+        LSMTypeAwareTupleWriterFactory deleteTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(typeTraits, true);
+
+        ITreeIndexFrameFactory insertLeafFrameFactory = new BTreeNSMLeafFrameFactory(insertTupleWriterFactory);
+        ITreeIndexFrameFactory deleteLeafFrameFactory = new BTreeNSMLeafFrameFactory(deleteTupleWriterFactory);
+        ITreeIndexFrameFactory interiorFrameFactory = new BTreeNSMInteriorFrameFactory(insertTupleWriterFactory);
+        ITreeIndexMetaDataFrameFactory metaFrameFactory = new LIFOMetaDataFrameFactory();
+
+        IFreePageManager memFreePageManager = new InMemoryFreePageManager(100, metaFrameFactory);
+
+        FreePageManagerFactory freePageManagerFactory = new FreePageManagerFactory(bufferCache, metaFrameFactory);
+        BTreeFactory bTreeFactory = new BTreeFactory(bufferCache, freePageManagerFactory, cmp, fieldCount,
+                interiorFrameFactory, insertLeafFrameFactory);
+
+        LSMTree lsmtree = new LSMTree(memBufferCache, bufferCache, fieldCount, cmp, memFreePageManager,
+                interiorFrameFactory, insertLeafFrameFactory, deleteLeafFrameFactory, bTreeFactory,
+                (IFileMapManager) fmp);
+
+        lsmtree.create(fileId);
+        lsmtree.open(fileId);
+
+        ByteBuffer frame = ctx.allocateFrame();
+        FrameTupleAppender appender = new FrameTupleAppender(ctx.getFrameSize());
+
+        ArrayTupleBuilder tb = new ArrayTupleBuilder(fieldCount);
+        DataOutput dos = tb.getDataOutput();
+
+        ISerializerDeserializer[] recDescSers = { IntegerSerializerDeserializer.INSTANCE,
+                IntegerSerializerDeserializer.INSTANCE };
+        RecordDescriptor recDesc = new RecordDescriptor(recDescSers);
+
+        IFrameTupleAccessor accessor = new FrameTupleAccessor(ctx.getFrameSize(), recDesc);
+        accessor.reset(frame);
+
+        FrameTupleReference tuple = new FrameTupleReference();
+
+        ITreeIndexAccessor lsmTreeAccessor = lsmtree.createAccessor();
+
+        int resultSize = 180;
+        int[][] resultArray = new int[resultSize][3];
+
+        // insert
+        for (int i = 0; i < 180; i++) {
+            int f0 = i % 100;
+            int f1;
+            if (i >= 100) {
+                f1 = 6;
+            } else {
+                f1 = 5;
+            }
+
+            resultArray[f0][0] = f0;
+            resultArray[f0][1] = f1;
+            resultArray[f0][2] = 0;
+
+            tb.reset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+            tb.addFieldEndOffset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+            tb.addFieldEndOffset();
+
+            appender.reset(frame, true);
+            appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+
+            tuple.reset(accessor, 0);
+
+            ArrayTupleReference t = new ArrayTupleReference();
+            t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+
+            try {
+                lsmTreeAccessor.insert(t);
+            } catch (TreeIndexException e) {
+                System.out.println("test06:" + e);
+                e.printStackTrace();
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+
+        // delete
+        for (int i = 0; i < 100; i++) {
+            int f0 = i % 60;
+            int f1;
+            if (i >= 60) {
+                f1 = 8;
+            } else {
+                f1 = 7;
+            }
+
+            resultArray[f0][0] = f0;
+            resultArray[f0][1] = f1;
+            resultArray[f0][2] = 1;
+
+            tb.reset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+            tb.addFieldEndOffset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+            tb.addFieldEndOffset();
+
+            appender.reset(frame, true);
+            appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+
+            tuple.reset(accessor, 0);
+
+            ArrayTupleReference t = new ArrayTupleReference();
+            t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+
+            try {
+                lsmTreeAccessor.delete(t);
+            } catch (TreeIndexException e) {
+                System.out.println("test06:" + e);
+                e.printStackTrace();
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+
+        }
+
+        // reinsert
+        for (int i = 0; i < 30; i++) {
+            int f0 = i % 20;
+            int f1;
+            if (i >= 20) {
+                f1 = 10;
+            } else {
+                f1 = 9;
+            }
+
+            resultArray[f0][0] = f0;
+            resultArray[f0][1] = f1;
+            resultArray[f0][2] = 0;
+
+            tb.reset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+            tb.addFieldEndOffset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+            tb.addFieldEndOffset();
+
+            appender.reset(frame, true);
+            appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+
+            tuple.reset(accessor, 0);
+
+            ArrayTupleReference t = new ArrayTupleReference();
+            t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+
+            try {
+                lsmTreeAccessor.insert(t);
+            } catch (TreeIndexException e) {
+                System.out.println("test06:" + e);
+                e.printStackTrace();
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+
+        // scan
+        ITreeIndexCursor scanCursor = new LSMTreeRangeSearchCursor();
+        RangePredicate nullPred = new RangePredicate(true, null, null, true, true, null, null);
+        lsmTreeAccessor.search(scanCursor, nullPred);
+
+        try {
+            int scanTupleIndex = 0;
+            int arrayIndex = 0;
+            Object o = null;
+            while (scanCursor.hasNext()) {
+                scanCursor.next();
+                ITupleReference frameTuple = scanCursor.getTuple();
+                int numPrintFields = Math.min(frameTuple.getFieldCount(), recDescSers.length);
+
+                for (int i = 0; i < numPrintFields; i++) {
+                    ByteArrayInputStream inStream = new ByteArrayInputStream(frameTuple.getFieldData(i),
+                            frameTuple.getFieldStart(i), frameTuple.getFieldLength(i));
+                    DataInput dataIn = new DataInputStream(inStream);
+                    o = recDescSers[i].deserialize(dataIn);
+                }
+                while (resultArray[arrayIndex][2] != 0) {
+                    arrayIndex++;
+                }
+                if (Integer.parseInt(o.toString()) != resultArray[arrayIndex][1]) {
+                    fail("Input value and Output value doesn't match on the " + scanTupleIndex + " tuple\n");
+                }
+
+                scanTupleIndex++;
+                arrayIndex++;
+            }
+
+        } catch (Exception e) {
+            e.printStackTrace();
+        } finally {
+            scanCursor.close();
+        }
+
+        lsmtree.close();
+        bufferCache.closeFile(fileId);
+        memBufferCache.close();
+    }
+}
diff --git a/hyracks-tests/hyracks-storage-am-lsmtree-common-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsmtree/common/LSMTreeFlushTest.java b/hyracks-tests/hyracks-storage-am-lsmtree-common-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsmtree/common/LSMTreeFlushTest.java
new file mode 100644
index 0000000..079fc23
--- /dev/null
+++ b/hyracks-tests/hyracks-storage-am-lsmtree-common-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsmtree/common/LSMTreeFlushTest.java
@@ -0,0 +1,755 @@
+package edu.uci.ics.hyracks.storage.am.lsmtree.btree;
+
+import java.io.DataOutput;
+import java.io.File;
+import java.nio.ByteBuffer;
+
+import org.junit.Test;
+
+import edu.uci.ics.hyracks.api.comm.IFrameTupleAccessor;
+import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
+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.ITypeTraits;
+import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
+import edu.uci.ics.hyracks.api.io.FileReference;
+import edu.uci.ics.hyracks.data.std.accessors.PointableBinaryComparatorFactory;
+import edu.uci.ics.hyracks.data.std.primitive.IntegerPointable;
+import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
+import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleReference;
+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.marshalling.IntegerSerializerDeserializer;
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeLeafFrame;
+import edu.uci.ics.hyracks.storage.am.btree.frames.BTreeNSMInteriorFrameFactory;
+import edu.uci.ics.hyracks.storage.am.btree.frames.BTreeNSMLeafFrameFactory;
+import edu.uci.ics.hyracks.storage.am.common.api.IFreePageManager;
+import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexAccessor;
+import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexFrameFactory;
+import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexMetaDataFrameFactory;
+import edu.uci.ics.hyracks.storage.am.common.api.TreeIndexException;
+import edu.uci.ics.hyracks.storage.am.common.frames.LIFOMetaDataFrameFactory;
+import edu.uci.ics.hyracks.storage.am.common.freepage.LinkedListFreePageManager;
+import edu.uci.ics.hyracks.storage.am.common.ophelpers.MultiComparator;
+import edu.uci.ics.hyracks.storage.am.lsmtree.freepage.FreePageManagerFactory;
+import edu.uci.ics.hyracks.storage.am.lsmtree.freepage.InMemoryBufferCacheFactory;
+import edu.uci.ics.hyracks.storage.am.lsmtree.freepage.InMemoryFreePageManager;
+import edu.uci.ics.hyracks.storage.am.lsmtree.impls.BTreeFactory;
+import edu.uci.ics.hyracks.storage.am.lsmtree.impls.LSMTree;
+import edu.uci.ics.hyracks.storage.am.lsmtree.tuples.LSMEntireTupleWriterFactory;
+import edu.uci.ics.hyracks.storage.am.lsmtree.tuples.LSMTypeAwareTupleWriterFactory;
+import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
+import edu.uci.ics.hyracks.storage.common.file.IFileMapManager;
+import edu.uci.ics.hyracks.storage.common.file.IFileMapProvider;
+import edu.uci.ics.hyracks.test.support.TestStorageManagerComponentHolder;
+import edu.uci.ics.hyracks.test.support.TestUtils;
+
+public class LSMTreeFlushTest extends AbstractLSMTreeTest {
+    private static final int PAGE_SIZE = 256;
+    private static final int NUM_PAGES = 100;
+    private static final int MAX_OPEN_FILES = 10000;
+    private static final int HYRACKS_FRAME_SIZE = 128;
+    private IHyracksTaskContext ctx = TestUtils.create(HYRACKS_FRAME_SIZE);
+
+    // BASIC TEST
+    // @Test
+    // public void Test1() throws Exception {
+    // System.out.printf("TEST1 START\n");
+    // //in disk
+    // TestStorageManagerComponentHolder.init(PAGE_SIZE, NUM_PAGES,
+    // MAX_OPEN_FILES);
+    // IBufferCache bufferCache =
+    // TestStorageManagerComponentHolder.getBufferCache(ctx);
+    // IFileMapProvider fmp =
+    // TestStorageManagerComponentHolder.getFileMapProvider(ctx);
+    // FileReference file = new FileReference(new File(fileName));
+    // bufferCache.createFile(file);
+    // int fileId = fmp.lookupFileId(file);
+    // bufferCache.openFile(fileId);
+    //
+    // //in memory
+    // InMemoryBufferCacheFactory InMemBufferCacheFactory = new
+    // InMemoryBufferCacheFactory(PAGE_SIZE, NUM_PAGES);
+    // IBufferCache memBufferCache =
+    // InMemBufferCacheFactory.createInMemoryBufferCache();
+    //
+    // // declare fields
+    // int fieldCount = 2;
+    // ITypeTrait[] typeTraits = new ITypeTrait[fieldCount];
+    // typeTraits[0] = new TypeTrait(4);
+    // typeTraits[1] = new TypeTrait(4);
+    //
+    // // declare keys
+    // int keyFieldCount = 1;
+    // IBinaryComparatorFactory[] cmpFactories = new
+    // IBinaryComparatorFactory[keyFieldCount];
+    // cmpFactories[0] = IntegerBinaryComparatorFactory.INSTANCE;
+    //
+    // MultiComparator cmp = BTreeUtils.createMultiComparator(cmpFactories);
+    //
+    // LSMTypeAwareTupleWriterFactory insertTupleWriterFactory = new
+    // LSMTypeAwareTupleWriterFactory(typeTraits, false);
+    // LSMTypeAwareTupleWriterFactory deleteTupleWriterFactory = new
+    // LSMTypeAwareTupleWriterFactory(typeTraits, true);
+    //
+    // ITreeIndexFrameFactory insertLeafFrameFactory = new
+    // BTreeNSMLeafFrameFactory(insertTupleWriterFactory);
+    // ITreeIndexFrameFactory deleteLeafFrameFactory = new
+    // BTreeNSMLeafFrameFactory(deleteTupleWriterFactory);
+    // ITreeIndexFrameFactory interiorFrameFactory = new
+    // BTreeNSMInteriorFrameFactory(insertTupleWriterFactory);
+    // ITreeIndexMetaDataFrameFactory metaFrameFactory = new
+    // LIFOMetaDataFrameFactory();
+    //
+    // IBTreeLeafFrame insertLeafFrame = (IBTreeLeafFrame)
+    // insertLeafFrameFactory.createFrame();
+    //
+    // IFreePageManager freePageManager = new
+    // LinkedListFreePageManager(bufferCache, fileId, 0, metaFrameFactory);
+    // IFreePageManager memFreePageManager = new InMemoryFreePageManager(100,
+    // metaFrameFactory);
+    //
+    // // For the Flush Mechanism
+    // LSMEntireTupleWriterFactory flushTupleWriterFactory = new
+    // LSMEntireTupleWriterFactory(typeTraits);
+    // ITreeIndexFrameFactory flushLeafFrameFactory = new
+    // BTreeNSMLeafFrameFactory(flushTupleWriterFactory);
+    // FreePageManagerFactory freePageManagerFactory = new
+    // FreePageManagerFactory(bufferCache, metaFrameFactory);
+    // BTreeFactory bTreeFactory = new BTreeFactory(bufferCache,
+    // freePageManagerFactory, cmp, fieldCount, interiorFrameFactory,
+    // flushLeafFrameFactory);
+    //
+    //
+    //
+    // // LSMTree lsmtree = new LSMTree(3, 100, 2, memBufferCache, bufferCache,
+    // fieldCount, cmp, memFreePageManager,
+    // // freePageManager, interiorFrameFactory, insertLeafFrameFactory,
+    // deleteLeafFrameFactory, bTreeFactory, flushLeafFrameFactory,
+    // (IFileMapManager)fmp);
+    // //
+    // LSMTree lsmtree = LSMTreeUtils.createLSMTree(memBufferCache, bufferCache,
+    // fileId, typeTraits, cmp.getComparators(), BTreeLeafFrameType.REGULAR_NSM,
+    // (IFileMapManager)fmp);
+    // lsmtree.create(fileId);
+    // lsmtree.open(fileId);
+    //
+    // ByteBuffer frame = ctx.allocateFrame();
+    // FrameTupleAppender appender = new FrameTupleAppender(ctx.getFrameSize());
+    //
+    // ArrayTupleBuilder tb = new ArrayTupleBuilder(fieldCount);
+    // DataOutput dos = tb.getDataOutput();
+    //
+    // ISerializerDeserializer[] recDescSers = {
+    // IntegerSerializerDeserializer.INSTANCE,
+    // IntegerSerializerDeserializer.INSTANCE };
+    // RecordDescriptor recDesc = new RecordDescriptor(recDescSers);
+    //
+    // IFrameTupleAccessor accessor = new FrameTupleAccessor(ctx.getFrameSize(),
+    // recDesc);
+    // accessor.reset(frame);
+    //
+    // FrameTupleReference tuple = new FrameTupleReference();
+    //
+    // int resultSize = 100;
+    // int[][] resultArray = new int[resultSize][2];
+    //
+    //
+    // //insert 100 tuples
+    // for (int i = 0; i < resultSize; i++){
+    // resultArray[i][0] = i;
+    // resultArray[i][1] = 1;
+    // }
+    //
+    //
+    // LSMTreeOpContext insertOpCtx = lsmtree.createOpContext(IndexOp.INSERT);
+    // for (int i = 0; i < resultSize; i++) {
+    //
+    // int f0 = resultArray[i][0];
+    // int f1 = resultArray[i][1];
+    //
+    // tb.reset();
+    // IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+    // tb.addFieldEndOffset();
+    // IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+    // tb.addFieldEndOffset();
+    //
+    // appender.reset(frame, true);
+    // appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0,
+    // tb.getSize());
+    //
+    // tuple.reset(accessor, 0);
+    //
+    // ArrayTupleReference t = new ArrayTupleReference();
+    // t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+    //
+    // try {
+    // lsmtree.insert(t, insertOpCtx);
+    // } catch (TreeIndexException e) {
+    // System.out.println("test01:" + e);
+    // e.printStackTrace();
+    // } catch (Exception e) {
+    // e.printStackTrace();
+    // }
+    // }
+    // // Delete the first 50 keys in the in-memory tree
+    // insertOpCtx = lsmtree.createOpContext(IndexOp.DELETE);
+    // for (int i = 0; i < 50; i++){
+    // resultArray[i][0] = i;
+    // resultArray[i][1] = 1;
+    // }
+    //
+    // for (int i = 0; i < 50; i++) {
+    //
+    // int f0 = resultArray[i][0];
+    // int f1 = resultArray[i][1];
+    //
+    // tb.reset();
+    // IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+    // tb.addFieldEndOffset();
+    // IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+    // tb.addFieldEndOffset();
+    //
+    // appender.reset(frame, true);
+    // appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0,
+    // tb.getSize());
+    //
+    // tuple.reset(accessor, 0);
+    //
+    // ArrayTupleReference t = new ArrayTupleReference();
+    // t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+    //
+    // try {
+    // lsmtree.delete(t, insertOpCtx);
+    // } catch (TreeIndexException e) {
+    // System.out.println("test01:" + e);
+    // e.printStackTrace();
+    // } catch (Exception e) {
+    // e.printStackTrace();
+    // }
+    // }
+    //
+    //
+    // //Flush the tree into the first in Disk tree
+    // lsmtree.flushInMemoryBtree();
+    //
+    // //insert 50 delete nodes
+    // insertOpCtx = lsmtree.createOpContext(IndexOp.DELETE);
+    // for (int i = 0; i < 50; i++){
+    // resultArray[i][0] = i;
+    // resultArray[i][1] = 2;
+    // }
+    //
+    // for (int i = 0; i < 50; i++) {
+    //
+    // int f0 = resultArray[i][0];
+    // int f1 = resultArray[i][1];
+    //
+    // tb.reset();
+    // IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+    // tb.addFieldEndOffset();
+    // IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+    // tb.addFieldEndOffset();
+    //
+    // appender.reset(frame, true);
+    // appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0,
+    // tb.getSize());
+    //
+    // tuple.reset(accessor, 0);
+    //
+    // ArrayTupleReference t = new ArrayTupleReference();
+    // t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+    //
+    // try {
+    // lsmtree.delete(t, insertOpCtx);
+    // } catch (TreeIndexException e) {
+    // System.out.println("test01:" + e);
+    // e.printStackTrace();
+    // } catch (Exception e) {
+    // e.printStackTrace();
+    // }
+    // }
+    //
+    // // insert 25 nodes
+    // insertOpCtx = lsmtree.createOpContext(IndexOp.INSERT);
+    // for (int i = 0; i < resultSize; i++){
+    // resultArray[i][0] = i;
+    // resultArray[i][1] = 2;
+    // }
+    // for (int i = 0; i < 25; i++) {
+    //
+    // int f0 = resultArray[i][0];
+    // int f1 = resultArray[i][1];
+    //
+    // tb.reset();
+    // IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+    // tb.addFieldEndOffset();
+    // IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+    // tb.addFieldEndOffset();
+    //
+    // appender.reset(frame, true);
+    // appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0,
+    // tb.getSize());
+    //
+    // tuple.reset(accessor, 0);
+    //
+    // ArrayTupleReference t = new ArrayTupleReference();
+    // t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+    //
+    // try {
+    // lsmtree.insert(t, insertOpCtx);
+    // } catch (TreeIndexException e) {
+    // System.out.println("test01:" + e);
+    // e.printStackTrace();
+    // } catch (Exception e) {
+    // e.printStackTrace();
+    // }
+    // }
+    //
+    // //Flush the tree into the fist in Disk tree, which have fieldId as "1"
+    // lsmtree.flushInMemoryBtree();
+    //
+    // //Print out the first in Disk Btree
+    // System.out.println("LSMTreeFlushTest: start print the first tree");
+    // lsmtree.scanDiskTree(0);
+    // //Print out the second in Disk Btree
+    // System.out.println("LSMTreeFlushTest: start print the second tree");
+    // lsmtree.scanDiskTree(1);
+    //
+    //
+    // lsmtree.close();
+    // bufferCache.closeFile(fileId);
+    // memBufferCache.close();
+    //
+    // System.out.printf("End of TEST1()\n");
+    //
+    // }
+    // TEST auto Flush
+    @Test
+    public void Test2() throws Exception {
+        System.out.printf("TEST2 START\n");
+        // in disk
+        TestStorageManagerComponentHolder.init(PAGE_SIZE, NUM_PAGES, MAX_OPEN_FILES);
+        IBufferCache bufferCache = TestStorageManagerComponentHolder.getBufferCache(ctx);
+        IFileMapProvider fmp = TestStorageManagerComponentHolder.getFileMapProvider(ctx);
+        FileReference file = new FileReference(new File(fileName));
+        bufferCache.createFile(file);
+        int fileId = fmp.lookupFileId(file);
+        bufferCache.openFile(fileId);
+
+        // in memory
+        InMemoryBufferCacheFactory InMemBufferCacheFactory = new InMemoryBufferCacheFactory(PAGE_SIZE, NUM_PAGES);
+        IBufferCache memBufferCache = InMemBufferCacheFactory.createInMemoryBufferCache();
+
+        // declare fields
+        int fieldCount = 2;
+        ITypeTraits[] typeTraits = new ITypeTraits[fieldCount];
+        typeTraits[0] = IntegerPointable.TYPE_TRAITS;
+        typeTraits[1] = IntegerPointable.TYPE_TRAITS;
+
+        // declare keys
+        int keyFieldCount = 1;
+        IBinaryComparator[] cmps = new IBinaryComparator[keyFieldCount];
+        cmps[0] = PointableBinaryComparatorFactory.of(IntegerPointable.FACTORY).createBinaryComparator();
+
+        MultiComparator cmp = new MultiComparator(cmps);
+
+        LSMTypeAwareTupleWriterFactory insertTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(typeTraits, false);
+        LSMTypeAwareTupleWriterFactory deleteTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(typeTraits, true);
+
+        ITreeIndexFrameFactory insertLeafFrameFactory = new BTreeNSMLeafFrameFactory(insertTupleWriterFactory);
+        ITreeIndexFrameFactory deleteLeafFrameFactory = new BTreeNSMLeafFrameFactory(deleteTupleWriterFactory);
+        ITreeIndexFrameFactory interiorFrameFactory = new BTreeNSMInteriorFrameFactory(insertTupleWriterFactory);
+        ITreeIndexMetaDataFrameFactory metaFrameFactory = new LIFOMetaDataFrameFactory();
+
+        IBTreeLeafFrame insertLeafFrame = (IBTreeLeafFrame) insertLeafFrameFactory.createFrame();
+
+        IFreePageManager freePageManager = new LinkedListFreePageManager(bufferCache, fileId, 0, metaFrameFactory);
+        IFreePageManager memFreePageManager = new InMemoryFreePageManager(NUM_PAGES, metaFrameFactory);
+
+        // For the Flush Mechanism
+        LSMEntireTupleWriterFactory flushTupleWriterFactory = new LSMEntireTupleWriterFactory(typeTraits);
+        ITreeIndexFrameFactory flushLeafFrameFactory = new BTreeNSMLeafFrameFactory(flushTupleWriterFactory);
+        FreePageManagerFactory freePageManagerFactory = new FreePageManagerFactory(bufferCache, metaFrameFactory);
+        BTreeFactory bTreeFactory = new BTreeFactory(bufferCache, freePageManagerFactory, cmp, fieldCount,
+                interiorFrameFactory, flushLeafFrameFactory);
+
+        LSMTree lsmtree = new LSMTree(memBufferCache, bufferCache, fieldCount, cmp, memFreePageManager,
+                interiorFrameFactory, insertLeafFrameFactory, deleteLeafFrameFactory, bTreeFactory,
+                (IFileMapManager) fmp);
+
+        lsmtree.create(fileId);
+        lsmtree.open(fileId);
+
+        ByteBuffer frame = ctx.allocateFrame();
+        FrameTupleAppender appender = new FrameTupleAppender(ctx.getFrameSize());
+
+        ArrayTupleBuilder tb = new ArrayTupleBuilder(fieldCount);
+        DataOutput dos = tb.getDataOutput();
+
+        ISerializerDeserializer[] recDescSers = { IntegerSerializerDeserializer.INSTANCE,
+                IntegerSerializerDeserializer.INSTANCE };
+        RecordDescriptor recDesc = new RecordDescriptor(recDescSers);
+
+        IFrameTupleAccessor accessor = new FrameTupleAccessor(ctx.getFrameSize(), recDesc);
+        accessor.reset(frame);
+
+        FrameTupleReference tuple = new FrameTupleReference();
+
+        int resultSize = 820;
+        int[][] resultArray = new int[resultSize][2];
+
+        // insert 820 tuples
+        for (int i = 0; i < resultSize; i++) {
+            resultArray[i][0] = i;
+            resultArray[i][1] = i;
+        }
+
+        ITreeIndexAccessor lsmTreeAccessor = lsmtree.createAccessor();
+        for (int i = 0; i < resultSize; i++) {
+
+            int f0 = resultArray[i][0];
+            int f1 = resultArray[i][1];
+
+            tb.reset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+            tb.addFieldEndOffset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+            tb.addFieldEndOffset();
+
+            appender.reset(frame, true);
+            appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+
+            tuple.reset(accessor, 0);
+
+            ArrayTupleReference t = new ArrayTupleReference();
+            t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+
+            try {
+                lsmTreeAccessor.insert(t);
+            } catch (TreeIndexException e) {
+                System.out.println("test02:" + e);
+                e.printStackTrace();
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+
+        // Print out the third in Disk Btree
+        System.out.println("LSMTreeFlushTest: start print the first tree");
+        // lsmtree.scanDiskTree(2);
+        // Print out the second in Disk Btree
+        System.out.println("LSMTreeFlushTest: start print the first tree");
+        // lsmtree.scanDiskTree(1);
+        // Print out the first in Disk Btree
+        System.out.println("LSMTreeFlushTest: start print the first tree");
+        lsmtree.scanDiskTree(0);
+
+        lsmtree.close();
+        bufferCache.closeFile(fileId);
+        memBufferCache.close();
+
+        System.out.printf("End of TEST2()\n");
+
+    }
+
+    // @Test
+    // public void Test3() throws Exception {
+    // System.out.printf("TEST3 START\n");
+    // //in disk
+    // TestStorageManagerComponentHolder.init(PAGE_SIZE, NUM_PAGES,
+    // MAX_OPEN_FILES);
+    // IBufferCache bufferCache =
+    // TestStorageManagerComponentHolder.getBufferCache(ctx);
+    // IFileMapProvider fmp =
+    // TestStorageManagerComponentHolder.getFileMapProvider(ctx);
+    // FileReference file = new FileReference(new File(fileName));
+    // bufferCache.createFile(file);
+    // int fileId = fmp.lookupFileId(file);
+    // bufferCache.openFile(fileId);
+    //
+    // //in memory
+    // InMemoryBufferCacheFactory InMemBufferCacheFactory = new
+    // InMemoryBufferCacheFactory(PAGE_SIZE, NUM_PAGES);
+    // IBufferCache memBufferCache =
+    // InMemBufferCacheFactory.createInMemoryBufferCache();
+    //
+    // // declare fields
+    // int fieldCount = 2;
+    // ITypeTrait[] typeTraits = new ITypeTrait[fieldCount];
+    // typeTraits[0] = new TypeTrait(4);
+    // typeTraits[1] = new TypeTrait(4);
+    //
+    // // declare keys
+    // int keyFieldCount = 1;
+    // IBinaryComparatorFactory[] cmpFactories = new
+    // IBinaryComparatorFactory[keyFieldCount];
+    // cmpFactories[0] = IntegerBinaryComparatorFactory.INSTANCE;
+    //
+    // MultiComparator cmp = BTreeUtils.createMultiComparator(cmpFactories);
+    //
+    // LSMTypeAwareTupleWriterFactory insertTupleWriterFactory = new
+    // LSMTypeAwareTupleWriterFactory(typeTraits, false);
+    // LSMTypeAwareTupleWriterFactory deleteTupleWriterFactory = new
+    // LSMTypeAwareTupleWriterFactory(typeTraits, true);
+    //
+    // ITreeIndexFrameFactory insertLeafFrameFactory = new
+    // BTreeNSMLeafFrameFactory(insertTupleWriterFactory);
+    // ITreeIndexFrameFactory deleteLeafFrameFactory = new
+    // BTreeNSMLeafFrameFactory(deleteTupleWriterFactory);
+    // ITreeIndexFrameFactory interiorFrameFactory = new
+    // BTreeNSMInteriorFrameFactory(insertTupleWriterFactory);
+    // ITreeIndexMetaDataFrameFactory metaFrameFactory = new
+    // LIFOMetaDataFrameFactory();
+    //
+    // IBTreeLeafFrame insertLeafFrame = (IBTreeLeafFrame)
+    // insertLeafFrameFactory.createFrame();
+    //
+    // IFreePageManager freePageManager = new
+    // LinkedListFreePageManager(bufferCache, fileId, 0, metaFrameFactory);
+    // IFreePageManager memFreePageManager = new InMemoryFreePageManager(30,
+    // metaFrameFactory);
+    //
+    // // For the Flush Mechanism
+    // LSMEntireTupleWriterFactory flushTupleWriterFactory = new
+    // LSMEntireTupleWriterFactory(typeTraits);
+    // ITreeIndexFrameFactory flushLeafFrameFactory = new
+    // BTreeNSMLeafFrameFactory(flushTupleWriterFactory);
+    // FreePageManagerFactory freePageManagerFactory = new
+    // FreePageManagerFactory(bufferCache, metaFrameFactory);
+    // BTreeFactory bTreeFactory = new BTreeFactory(bufferCache,
+    // freePageManagerFactory, cmp, fieldCount, interiorFrameFactory,
+    // flushLeafFrameFactory);
+    //
+    //
+    //
+    // LSMTree lsmtree = new LSMTree(memBufferCache, bufferCache, fieldCount,
+    // cmp, memFreePageManager, interiorFrameFactory, insertLeafFrameFactory,
+    // deleteLeafFrameFactory, bTreeFactory, (IFileMapManager)fmp);
+    //
+    // lsmtree.create(fileId);
+    // lsmtree.open(fileId);
+    //
+    // ByteBuffer frame = ctx.allocateFrame();
+    // FrameTupleAppender appender = new FrameTupleAppender(ctx.getFrameSize());
+    //
+    // ArrayTupleBuilder tb = new ArrayTupleBuilder(fieldCount);
+    // DataOutput dos = tb.getDataOutput();
+    //
+    // ISerializerDeserializer[] recDescSers = {
+    // IntegerSerializerDeserializer.INSTANCE,
+    // IntegerSerializerDeserializer.INSTANCE };
+    // RecordDescriptor recDesc = new RecordDescriptor(recDescSers);
+    //
+    // IFrameTupleAccessor accessor = new FrameTupleAccessor(ctx.getFrameSize(),
+    // recDesc);
+    // accessor.reset(frame);
+    //
+    // FrameTupleReference tuple = new FrameTupleReference();
+    //
+    // int resultSize = 500;
+    // int[][] resultArray = new int[resultSize][2];
+    //
+    //
+    // //insert 250 tuples
+    // System.out.printf("Start for 1st Insert\n");
+    // LSMTreeOpContext insertOpCtx = lsmtree.createOpContext(IndexOp.INSERT);
+    // for (int i = 0; i < 252; i++){
+    // resultArray[i][0] = i;
+    // resultArray[i][1] = i;
+    // }
+    // for (int i = 0; i < 252; i++) {
+    //
+    // int f0 = resultArray[i][0];
+    // int f1 = resultArray[i][1];
+    //
+    // tb.reset();
+    // IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+    // tb.addFieldEndOffset();
+    // IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+    // tb.addFieldEndOffset();
+    //
+    // appender.reset(frame, true);
+    // appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0,
+    // tb.getSize());
+    //
+    // tuple.reset(accessor, 0);
+    //
+    // ArrayTupleReference t = new ArrayTupleReference();
+    // t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+    //
+    // try {
+    // lsmtree.insert(t, insertOpCtx);
+    // } catch (TreeIndexException e) {
+    // System.out.println("test03:" + e);
+    // e.printStackTrace();
+    // } catch (Exception e) {
+    // e.printStackTrace();
+    // }
+    // }
+    // System.out.printf("Start for 2nd Insert\n");
+    // //delete 126~251. Deletion of 251 cause the flush
+    // insertOpCtx.reset(IndexOp.DELETE);
+    // // LSMTreeOpContext insertOpCtx =
+    // lsmtree.createOpContext(IndexOp.DELETE);
+    // for (int i = 125; i < 253; i++){
+    // resultArray[i][0] = i;
+    // resultArray[i][1] = i;
+    // }
+    // for (int i = 126; i < 253; i++) {
+    // int f0 = resultArray[i][0];
+    // int f1 = resultArray[i][1];
+    //
+    // tb.reset();
+    // IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+    // tb.addFieldEndOffset();
+    // IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+    // tb.addFieldEndOffset();
+    //
+    // appender.reset(frame, true);
+    // appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0,
+    // tb.getSize());
+    //
+    // tuple.reset(accessor, 0);
+    //
+    // ArrayTupleReference t = new ArrayTupleReference();
+    // t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+    //
+    // try {
+    // lsmtree.delete(t, insertOpCtx);
+    // } catch (TreeIndexException e) {
+    // System.out.println("test03:" + e);
+    // e.printStackTrace();
+    // } catch (Exception e) {
+    // e.printStackTrace();
+    // }
+    // }
+    // //delete 0~250
+    // insertOpCtx = lsmtree.createOpContext(IndexOp.INSERT);
+    // for (int i = 130; i > 0; i--){
+    // resultArray[i][0] = i;
+    // resultArray[i][1] = i;
+    // }
+    // for (int i = 130; i > 0; i--) {
+    //
+    // int f0 = resultArray[i][0];
+    // int f1 = resultArray[i][1];
+    //
+    // tb.reset();
+    // IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+    // tb.addFieldEndOffset();
+    // IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+    // tb.addFieldEndOffset();
+    //
+    // appender.reset(frame, true);
+    // appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0,
+    // tb.getSize());
+    //
+    // tuple.reset(accessor, 0);
+    //
+    // ArrayTupleReference t = new ArrayTupleReference();
+    // t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+    //
+    // try {
+    // lsmtree.insert(t, insertOpCtx);
+    // } catch (TreeIndexException e) {
+    // System.out.println("test03:" + e);
+    // e.printStackTrace();
+    // } catch (Exception e) {
+    // e.printStackTrace();
+    // }
+    // }
+    //
+    // //
+    // //
+    // //
+    // // //Print out the second in Disk Btree
+    // // System.out.println("LSMTreeFlushTest: start print the second tree");
+    // // lsmtree.scanDiskTree(1);
+    // // //Print out the first in Disk Btree
+    // // System.out.println("LSMTreeFlushTest: start print the first tree");
+    // // lsmtree.scanDiskTree(0);
+    // //
+    // // //Print out the In-memory Tree
+    // //
+    // System.out.println("LSMTreeFlushTest: start print the In-memory tree");
+    // // lsmtree.scanInMemoryTree();
+    // // //TODO: scan whole tree
+    //
+    // LOGGER.info("RANGE SEARCH:");
+    //
+    // BTreeOpContext searchOpCtx = lsmtree.createOpContext(IndexOp.SEARCH);
+    // ITreeIndexCursor rangeCursor = new LSMTreeRangeSearchCursor();
+    //
+    // // build low and high keys
+    // ArrayTupleBuilder ktb = new ArrayTupleBuilder(cmp.getKeyFieldCount());
+    // DataOutput kdos = ktb.getDataOutput();
+    //
+    // ISerializerDeserializer[] keyDescSers = {
+    // IntegerSerializerDeserializer.INSTANCE };
+    // RecordDescriptor keyDesc = new RecordDescriptor(keyDescSers);
+    // IFrameTupleAccessor keyAccessor = new
+    // FrameTupleAccessor(ctx.getFrameSize(), keyDesc);
+    // keyAccessor.reset(frame);
+    //
+    // appender.reset(frame, true);
+    //
+    // // build and append low key
+    // ktb.reset();
+    // IntegerSerializerDeserializer.INSTANCE.serialize(-1, kdos);
+    // ktb.addFieldEndOffset();
+    // appender.append(ktb.getFieldEndOffsets(), ktb.getByteArray(), 0,
+    // ktb.getSize());
+    //
+    // // build and append high key
+    // ktb.reset();
+    // IntegerSerializerDeserializer.INSTANCE.serialize(300, 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(searchCmps);
+    //
+    // RangePredicate rangePred = new RangePredicate(true, lowKey, highKey,
+    // true, true, searchCmp, searchCmp);
+    // lsmtree.search(rangeCursor, rangePred, searchOpCtx);
+    //
+    // try {
+    // while (rangeCursor.hasNext()) {
+    // rangeCursor.next();
+    // ITupleReference frameTuple = rangeCursor.getTuple();
+    // String rec = TupleUtils.printTuple(frameTuple, recDescSers);
+    // if(((LSMTypeAwareTupleReference)frameTuple).isDelete()) {
+    // System.out.println("del " + rec);
+    // }
+    // else {
+    // System.out.println("ins " + rec);
+    // }
+    // // System.out.println("------------------");
+    // }
+    // } catch (Exception e) {
+    // e.printStackTrace();
+    // } finally {
+    // rangeCursor.close();
+    // }
+    //
+    // lsmtree.close();
+    // bufferCache.closeFile(fileId);
+    // memBufferCache.close();
+    //
+    // System.out.printf("End of TEST3()\n");
+    //
+    // }
+
+}
\ No newline at end of file
diff --git a/hyracks-tests/hyracks-storage-am-lsmtree-common-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsmtree/common/LSMTreeMergeTest.java b/hyracks-tests/hyracks-storage-am-lsmtree-common-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsmtree/common/LSMTreeMergeTest.java
new file mode 100644
index 0000000..03924f0
--- /dev/null
+++ b/hyracks-tests/hyracks-storage-am-lsmtree-common-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsmtree/common/LSMTreeMergeTest.java
@@ -0,0 +1,378 @@
+package edu.uci.ics.hyracks.storage.am.lsmtree.btree;
+
+import java.io.DataOutput;
+import java.io.File;
+import java.nio.ByteBuffer;
+
+import org.junit.Test;
+
+import edu.uci.ics.hyracks.api.comm.IFrameTupleAccessor;
+import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
+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.ITypeTraits;
+import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
+import edu.uci.ics.hyracks.api.io.FileReference;
+import edu.uci.ics.hyracks.data.std.accessors.PointableBinaryComparatorFactory;
+import edu.uci.ics.hyracks.data.std.primitive.IntegerPointable;
+import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
+import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleReference;
+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.marshalling.IntegerSerializerDeserializer;
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeLeafFrame;
+import edu.uci.ics.hyracks.storage.am.btree.frames.BTreeNSMInteriorFrameFactory;
+import edu.uci.ics.hyracks.storage.am.btree.frames.BTreeNSMLeafFrameFactory;
+import edu.uci.ics.hyracks.storage.am.common.api.IFreePageManager;
+import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexAccessor;
+import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexFrameFactory;
+import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexMetaDataFrameFactory;
+import edu.uci.ics.hyracks.storage.am.common.api.TreeIndexException;
+import edu.uci.ics.hyracks.storage.am.common.frames.LIFOMetaDataFrameFactory;
+import edu.uci.ics.hyracks.storage.am.common.freepage.LinkedListFreePageManager;
+import edu.uci.ics.hyracks.storage.am.common.ophelpers.MultiComparator;
+import edu.uci.ics.hyracks.storage.am.lsmtree.freepage.FreePageManagerFactory;
+import edu.uci.ics.hyracks.storage.am.lsmtree.freepage.InMemoryBufferCacheFactory;
+import edu.uci.ics.hyracks.storage.am.lsmtree.freepage.InMemoryFreePageManager;
+import edu.uci.ics.hyracks.storage.am.lsmtree.impls.BTreeFactory;
+import edu.uci.ics.hyracks.storage.am.lsmtree.impls.LSMTree;
+import edu.uci.ics.hyracks.storage.am.lsmtree.tuples.LSMEntireTupleWriterFactory;
+import edu.uci.ics.hyracks.storage.am.lsmtree.tuples.LSMTypeAwareTupleWriterFactory;
+import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
+import edu.uci.ics.hyracks.storage.common.file.IFileMapManager;
+import edu.uci.ics.hyracks.storage.common.file.IFileMapProvider;
+import edu.uci.ics.hyracks.test.support.TestStorageManagerComponentHolder;
+import edu.uci.ics.hyracks.test.support.TestUtils;
+
+public class LSMTreeMergeTest extends AbstractLSMTreeTest {
+    private static final int PAGE_SIZE = 256;
+    private static final int NUM_PAGES = 30;
+    private static final int MAX_OPEN_FILES = 100;
+    private static final int HYRACKS_FRAME_SIZE = 128;
+    private IHyracksTaskContext ctx = TestUtils.create(HYRACKS_FRAME_SIZE);
+
+    @Test
+    public void Test1() throws Exception {
+        System.out.printf("TEST1 START\n");
+        // in disk
+        TestStorageManagerComponentHolder.init(PAGE_SIZE, NUM_PAGES, MAX_OPEN_FILES);
+        IBufferCache bufferCache = TestStorageManagerComponentHolder.getBufferCache(ctx);
+        IFileMapProvider fmp = TestStorageManagerComponentHolder.getFileMapProvider(ctx);
+        FileReference file = new FileReference(new File(fileName));
+        bufferCache.createFile(file);
+        int fileId = fmp.lookupFileId(file);
+        bufferCache.openFile(fileId);
+
+        // in memory
+        InMemoryBufferCacheFactory InMemBufferCacheFactory = new InMemoryBufferCacheFactory(PAGE_SIZE, NUM_PAGES);
+        IBufferCache memBufferCache = InMemBufferCacheFactory.createInMemoryBufferCache();
+
+        // declare fields
+        int fieldCount = 2;
+        ITypeTraits[] typeTraits = new ITypeTraits[fieldCount];
+        typeTraits[0] = IntegerPointable.TYPE_TRAITS;
+        typeTraits[1] = IntegerPointable.TYPE_TRAITS;
+
+        // declare keys
+        int keyFieldCount = 1;
+        IBinaryComparator[] cmps = new IBinaryComparator[keyFieldCount];
+        cmps[0] = PointableBinaryComparatorFactory.of(IntegerPointable.FACTORY).createBinaryComparator();
+
+        MultiComparator cmp = new MultiComparator(cmps);
+
+        LSMTypeAwareTupleWriterFactory insertTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(typeTraits, false);
+        LSMTypeAwareTupleWriterFactory deleteTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(typeTraits, true);
+
+        ITreeIndexFrameFactory insertLeafFrameFactory = new BTreeNSMLeafFrameFactory(insertTupleWriterFactory);
+        ITreeIndexFrameFactory deleteLeafFrameFactory = new BTreeNSMLeafFrameFactory(deleteTupleWriterFactory);
+        ITreeIndexFrameFactory interiorFrameFactory = new BTreeNSMInteriorFrameFactory(insertTupleWriterFactory);
+        ITreeIndexMetaDataFrameFactory metaFrameFactory = new LIFOMetaDataFrameFactory();
+
+        IBTreeLeafFrame insertLeafFrame = (IBTreeLeafFrame) insertLeafFrameFactory.createFrame();
+
+        IFreePageManager freePageManager = new LinkedListFreePageManager(bufferCache, fileId, 0, metaFrameFactory);
+        IFreePageManager memFreePageManager = new InMemoryFreePageManager(NUM_PAGES, metaFrameFactory);
+
+        // For the Flush Mechanism
+        LSMEntireTupleWriterFactory flushTupleWriterFactory = new LSMEntireTupleWriterFactory(typeTraits);
+        ITreeIndexFrameFactory flushLeafFrameFactory = new BTreeNSMLeafFrameFactory(flushTupleWriterFactory);
+        FreePageManagerFactory freePageManagerFactory = new FreePageManagerFactory(bufferCache, metaFrameFactory);
+        BTreeFactory bTreeFactory = new BTreeFactory(bufferCache, freePageManagerFactory, cmp, fieldCount,
+                interiorFrameFactory, flushLeafFrameFactory);
+
+        LSMTree lsmtree = new LSMTree(memBufferCache, bufferCache, fieldCount, cmp, memFreePageManager,
+                interiorFrameFactory, insertLeafFrameFactory, deleteLeafFrameFactory, bTreeFactory,
+                (IFileMapManager) fmp);
+
+        // LSMTree lsmtree = LSMTreeUtils.createLSMTree(10, 30, 2,
+        // memBufferCache, bufferCache, fileId, typeTraits,
+        // cmp.getComparators(), BTreeLeafFrameType.REGULAR_NSM,
+        // (IFileMapManager)fmp);
+
+        lsmtree.create(fileId);
+        lsmtree.open(fileId);
+
+        ByteBuffer frame = ctx.allocateFrame();
+        FrameTupleAppender appender = new FrameTupleAppender(ctx.getFrameSize());
+
+        ArrayTupleBuilder tb = new ArrayTupleBuilder(fieldCount);
+        DataOutput dos = tb.getDataOutput();
+
+        ISerializerDeserializer[] recDescSers = { IntegerSerializerDeserializer.INSTANCE,
+                IntegerSerializerDeserializer.INSTANCE };
+        RecordDescriptor recDesc = new RecordDescriptor(recDescSers);
+
+        IFrameTupleAccessor accessor = new FrameTupleAccessor(ctx.getFrameSize(), recDesc);
+        accessor.reset(frame);
+
+        FrameTupleReference tuple = new FrameTupleReference();
+
+        int resultSize = 100000;
+        int[][] resultArray = new int[resultSize][2];
+
+        // insert 0~250 tuples
+        System.out.printf("Start for 1st Insert\n");
+        ITreeIndexAccessor lsmTreeAccessor = lsmtree.createAccessor();
+        for (int i = 0; i < 251; i++) {
+            resultArray[i][0] = i;
+            resultArray[i][1] = i;
+        }
+        for (int i = 0; i < 251; i++) {
+
+            int f0 = resultArray[i][0];
+            int f1 = resultArray[i][1];
+
+            tb.reset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+            tb.addFieldEndOffset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+            tb.addFieldEndOffset();
+
+            appender.reset(frame, true);
+            appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+
+            tuple.reset(accessor, 0);
+
+            ArrayTupleReference t = new ArrayTupleReference();
+            t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+
+            try {
+                lsmTreeAccessor.insert(t);
+            } catch (TreeIndexException e) {
+                System.out.println("test03:" + e);
+                e.printStackTrace();
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+        System.out.printf("Start for 2nd Insert\n");
+        // delete 126~250.
+        for (int i = 126; i < 251; i++) {
+            resultArray[i][0] = i;
+            resultArray[i][1] = i;
+        }
+        for (int i = 126; i < 251; i++) {
+            int f0 = resultArray[i][0];
+            int f1 = resultArray[i][1];
+
+            tb.reset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+            tb.addFieldEndOffset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+            tb.addFieldEndOffset();
+
+            appender.reset(frame, true);
+            appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+
+            tuple.reset(accessor, 0);
+
+            ArrayTupleReference t = new ArrayTupleReference();
+            t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+
+            try {
+                lsmTreeAccessor.delete(t);
+            } catch (TreeIndexException e) {
+                System.out.println("test03:" + e);
+                e.printStackTrace();
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+        // insert 251~1
+        for (int i = 251; i > 0; i--) {
+            resultArray[i][0] = i;
+            resultArray[i][1] = i;
+        }
+        for (int i = 251; i > 0; i--) {
+
+            int f0 = resultArray[i][0];
+            int f1 = resultArray[i][1];
+
+            tb.reset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+            tb.addFieldEndOffset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+            tb.addFieldEndOffset();
+
+            appender.reset(frame, true);
+            appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+
+            tuple.reset(accessor, 0);
+
+            ArrayTupleReference t = new ArrayTupleReference();
+            t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+
+            try {
+                lsmTreeAccessor.insert(t);
+            } catch (TreeIndexException e) {
+                System.out.println("test03:" + e);
+                e.printStackTrace();
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+        // delete 100~0
+        for (int i = 100; i >= 0; i--) {
+            resultArray[i][0] = i;
+            resultArray[i][1] = i;
+        }
+        for (int i = 100; i >= 0; i--) {
+
+            int f0 = resultArray[i][0];
+            int f1 = resultArray[i][1];
+
+            tb.reset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+            tb.addFieldEndOffset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+            tb.addFieldEndOffset();
+
+            appender.reset(frame, true);
+            appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+
+            tuple.reset(accessor, 0);
+
+            ArrayTupleReference t = new ArrayTupleReference();
+            t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+
+            try {
+                lsmTreeAccessor.delete(t);
+            } catch (TreeIndexException e) {
+                System.out.println("test03:" + e);
+                e.printStackTrace();
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+        // insert 1~50
+        for (int i = 1; i < 51; i++) {
+            resultArray[i][0] = i;
+            resultArray[i][1] = i;
+        }
+        for (int i = 1; i < 51; i++) {
+
+            int f0 = resultArray[i][0];
+            int f1 = resultArray[i][1];
+
+            tb.reset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+            tb.addFieldEndOffset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+            tb.addFieldEndOffset();
+
+            appender.reset(frame, true);
+            appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+
+            tuple.reset(accessor, 0);
+
+            ArrayTupleReference t = new ArrayTupleReference();
+            t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+
+            try {
+                lsmTreeAccessor.insert(t);
+            } catch (TreeIndexException e) {
+                System.out.println("test03:" + e);
+                e.printStackTrace();
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+
+        lsmtree.merge();
+
+        // Output should be:
+        // In memory tree = 0->del, 1~50 ins
+        // MergedTree = 0->ins, 1~100->del, 101~251->ins
+        // Whole search = 1~50,101~251
+
+        // System.out.println("LSMTreeFlushTest: start print the first tree");
+        // lsmtree.scanDiskTree(1);
+        //
+        // Print out the first in Disk Btree
+        // System.out.println("LSMTreeFlushTest: start print the first tree");
+        // lsmtree.scanDiskTree(0);
+        // Print out the In-memory Tree
+        System.out.println("LSMTreeFlushTest: start print the In-memory tree");
+        lsmtree.scanInMemoryTree();
+        // TODO: scan whole tree
+        /*
+         * System.out.println("Range SEARCH:");
+         * 
+         * BTreeOpContext searchOpCtx = lsmtree.createOpContext(IndexOp.SEARCH);
+         * ITreeIndexCursor rangeCursor = new LSMTreeRangeSearchCursor();
+         * 
+         * // build low and high keys ArrayTupleBuilder ktb = new
+         * ArrayTupleBuilder(cmp.getKeyFieldCount()); DataOutput kdos =
+         * ktb.getDataOutput();
+         * 
+         * ISerializerDeserializer[] keyDescSers = {
+         * IntegerSerializerDeserializer.INSTANCE }; RecordDescriptor keyDesc =
+         * new RecordDescriptor(keyDescSers); IFrameTupleAccessor keyAccessor =
+         * new FrameTupleAccessor( ctx.getFrameSize(), keyDesc);
+         * keyAccessor.reset(frame);
+         * 
+         * appender.reset(frame, true);
+         * 
+         * // build and append low key ktb.reset();
+         * IntegerSerializerDeserializer.INSTANCE.serialize(-1, kdos);
+         * ktb.addFieldEndOffset(); appender.append(ktb.getFieldEndOffsets(),
+         * ktb.getByteArray(), 0, ktb.getSize());
+         * 
+         * // build and append high key ktb.reset();
+         * IntegerSerializerDeserializer.INSTANCE.serialize(300, 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(searchCmps);
+         * 
+         * RangePredicate rangePred = new RangePredicate(true, lowKey, highKey,
+         * true, true, searchCmp, searchCmp); lsmtree.search(rangeCursor,
+         * rangePred, searchOpCtx);
+         * 
+         * try { while (rangeCursor.hasNext()) { rangeCursor.next();
+         * ITupleReference frameTuple = rangeCursor.getTuple(); String rec =
+         * TupleUtils.printTuple(frameTuple, recDescSers); if
+         * (((LSMTypeAwareTupleReference) frameTuple).isDelete()) {
+         * System.out.println("del " + rec); } else { System.out.println("ins "
+         * + rec); } // System.out.println("------------------"); } } catch
+         * (Exception e) { e.printStackTrace(); } finally { rangeCursor.close();
+         * }
+         */
+        lsmtree.close();
+        bufferCache.closeFile(fileId);
+        memBufferCache.close();
+
+        System.out.printf("End of TEST1()\n");
+
+    }
+}
diff --git a/hyracks-tests/hyracks-storage-am-lsmtree-common-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsmtree/common/LSMTreeSearchTest.java b/hyracks-tests/hyracks-storage-am-lsmtree-common-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsmtree/common/LSMTreeSearchTest.java
new file mode 100644
index 0000000..9b48303
--- /dev/null
+++ b/hyracks-tests/hyracks-storage-am-lsmtree-common-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsmtree/common/LSMTreeSearchTest.java
@@ -0,0 +1,419 @@
+package edu.uci.ics.hyracks.storage.am.lsmtree.btree;
+
+import java.io.DataOutput;
+import java.io.File;
+import java.nio.ByteBuffer;
+import java.util.Date;
+import java.util.Random;
+
+import org.junit.Test;
+
+import edu.uci.ics.hyracks.api.comm.IFrameTupleAccessor;
+import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
+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.ITypeTraits;
+import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
+import edu.uci.ics.hyracks.api.io.FileReference;
+import edu.uci.ics.hyracks.data.std.accessors.PointableBinaryComparatorFactory;
+import edu.uci.ics.hyracks.data.std.primitive.IntegerPointable;
+import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
+import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleReference;
+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.marshalling.IntegerSerializerDeserializer;
+import edu.uci.ics.hyracks.dataflow.common.util.TupleUtils;
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeInteriorFrame;
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeLeafFrame;
+import edu.uci.ics.hyracks.storage.am.btree.frames.BTreeNSMInteriorFrameFactory;
+import edu.uci.ics.hyracks.storage.am.btree.frames.BTreeNSMLeafFrameFactory;
+import edu.uci.ics.hyracks.storage.am.btree.impls.BTree;
+import edu.uci.ics.hyracks.storage.am.btree.impls.RangePredicate;
+import edu.uci.ics.hyracks.storage.am.common.api.IFreePageManager;
+import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexAccessor;
+import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexCursor;
+import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexFrameFactory;
+import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexMetaDataFrame;
+import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexMetaDataFrameFactory;
+import edu.uci.ics.hyracks.storage.am.common.api.TreeIndexException;
+import edu.uci.ics.hyracks.storage.am.common.frames.LIFOMetaDataFrameFactory;
+import edu.uci.ics.hyracks.storage.am.common.freepage.LinkedListFreePageManager;
+import edu.uci.ics.hyracks.storage.am.common.ophelpers.MultiComparator;
+import edu.uci.ics.hyracks.storage.am.common.tuples.TypeAwareTupleWriterFactory;
+import edu.uci.ics.hyracks.storage.am.lsmtree.freepage.FreePageManagerFactory;
+import edu.uci.ics.hyracks.storage.am.lsmtree.freepage.InMemoryBufferCacheFactory;
+import edu.uci.ics.hyracks.storage.am.lsmtree.freepage.InMemoryFreePageManager;
+import edu.uci.ics.hyracks.storage.am.lsmtree.impls.BTreeFactory;
+import edu.uci.ics.hyracks.storage.am.lsmtree.impls.InDiskTreeInfo;
+import edu.uci.ics.hyracks.storage.am.lsmtree.impls.LSMTree;
+import edu.uci.ics.hyracks.storage.am.lsmtree.impls.LSMTreeRangeSearchCursor;
+import edu.uci.ics.hyracks.storage.am.lsmtree.tuples.LSMEntireTupleWriterFactory;
+import edu.uci.ics.hyracks.storage.am.lsmtree.tuples.LSMTypeAwareTupleReference;
+import edu.uci.ics.hyracks.storage.am.lsmtree.tuples.LSMTypeAwareTupleWriterFactory;
+import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
+import edu.uci.ics.hyracks.storage.common.file.IFileMapManager;
+import edu.uci.ics.hyracks.storage.common.file.IFileMapProvider;
+import edu.uci.ics.hyracks.test.support.TestStorageManagerComponentHolder;
+import edu.uci.ics.hyracks.test.support.TestUtils;
+
+// TODO: needs a big cleanup phase.
+public class LSMTreeSearchTest extends AbstractLSMTreeTest {
+
+    private static final int PAGE_SIZE = 256;
+    private static final int NUM_PAGES = 10;
+    private static final int MAX_OPEN_FILES = 100;
+    private static final int HYRACKS_FRAME_SIZE = 128;
+    private IHyracksTaskContext ctx = TestUtils.create(HYRACKS_FRAME_SIZE);
+
+    @Test
+    public void Test1() throws Exception {
+        // in disk
+        TestStorageManagerComponentHolder.init(PAGE_SIZE, NUM_PAGES, MAX_OPEN_FILES);
+        IBufferCache bufferCache = TestStorageManagerComponentHolder.getBufferCache(ctx);
+        IFileMapProvider fmp = TestStorageManagerComponentHolder.getFileMapProvider(ctx);
+        FileReference file = new FileReference(new File(fileName));
+        bufferCache.createFile(file);
+        int fileId = fmp.lookupFileId(file);
+        bufferCache.openFile(fileId);
+
+        // in memory
+        InMemoryBufferCacheFactory InMemBufferCacheFactory = new InMemoryBufferCacheFactory(PAGE_SIZE, NUM_PAGES);
+        IBufferCache memBufferCache = InMemBufferCacheFactory.createInMemoryBufferCache();
+
+        // declare fields
+        int fieldCount = 2;
+        ITypeTraits[] typeTraits = new ITypeTraits[fieldCount];
+        typeTraits[0] = IntegerPointable.TYPE_TRAITS;
+        typeTraits[1] = IntegerPointable.TYPE_TRAITS;
+
+        // declare keys
+        int keyFieldCount = 1;
+        IBinaryComparator[] cmps = new IBinaryComparator[keyFieldCount];
+        cmps[0] = PointableBinaryComparatorFactory.of(IntegerPointable.FACTORY).createBinaryComparator();
+
+        MultiComparator cmp = new MultiComparator(cmps);
+
+        LSMTypeAwareTupleWriterFactory insertTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(typeTraits, false);
+        LSMTypeAwareTupleWriterFactory deleteTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(typeTraits, true);
+
+        ITreeIndexFrameFactory insertLeafFrameFactory = new BTreeNSMLeafFrameFactory(insertTupleWriterFactory);
+        ITreeIndexFrameFactory deleteLeafFrameFactory = new BTreeNSMLeafFrameFactory(deleteTupleWriterFactory);
+        ITreeIndexFrameFactory interiorFrameFactory = new BTreeNSMInteriorFrameFactory(insertTupleWriterFactory);
+        ITreeIndexMetaDataFrameFactory metaFrameFactory = new LIFOMetaDataFrameFactory();
+
+        IBTreeLeafFrame insertLeafFrame = (IBTreeLeafFrame) insertLeafFrameFactory.createFrame();
+
+        IFreePageManager freePageManager = new LinkedListFreePageManager(bufferCache, fileId, 0, metaFrameFactory);
+        IFreePageManager memFreePageManager = new InMemoryFreePageManager(100, metaFrameFactory);
+
+        LSMEntireTupleWriterFactory flushTupleWriterFactory = new LSMEntireTupleWriterFactory(typeTraits);
+        ITreeIndexFrameFactory flushLeafFrameFactory = new BTreeNSMLeafFrameFactory(flushTupleWriterFactory);
+        FreePageManagerFactory freePageManagerFactory = new FreePageManagerFactory(bufferCache, metaFrameFactory);
+        BTreeFactory bTreeFactory = new BTreeFactory(bufferCache, freePageManagerFactory, cmp, fieldCount,
+                interiorFrameFactory, flushLeafFrameFactory);
+
+        LSMTree lsmtree = new LSMTree(memBufferCache, bufferCache, fieldCount, cmp, memFreePageManager,
+                interiorFrameFactory, insertLeafFrameFactory, deleteLeafFrameFactory, bTreeFactory,
+                (IFileMapManager) fmp);
+
+        lsmtree.create(fileId);
+        lsmtree.open(fileId);
+
+        ByteBuffer frame = ctx.allocateFrame();
+        FrameTupleAppender appender = new FrameTupleAppender(ctx.getFrameSize());
+
+        ArrayTupleBuilder tb = new ArrayTupleBuilder(fieldCount);
+        DataOutput dos = tb.getDataOutput();
+
+        ISerializerDeserializer[] recDescSers = { IntegerSerializerDeserializer.INSTANCE,
+                IntegerSerializerDeserializer.INSTANCE };
+        RecordDescriptor recDesc = new RecordDescriptor(recDescSers);
+
+        IFrameTupleAccessor accessor = new FrameTupleAccessor(ctx.getFrameSize(), recDesc);
+        accessor.reset(frame);
+
+        FrameTupleReference tuple = new FrameTupleReference();
+
+        ITreeIndexAccessor lsmTreeAccessor = lsmtree.createAccessor();
+
+        // delete
+        for (int i = 26; i < 36; i++) {
+
+            int f0 = i;
+            int f1 = -1;
+
+            tb.reset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+            tb.addFieldEndOffset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+            tb.addFieldEndOffset();
+
+            appender.reset(frame, true);
+            appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+
+            tuple.reset(accessor, 0);
+
+            ArrayTupleReference t = new ArrayTupleReference();
+            t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+
+            try {
+                lsmTreeAccessor.delete(t);
+            } catch (TreeIndexException e) {
+                System.out.println("test01:" + e);
+                e.printStackTrace();
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+
+        for (int i = 21; i < 31; i++) {
+            int f0 = i;
+            int f1 = 0;
+
+            tb.reset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+            tb.addFieldEndOffset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+            tb.addFieldEndOffset();
+
+            appender.reset(frame, true);
+            appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+
+            tuple.reset(accessor, 0);
+
+            ArrayTupleReference t = new ArrayTupleReference();
+            t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+
+            try {
+                lsmTreeAccessor.insert(t);
+            } catch (TreeIndexException e) {
+                System.out.println("test01:" + e);
+                e.printStackTrace();
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+
+        // In disk insertions 1
+
+        LOGGER.info("Start in-disk insertions");
+
+        fileName = tmpDir + sep + simpleDateFormat.format(new Date());
+        FileReference file_1 = new FileReference(new File(fileName));
+        bufferCache.createFile(file_1);
+        int fileId_1 = fmp.lookupFileId(file_1);
+        bufferCache.openFile(fileId_1);
+
+        TypeAwareTupleWriterFactory tupleWriterFactory = new TypeAwareTupleWriterFactory(typeTraits);
+        ITreeIndexFrameFactory leafFrameFactory = new BTreeNSMLeafFrameFactory(tupleWriterFactory);
+
+        IBTreeLeafFrame leafFrame = (IBTreeLeafFrame) leafFrameFactory.createFrame();
+        IBTreeInteriorFrame interiorFrame = (IBTreeInteriorFrame) interiorFrameFactory.createFrame();
+        ITreeIndexMetaDataFrame metaFrame = metaFrameFactory.createFrame();
+
+        IFreePageManager freePageManager_1 = new LinkedListFreePageManager(bufferCache, fileId_1, 0, metaFrameFactory);
+
+        BTree btree_1 = new BTree(bufferCache, fieldCount, cmp, freePageManager_1, interiorFrameFactory,
+                leafFrameFactory);
+        btree_1.create(fileId_1);
+        btree_1.open(fileId_1);
+
+        // TODO: rename this one.
+        InDiskTreeInfo info_1 = new InDiskTreeInfo(btree_1);
+        lsmtree.inDiskTreeInfoList.add(info_1);
+
+        Random rnd = new Random();
+        rnd.setSeed(50);
+
+        LOGGER.info("INSERTING INTO TREE");
+
+        // ByteBuffer
+        frame = ctx.allocateFrame();
+        // FrameTupleAppender
+        appender = new FrameTupleAppender(ctx.getFrameSize());
+        // ArrayTupleBuilder
+        tb = new ArrayTupleBuilder(fieldCount);
+        // DataOutput
+        dos = tb.getDataOutput();
+
+        recDesc = new RecordDescriptor(recDescSers);
+
+        accessor.reset(frame);
+
+        tuple = new FrameTupleReference();
+
+        ITreeIndexAccessor indexAccessor_1 = btree_1.createAccessor();
+
+        // 10000
+        for (int i = 16; i < 41; i++) {
+
+            int f0 = i;
+            int f1 = 1;
+
+            tb.reset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+            tb.addFieldEndOffset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+            tb.addFieldEndOffset();
+
+            appender.reset(frame, true);
+            appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+
+            tuple.reset(accessor, 0);
+
+            ArrayTupleReference t = new ArrayTupleReference();
+            t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+
+            if (i % 10 == 0) {
+                System.out.println("INSERTING " + i + " : " + f0 + " " + f1);
+            }
+
+            try {
+                indexAccessor_1.insert(t);
+            } catch (TreeIndexException e) {
+                e.printStackTrace();
+                System.out.println("Error: " + e);
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+        // btree_1.close();
+
+        // In disk insertions 2
+
+        LOGGER.info("Start in-disk insertions");
+
+        fileName = tmpDir + sep + simpleDateFormat.format(new Date());
+        FileReference file_2 = new FileReference(new File(fileName));
+        bufferCache.createFile(file_2);
+        int fileId_2 = fmp.lookupFileId(file_2);
+        bufferCache.openFile(fileId_2);
+
+        IFreePageManager freePageManager_2 = new LinkedListFreePageManager(bufferCache, fileId_2, 0, metaFrameFactory);
+        BTree btree_2 = new BTree(bufferCache, fieldCount, cmp, freePageManager_2, interiorFrameFactory,
+                leafFrameFactory);
+        btree_2.create(fileId_2);
+        btree_2.open(fileId_2);
+
+        InDiskTreeInfo info_2 = new InDiskTreeInfo(btree_2);
+        lsmtree.inDiskTreeInfoList.add(info_2);
+
+        LOGGER.info("INSERTING INTO TREE");
+
+        // ByteBuffer
+        frame = ctx.allocateFrame();
+        // FrameTupleAppender
+        appender = new FrameTupleAppender(ctx.getFrameSize());
+        // ArrayTupleBuilder
+        tb = new ArrayTupleBuilder(fieldCount);
+        // DataOutput
+        dos = tb.getDataOutput();
+
+        recDesc = new RecordDescriptor(recDescSers);
+
+        accessor.reset(frame);
+
+        tuple = new FrameTupleReference();
+
+        ITreeIndexAccessor indexAccessor_2 = btree_2.createAccessor();
+
+        // 10000
+        for (int i = 11; i < 51; i++) {
+
+            int f0 = i;
+            int f1 = 2;
+
+            tb.reset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f0, dos);
+            tb.addFieldEndOffset();
+            IntegerSerializerDeserializer.INSTANCE.serialize(f1, dos);
+            tb.addFieldEndOffset();
+
+            appender.reset(frame, true);
+            appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize());
+
+            tuple.reset(accessor, 0);
+
+            ArrayTupleReference t = new ArrayTupleReference();
+            t.reset(tb.getFieldEndOffsets(), tb.getByteArray());
+
+            if (i % 10 == 0) {
+                System.out.println("INSERTING " + i + " : " + f0 + " " + f1);
+            }
+
+            try {
+                indexAccessor_2.insert(t);
+            } catch (TreeIndexException e) {
+                e.printStackTrace();
+                System.out.println("Error: " + e);
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+        // btree_2.close();
+
+        // range search in [-1000, 1000]
+        LOGGER.info("RANGE SEARCH:");
+
+        ITreeIndexCursor rangeCursor = new LSMTreeRangeSearchCursor();
+
+        // build low and high keys
+        ArrayTupleBuilder ktb = new ArrayTupleBuilder(cmp.getKeyFieldCount());
+        DataOutput kdos = ktb.getDataOutput();
+
+        ISerializerDeserializer[] keyDescSers = { IntegerSerializerDeserializer.INSTANCE };
+        RecordDescriptor keyDesc = new RecordDescriptor(keyDescSers);
+        IFrameTupleAccessor keyAccessor = new FrameTupleAccessor(ctx.getFrameSize(), keyDesc);
+        keyAccessor.reset(frame);
+
+        appender.reset(frame, true);
+
+        // build and append low key
+        ktb.reset();
+        IntegerSerializerDeserializer.INSTANCE.serialize(-100, kdos);
+        ktb.addFieldEndOffset();
+        appender.append(ktb.getFieldEndOffsets(), ktb.getByteArray(), 0, ktb.getSize());
+
+        // build and append high key
+        ktb.reset();
+        IntegerSerializerDeserializer.INSTANCE.serialize(100, 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 = cmps;
+        MultiComparator searchCmp = new MultiComparator(searchCmps);
+
+        RangePredicate rangePred = new RangePredicate(true, lowKey, highKey, true, true, searchCmp, searchCmp);
+        lsmTreeAccessor.search(rangeCursor, rangePred);
+
+        try {
+            while (rangeCursor.hasNext()) {
+                rangeCursor.next();
+                ITupleReference frameTuple = rangeCursor.getTuple();
+                String rec = TupleUtils.printTuple(frameTuple, recDescSers);
+                if (((LSMTypeAwareTupleReference) frameTuple).isDelete()) {
+                    System.out.println("del " + rec);
+                } else {
+                    System.out.println("ins " + rec);
+                }
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+        } finally {
+            rangeCursor.close();
+        }
+
+        lsmtree.close();
+        bufferCache.closeFile(fileId);
+        memBufferCache.close();
+    }
+}
\ No newline at end of file
diff --git a/hyracks-tests/pom.xml b/hyracks-tests/pom.xml
index 0e46b3f..4e6e4db 100644
--- a/hyracks-tests/pom.xml
+++ b/hyracks-tests/pom.xml
@@ -16,6 +16,7 @@
     <module>hyracks-storage-am-btree-test</module>
     <module>hyracks-storage-am-invertedindex-test</module>
     <module>hyracks-storage-am-rtree-test</module>
+    <module>hyracks-storage-am-lsmtree-common-test</module>
     <module>hyracks-storage-am-lsmtree-btree-test</module>
     <module>hyracks-storage-am-lsmtree-rtree-test</module>
   </modules>