diff --git a/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/btree/impls/LSMBTree.java b/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/btree/impls/LSMBTree.java
index 4a916ce..651083c 100644
--- a/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/btree/impls/LSMBTree.java
+++ b/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/btree/impls/LSMBTree.java
@@ -16,7 +16,6 @@
 package edu.uci.ics.hyracks.storage.am.lsm.btree.impls;
 
 import java.io.File;
-import java.util.LinkedList;
 import java.util.List;
 import java.util.ListIterator;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -48,7 +47,7 @@
 import edu.uci.ics.hyracks.storage.am.common.ophelpers.MultiComparator;
 import edu.uci.ics.hyracks.storage.am.lsm.btree.tuples.LSMBTreeTupleReference;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.IInMemoryBufferCache;
-import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMComponentFinalizer;
+import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMComponent;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMFlushController;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMHarness;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIOOperation;
@@ -56,74 +55,53 @@
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIOOperationScheduler;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIndexAccessor;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIndexFileManager;
-import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIndexInternal;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIndexOperationContext;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMMergePolicy;
-import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMOperationTracker;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMOperationTrackerFactory;
 import edu.uci.ics.hyracks.storage.am.lsm.common.freepage.InMemoryBufferCache;
+import edu.uci.ics.hyracks.storage.am.lsm.common.impls.AbstractLSMIndex;
 import edu.uci.ics.hyracks.storage.am.lsm.common.impls.BlockingIOOperationCallback;
+import edu.uci.ics.hyracks.storage.am.lsm.common.impls.LSMComponentFileReferences;
 import edu.uci.ics.hyracks.storage.am.lsm.common.impls.LSMFlushOperation;
-import edu.uci.ics.hyracks.storage.am.lsm.common.impls.LSMHarness;
-import edu.uci.ics.hyracks.storage.am.lsm.common.impls.LSMMergeOperation;
 import edu.uci.ics.hyracks.storage.am.lsm.common.impls.LSMTreeIndexAccessor;
-import edu.uci.ics.hyracks.storage.am.lsm.common.impls.TreeIndexComponentFinalizer;
 import edu.uci.ics.hyracks.storage.am.lsm.common.impls.TreeIndexFactory;
 import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
 import edu.uci.ics.hyracks.storage.common.file.IFileMapProvider;
 
-public class LSMBTree implements ILSMIndexInternal, ITreeIndex {
-    private final ILSMHarness lsmHarness;
+public class LSMBTree extends AbstractLSMIndex implements ITreeIndex {
 
     // In-memory components.   
-    private final BTree memBTree;
-    private final IInMemoryFreePageManager memFreePageManager;
+    private final LSMBTreeComponent mutableComponent;
 
     // On-disk components.    
-    private final ILSMIndexFileManager fileManager;
     // For creating BTree's used in flush and merge.
-    private final TreeIndexFactory<BTree> diskBTreeFactory;
+    private final LSMBTreeComponentFactory componentFactory;
     // For creating BTree's used in bulk load. Different from diskBTreeFactory
     // because it should have a different tuple writer in it's leaf frames.
-    private final TreeIndexFactory<BTree> bulkLoadBTreeFactory;
-    private final IBufferCache diskBufferCache;
-    private final IFileMapProvider diskFileMapProvider;
-    // List of BTree instances. Using Object for better sharing via ILSMTree + LSMHarness.
-    private LinkedList<Object> diskBTrees;
-    // Helps to guarantees physical consistency of LSM components.
-    private final ILSMComponentFinalizer componentFinalizer;
+    private final LSMBTreeComponentFactory bulkLoadComponentFactory;
 
     // Common for in-memory and on-disk components.
     private final ITreeIndexFrameFactory insertLeafFrameFactory;
     private final ITreeIndexFrameFactory deleteLeafFrameFactory;
     private final IBinaryComparatorFactory[] cmpFactories;
 
-    private boolean isActivated;
-
     public LSMBTree(IInMemoryBufferCache memBufferCache, IInMemoryFreePageManager memFreePageManager,
             ITreeIndexFrameFactory interiorFrameFactory, ITreeIndexFrameFactory insertLeafFrameFactory,
-            ITreeIndexFrameFactory deleteLeafFrameFactory, ILSMIndexFileManager fileNameManager,
+            ITreeIndexFrameFactory deleteLeafFrameFactory, ILSMIndexFileManager fileManager,
             TreeIndexFactory<BTree> diskBTreeFactory, TreeIndexFactory<BTree> bulkLoadBTreeFactory,
             IFileMapProvider diskFileMapProvider, int fieldCount, IBinaryComparatorFactory[] cmpFactories,
             ILSMFlushController flushController, ILSMMergePolicy mergePolicy,
             ILSMOperationTrackerFactory opTrackerFactory, ILSMIOOperationScheduler ioScheduler) {
-        memBTree = new BTree(memBufferCache, ((InMemoryBufferCache) memBufferCache).getFileMapProvider(),
-                memFreePageManager, interiorFrameFactory, insertLeafFrameFactory, cmpFactories, fieldCount,
-                new FileReference(new File("membtree")));
-        this.memFreePageManager = memFreePageManager;
+        super(memFreePageManager, diskBTreeFactory.getBufferCache(), fileManager, diskFileMapProvider, flushController,
+                mergePolicy, opTrackerFactory, ioScheduler);
+        mutableComponent = new LSMBTreeComponent(new BTree(memBufferCache,
+                ((InMemoryBufferCache) memBufferCache).getFileMapProvider(), memFreePageManager, interiorFrameFactory,
+                insertLeafFrameFactory, cmpFactories, fieldCount, new FileReference(new File("membtree"))));
         this.insertLeafFrameFactory = insertLeafFrameFactory;
         this.deleteLeafFrameFactory = deleteLeafFrameFactory;
-        this.diskBufferCache = diskBTreeFactory.getBufferCache();
-        this.diskFileMapProvider = diskFileMapProvider;
-        this.diskBTreeFactory = diskBTreeFactory;
-        this.bulkLoadBTreeFactory = bulkLoadBTreeFactory;
         this.cmpFactories = cmpFactories;
-        this.diskBTrees = new LinkedList<Object>();
-        this.fileManager = fileNameManager;
-        ILSMOperationTracker opTracker = opTrackerFactory.createOperationTracker(this);
-        lsmHarness = new LSMHarness(this, flushController, mergePolicy, opTracker, ioScheduler);
-        componentFinalizer = new TreeIndexComponentFinalizer(diskFileMapProvider);
-        isActivated = false;
+        componentFactory = new LSMBTreeComponentFactory(diskBTreeFactory);
+        bulkLoadComponentFactory = new LSMBTreeComponentFactory(bulkLoadBTreeFactory);
     }
 
     @Override
@@ -134,7 +112,7 @@
 
         fileManager.deleteDirs();
         fileManager.createDirs();
-        diskBTrees.clear();
+        immutableComponents.clear();
     }
 
     @Override
@@ -143,16 +121,25 @@
             return;
         }
 
-        ((InMemoryBufferCache) memBTree.getBufferCache()).open();
-        memBTree.create();
-        memBTree.activate();
-        diskBTrees.clear();
-        List<Object> validFileNames = fileManager.cleanupAndGetValidFiles(componentFinalizer);
-        for (Object o : validFileNames) {
-            String fileName = (String) o;
-            FileReference fileRef = new FileReference(new File(fileName));
-            BTree btree = createDiskBTree(diskBTreeFactory, fileRef, false);
-            diskBTrees.add(btree);
+        ((InMemoryBufferCache) mutableComponent.getBTree().getBufferCache()).open();
+        mutableComponent.getBTree().create();
+        mutableComponent.getBTree().activate();
+        immutableComponents.clear();
+        List<LSMComponentFileReferences> validFileReferences;
+        try {
+            validFileReferences = fileManager.cleanupAndGetValidFiles();
+        } catch (IndexException e) {
+            throw new HyracksDataException(e);
+        }
+        for (LSMComponentFileReferences lsmComonentFileReference : validFileReferences) {
+            LSMBTreeComponent btree;
+            try {
+                btree = createDiskComponent(componentFactory, lsmComonentFileReference.getInsertIndexFileReference(),
+                        false);
+            } catch (IndexException e) {
+                throw new HyracksDataException(e);
+            }
+            immutableComponents.add(btree);
         }
         isActivated = true;
     }
@@ -172,13 +159,13 @@
             throw new HyracksDataException(e);
         }
 
-        for (Object o : diskBTrees) {
-            BTree btree = (BTree) o;
+        for (ILSMComponent c : immutableComponents) {
+            BTree btree = (BTree) ((LSMBTreeComponent) c).getBTree();
             btree.deactivate();
-        }        
-        memBTree.deactivate();
-        memBTree.destroy();
-        ((InMemoryBufferCache) memBTree.getBufferCache()).close();
+        }
+        mutableComponent.getBTree().deactivate();
+        mutableComponent.getBTree().destroy();
+        ((InMemoryBufferCache) mutableComponent.getBTree().getBufferCache()).close();
         isActivated = false;
     }
 
@@ -188,11 +175,11 @@
             throw new HyracksDataException("Failed to destroy the index since it is activated.");
         }
 
-        for (Object o : diskBTrees) {
-            BTree btree = (BTree) o;
+        for (ILSMComponent c : immutableComponents) {
+            BTree btree = (BTree) ((LSMBTreeComponent) c).getBTree();
             btree.destroy();
         }
-        memBTree.destroy();
+        mutableComponent.getBTree().destroy();
         fileManager.deleteDirs();
     }
 
@@ -202,13 +189,13 @@
             throw new HyracksDataException("Failed to clear the index since it is not activated.");
         }
 
-        memBTree.clear();
-        for (Object o : diskBTrees) {
-            BTree btree = (BTree) o;
+        mutableComponent.getBTree().clear();
+        for (ILSMComponent c : immutableComponents) {
+            BTree btree = (BTree) ((LSMBTreeComponent) c).getBTree();
             btree.deactivate();
             btree.destroy();
         }
-        diskBTrees.clear();
+        immutableComponents.clear();
     }
 
     @Override
@@ -229,7 +216,8 @@
     }
 
     private boolean insert(ITupleReference tuple, LSMBTreeOpContext ctx) throws HyracksDataException, IndexException {
-        MultiComparator comparator = MultiComparator.createIgnoreFieldLength(memBTree.getComparatorFactories());
+        MultiComparator comparator = MultiComparator.createIgnoreFieldLength(mutableComponent.getBTree()
+                .getComparatorFactories());
         LSMBTreeRangeSearchCursor searchCursor = new LSMBTreeRangeSearchCursor(ctx);
         IIndexCursor memCursor = new BTreeRangeSearchCursor(ctx.memBTreeOpCtx.leafFrame, false);
         RangePredicate predicate = new RangePredicate(tuple, tuple, true, true, comparator, comparator);
@@ -267,17 +255,17 @@
     }
 
     @Override
-    public ITreeIndex flush(ILSMIOOperation operation) throws HyracksDataException, IndexException {
+    public ILSMComponent flush(ILSMIOOperation operation) throws HyracksDataException, IndexException {
         LSMFlushOperation flushOp = (LSMFlushOperation) operation;
         // Bulk load a new on-disk BTree from the in-memory BTree.        
         RangePredicate nullPred = new RangePredicate(null, null, true, true, null, null);
-        ITreeIndexAccessor memBTreeAccessor = memBTree.createAccessor(NoOpOperationCallback.INSTANCE,
-                NoOpOperationCallback.INSTANCE);
+        ITreeIndexAccessor memBTreeAccessor = mutableComponent.getBTree().createAccessor(
+                NoOpOperationCallback.INSTANCE, NoOpOperationCallback.INSTANCE);
         IIndexCursor scanCursor = memBTreeAccessor.createSearchCursor();
         memBTreeAccessor.search(scanCursor, nullPred);
-        BTree diskBTree = createDiskBTree(diskBTreeFactory, flushOp.getFlushTarget(), true);
+        LSMBTreeComponent component = createDiskComponent(componentFactory, flushOp.getFlushTarget(), true);
         // Bulk load the tuples from the in-memory BTree into the new disk BTree.
-        IIndexBulkLoader bulkLoader = diskBTree.createBulkLoader(1.0f, false);
+        IIndexBulkLoader bulkLoader = component.getBTree().createBulkLoader(1.0f, false);
         try {
             while (scanCursor.hasNext()) {
                 scanCursor.next();
@@ -287,45 +275,41 @@
             scanCursor.close();
         }
         bulkLoader.end();
-        return diskBTree;
+        return component;
     }
 
     @Override
-    public void addFlushedComponent(Object index) {
-        diskBTrees.addFirst(index);
-    }
-
-    @Override
-    public void resetInMemoryComponent() throws HyracksDataException {
+    public void resetMutableComponent() throws HyracksDataException {
         memFreePageManager.reset();
-        memBTree.clear();
+        mutableComponent.getBTree().clear();
     }
 
-    private BTree createBulkLoadTarget() throws HyracksDataException {
-        String relFlushFileName = (String) fileManager.getRelFlushFileName();
-        FileReference fileRef = fileManager.createFlushFile(relFlushFileName);
-        return createDiskBTree(bulkLoadBTreeFactory, fileRef, true);
+    private ILSMComponent createBulkLoadTarget() throws HyracksDataException, IndexException {
+        LSMComponentFileReferences componentFileRefs = fileManager.getRelFlushFileReference();
+        return createDiskComponent(bulkLoadComponentFactory, componentFileRefs.getInsertIndexFileReference(), true);
     }
 
-    private BTree createDiskBTree(TreeIndexFactory<BTree> factory, FileReference fileRef, boolean createBTree)
-            throws HyracksDataException {
+    private LSMBTreeComponent createDiskComponent(LSMBTreeComponentFactory factory, FileReference fileRef,
+            boolean createComponent) throws HyracksDataException, IndexException {
         // Create new BTree instance.
-        BTree diskBTree = factory.createIndexInstance(fileRef);
-        if (createBTree) {
-            diskBTree.create();
+        LSMBTreeComponent component = (LSMBTreeComponent) factory
+                .createLSMComponentInstance(new LSMComponentFileReferences(fileRef, null));
+        if (createComponent) {
+            component.getBTree().create();
         }
         // BTree will be closed during cleanup of merge().
-        diskBTree.activate();
-        return diskBTree;
+        component.getBTree().activate();
+        return component;
     }
 
-    public void search(IIndexCursor cursor, List<Object> diskComponents, ISearchPredicate pred,
+    @Override
+    public void search(IIndexCursor cursor, List<ILSMComponent> diskComponents, ISearchPredicate pred,
             IIndexOperationContext ictx, boolean includeMemComponent, AtomicInteger searcherRefCount)
             throws HyracksDataException, IndexException {
         LSMBTreeOpContext ctx = (LSMBTreeOpContext) ictx;
         LSMBTreeRangeSearchCursor lsmTreeCursor = (LSMBTreeRangeSearchCursor) cursor;
-        int numDiskBTrees = diskComponents.size();
-        int numBTrees = (includeMemComponent) ? numDiskBTrees + 1 : numDiskBTrees;
+        int numDiskComponents = diskComponents.size();
+        int numBTrees = (includeMemComponent) ? numDiskComponents + 1 : numDiskComponents;
         LSMBTreeCursorInitialState initialState = new LSMBTreeCursorInitialState(numBTrees, insertLeafFrameFactory,
                 ctx.cmp, includeMemComponent, searcherRefCount, lsmHarness, ctx.memBTreeAccessor, pred,
                 ctx.searchCallback);
@@ -342,11 +326,11 @@
         }
 
         // Open cursors of on-disk BTrees.
-        ITreeIndexAccessor[] diskBTreeAccessors = new ITreeIndexAccessor[numDiskBTrees];
+        ITreeIndexAccessor[] diskBTreeAccessors = new ITreeIndexAccessor[numDiskComponents];
         int diskBTreeIx = 0;
-        ListIterator<Object> diskBTreesIter = diskComponents.listIterator();
+        ListIterator<ILSMComponent> diskBTreesIter = diskComponents.listIterator();
         while (diskBTreesIter.hasNext()) {
-            BTree diskBTree = (BTree) diskBTreesIter.next();
+            BTree diskBTree = (BTree) ((LSMBTreeComponent) diskBTreesIter.next()).getBTree();
             diskBTreeAccessors[diskBTreeIx] = diskBTree.createAccessor(NoOpOperationCallback.INSTANCE,
                     NoOpOperationCallback.INSTANCE);
             diskBTreeAccessors[diskBTreeIx].search(lsmTreeCursor.getCursor(cursorIx), pred);
@@ -356,9 +340,10 @@
         lsmTreeCursor.initPriorityQueue();
     }
 
-    public ITreeIndex merge(List<Object> mergedComponents, ILSMIOOperation operation) throws HyracksDataException,
-            IndexException {
-        LSMMergeOperation mergeOp = (LSMMergeOperation) operation;
+    @Override
+    public ILSMComponent merge(List<ILSMComponent> mergedComponents, ILSMIOOperation operation)
+            throws HyracksDataException, IndexException {
+        LSMBTreeMergeOperation mergeOp = (LSMBTreeMergeOperation) operation;
         ITreeIndexCursor cursor = mergeOp.getCursor();
 
         mergedComponents.addAll(mergeOp.getMergingComponents());
@@ -370,8 +355,8 @@
         }
 
         // Bulk load the tuples from all on-disk BTrees into the new BTree.
-        BTree mergedBTree = createDiskBTree(diskBTreeFactory, mergeOp.getMergeTarget(), true);
-        IIndexBulkLoader bulkLoader = mergedBTree.createBulkLoader(1.0f, false);
+        LSMBTreeComponent mergedBTree = createDiskComponent(componentFactory, mergeOp.getMergeTarget(), true);
+        IIndexBulkLoader bulkLoader = mergedBTree.getBTree().createBulkLoader(1.0f, false);
         try {
             while (cursor.hasNext()) {
                 cursor.next();
@@ -386,46 +371,33 @@
     }
 
     @Override
-    public void addMergedComponent(Object newComponent, List<Object> mergedComponents) {
-        diskBTrees.removeAll(mergedComponents);
-        diskBTrees.addLast(newComponent);
-    }
-
-    @Override
-    public void cleanUpAfterMerge(List<Object> mergedComponents) throws HyracksDataException {
-        for (Object o : mergedComponents) {
-            BTree oldBTree = (BTree) o;
+    public void cleanUpAfterMerge(List<ILSMComponent> mergedComponents) throws HyracksDataException {
+        for (ILSMComponent c : mergedComponents) {
+            BTree oldBTree = (BTree) ((LSMBTreeComponent) c).getBTree();
             oldBTree.deactivate();
             oldBTree.destroy();
         }
     }
 
     @Override
-    public IInMemoryFreePageManager getInMemoryFreePageManager() {
-        return memFreePageManager;
-    }
-
-    @Override
-    public List<Object> getDiskComponents() {
-        return diskBTrees;
-    }
-
-    @Override
     public IIndexBulkLoader createBulkLoader(float fillLevel, boolean verifyInput) throws TreeIndexException {
         return new LSMBTreeBulkLoader(fillLevel, verifyInput);
     }
 
     public class LSMBTreeBulkLoader implements IIndexBulkLoader {
-        private final BTree diskBTree;
+        private final ILSMComponent component;
         private final BTreeBulkLoader bulkLoader;
 
         public LSMBTreeBulkLoader(float fillFactor, boolean verifyInput) throws TreeIndexException {
             try {
-                diskBTree = createBulkLoadTarget();
+                component = createBulkLoadTarget();
             } catch (HyracksDataException e) {
                 throw new TreeIndexException(e);
+            } catch (IndexException e) {
+                throw new TreeIndexException(e);
             }
-            bulkLoader = (BTreeBulkLoader) diskBTree.createBulkLoader(1.0f, verifyInput);
+            bulkLoader = (BTreeBulkLoader) ((LSMBTreeComponent) component).getBTree().createBulkLoader(fillFactor,
+                    verifyInput);
         }
 
         @Override
@@ -445,46 +417,46 @@
         }
 
         protected void handleException() throws HyracksDataException {
-            diskBTree.deactivate();
-            diskBTree.destroy();
+            ((LSMBTreeComponent) component).getBTree().deactivate();
+            ((LSMBTreeComponent) component).getBTree().destroy();
         }
 
         @Override
         public void end() throws HyracksDataException, IndexException {
             bulkLoader.end();
-            lsmHarness.addBulkLoadedComponent(diskBTree);
+            lsmHarness.addBulkLoadedComponent(component);
         }
 
     }
 
     @Override
     public ITreeIndexFrameFactory getLeafFrameFactory() {
-        return memBTree.getLeafFrameFactory();
+        return mutableComponent.getBTree().getLeafFrameFactory();
     }
 
     @Override
     public ITreeIndexFrameFactory getInteriorFrameFactory() {
-        return memBTree.getInteriorFrameFactory();
+        return mutableComponent.getBTree().getInteriorFrameFactory();
     }
 
     @Override
     public IFreePageManager getFreePageManager() {
-        return memBTree.getFreePageManager();
+        return mutableComponent.getBTree().getFreePageManager();
     }
 
     @Override
     public int getFieldCount() {
-        return memBTree.getFieldCount();
+        return mutableComponent.getBTree().getFieldCount();
     }
 
     @Override
     public int getRootPageId() {
-        return memBTree.getRootPageId();
+        return mutableComponent.getBTree().getRootPageId();
     }
 
     @Override
     public int getFileId() {
-        return memBTree.getFileId();
+        return mutableComponent.getBTree().getFileId();
     }
 
     public IBinaryComparatorFactory[] getComparatorFactories() {
@@ -493,8 +465,8 @@
 
     public LSMBTreeOpContext createOpContext(IModificationOperationCallback modificationCallback,
             ISearchOperationCallback searchCallback) {
-        return new LSMBTreeOpContext(memBTree, insertLeafFrameFactory, deleteLeafFrameFactory, modificationCallback,
-                searchCallback);
+        return new LSMBTreeOpContext(mutableComponent.getBTree(), insertLeafFrameFactory, deleteLeafFrameFactory,
+                modificationCallback, searchCallback);
     }
 
     @Override
@@ -520,9 +492,9 @@
 
         @Override
         public ILSMIOOperation createFlushOperation(ILSMIOOperationCallback callback) {
-            String relFlushFileName = (String) fileManager.getRelFlushFileName();
-            FileReference fileRef = fileManager.createFlushFile(relFlushFileName);
-            return new LSMFlushOperation(lsmHarness.getIndex(), fileRef, callback);
+            LSMComponentFileReferences componentFileRefs = fileManager.getRelFlushFileReference();
+            return new LSMFlushOperation(lsmHarness.getIndex(), componentFileRefs.getInsertIndexFileReference(),
+                    callback);
         }
     }
 
@@ -533,10 +505,10 @@
         // Ordered scan, ignoring the in-memory BTree.
         // We get back a snapshot of the on-disk BTrees that are going to be
         // merged now, so we can clean them up after the merge has completed.
-        List<Object> mergingDiskBTrees;
+        List<ILSMComponent> mergingDiskComponents;
         try {
-            mergingDiskBTrees = lsmHarness.search(cursor, (RangePredicate) rangePred, ctx, false);
-            if (mergingDiskBTrees.size() <= 1) {
+            mergingDiskComponents = lsmHarness.search(cursor, (RangePredicate) rangePred, ctx, false);
+            if (mergingDiskComponents.size() <= 1) {
                 cursor.close();
                 return null;
             }
@@ -544,19 +516,22 @@
             throw new HyracksDataException(e);
         }
 
-        BTree firstBTree = (BTree) mergingDiskBTrees.get(0);
-        BTree lastBTree = (BTree) mergingDiskBTrees.get(mergingDiskBTrees.size() - 1);
+        BTree firstBTree = (BTree) ((LSMBTreeComponent) mergingDiskComponents.get(0)).getBTree();
+        BTree lastBTree = (BTree) ((LSMBTreeComponent) mergingDiskComponents.get(mergingDiskComponents.size() - 1))
+                .getBTree();
         FileReference firstFile = diskFileMapProvider.lookupFileName(firstBTree.getFileId());
         FileReference lastFile = diskFileMapProvider.lookupFileName(lastBTree.getFileId());
-        String relMergeFileName = (String) fileManager.getRelMergeFileName(firstFile.getFile().getName(), lastFile
-                .getFile().getName());
-        FileReference fileRef = fileManager.createMergeFile(relMergeFileName);
-        return new LSMMergeOperation(lsmHarness.getIndex(), mergingDiskBTrees, cursor, fileRef, callback);
+        LSMComponentFileReferences relMergeFileRefs = fileManager.getRelMergeFileReference(firstFile.getFile()
+                .getName(), lastFile.getFile().getName());
+        return new LSMBTreeMergeOperation(lsmHarness.getIndex(), mergingDiskComponents, cursor,
+                relMergeFileRefs.getInsertIndexFileReference(), callback);
     }
 
     @Override
-    public ILSMComponentFinalizer getComponentFinalizer() {
-        return componentFinalizer;
+    public void markAsValid(ILSMComponent lsmComponent) throws HyracksDataException {
+        BTree btree = ((LSMBTreeComponent) lsmComponent).getBTree();
+        forceFlushDirtyPages(btree);
+        markAsValidInternal(btree);
     }
 
     @Override
@@ -564,37 +539,24 @@
         return diskBufferCache;
     }
 
-    @Override
-    public ILSMFlushController getFlushController() {
-        return lsmHarness.getFlushController();
-    }
-
-    @Override
-    public ILSMOperationTracker getOperationTracker() {
-        return lsmHarness.getOperationTracker();
-    }
-
-    @Override
-    public ILSMIOOperationScheduler getIOScheduler() {
-        return lsmHarness.getIOScheduler();
-    }
-
     public boolean isEmptyIndex() throws HyracksDataException {
-        return diskBTrees.isEmpty() && memBTree.isEmptyTree(memBTree.getInteriorFrameFactory().createFrame());
+        return immutableComponents.isEmpty()
+                && mutableComponent.getBTree().isEmptyTree(
+                        mutableComponent.getBTree().getInteriorFrameFactory().createFrame());
     }
 
     @Override
     public void validate() throws HyracksDataException {
-        memBTree.validate();
-        for (Object o : diskBTrees) {
-            BTree btree = (BTree) o;
+        mutableComponent.getBTree().validate();
+        for (ILSMComponent c : immutableComponents) {
+            BTree btree = (BTree) ((LSMBTreeComponent) c).getBTree();
             btree.validate();
         }
     }
 
     @Override
     public long getMemoryAllocationSize() {
-        InMemoryBufferCache memBufferCache = (InMemoryBufferCache) memBTree.getBufferCache();
+        InMemoryBufferCache memBufferCache = (InMemoryBufferCache) mutableComponent.getBTree().getBufferCache();
         return memBufferCache.getNumPages() * memBufferCache.getPageSize();
     }
 }
diff --git a/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/btree/impls/LSMBTreeComponent.java b/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/btree/impls/LSMBTreeComponent.java
new file mode 100644
index 0000000..b89e332
--- /dev/null
+++ b/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/btree/impls/LSMBTreeComponent.java
@@ -0,0 +1,70 @@
+/*
+ * Copyright 2009-2012 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.lsm.btree.impls;
+
+import edu.uci.ics.hyracks.storage.am.btree.impls.BTree;
+import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMComponent;
+import edu.uci.ics.hyracks.storage.am.lsm.common.impls.LSMComponentState;
+
+public class LSMBTreeComponent implements ILSMComponent {
+
+    private final BTree btree;
+
+    public LSMBTreeComponent(BTree btree) {
+        this.btree = btree;
+    }
+
+    @Override
+    public void activate() {
+        // TODO Auto-generated method stub
+
+    }
+
+    @Override
+    public void deactivate() {
+        // TODO Auto-generated method stub
+
+    }
+
+    @Override
+    public void threadEnter() {
+        // TODO Auto-generated method stub
+
+    }
+
+    @Override
+    public void threadExit() {
+        // TODO Auto-generated method stub
+
+    }
+
+    @Override
+    public void setState(LSMComponentState state) {
+        // TODO Auto-generated method stub
+
+    }
+
+    @Override
+    public LSMComponentState getState() {
+        // TODO Auto-generated method stub
+        return null;
+    }
+
+    public BTree getBTree() {
+        return btree;
+    }
+
+}
diff --git a/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/btree/impls/LSMBTreeComponentFactory.java b/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/btree/impls/LSMBTreeComponentFactory.java
new file mode 100644
index 0000000..23e0d83
--- /dev/null
+++ b/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/btree/impls/LSMBTreeComponentFactory.java
@@ -0,0 +1,42 @@
+/*
+ * Copyright 2009-2012 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.lsm.btree.impls;
+
+import edu.uci.ics.hyracks.storage.am.btree.impls.BTree;
+import edu.uci.ics.hyracks.storage.am.common.api.IndexException;
+import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMComponent;
+import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMComponentFactory;
+import edu.uci.ics.hyracks.storage.am.lsm.common.impls.LSMComponentFileReferences;
+import edu.uci.ics.hyracks.storage.am.lsm.common.impls.TreeIndexFactory;
+import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
+
+public class LSMBTreeComponentFactory implements ILSMComponentFactory {
+    private final TreeIndexFactory<BTree> btreeFactory;
+
+    public LSMBTreeComponentFactory(TreeIndexFactory<BTree> btreeFactory) {
+        this.btreeFactory = btreeFactory;
+    }
+
+    @Override
+    public ILSMComponent createLSMComponentInstance(LSMComponentFileReferences cfr) throws IndexException {
+        return new LSMBTreeComponent(btreeFactory.createIndexInstance(cfr.getInsertIndexFileReference()));
+    }
+
+    @Override
+    public IBufferCache getBufferCache() {
+        return btreeFactory.getBufferCache();
+    }
+}
diff --git a/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/LSMMergeOperation.java b/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/btree/impls/LSMBTreeMergeOperation.java
similarity index 62%
rename from hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/LSMMergeOperation.java
rename to hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/btree/impls/LSMBTreeMergeOperation.java
index a38f82b..ec8afaf 100644
--- a/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/LSMMergeOperation.java
+++ b/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/btree/impls/LSMBTreeMergeOperation.java
@@ -1,4 +1,19 @@
-package edu.uci.ics.hyracks.storage.am.lsm.common.impls;
+/*
+ * Copyright 2009-2012 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.lsm.btree.impls;
 
 import java.util.ArrayList;
 import java.util.Collections;
@@ -9,22 +24,22 @@
 import edu.uci.ics.hyracks.api.io.IODeviceHandle;
 import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexCursor;
 import edu.uci.ics.hyracks.storage.am.common.api.IndexException;
-import edu.uci.ics.hyracks.storage.am.common.impls.AbstractTreeIndex;
 import edu.uci.ics.hyracks.storage.am.common.impls.NoOpOperationCallback;
+import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMComponent;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIOOperation;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallback;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIndex;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIndexAccessor;
 
-public class LSMMergeOperation implements ILSMIOOperation {
+public class LSMBTreeMergeOperation implements ILSMIOOperation {
 
     private final ILSMIndex index;
-    private final List<Object> mergingComponents;
+    private final List<ILSMComponent> mergingComponents;
     private final ITreeIndexCursor cursor;
     private final FileReference mergeTarget;
     private final ILSMIOOperationCallback callback;
 
-    public LSMMergeOperation(ILSMIndex index, List<Object> mergingComponents, ITreeIndexCursor cursor,
+    public LSMBTreeMergeOperation(ILSMIndex index, List<ILSMComponent> mergingComponents, ITreeIndexCursor cursor,
             FileReference mergeTarget, ILSMIOOperationCallback callback) {
         this.index = index;
         this.mergingComponents = mergingComponents;
@@ -36,9 +51,9 @@
     @Override
     public List<IODeviceHandle> getReadDevices() {
         List<IODeviceHandle> devs = new ArrayList<IODeviceHandle>();
-        for (Object o : mergingComponents) {
-            AbstractTreeIndex idx = (AbstractTreeIndex) o;
-            devs.add(idx.getFileReference().getDevideHandle());
+        for (ILSMComponent o : mergingComponents) {
+            LSMBTreeComponent component = (LSMBTreeComponent) o;
+            devs.add(component.getBTree().getFileReference().getDevideHandle());
         }
         return devs;
     }
@@ -68,7 +83,7 @@
         return cursor;
     }
 
-    public List<Object> getMergingComponents() {
+    public List<ILSMComponent> getMergingComponents() {
         return mergingComponents;
     }
 }
diff --git a/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/btree/util/LSMBTreeUtils.java b/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/btree/util/LSMBTreeUtils.java
index 9fc7b92..df9045f 100644
--- a/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/btree/util/LSMBTreeUtils.java
+++ b/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/btree/util/LSMBTreeUtils.java
@@ -62,6 +62,7 @@
         ITreeIndexMetaDataFrameFactory metaFrameFactory = new LIFOMetaDataFrameFactory();
         IFreePageManagerFactory freePageManagerFactory = new LinkedListFreePageManagerFactory(diskBufferCache,
                 metaFrameFactory);
+
         TreeIndexFactory<BTree> diskBTreeFactory = new BTreeFactory(diskBufferCache, diskFileMapProvider,
                 freePageManagerFactory, interiorFrameFactory, copyTupleLeafFrameFactory, cmpFactories,
                 typeTraits.length);
@@ -69,6 +70,7 @@
                 freePageManagerFactory, interiorFrameFactory, insertLeafFrameFactory, cmpFactories, typeTraits.length);
         ILSMIndexFileManager fileNameManager = new LSMIndexFileManager(ioManager, diskFileMapProvider, file,
                 diskBTreeFactory);
+
         LSMBTree lsmTree = new LSMBTree(memBufferCache, memFreePageManager, interiorFrameFactory,
                 insertLeafFrameFactory, deleteLeafFrameFactory, fileNameManager, diskBTreeFactory,
                 bulkLoadBTreeFactory, diskFileMapProvider, typeTraits.length, cmpFactories, flushController,
diff --git a/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/api/ILSMComponent.java b/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/api/ILSMComponent.java
new file mode 100644
index 0000000..c1e5eb2
--- /dev/null
+++ b/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/api/ILSMComponent.java
@@ -0,0 +1,17 @@
+package edu.uci.ics.hyracks.storage.am.lsm.common.api;
+
+import edu.uci.ics.hyracks.storage.am.lsm.common.impls.LSMComponentState;
+
+public interface ILSMComponent {
+    public void activate();
+
+    public void deactivate();
+
+    public void threadEnter();
+
+    public void threadExit();
+
+    public void setState(LSMComponentState state);
+
+    public LSMComponentState getState();
+}
diff --git a/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/api/ILSMComponentFactory.java b/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/api/ILSMComponentFactory.java
new file mode 100644
index 0000000..d00b805
--- /dev/null
+++ b/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/api/ILSMComponentFactory.java
@@ -0,0 +1,11 @@
+package edu.uci.ics.hyracks.storage.am.lsm.common.api;
+
+import edu.uci.ics.hyracks.storage.am.common.api.IndexException;
+import edu.uci.ics.hyracks.storage.am.lsm.common.impls.LSMComponentFileReferences;
+import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
+
+public interface ILSMComponentFactory {
+    public ILSMComponent createLSMComponentInstance(LSMComponentFileReferences cfr) throws IndexException;
+
+    public IBufferCache getBufferCache();
+}
diff --git a/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/api/ILSMComponentFinalizer.java b/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/api/ILSMComponentFinalizer.java
deleted file mode 100644
index 86f709a..0000000
--- a/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/api/ILSMComponentFinalizer.java
+++ /dev/null
@@ -1,34 +0,0 @@
-/*
- * Copyright 2009-2012 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.lsm.common.api;
-
-
-import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
-
-public interface ILSMComponentFinalizer {
-
-    /**
-     * Checks whether the given file is valid with respect to the given LSM component.
-     * Used for guaranteeing the atomicity of LSM component writes.
-     */
-    public boolean isValid(Object lsmComponent) throws HyracksDataException;
-
-    /**
-     * Marks the given LSM component as physically valid, synchronously forcing
-     * the necessary information to disk.
-     */
-    public void finalize(Object lsmComponent) throws HyracksDataException;
-}
diff --git a/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/api/ILSMHarness.java b/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/api/ILSMHarness.java
index d46c5c2..047b59b 100644
--- a/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/api/ILSMHarness.java
+++ b/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/api/ILSMHarness.java
@@ -30,7 +30,7 @@
     
     public boolean noOp(ILSMIndexOperationContext ictx, boolean tryOperation) throws HyracksDataException;            
 
-    public List<Object> search(IIndexCursor cursor, ISearchPredicate pred, ILSMIndexOperationContext ctx,
+    public List<ILSMComponent> search(IIndexCursor cursor, ISearchPredicate pred, ILSMIndexOperationContext ctx,
             boolean includeMemComponent) throws HyracksDataException, IndexException;
 
     public void closeSearchCursor(AtomicInteger searcherRefCount, boolean includeMemComponent, ILSMIndexOperationContext ctx)
@@ -43,7 +43,7 @@
 
     public void flush(ILSMIOOperation operation) throws HyracksDataException, IndexException;
 
-    public void addBulkLoadedComponent(Object index) throws HyracksDataException, IndexException;
+    public void addBulkLoadedComponent(ILSMComponent index) throws HyracksDataException, IndexException;
 
     public ILSMIndex getIndex();
 
diff --git a/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/api/ILSMIOOperationCallback.java b/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/api/ILSMIOOperationCallback.java
index f6e2b75..48416fd 100644
--- a/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/api/ILSMIOOperationCallback.java
+++ b/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/api/ILSMIOOperationCallback.java
@@ -7,8 +7,8 @@
 public interface ILSMIOOperationCallback {
     public void beforeOperation(ILSMIOOperation operation) throws HyracksDataException;
 
-    public void afterOperation(ILSMIOOperation operation, List<Object> oldComponents, Object newComponent)
+    public void afterOperation(ILSMIOOperation operation, List<ILSMComponent> oldComponents, ILSMComponent newComponent)
             throws HyracksDataException;
 
-    public void afterFinalize(ILSMIOOperation operation, Object newComponent) throws HyracksDataException;
+    public void afterFinalize(ILSMIOOperation operation, ILSMComponent newComponent) throws HyracksDataException;
 }
diff --git a/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/api/ILSMIOOperationCallbackFactory.java b/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/api/ILSMIOOperationCallbackFactory.java
deleted file mode 100644
index 52361ee..0000000
--- a/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/api/ILSMIOOperationCallbackFactory.java
+++ /dev/null
@@ -1,7 +0,0 @@
-package edu.uci.ics.hyracks.storage.am.lsm.common.api;
-
-import java.io.Serializable;
-
-public interface ILSMIOOperationCallbackFactory extends Serializable {
-    public ILSMIOOperationCallback createIOOperationCallback(Object syncObj);
-}
diff --git a/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/api/ILSMIndexFileManager.java b/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/api/ILSMIndexFileManager.java
index 9320053..b3f7913 100644
--- a/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/api/ILSMIndexFileManager.java
+++ b/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/api/ILSMIndexFileManager.java
@@ -20,6 +20,8 @@
 
 import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
 import edu.uci.ics.hyracks.api.io.FileReference;
+import edu.uci.ics.hyracks.storage.am.common.api.IndexException;
+import edu.uci.ics.hyracks.storage.am.lsm.common.impls.LSMComponentFileReferences;
 
 /**
  * Provides file names for LSM on-disk components. Also cleans up invalid files.
@@ -37,15 +39,16 @@
 
     public FileReference createMergeFile(String relMergeFileName);
 
-    public Object getRelFlushFileName();
+    public LSMComponentFileReferences getRelFlushFileReference();
 
-    public Object getRelMergeFileName(String firstFileName, String lastFileName) throws HyracksDataException;
+    public LSMComponentFileReferences getRelMergeFileReference(String firstFileName, String lastFileName)
+            throws HyracksDataException;
 
     public String getBaseDir();
 
     // Deletes invalid files, and returns list of valid files from baseDir.
     // The returned valid files are correctly sorted (based on the recency of data). 
-    public List<Object> cleanupAndGetValidFiles(ILSMComponentFinalizer componentFinalizer) throws HyracksDataException;
+    public List<LSMComponentFileReferences> cleanupAndGetValidFiles() throws HyracksDataException, IndexException;
 
     public Comparator<String> getFileNameComparator();
 }
diff --git a/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/api/ILSMIndexInternal.java b/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/api/ILSMIndexInternal.java
index 62f5a15..4dc442f 100644
--- a/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/api/ILSMIndexInternal.java
+++ b/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/api/ILSMIndexInternal.java
@@ -1,3 +1,18 @@
+/*
+ * Copyright 2009-2012 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.lsm.common.api;
 
 import java.util.List;
@@ -13,32 +28,33 @@
 
 public interface ILSMIndexInternal extends ILSMIndex {
 
-    public Object merge(List<Object> mergedComponents, ILSMIOOperation operation) throws HyracksDataException,
-            IndexException;
+    public ILSMComponent merge(List<ILSMComponent> mergedComponents, ILSMIOOperation operation)
+            throws HyracksDataException, IndexException;
 
     public void insertUpdateOrDelete(ITupleReference tuple, IIndexOperationContext ictx) throws HyracksDataException,
             IndexException;
 
-    public void search(IIndexCursor cursor, List<Object> diskComponents, ISearchPredicate pred,
+    public void search(IIndexCursor cursor, List<ILSMComponent> diskComponents, ISearchPredicate pred,
             IIndexOperationContext ictx, boolean includeMemComponent, AtomicInteger searcherRefCount)
             throws HyracksDataException, IndexException;
 
     public ILSMIOOperation createMergeOperation(ILSMIOOperationCallback callback) throws HyracksDataException,
             IndexException;
 
-    public void addMergedComponent(Object newComponent, List<Object> mergedComponents);
+    public void addMergedComponent(ILSMComponent newComponent, List<ILSMComponent> mergedComponents);
 
-    public void cleanUpAfterMerge(List<Object> mergedComponents) throws HyracksDataException;
+    public void cleanUpAfterMerge(List<ILSMComponent> mergedComponents) throws HyracksDataException;
 
-    public Object flush(ILSMIOOperation operation) throws HyracksDataException, IndexException;
+    public ILSMComponent flush(ILSMIOOperation operation) throws HyracksDataException, IndexException;
 
-    public void addFlushedComponent(Object index);
+    public void addFlushedComponent(ILSMComponent index);
 
     public IInMemoryFreePageManager getInMemoryFreePageManager();
 
-    public void resetInMemoryComponent() throws HyracksDataException;
+    public void resetMutableComponent() throws HyracksDataException;
 
-    public List<Object> getDiskComponents();
+    public List<ILSMComponent> getImmutableComponents();
 
-    public ILSMComponentFinalizer getComponentFinalizer();
+    public void markAsValid(ILSMComponent lsmComponent) throws HyracksDataException;
+
 }
diff --git a/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/AbstractLSMIndex.java b/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/AbstractLSMIndex.java
new file mode 100644
index 0000000..9ba101b
--- /dev/null
+++ b/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/AbstractLSMIndex.java
@@ -0,0 +1,160 @@
+/*
+ * Copyright 2009-2012 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.lsm.common.impls;
+
+import java.util.LinkedList;
+import java.util.List;
+
+import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
+import edu.uci.ics.hyracks.storage.am.common.api.IInMemoryFreePageManager;
+import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndex;
+import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexMetaDataFrame;
+import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMComponent;
+import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMFlushController;
+import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMHarness;
+import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIOOperationScheduler;
+import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIndexFileManager;
+import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIndexInternal;
+import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMMergePolicy;
+import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMOperationTracker;
+import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMOperationTrackerFactory;
+import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
+import edu.uci.ics.hyracks.storage.common.buffercache.ICachedPage;
+import edu.uci.ics.hyracks.storage.common.file.BufferedFileHandle;
+import edu.uci.ics.hyracks.storage.common.file.IFileMapProvider;
+
+public abstract class AbstractLSMIndex implements ILSMIndexInternal {
+    protected final ILSMHarness lsmHarness;
+
+    // In-memory components.   
+    protected final IInMemoryFreePageManager memFreePageManager;
+
+    // On-disk components.    
+    protected final IBufferCache diskBufferCache;
+    protected final ILSMIndexFileManager fileManager;
+    protected final IFileMapProvider diskFileMapProvider;
+    protected final LinkedList<ILSMComponent> immutableComponents;
+
+    protected boolean isActivated;
+
+    public AbstractLSMIndex(IInMemoryFreePageManager memFreePageManager, IBufferCache diskBufferCache,
+            ILSMIndexFileManager fileManager, IFileMapProvider diskFileMapProvider,
+            ILSMFlushController flushController, ILSMMergePolicy mergePolicy,
+            ILSMOperationTrackerFactory opTrackerFactory, ILSMIOOperationScheduler ioScheduler) {
+        this.memFreePageManager = memFreePageManager;
+        this.diskBufferCache = diskBufferCache;
+        this.diskFileMapProvider = diskFileMapProvider;
+        this.immutableComponents = new LinkedList<ILSMComponent>();
+        this.fileManager = fileManager;
+        ILSMOperationTracker opTracker = opTrackerFactory.createOperationTracker(this);
+        lsmHarness = new LSMHarness(this, flushController, mergePolicy, opTracker, ioScheduler);
+        isActivated = false;
+    }
+
+    protected void forceFlushDirtyPages(ITreeIndex treeIndex) throws HyracksDataException {
+        int fileId = treeIndex.getFileId();
+        IBufferCache bufferCache = treeIndex.getBufferCache();
+        // Flush all dirty pages of the tree. 
+        // By default, metadata and data are flushed asynchronously in the buffercache.
+        // This means that the flush issues writes to the OS, but the data may still lie in filesystem buffers.
+        ITreeIndexMetaDataFrame metadataFrame = treeIndex.getFreePageManager().getMetaDataFrameFactory().createFrame();
+        int startPage = 0;
+        int maxPage = treeIndex.getFreePageManager().getMaxPage(metadataFrame);
+        forceFlushDirtyPages(bufferCache, fileId, startPage, maxPage);
+    }
+
+    protected void forceFlushDirtyPages(IBufferCache bufferCache, int fileId, int startPageId, int endPageId)
+            throws HyracksDataException {
+        for (int i = startPageId; i <= endPageId; i++) {
+            ICachedPage page = bufferCache.tryPin(BufferedFileHandle.getDiskPageId(fileId, i));
+            // If tryPin returns null, it means the page is not cached, and therefore cannot be dirty.
+            if (page == null) {
+                continue;
+            }
+            try {
+                bufferCache.flushDirtyPage(page);
+            } finally {
+                bufferCache.unpin(page);
+            }
+        }
+        // Forces all pages of given file to disk. This guarantees the data makes it to disk.
+        bufferCache.force(fileId, true);
+    }
+
+    protected void markAsValidInternal(ITreeIndex treeIndex) throws HyracksDataException {
+        int fileId = treeIndex.getFileId();
+        IBufferCache bufferCache = treeIndex.getBufferCache();
+        ITreeIndexMetaDataFrame metadataFrame = treeIndex.getFreePageManager().getMetaDataFrameFactory().createFrame();
+        // Mark the component as a valid component by flushing the metadata page to disk
+        int metadataPageId = treeIndex.getFreePageManager().getFirstMetadataPage();
+        ICachedPage metadataPage = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, metadataPageId), false);
+        metadataPage.acquireWriteLatch();
+        try {
+            metadataFrame.setPage(metadataPage);
+            metadataFrame.setValid(true);
+
+            // Flush the single modified page to disk.
+            bufferCache.flushDirtyPage(metadataPage);
+
+            // Force modified metadata page to disk.
+            bufferCache.force(fileId, true);
+        } finally {
+            metadataPage.releaseWriteLatch();
+            bufferCache.unpin(metadataPage);
+        }
+    }
+
+    @Override
+    public void addFlushedComponent(ILSMComponent index) {
+        immutableComponents.addFirst(index);
+    }
+
+    @Override
+    public void addMergedComponent(ILSMComponent newComponent, List<ILSMComponent> mergedComponents) {
+        immutableComponents.removeAll(mergedComponents);
+        immutableComponents.addLast(newComponent);
+    }
+
+    @Override
+    public IInMemoryFreePageManager getInMemoryFreePageManager() {
+        return memFreePageManager;
+    }
+
+    @Override
+    public List<ILSMComponent> getImmutableComponents() {
+        return immutableComponents;
+    }
+
+    @Override
+    public ILSMFlushController getFlushController() {
+        return lsmHarness.getFlushController();
+    }
+
+    @Override
+    public ILSMOperationTracker getOperationTracker() {
+        return lsmHarness.getOperationTracker();
+    }
+
+    @Override
+    public ILSMIOOperationScheduler getIOScheduler() {
+        return lsmHarness.getIOScheduler();
+    }
+
+    @Override
+    public IBufferCache getBufferCache() {
+        return diskBufferCache;
+    }
+}
diff --git a/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/AbstractLSMIndexFileManager.java b/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/AbstractLSMIndexFileManager.java
new file mode 100644
index 0000000..38694a3
--- /dev/null
+++ b/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/AbstractLSMIndexFileManager.java
@@ -0,0 +1,277 @@
+/*
+ * Copyright 2009-2012 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.lsm.common.impls;
+
+import java.io.File;
+import java.io.FilenameFilter;
+import java.text.Format;
+import java.text.SimpleDateFormat;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.Date;
+import java.util.List;
+
+import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
+import edu.uci.ics.hyracks.api.io.FileReference;
+import edu.uci.ics.hyracks.api.io.IIOManager;
+import edu.uci.ics.hyracks.api.io.IODeviceHandle;
+import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndex;
+import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexMetaDataFrame;
+import edu.uci.ics.hyracks.storage.am.common.api.IndexException;
+import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIndexFileManager;
+import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
+import edu.uci.ics.hyracks.storage.common.buffercache.ICachedPage;
+import edu.uci.ics.hyracks.storage.common.file.BufferedFileHandle;
+import edu.uci.ics.hyracks.storage.common.file.IFileMapProvider;
+
+public abstract class AbstractLSMIndexFileManager implements ILSMIndexFileManager {
+
+    protected static final String SPLIT_STRING = "_";
+
+    // Use all IODevices registered in ioManager in a round-robin fashion to choose
+    // where to flush and merge
+    protected final IIOManager ioManager;
+    protected final IFileMapProvider fileMapProvider;
+
+    // baseDir should reflect dataset name and partition name.
+    protected FileReference file;
+    protected String baseDir;
+    protected final Format formatter = new SimpleDateFormat("yyyy-MM-dd-HH-mm-ss-SSS");
+    protected final Comparator<String> cmp = new FileNameComparator();
+    protected final Comparator<ComparableFileName> recencyCmp = new RecencyComparator();
+
+    protected final TreeIndexFactory<? extends ITreeIndex> treeFactory;
+
+    // The current index for the round-robin file assignment
+    private int ioDeviceIndex = 0;
+
+    public AbstractLSMIndexFileManager(IIOManager ioManager, IFileMapProvider fileMapProvider, FileReference file,
+            TreeIndexFactory<? extends ITreeIndex> treeFactory) {
+        this.file = file;
+        this.baseDir = file.getFile().getPath();
+        if (!baseDir.endsWith(System.getProperty("file.separator"))) {
+            baseDir += System.getProperty("file.separator");
+        }
+        this.fileMapProvider = fileMapProvider;
+        this.ioManager = ioManager;
+        this.treeFactory = treeFactory;
+    }
+
+    private static FilenameFilter fileNameFilter = new FilenameFilter() {
+        public boolean accept(File dir, String name) {
+            return !name.startsWith(".");
+        }
+    };
+
+    protected boolean isValidTreeIndex(ITreeIndex treeIndex) throws HyracksDataException {
+        IBufferCache bufferCache = treeIndex.getBufferCache();
+        treeIndex.activate();
+        try {
+            int metadataPage = treeIndex.getFreePageManager().getFirstMetadataPage();
+            ITreeIndexMetaDataFrame metadataFrame = treeIndex.getFreePageManager().getMetaDataFrameFactory()
+                    .createFrame();
+            ICachedPage page = bufferCache.pin(BufferedFileHandle.getDiskPageId(treeIndex.getFileId(), metadataPage),
+                    false);
+            page.acquireReadLatch();
+            try {
+                metadataFrame.setPage(page);
+                return metadataFrame.isValid();
+            } finally {
+                page.releaseReadLatch();
+                bufferCache.unpin(page);
+            }
+        } finally {
+            treeIndex.deactivate();
+        }
+    }
+
+    abstract protected void cleanupAndGetValidFilesInternal(IODeviceHandle dev, FilenameFilter filter,
+            TreeIndexFactory<? extends ITreeIndex> treeFactory, ArrayList<ComparableFileName> allFiles)
+            throws HyracksDataException, IndexException;
+
+    @Override
+    public void createDirs() {
+        for (IODeviceHandle dev : ioManager.getIODevices()) {
+            File f = new File(dev.getPath(), baseDir);
+            f.mkdirs();
+        }
+    }
+
+    @Override
+    public void deleteDirs() {
+        for (IODeviceHandle dev : ioManager.getIODevices()) {
+            File f = new File(dev.getPath(), baseDir);
+            delete(f);
+        }
+    }
+
+    private void delete(File f) {
+        if (f.isDirectory()) {
+            for (File c : f.listFiles()) {
+                delete(c);
+            }
+        }
+        f.delete();
+    }
+
+    public FileReference createFlushFile(String relFlushFileName) {
+        // Assigns new files to I/O devices in round-robin fashion.
+        IODeviceHandle dev = ioManager.getIODevices().get(ioDeviceIndex);
+        ioDeviceIndex = (ioDeviceIndex + 1) % ioManager.getIODevices().size();
+        return dev.createFileReference(relFlushFileName);
+    }
+
+    public FileReference createMergeFile(String relMergeFileName) {
+        return createFlushFile(relMergeFileName);
+    }
+
+    @Override
+    public LSMComponentFileReferences getRelFlushFileReference() {
+        Date date = new Date();
+        String ts = formatter.format(date);
+        // Begin timestamp and end timestamp are identical since it is a flush
+        return new LSMComponentFileReferences(createFlushFile(baseDir + ts + SPLIT_STRING + ts), null);
+    }
+
+    @Override
+    public LSMComponentFileReferences getRelMergeFileReference(String firstFileName, String lastFileName)
+            throws HyracksDataException {
+        String[] firstTimestampRange = firstFileName.split(SPLIT_STRING);
+        String[] lastTimestampRange = lastFileName.split(SPLIT_STRING);
+        // Get the range of timestamps by taking the earliest and the latest timestamps
+        return new LSMComponentFileReferences(createMergeFile(baseDir + firstTimestampRange[0] + SPLIT_STRING
+                + lastTimestampRange[1]), null);
+    }
+
+    @Override
+    public List<LSMComponentFileReferences> cleanupAndGetValidFiles() throws HyracksDataException, IndexException {
+        List<LSMComponentFileReferences> validFiles = new ArrayList<LSMComponentFileReferences>();
+        ArrayList<ComparableFileName> allFiles = new ArrayList<ComparableFileName>();
+
+        // Gather files from all IODeviceHandles and delete invalid files
+        // There are two types of invalid files:
+        // (1) The isValid flag is not set
+        // (2) The file's interval is contained by some other file
+        // Here, we only filter out (1).
+        for (IODeviceHandle dev : ioManager.getIODevices()) {
+            cleanupAndGetValidFilesInternal(dev, fileNameFilter, treeFactory, allFiles);
+        }
+
+        if (allFiles.isEmpty()) {
+            return validFiles;
+        }
+
+        if (allFiles.size() == 1) {
+            validFiles.add(new LSMComponentFileReferences(allFiles.get(0).fileRef, null));
+            return validFiles;
+        }
+
+        // Sorts files names from earliest to latest timestamp.
+        Collections.sort(allFiles);
+
+        List<ComparableFileName> validComparableFiles = new ArrayList<ComparableFileName>();
+        ComparableFileName last = allFiles.get(0);
+        validComparableFiles.add(last);
+        for (int i = 1; i < allFiles.size(); i++) {
+            ComparableFileName current = allFiles.get(i);
+            // The current start timestamp is greater than last stop timestamp so current is valid.
+            if (current.interval[0].compareTo(last.interval[1]) > 0) {
+                validComparableFiles.add(current);
+                last = current;
+            } else if (current.interval[0].compareTo(last.interval[0]) >= 0
+                    && current.interval[1].compareTo(last.interval[1]) <= 0) {
+                // The current file is completely contained in the interval of the 
+                // last file. Thus the last file must contain at least as much information 
+                // as the current file, so delete the current file.
+                current.fileRef.delete();
+            } else {
+                // This scenario should not be possible since timestamps are monotonically increasing.
+                throw new HyracksDataException("Found LSM files with overlapping timestamp intervals, "
+                        + "but the intervals were not contained by another file.");
+            }
+        }
+
+        // Sort valid files in reverse lexicographical order, such that newer files come first.
+        Collections.sort(validComparableFiles, recencyCmp);
+        for (ComparableFileName cmpFileName : validComparableFiles) {
+            validFiles.add(new LSMComponentFileReferences(cmpFileName.fileRef, null));
+        }
+
+        return validFiles;
+    }
+
+    @Override
+    public Comparator<String> getFileNameComparator() {
+        return cmp;
+    }
+
+    /**
+     * Sorts strings in reverse lexicographical order. The way we construct the
+     * file names above guarantees that:
+     * 1. Flushed files sort lower than merged files
+     * 2. Flushed files are sorted from newest to oldest (based on the timestamp
+     * string)
+     */
+    private class FileNameComparator implements Comparator<String> {
+        @Override
+        public int compare(String a, String b) {
+            // Consciously ignoring locale.
+            return -a.compareTo(b);
+        }
+    }
+
+    @Override
+    public String getBaseDir() {
+        return baseDir;
+    }
+
+    protected class ComparableFileName implements Comparable<ComparableFileName> {
+        public final FileReference fileRef;
+        public final String fullPath;
+        public final String fileName;
+
+        // Timestamp interval.
+        public final String[] interval;
+
+        public ComparableFileName(FileReference fileRef) {
+            this.fileRef = fileRef;
+            this.fullPath = fileRef.getFile().getAbsolutePath();
+            this.fileName = fileRef.getFile().getName();
+            interval = fileName.split(SPLIT_STRING);
+        }
+
+        @Override
+        public int compareTo(ComparableFileName b) {
+            int startCmp = interval[0].compareTo(b.interval[0]);
+            if (startCmp != 0) {
+                return startCmp;
+            }
+            return b.interval[1].compareTo(interval[1]);
+        }
+    }
+
+    private class RecencyComparator implements Comparator<ComparableFileName> {
+        @Override
+        public int compare(ComparableFileName a, ComparableFileName b) {
+            int cmp = -a.interval[0].compareTo(b.interval[0]);
+            if (cmp != 0) {
+                return cmp;
+            }
+            return -a.interval[1].compareTo(b.interval[1]);
+        }
+    }
+}
diff --git a/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/BTreeFactory.java b/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/BTreeFactory.java
index 70f8986..008c418 100644
--- a/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/BTreeFactory.java
+++ b/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/BTreeFactory.java
@@ -20,6 +20,7 @@
 import edu.uci.ics.hyracks.storage.am.btree.impls.BTree;
 import edu.uci.ics.hyracks.storage.am.common.api.IFreePageManagerFactory;
 import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexFrameFactory;
+import edu.uci.ics.hyracks.storage.am.common.api.IndexException;
 import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
 import edu.uci.ics.hyracks.storage.common.file.IFileMapProvider;
 
@@ -33,9 +34,9 @@
     }
 
     @Override
-    public BTree createIndexInstance(FileReference file) {
+    public BTree createIndexInstance(FileReference file) throws IndexException {
         return new BTree(bufferCache, fileMapProvider, freePageManagerFactory.createFreePageManager(),
                 interiorFrameFactory, leafFrameFactory, cmpFactories, fieldCount, file);
     }
 
-}
+}
\ No newline at end of file
diff --git a/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/BlockingIOOperationCallback.java b/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/BlockingIOOperationCallback.java
index d8d4d81..6b6bb22 100644
--- a/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/BlockingIOOperationCallback.java
+++ b/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/BlockingIOOperationCallback.java
@@ -3,6 +3,7 @@
 import java.util.List;
 
 import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
+import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMComponent;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIOOperation;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallback;
 
@@ -23,13 +24,14 @@
     }
 
     @Override
-    public void afterOperation(ILSMIOOperation operation, List<Object> oldComponents, Object newComponent)
+    public void afterOperation(ILSMIOOperation operation, List<ILSMComponent> oldComponents, ILSMComponent newComponent)
             throws HyracksDataException {
         // Do nothing.
     }
 
     @Override
-    public synchronized void afterFinalize(ILSMIOOperation operation, Object newComponent) throws HyracksDataException {
+    public synchronized void afterFinalize(ILSMIOOperation operation, ILSMComponent newComponent)
+            throws HyracksDataException {
         this.notifyAll();
         notified = true;
     }
diff --git a/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/IndexFactory.java b/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/IndexFactory.java
index f53756a..3feaecf 100644
--- a/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/IndexFactory.java
+++ b/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/IndexFactory.java
@@ -40,4 +40,4 @@
     public IBufferCache getBufferCache() {
         return bufferCache;
     }
-}
+}
\ No newline at end of file
diff --git a/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/LSMComponentFileReferences.java b/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/LSMComponentFileReferences.java
new file mode 100644
index 0000000..ac6ddcf
--- /dev/null
+++ b/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/LSMComponentFileReferences.java
@@ -0,0 +1,39 @@
+/*
+ * Copyright 2009-2012 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.lsm.common.impls;
+
+import edu.uci.ics.hyracks.api.io.FileReference;
+
+public final class LSMComponentFileReferences {
+
+    // The FileReference for the index that is used for inserting records of the component. For instance, this will be the FileReference of the RTree in one component of the LSM-RTree.
+    private final FileReference insertIndexFileReference;
+    // This FileReference for the delete index (if any). For example, this will be the the FileReference of the buddy BTree in one component of the LSM-RTree.
+    private final FileReference deleteIndexFileReference;
+
+    public LSMComponentFileReferences(FileReference insertIndexFileReference, FileReference deleteIndexFileReference) {
+        this.insertIndexFileReference = insertIndexFileReference;
+        this.deleteIndexFileReference = deleteIndexFileReference;
+    }
+
+    public FileReference getInsertIndexFileReference() {
+        return insertIndexFileReference;
+    }
+
+    public FileReference getDeleteIndexFileReference() {
+        return deleteIndexFileReference;
+    }
+}
diff --git a/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/LSMComponentState.java b/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/LSMComponentState.java
new file mode 100644
index 0000000..df4f07d
--- /dev/null
+++ b/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/LSMComponentState.java
@@ -0,0 +1,20 @@
+/*
+ * Copyright 2009-2012 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.lsm.common.impls;
+
+public enum LSMComponentState {
+
+}
diff --git a/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/LSMHarness.java b/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/LSMHarness.java
index 05a6fa3..bf0f90b 100644
--- a/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/LSMHarness.java
+++ b/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/LSMHarness.java
@@ -27,6 +27,7 @@
 import edu.uci.ics.hyracks.storage.am.common.api.IIndexCursor;
 import edu.uci.ics.hyracks.storage.am.common.api.ISearchPredicate;
 import edu.uci.ics.hyracks.storage.am.common.api.IndexException;
+import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMComponent;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMFlushController;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMHarness;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIOOperation;
@@ -119,7 +120,7 @@
         if (LOGGER.isLoggable(Level.INFO)) {
             LOGGER.info("Flushing LSM-Index: " + lsmIndex);
         }
-        Object newComponent = null;
+        ILSMComponent newComponent = null;
         try {
             operation.getCallback().beforeOperation(operation);
             newComponent = lsmIndex.flush(operation);
@@ -129,21 +130,21 @@
             // the new component permanent, and mark it as valid (usually this means
             // forcing all pages of the tree to disk, possibly with some extra
             // information to mark the tree as valid).
-            lsmIndex.getComponentFinalizer().finalize(newComponent);
+            lsmIndex.markAsValid(newComponent);
         } finally {
             operation.getCallback().afterFinalize(operation, newComponent);
         }
-        lsmIndex.resetInMemoryComponent();
+        lsmIndex.resetMutableComponent();
         synchronized (diskComponentsSync) {
             lsmIndex.addFlushedComponent(newComponent);
-            mergePolicy.diskComponentAdded(lsmIndex, lsmIndex.getDiskComponents().size());
+            mergePolicy.diskComponentAdded(lsmIndex, lsmIndex.getImmutableComponents().size());
         }
 
         // Unblock entering threads waiting for the flush
         flushController.setFlushStatus(lsmIndex, false);
     }
 
-    public List<Object> search(IIndexCursor cursor, ISearchPredicate pred, ILSMIndexOperationContext ctx,
+    public List<ILSMComponent> search(IIndexCursor cursor, ISearchPredicate pred, ILSMIndexOperationContext ctx,
             boolean includeMemComponent) throws HyracksDataException, IndexException {
         // If the search doesn't include the in-memory component, then we don't have
         // to synchronize with a flush.
@@ -158,10 +159,10 @@
         // flush adds another on-disk Tree.
         // Since this mode is only used for merging trees, it doesn't really
         // matter if the merge excludes the new on-disk Tree.
-        List<Object> diskComponentSnapshot = new ArrayList<Object>();
+        List<ILSMComponent> diskComponentSnapshot = new ArrayList<ILSMComponent>();
         AtomicInteger localSearcherRefCount = null;
         synchronized (diskComponentsSync) {
-            diskComponentSnapshot.addAll(lsmIndex.getDiskComponents());
+            diskComponentSnapshot.addAll(lsmIndex.getImmutableComponents());
             localSearcherRefCount = searcherRefCount;
             localSearcherRefCount.incrementAndGet();
         }
@@ -193,8 +194,8 @@
         // (after we swap the searcher ref count).
         AtomicInteger localSearcherRefCount = searcherRefCount;
 
-        List<Object> mergedComponents = new ArrayList<Object>();
-        Object newComponent = null;
+        List<ILSMComponent> mergedComponents = new ArrayList<ILSMComponent>();
+        ILSMComponent newComponent = null;
         try {
             operation.getCallback().beforeOperation(operation);
             newComponent = lsmIndex.merge(mergedComponents, operation);
@@ -211,7 +212,7 @@
             // the new component permanent, and mark it as valid (usually this means
             // forcing all pages of the tree to disk, possibly with some extra
             // information to mark the tree as valid).
-            lsmIndex.getComponentFinalizer().finalize(newComponent);
+            lsmIndex.markAsValid(newComponent);
         } finally {
             operation.getCallback().afterFinalize(operation, newComponent);
         }
@@ -262,30 +263,34 @@
     }
 
     @Override
-    public void addBulkLoadedComponent(Object index) throws HyracksDataException, IndexException {
+    public void addBulkLoadedComponent(ILSMComponent index) throws HyracksDataException, IndexException {
         // The implementation of this call must take any necessary steps to make
         // the new component permanent, and mark it as valid (usually this means
         // forcing all pages of the tree to disk, possibly with some extra
         // information to mark the tree as valid).
-        lsmIndex.getComponentFinalizer().finalize(index);
+        lsmIndex.markAsValid(index);
         synchronized (diskComponentsSync) {
             lsmIndex.addFlushedComponent(index);
-            mergePolicy.diskComponentAdded(lsmIndex, lsmIndex.getDiskComponents().size());
+            mergePolicy.diskComponentAdded(lsmIndex, lsmIndex.getImmutableComponents().size());
         }
     }
 
+    @Override
     public ILSMFlushController getFlushController() {
         return flushController;
     }
 
+    @Override
     public ILSMOperationTracker getOperationTracker() {
         return opTracker;
     }
 
+    @Override
     public ILSMIOOperationScheduler getIOScheduler() {
         return ioScheduler;
     }
 
+    @Override
     public ILSMIndex getIndex() {
         return lsmIndex;
     }
diff --git a/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/LSMIndexFileManager.java b/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/LSMIndexFileManager.java
index 71c335a..19282a0 100644
--- a/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/LSMIndexFileManager.java
+++ b/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/LSMIndexFileManager.java
@@ -17,243 +17,36 @@
 
 import java.io.File;
 import java.io.FilenameFilter;
-import java.text.Format;
-import java.text.SimpleDateFormat;
 import java.util.ArrayList;
-import java.util.Collections;
-import java.util.Comparator;
-import java.util.Date;
-import java.util.List;
 
 import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
 import edu.uci.ics.hyracks.api.io.FileReference;
 import edu.uci.ics.hyracks.api.io.IIOManager;
 import edu.uci.ics.hyracks.api.io.IODeviceHandle;
 import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndex;
-import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMComponentFinalizer;
-import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIndexFileManager;
+import edu.uci.ics.hyracks.storage.am.common.api.IndexException;
 import edu.uci.ics.hyracks.storage.common.file.IFileMapProvider;
 
-public class LSMIndexFileManager implements ILSMIndexFileManager {
-
-    protected static final String SPLIT_STRING = "_";
-
-    // Use all IODevices registered in ioManager in a round-robin fashion to choose
-    // where to flush and merge
-    protected final IIOManager ioManager;
-    protected final IFileMapProvider fileMapProvider;
-    protected final TreeIndexFactory<? extends ITreeIndex> treeFactory;
-
-    // baseDir should reflect dataset name and partition name.
-    protected FileReference file;
-    protected String baseDir;
-    protected final Format formatter = new SimpleDateFormat("yyyy-MM-dd-HH-mm-ss-SSS");
-    protected final Comparator<String> cmp = new FileNameComparator();
-    protected final Comparator<ComparableFileName> recencyCmp = new RecencyComparator();
-
-    // The current index for the round-robin file assignment
-    private int ioDeviceIndex = 0;
-
-    private static FilenameFilter fileNameFilter = new FilenameFilter() {
-        public boolean accept(File dir, String name) {
-            return !name.startsWith(".");
-        }
-    };
+public class LSMIndexFileManager extends AbstractLSMIndexFileManager {
 
     public LSMIndexFileManager(IIOManager ioManager, IFileMapProvider fileMapProvider, FileReference file,
             TreeIndexFactory<? extends ITreeIndex> treeFactory) {
-        this.file = file;
-        this.baseDir = file.getFile().getPath();
-        if (!baseDir.endsWith(System.getProperty("file.separator"))) {
-            baseDir += System.getProperty("file.separator");
-        }
-        this.fileMapProvider = fileMapProvider;
-        this.ioManager = ioManager;
-        this.treeFactory = treeFactory;
-    }
-
-    @Override
-    public void createDirs() {
-        for (IODeviceHandle dev : ioManager.getIODevices()) {
-            File f = new File(dev.getPath(), baseDir);
-            f.mkdirs();
-        }
-    }
-
-    @Override
-    public void deleteDirs() {
-        for (IODeviceHandle dev : ioManager.getIODevices()) {
-            File f = new File(dev.getPath(), baseDir);
-            delete(f);
-        }
-    }
-
-    private void delete(File f) {
-        if (f.isDirectory()) {
-            for (File c : f.listFiles()) {
-                delete(c);
-            }
-        }
-        f.delete();
-    }
-    
-    public FileReference createFlushFile(String relFlushFileName) {
-        // Assigns new files to I/O devices in round-robin fashion.
-        IODeviceHandle dev = ioManager.getIODevices().get(ioDeviceIndex);
-        ioDeviceIndex = (ioDeviceIndex + 1) % ioManager.getIODevices().size();
-        return dev.createFileReference(relFlushFileName);
-    }
-
-    public FileReference createMergeFile(String relMergeFileName) {
-        return createFlushFile(relMergeFileName);
-    }
-
-    @Override
-    public Object getRelFlushFileName() {
-        Date date = new Date();
-        String ts = formatter.format(date);
-        // Begin timestamp and end timestamp are identical since it is a flush
-        return baseDir + ts + SPLIT_STRING + ts;
-    }
-
-    @Override
-    public Object getRelMergeFileName(String firstFileName, String lastFileName) throws HyracksDataException {
-        String[] firstTimestampRange = firstFileName.split(SPLIT_STRING);
-        String[] lastTimestampRange = lastFileName.split(SPLIT_STRING);
-        // Get the range of timestamps by taking the earliest and the latest timestamps
-        return baseDir + firstTimestampRange[0] + SPLIT_STRING + lastTimestampRange[1];
-    }
-
-    @Override
-    public Comparator<String> getFileNameComparator() {
-        return cmp;
-    }
-
-    /**
-     * Sorts strings in reverse lexicographical order. The way we construct the
-     * file names above guarantees that:
-     * 1. Flushed files sort lower than merged files
-     * 2. Flushed files are sorted from newest to oldest (based on the timestamp
-     * string)
-     */
-    private class FileNameComparator implements Comparator<String> {
-        @Override
-        public int compare(String a, String b) {
-            // Consciously ignoring locale.
-            return -a.compareTo(b);
-        }
-    }
-
-    @Override
-    public String getBaseDir() {
-        return baseDir;
+        super(ioManager, fileMapProvider, file, treeFactory);
     }
 
     protected void cleanupAndGetValidFilesInternal(IODeviceHandle dev, FilenameFilter filter,
-            TreeIndexFactory<? extends ITreeIndex> treeFactory, ILSMComponentFinalizer componentFinalizer,
-            ArrayList<ComparableFileName> allFiles) throws HyracksDataException {
+            TreeIndexFactory<? extends ITreeIndex> treeFactory, ArrayList<ComparableFileName> allFiles)
+            throws HyracksDataException, IndexException {
         File dir = new File(dev.getPath(), baseDir);
         String[] files = dir.list(filter);
         for (String fileName : files) {
             File file = new File(dir.getPath() + File.separator + fileName);
-            if (componentFinalizer.isValid(treeFactory.createIndexInstance(new FileReference(file)))) {
-                allFiles.add(new ComparableFileName(file.getAbsolutePath()));
+            FileReference fileRef = new FileReference(file);
+            if (isValidTreeIndex(treeFactory.createIndexInstance(fileRef))) {
+                allFiles.add(new ComparableFileName(fileRef));
             } else {
                 file.delete();
             }
         }
     }
-
-    @Override
-    public List<Object> cleanupAndGetValidFiles(ILSMComponentFinalizer componentFinalizer) throws HyracksDataException {
-        List<Object> validFiles = new ArrayList<Object>();
-        ArrayList<ComparableFileName> allFiles = new ArrayList<ComparableFileName>();
-
-        // Gather files from all IODeviceHandles and delete invalid files
-        // There are two types of invalid files:
-        // (1) The isValid flag is not set
-        // (2) The file's interval is contained by some other file
-        // Here, we only filter out (1).
-        for (IODeviceHandle dev : ioManager.getIODevices()) {
-            cleanupAndGetValidFilesInternal(dev, fileNameFilter, treeFactory, componentFinalizer, allFiles);
-        }
-
-        if (allFiles.isEmpty()) {
-            return validFiles;
-        }
-
-        if (allFiles.size() == 1) {
-            validFiles.add(allFiles.get(0).fullPath);
-            return validFiles;
-        }
-
-        // Sorts files names from earliest to latest timestamp.
-        Collections.sort(allFiles);
-
-        List<ComparableFileName> validComparableFiles = new ArrayList<ComparableFileName>();
-        ComparableFileName last = allFiles.get(0);
-        validComparableFiles.add(last);
-        for (int i = 1; i < allFiles.size(); i++) {
-            ComparableFileName current = allFiles.get(i);
-            // The current start timestamp is greater than last stop timestamp so current is valid.
-            if (current.interval[0].compareTo(last.interval[1]) > 0) {
-                validComparableFiles.add(current);
-                last = current;
-            } else if (current.interval[0].compareTo(last.interval[0]) >= 0
-                    && current.interval[1].compareTo(last.interval[1]) <= 0) {
-                // The current file is completely contained in the interval of the 
-                // last file. Thus the last file must contain at least as much information 
-                // as the current file, so delete the current file.
-                File invalidFile = new File(current.fullPath);
-                invalidFile.delete();
-            } else {
-                // This scenario should not be possible since timestamps are monotonically increasing.
-                throw new HyracksDataException("Found LSM files with overlapping timestamp intervals, "
-                        + "but the intervals were not contained by another file.");
-            }
-        }
-
-        // Sort valid files in reverse lexicographical order, such that newer files come first.
-        Collections.sort(validComparableFiles, recencyCmp);
-        for (ComparableFileName cmpFileName : validComparableFiles) {
-            validFiles.add(cmpFileName.fullPath);
-        }
-
-        return validFiles;
-    }
-
-    protected class ComparableFileName implements Comparable<ComparableFileName> {
-        public final String fullPath;
-        public final String fileName;
-
-        // Timestamp interval.
-        public final String[] interval;
-
-        public ComparableFileName(String fullPath) {
-            this.fullPath = fullPath;
-            File f = new File(fullPath);
-            this.fileName = f.getName();
-            interval = fileName.split(SPLIT_STRING);
-        }
-
-        @Override
-        public int compareTo(ComparableFileName b) {
-            int startCmp = interval[0].compareTo(b.interval[0]);
-            if (startCmp != 0) {
-                return startCmp;
-            }
-            return b.interval[1].compareTo(interval[1]);
-        }
-    }
-
-    private class RecencyComparator implements Comparator<ComparableFileName> {
-        @Override
-        public int compare(ComparableFileName a, ComparableFileName b) {
-            int cmp = -a.interval[0].compareTo(b.interval[0]);
-            if (cmp != 0) {
-                return cmp;
-            }
-            return -a.interval[1].compareTo(b.interval[1]);
-        }
-    }
 }
diff --git a/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/NoOpIOOperationCallback.java b/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/NoOpIOOperationCallback.java
index 427c061..74cf4f1 100644
--- a/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/NoOpIOOperationCallback.java
+++ b/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/NoOpIOOperationCallback.java
@@ -3,6 +3,7 @@
 import java.util.List;
 
 import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
+import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMComponent;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIOOperation;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallback;
 
@@ -15,13 +16,13 @@
     }
 
     @Override
-    public void afterOperation(ILSMIOOperation operation, List<Object> oldComponents, Object newComponent)
+    public void afterOperation(ILSMIOOperation operation, List<ILSMComponent> oldComponents, ILSMComponent newComponent)
             throws HyracksDataException {
         // Do nothing.
     }
 
     @Override
-    public void afterFinalize(ILSMIOOperation operation, Object newComponent) throws HyracksDataException {
+    public void afterFinalize(ILSMIOOperation operation, ILSMComponent newComponent) throws HyracksDataException {
         // Do nothing.
     }
 }
diff --git a/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/ReferenceCountingOperationTracker.java b/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/ReferenceCountingOperationTracker.java
index 7674bba..2f41464 100644
--- a/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/ReferenceCountingOperationTracker.java
+++ b/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/ReferenceCountingOperationTracker.java
@@ -6,11 +6,11 @@
 import edu.uci.ics.hyracks.storage.am.common.api.IModificationOperationCallback;
 import edu.uci.ics.hyracks.storage.am.common.api.ISearchOperationCallback;
 import edu.uci.ics.hyracks.storage.am.common.impls.NoOpOperationCallback;
+import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMComponent;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIOOperation;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallback;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIndex;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIndexAccessor;
-import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIndexOperationContext;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMOperationTracker;
 
 public class ReferenceCountingOperationTracker implements ILSMOperationTracker {
@@ -25,8 +25,7 @@
 
     @Override
     public synchronized boolean beforeOperation(ISearchOperationCallback searchCallback,
-            IModificationOperationCallback modificationCallback, boolean tryOperation)
-            throws HyracksDataException {
+            IModificationOperationCallback modificationCallback, boolean tryOperation) throws HyracksDataException {
         // Wait for pending flushes to complete.
         // If flushFlag is set, then the flush is queued to occur by the last exiting thread.
         // This operation should wait for that flush to occur before proceeding.
@@ -71,13 +70,13 @@
         }
 
         @Override
-        public void afterOperation(ILSMIOOperation operation, List<Object> oldComponents, Object newComponent)
-                throws HyracksDataException {
+        public void afterOperation(ILSMIOOperation operation, List<ILSMComponent> oldComponents,
+                ILSMComponent newComponent) throws HyracksDataException {
             // Do nothing.
         }
 
         @Override
-        public void afterFinalize(ILSMIOOperation operation, Object newComponent) throws HyracksDataException {
+        public void afterFinalize(ILSMIOOperation operation, ILSMComponent newComponent) throws HyracksDataException {
             ReferenceCountingOperationTracker.this.notifyAll();
         }
     }
diff --git a/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/TreeIndexComponentFinalizer.java b/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/TreeIndexComponentFinalizer.java
deleted file mode 100644
index 3eac3dd..0000000
--- a/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/TreeIndexComponentFinalizer.java
+++ /dev/null
@@ -1,118 +0,0 @@
-/*
- * Copyright 2009-2012 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.lsm.common.impls;
-
-import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
-import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndex;
-import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexMetaDataFrame;
-import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMComponentFinalizer;
-import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
-import edu.uci.ics.hyracks.storage.common.buffercache.ICachedPage;
-import edu.uci.ics.hyracks.storage.common.file.BufferedFileHandle;
-import edu.uci.ics.hyracks.storage.common.file.IFileMapProvider;
-
-public class TreeIndexComponentFinalizer implements ILSMComponentFinalizer {
-
-    protected final IFileMapProvider fileMapProvider;
-
-    public TreeIndexComponentFinalizer(IFileMapProvider fileMapProvider) {
-        this.fileMapProvider = fileMapProvider;
-    }
-
-    @Override
-    public boolean isValid(Object lsmComponent) throws HyracksDataException {
-        ITreeIndex treeIndex = (ITreeIndex) lsmComponent;
-        IBufferCache bufferCache = treeIndex.getBufferCache();
-        treeIndex.activate();
-        try {
-            int metadataPage = treeIndex.getFreePageManager().getFirstMetadataPage();
-            ITreeIndexMetaDataFrame metadataFrame = treeIndex.getFreePageManager().getMetaDataFrameFactory()
-                    .createFrame();
-            ICachedPage page = bufferCache.pin(BufferedFileHandle.getDiskPageId(treeIndex.getFileId(), metadataPage),
-                    false);
-            page.acquireReadLatch();
-            try {
-                metadataFrame.setPage(page);
-                return metadataFrame.isValid();
-            } finally {
-                page.releaseReadLatch();
-                bufferCache.unpin(page);
-            }
-        } finally {
-            treeIndex.deactivate();
-        }
-    }
-
-    @Override
-    public void finalize(Object lsmComponent) throws HyracksDataException {
-        ITreeIndex treeIndex = (ITreeIndex) lsmComponent;
-        forceFlushDirtyPages(treeIndex);
-        markAsValid(treeIndex);
-    }
-    
-    protected void forceFlushDirtyPages(ITreeIndex treeIndex) throws HyracksDataException {
-        int fileId = treeIndex.getFileId();
-        IBufferCache bufferCache = treeIndex.getBufferCache();
-        // Flush all dirty pages of the tree. 
-        // By default, metadata and data are flushed asynchronously in the buffercache.
-        // This means that the flush issues writes to the OS, but the data may still lie in filesystem buffers.
-        ITreeIndexMetaDataFrame metadataFrame = treeIndex.getFreePageManager().getMetaDataFrameFactory().createFrame();
-        int startPage = 0;
-        int maxPage = treeIndex.getFreePageManager().getMaxPage(metadataFrame);
-        forceFlushDirtyPages(bufferCache, fileId, startPage, maxPage);
-    }
-
-    protected void forceFlushDirtyPages(IBufferCache bufferCache, int fileId, int startPageId, int endPageId)
-            throws HyracksDataException {
-        for (int i = startPageId; i <= endPageId; i++) {
-            ICachedPage page = bufferCache.tryPin(BufferedFileHandle.getDiskPageId(fileId, i));
-            // If tryPin returns null, it means the page is not cached, and therefore cannot be dirty.
-            if (page == null) {
-                continue;
-            }
-            try {
-                bufferCache.flushDirtyPage(page);
-            } finally {
-                bufferCache.unpin(page);
-            }
-        }
-        // Forces all pages of given file to disk. This guarantees the data makes it to disk.
-        bufferCache.force(fileId, true);
-    }
-
-    protected void markAsValid(ITreeIndex treeIndex) throws HyracksDataException {
-        int fileId = treeIndex.getFileId();
-        IBufferCache bufferCache = treeIndex.getBufferCache();
-        ITreeIndexMetaDataFrame metadataFrame = treeIndex.getFreePageManager().getMetaDataFrameFactory().createFrame();
-        // Mark the component as a valid component by flushing the metadata page to disk
-        int metadataPageId = treeIndex.getFreePageManager().getFirstMetadataPage();
-        ICachedPage metadataPage = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, metadataPageId), false);
-        metadataPage.acquireWriteLatch();
-        try {
-            metadataFrame.setPage(metadataPage);
-            metadataFrame.setValid(true);
-
-            // Flush the single modified page to disk.
-            bufferCache.flushDirtyPage(metadataPage);
-
-            // Force modified metadata page to disk.
-            bufferCache.force(fileId, true);
-        } finally {
-            metadataPage.releaseWriteLatch();
-            bufferCache.unpin(metadataPage);
-        }
-    }
-}
diff --git a/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/TreeIndexFactory.java b/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/TreeIndexFactory.java
index 1ea174d..f570058 100644
--- a/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/TreeIndexFactory.java
+++ b/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/TreeIndexFactory.java
@@ -16,7 +16,6 @@
 package edu.uci.ics.hyracks.storage.am.lsm.common.impls;
 
 import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparatorFactory;
-import edu.uci.ics.hyracks.api.io.FileReference;
 import edu.uci.ics.hyracks.storage.am.common.api.IFreePageManagerFactory;
 import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndex;
 import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexFrameFactory;
@@ -39,10 +38,4 @@
         this.cmpFactories = cmpFactories;
         this.fieldCount = fieldCount;
     }
-
-    public abstract T createIndexInstance(FileReference file);
-
-    public IBufferCache getBufferCache() {
-        return bufferCache;
-    }
-}
+}
\ No newline at end of file
diff --git a/hyracks-storage-am-lsm-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndex.java b/hyracks-storage-am-lsm-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndex.java
index 9a7b673..1b67406 100644
--- a/hyracks-storage-am-lsm-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndex.java
+++ b/hyracks-storage-am-lsm-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndex.java
@@ -16,7 +16,6 @@
 
 import java.io.File;
 import java.util.ArrayList;
-import java.util.LinkedList;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicInteger;
 
@@ -48,25 +47,23 @@
 import edu.uci.ics.hyracks.storage.am.common.ophelpers.MultiComparator;
 import edu.uci.ics.hyracks.storage.am.common.tuples.PermutingTupleReference;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.IInMemoryBufferCache;
-import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMComponentFinalizer;
+import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMComponent;
+import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMComponentFactory;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMFlushController;
-import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMHarness;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIOOperation;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallback;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIOOperationScheduler;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIndexAccessor;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIndexFileManager;
-import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIndexInternal;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMMergePolicy;
-import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMOperationTracker;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMOperationTrackerFactory;
 import edu.uci.ics.hyracks.storage.am.lsm.common.freepage.InMemoryBufferCache;
+import edu.uci.ics.hyracks.storage.am.lsm.common.impls.AbstractLSMIndex;
 import edu.uci.ics.hyracks.storage.am.lsm.common.impls.BTreeFactory;
 import edu.uci.ics.hyracks.storage.am.lsm.common.impls.BlockingIOOperationCallback;
-import edu.uci.ics.hyracks.storage.am.lsm.common.impls.LSMHarness;
+import edu.uci.ics.hyracks.storage.am.lsm.common.impls.LSMComponentFileReferences;
 import edu.uci.ics.hyracks.storage.am.lsm.invertedindex.api.IInvertedIndex;
 import edu.uci.ics.hyracks.storage.am.lsm.invertedindex.api.IInvertedListCursor;
-import edu.uci.ics.hyracks.storage.am.lsm.invertedindex.impls.LSMInvertedIndexFileManager.LSMInvertedIndexFileNameComponent;
 import edu.uci.ics.hyracks.storage.am.lsm.invertedindex.inmemory.InMemoryInvertedIndex;
 import edu.uci.ics.hyracks.storage.am.lsm.invertedindex.inmemory.InMemoryInvertedIndexAccessor;
 import edu.uci.ics.hyracks.storage.am.lsm.invertedindex.ondisk.OnDiskInvertedIndex;
@@ -77,44 +74,16 @@
 import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
 import edu.uci.ics.hyracks.storage.common.file.IFileMapProvider;
 
-public class LSMInvertedIndex implements ILSMIndexInternal, IInvertedIndex {
-    public class LSMInvertedIndexComponent {
-        private final IInvertedIndex invIndex;
-        private final BTree deletedKeysBTree;
-
-        LSMInvertedIndexComponent(IInvertedIndex invIndex, BTree deletedKeysBTree) {
-            this.invIndex = invIndex;
-            this.deletedKeysBTree = deletedKeysBTree;
-        }
-
-        public IInvertedIndex getInvIndex() {
-            return invIndex;
-        }
-
-        public BTree getDeletedKeysBTree() {
-            return deletedKeysBTree;
-        }
-    }
-
-    protected final ILSMHarness lsmHarness;
+public class LSMInvertedIndex extends AbstractLSMIndex implements IInvertedIndex {
 
     // In-memory components.
-    protected final LSMInvertedIndexComponent memComponent;
+    protected final LSMInvertedIndexComponent mutableComponent;
     protected final IInMemoryFreePageManager memFreePageManager;
     protected final IBinaryTokenizerFactory tokenizerFactory;
 
     // On-disk components.
-    protected final ILSMIndexFileManager fileManager;
     // For creating inverted indexes in flush and merge.
-    protected final OnDiskInvertedIndexFactory diskInvIndexFactory;
-    // For creating deleted-keys BTrees in flush and merge.
-    protected final BTreeFactory deletedKeysBTreeFactory;
-    protected final IBufferCache diskBufferCache;
-    // List of LSMInvertedIndexComponent instances. Using Object for better sharing via
-    // ILSMIndex + LSMHarness.
-    protected final LinkedList<Object> diskComponents;
-    // Helps to guarantees physical consistency of LSM components.
-    protected final ILSMComponentFinalizer componentFinalizer;
+    protected final ILSMComponentFactory componentFactory;
 
     // Type traits and comparators for tokens and inverted-list elements.
     protected final ITypeTraits[] invListTypeTraits;
@@ -122,39 +91,30 @@
     protected final ITypeTraits[] tokenTypeTraits;
     protected final IBinaryComparatorFactory[] tokenCmpFactories;
 
-    protected boolean isActivated;
-
     public LSMInvertedIndex(IInMemoryBufferCache memBufferCache, IInMemoryFreePageManager memFreePageManager,
             OnDiskInvertedIndexFactory diskInvIndexFactory, BTreeFactory deletedKeysBTreeFactory,
             ILSMIndexFileManager fileManager, IFileMapProvider diskFileMapProvider, ITypeTraits[] invListTypeTraits,
             IBinaryComparatorFactory[] invListCmpFactories, ITypeTraits[] tokenTypeTraits,
             IBinaryComparatorFactory[] tokenCmpFactories, IBinaryTokenizerFactory tokenizerFactory,
-            ILSMFlushController flushController, ILSMMergePolicy mergePolicy, ILSMOperationTrackerFactory opTrackerFactory,
-            ILSMIOOperationScheduler ioScheduler) throws IndexException {
+            ILSMFlushController flushController, ILSMMergePolicy mergePolicy,
+            ILSMOperationTrackerFactory opTrackerFactory, ILSMIOOperationScheduler ioScheduler) throws IndexException {
+        super(memFreePageManager, diskInvIndexFactory.getBufferCache(), fileManager, diskFileMapProvider,
+                flushController, mergePolicy, opTrackerFactory, ioScheduler);
         this.memFreePageManager = memFreePageManager;
         this.tokenizerFactory = tokenizerFactory;
-        this.fileManager = fileManager;
-        this.diskInvIndexFactory = diskInvIndexFactory;
-        this.deletedKeysBTreeFactory = deletedKeysBTreeFactory;
-        this.diskBufferCache = diskInvIndexFactory.getBufferCache();
         this.invListTypeTraits = invListTypeTraits;
         this.invListCmpFactories = invListCmpFactories;
         this.tokenTypeTraits = tokenTypeTraits;
         this.tokenCmpFactories = tokenCmpFactories;
-        diskComponents = new LinkedList<Object>();
-        isActivated = false;
         // Create in-memory component.
         InMemoryInvertedIndex memInvIndex = createInMemoryInvertedIndex(memBufferCache);
         BTree deleteKeysBTree = BTreeUtils.createBTree(memBufferCache, memFreePageManager,
                 ((InMemoryBufferCache) memBufferCache).getFileMapProvider(), invListTypeTraits, invListCmpFactories,
                 BTreeLeafFrameType.REGULAR_NSM, new FileReference(new File("membtree")));
-        memComponent = new LSMInvertedIndexComponent(memInvIndex, deleteKeysBTree);
-        // The operation tracker may need to have the in-memory component created.
-        ILSMOperationTracker opTracker = opTrackerFactory.createOperationTracker(this);
-        this.lsmHarness = new LSMHarness(this, flushController, mergePolicy, opTracker, ioScheduler);
-        this.componentFinalizer = new LSMInvertedIndexComponentFinalizer(diskFileMapProvider);
+        mutableComponent = new LSMInvertedIndexComponent(memInvIndex, deleteKeysBTree);
+        componentFactory = new LSMInvertedIndexComponentFactory(diskInvIndexFactory, deletedKeysBTreeFactory);
     }
-    
+
     protected InMemoryInvertedIndex createInMemoryInvertedIndex(IInMemoryBufferCache memBufferCache)
             throws IndexException {
         return InvertedIndexUtils.createInMemoryBTreeInvertedindex(memBufferCache, memFreePageManager,
@@ -169,7 +129,7 @@
 
         fileManager.deleteDirs();
         fileManager.createDirs();
-        diskComponents.clear();
+        immutableComponents.clear();
     }
 
     @Override
@@ -178,22 +138,23 @@
             return;
         }
         try {
-            ((InMemoryBufferCache) memComponent.getInvIndex().getBufferCache()).open();
-            memComponent.getInvIndex().create();
-            memComponent.getInvIndex().activate();
-            memComponent.getDeletedKeysBTree().create();
-            memComponent.getDeletedKeysBTree().activate();
-            diskComponents.clear();
-            List<Object> validFileNames = fileManager.cleanupAndGetValidFiles(componentFinalizer);
-            for (Object o : validFileNames) {
-                LSMInvertedIndexFileNameComponent component = (LSMInvertedIndexFileNameComponent) o;
-                FileReference dictBTreeFile = new FileReference(new File(component.getDictBTreeFileName()));
-                FileReference deletedKeysBTreeFile = new FileReference(
-                        new File(component.getDeletedKeysBTreeFileName()));
-                IInvertedIndex invIndex = createDiskInvIndex(diskInvIndexFactory, dictBTreeFile, false);
-                BTree deletedKeysBTree = (BTree) createDiskTree(deletedKeysBTreeFactory, deletedKeysBTreeFile, false);
-                LSMInvertedIndexComponent diskComponent = new LSMInvertedIndexComponent(invIndex, deletedKeysBTree);
-                diskComponents.add(diskComponent);
+            ((InMemoryBufferCache) mutableComponent.getInvIndex().getBufferCache()).open();
+            mutableComponent.getInvIndex().create();
+            mutableComponent.getInvIndex().activate();
+            mutableComponent.getDeletedKeysBTree().create();
+            mutableComponent.getDeletedKeysBTree().activate();
+            immutableComponents.clear();
+            List<LSMComponentFileReferences> validFileReferences = fileManager.cleanupAndGetValidFiles();
+            for (LSMComponentFileReferences lsmComonentFileReference : validFileReferences) {
+                LSMInvertedIndexComponent component;
+                try {
+                    component = createDiskInvIndexComponent(componentFactory,
+                            lsmComonentFileReference.getInsertIndexFileReference(),
+                            lsmComonentFileReference.getDeleteIndexFileReference(), false);
+                } catch (IndexException e) {
+                    throw new HyracksDataException(e);
+                }
+                immutableComponents.add(component);
             }
             isActivated = true;
             // TODO: Maybe we can make activate throw an index exception?
@@ -202,26 +163,19 @@
         }
     }
 
-    protected IInvertedIndex createDiskInvIndex(OnDiskInvertedIndexFactory invIndexFactory,
-            FileReference dictBTreeFileRef, boolean create) throws HyracksDataException, IndexException {
-        IInvertedIndex invIndex = invIndexFactory.createIndexInstance(dictBTreeFileRef);
+    protected LSMInvertedIndexComponent createDiskInvIndexComponent(ILSMComponentFactory factory,
+            FileReference dictBTreeFileRef, FileReference btreeFileRef, boolean create) throws HyracksDataException,
+            IndexException {
+        LSMInvertedIndexComponent component = (LSMInvertedIndexComponent) factory
+                .createLSMComponentInstance(new LSMComponentFileReferences(dictBTreeFileRef, btreeFileRef));
         if (create) {
-            invIndex.create();
+            component.getInvIndex().create();
+            component.getDeletedKeysBTree().create();
         }
         // Will be closed during cleanup of merge().
-        invIndex.activate();
-        return invIndex;
-    }
-
-    protected ITreeIndex createDiskTree(BTreeFactory btreeFactory, FileReference btreeFileRef, boolean create)
-            throws HyracksDataException, IndexException {
-        ITreeIndex btree = btreeFactory.createIndexInstance(btreeFileRef);
-        if (create) {
-            btree.create();
-        }
-        // Will be closed during cleanup of merge().
-        btree.activate();
-        return btree;
+        component.getInvIndex().activate();
+        component.getDeletedKeysBTree().activate();
+        return component;
     }
 
     @Override
@@ -229,15 +183,15 @@
         if (!isActivated) {
             throw new HyracksDataException("Failed to clear the index since it is not activated.");
         }
-        resetInMemoryComponent();
-        for (Object o : diskComponents) {
-            LSMInvertedIndexComponent component = (LSMInvertedIndexComponent) o;
+        resetMutableComponent();
+        for (ILSMComponent c : immutableComponents) {
+            LSMInvertedIndexComponent component = (LSMInvertedIndexComponent) c;
             component.getInvIndex().deactivate();
             component.getDeletedKeysBTree().deactivate();
             component.getInvIndex().destroy();
             component.getDeletedKeysBTree().destroy();
         }
-        diskComponents.clear();
+        immutableComponents.clear();
     }
 
     @Override
@@ -258,16 +212,16 @@
             throw new HyracksDataException(e);
         }
 
-        for (Object o : diskComponents) {
-            LSMInvertedIndexComponent component = (LSMInvertedIndexComponent) o;
+        for (ILSMComponent c : immutableComponents) {
+            LSMInvertedIndexComponent component = (LSMInvertedIndexComponent) c;
             component.getInvIndex().deactivate();
             component.getDeletedKeysBTree().deactivate();
         }
-        memComponent.getInvIndex().deactivate();
-        memComponent.getDeletedKeysBTree().deactivate();
-        memComponent.getInvIndex().destroy();
-        memComponent.getDeletedKeysBTree().destroy();
-        ((InMemoryBufferCache) memComponent.getInvIndex().getBufferCache()).close();
+        mutableComponent.getInvIndex().deactivate();
+        mutableComponent.getDeletedKeysBTree().deactivate();
+        mutableComponent.getInvIndex().destroy();
+        mutableComponent.getDeletedKeysBTree().destroy();
+        ((InMemoryBufferCache) mutableComponent.getInvIndex().getBufferCache()).close();
     }
 
     @Override
@@ -276,10 +230,10 @@
             throw new HyracksDataException("Failed to destroy the index since it is activated.");
         }
 
-        memComponent.getInvIndex().destroy();
-        memComponent.getDeletedKeysBTree().destroy();
-        for (Object o : diskComponents) {
-            LSMInvertedIndexComponent component = (LSMInvertedIndexComponent) o;
+        mutableComponent.getInvIndex().destroy();
+        mutableComponent.getDeletedKeysBTree().destroy();
+        for (ILSMComponent c : immutableComponents) {
+            LSMInvertedIndexComponent component = (LSMInvertedIndexComponent) c;
             component.getInvIndex().destroy();
             component.getDeletedKeysBTree().destroy();
         }
@@ -288,10 +242,10 @@
 
     @Override
     public void validate() throws HyracksDataException {
-        memComponent.getInvIndex().validate();
-        memComponent.getDeletedKeysBTree().validate();
-        for (Object o : diskComponents) {
-            LSMInvertedIndexComponent component = (LSMInvertedIndexComponent) o;
+        mutableComponent.getInvIndex().validate();
+        mutableComponent.getDeletedKeysBTree().validate();
+        for (ILSMComponent c : immutableComponents) {
+            LSMInvertedIndexComponent component = (LSMInvertedIndexComponent) c;
             component.getInvIndex().validate();
             component.getDeletedKeysBTree().validate();
         }
@@ -306,7 +260,7 @@
 
     private LSMInvertedIndexOpContext createOpContext(IModificationOperationCallback modificationCallback,
             ISearchOperationCallback searchCallback) {
-        return new LSMInvertedIndexOpContext(memComponent.getInvIndex(), memComponent.getDeletedKeysBTree(),
+        return new LSMInvertedIndexOpContext(mutableComponent.getInvIndex(), mutableComponent.getDeletedKeysBTree(),
                 modificationCallback, searchCallback);
     }
 
@@ -319,7 +273,7 @@
      * The keys in the in-memory deleted-keys BTree only refer to on-disk components.
      * We delete documents from the in-memory inverted index by deleting its entries directly,
      * while still adding the deleted key to the deleted-keys BTree.
-     * Otherwise, inserts would have to remove keys from the in-memory deleted-keys BTree which 
+     * Otherwise, inserts would have to remove keys from the in-memory deleted-keys BTree which
      * may cause incorrect behavior (lost deletes) in the following pathological case:
      * Insert doc 1, flush, delete doc 1, insert doc 1
      * After the sequence above doc 1 will now appear twice because the delete of the on-disk doc 1 has been lost.
@@ -360,22 +314,22 @@
     }
 
     @Override
-    public void search(IIndexCursor cursor, List<Object> diskComponents, ISearchPredicate pred,
-            IIndexOperationContext ictx, boolean includeMemComponent, AtomicInteger searcherRefCount)
+    public void search(IIndexCursor cursor, List<ILSMComponent> immutableComponents, ISearchPredicate pred,
+            IIndexOperationContext ictx, boolean includemutableComponent, AtomicInteger searcherRefCount)
             throws HyracksDataException, IndexException {
-        int numComponents = (includeMemComponent) ? diskComponents.size() : diskComponents.size() + 1;
+        int numComponents = (includemutableComponent) ? immutableComponents.size() : immutableComponents.size() + 1;
         ArrayList<IIndexAccessor> indexAccessors = new ArrayList<IIndexAccessor>(numComponents);
         ArrayList<IIndexAccessor> deletedKeysBTreeAccessors = new ArrayList<IIndexAccessor>(numComponents);
-        if (includeMemComponent) {
-            IIndexAccessor invIndexAccessor = memComponent.getInvIndex().createAccessor(NoOpOperationCallback.INSTANCE,
-                    NoOpOperationCallback.INSTANCE);
+        if (includemutableComponent) {
+            IIndexAccessor invIndexAccessor = mutableComponent.getInvIndex().createAccessor(
+                    NoOpOperationCallback.INSTANCE, NoOpOperationCallback.INSTANCE);
             indexAccessors.add(invIndexAccessor);
-            IIndexAccessor deletedKeysAccessor = memComponent.getDeletedKeysBTree().createAccessor(
+            IIndexAccessor deletedKeysAccessor = mutableComponent.getDeletedKeysBTree().createAccessor(
                     NoOpOperationCallback.INSTANCE, NoOpOperationCallback.INSTANCE);
             deletedKeysBTreeAccessors.add(deletedKeysAccessor);
         }
-        for (int i = 0; i < diskComponents.size(); i++) {
-            LSMInvertedIndexComponent component = (LSMInvertedIndexComponent) diskComponents.get(i);
+        for (int i = 0; i < immutableComponents.size(); i++) {
+            LSMInvertedIndexComponent component = (LSMInvertedIndexComponent) immutableComponents.get(i);
             IIndexAccessor invIndexAccessor = component.getInvIndex().createAccessor(NoOpOperationCallback.INSTANCE,
                     NoOpOperationCallback.INSTANCE);
             indexAccessors.add(invIndexAccessor);
@@ -383,13 +337,13 @@
                     NoOpOperationCallback.INSTANCE, NoOpOperationCallback.INSTANCE);
             deletedKeysBTreeAccessors.add(deletedKeysAccessor);
         }
-        ICursorInitialState initState = createCursorInitialState(pred, ictx, includeMemComponent, searcherRefCount,
+        ICursorInitialState initState = createCursorInitialState(pred, ictx, includemutableComponent, searcherRefCount,
                 indexAccessors, deletedKeysBTreeAccessors);
         cursor.open(initState, pred);
     }
 
     private ICursorInitialState createCursorInitialState(ISearchPredicate pred, IIndexOperationContext ictx,
-            boolean includeMemComponent, AtomicInteger searcherRefCount, ArrayList<IIndexAccessor> indexAccessors,
+            boolean includemutableComponent, AtomicInteger searcherRefCount, ArrayList<IIndexAccessor> indexAccessors,
             ArrayList<IIndexAccessor> deletedKeysBTreeAccessors) {
         ICursorInitialState initState = null;
         PermutingTupleReference keysOnlyTuple = createKeysOnlyTupleReference();
@@ -398,12 +352,13 @@
         // Distinguish between regular searches and range searches (mostly used in merges).
         if (pred instanceof InvertedIndexSearchPredicate) {
             initState = new LSMInvertedIndexSearchCursorInitialState(keyCmp, keysOnlyTuple, indexAccessors,
-                    deletedKeysBTreeAccessors, ictx, includeMemComponent, searcherRefCount, lsmHarness);
+                    deletedKeysBTreeAccessors, ictx, includemutableComponent, searcherRefCount, lsmHarness);
         } else {
-            InMemoryInvertedIndex memInvIndex = (InMemoryInvertedIndex) memComponent.getInvIndex();
+            InMemoryInvertedIndex memInvIndex = (InMemoryInvertedIndex) mutableComponent.getInvIndex();
             MultiComparator tokensAndKeysCmp = MultiComparator.create(memInvIndex.getBTree().getComparatorFactories());
             initState = new LSMInvertedIndexRangeSearchCursorInitialState(tokensAndKeysCmp, keyCmp, keysOnlyTuple,
-                    includeMemComponent, searcherRefCount, lsmHarness, indexAccessors, deletedKeysBTreeAccessors, pred);
+                    includemutableComponent, searcherRefCount, lsmHarness, indexAccessors, deletedKeysBTreeAccessors,
+                    pred);
         }
         return initState;
     }
@@ -422,13 +377,15 @@
     }
 
     @Override
-    public Object merge(List<Object> mergedComponents, ILSMIOOperation operation) throws HyracksDataException,
-            IndexException {
+    public ILSMComponent merge(List<ILSMComponent> mergedComponents, ILSMIOOperation operation)
+            throws HyracksDataException, IndexException {
         LSMInvertedIndexMergeOperation mergeOp = (LSMInvertedIndexMergeOperation) operation;
 
         // Create an inverted index instance.
-        IInvertedIndex mergedDiskInvertedIndex = createDiskInvIndex(diskInvIndexFactory,
-                mergeOp.getDictBTreeMergeTarget(), true);
+        LSMInvertedIndexComponent component = createDiskInvIndexComponent(componentFactory,
+                mergeOp.getDictBTreeMergeTarget(), mergeOp.getDeletedKeysBTreeMergeTarget(), true);
+
+        IInvertedIndex mergedDiskInvertedIndex = component.getInvIndex();
         IIndexCursor cursor = mergeOp.getCursor();
         IIndexBulkLoader invIndexBulkLoader = mergedDiskInvertedIndex.createBulkLoader(1.0f, true);
         try {
@@ -443,8 +400,7 @@
         invIndexBulkLoader.end();
 
         // Create an empty deleted keys BTree (do nothing with the returned index).
-        BTree deletedKeysBTree = (BTree) createDiskTree(deletedKeysBTreeFactory,
-                mergeOp.getDeletedKeysBTreeMergeTarget(), true);
+        BTree deletedKeysBTree = component.getDeletedKeysBTree();
 
         // Add the merged components for cleanup.
         mergedComponents.addAll(mergeOp.getMergingComponents());
@@ -461,7 +417,7 @@
         RangePredicate mergePred = new RangePredicate(null, null, true, true, null, null);
 
         // Scan diskInvertedIndexes ignoring the memoryInvertedIndex.
-        List<Object> mergingComponents = lsmHarness.search(cursor, mergePred, ctx, false);
+        List<ILSMComponent> mergingComponents = lsmHarness.search(cursor, mergePred, ctx, false);
         if (mergingComponents.size() <= 1) {
             cursor.close();
             return null;
@@ -476,28 +432,18 @@
         OnDiskInvertedIndex lastInvIndex = (OnDiskInvertedIndex) lastComponent.getInvIndex();
         String lastFileName = lastInvIndex.getBTree().getFileReference().getFile().getName();
 
-        LSMInvertedIndexFileNameComponent fileNameComponent = (LSMInvertedIndexFileNameComponent) fileManager
-                .getRelMergeFileName(firstFileName, lastFileName);
-        FileReference dictBTreeFileRef = fileManager.createMergeFile(fileNameComponent.getDictBTreeFileName());
-        FileReference deletedKeysBTreeFileRef = fileManager.createMergeFile(fileNameComponent
-                .getDeletedKeysBTreeFileName());
-
+        LSMComponentFileReferences relMergeFileRefs = fileManager.getRelMergeFileReference(firstFileName, lastFileName);
         LSMInvertedIndexMergeOperation mergeOp = new LSMInvertedIndexMergeOperation(this, mergingComponents, cursor,
-                dictBTreeFileRef, deletedKeysBTreeFileRef, callback);
+                relMergeFileRefs.getInsertIndexFileReference(), relMergeFileRefs.getDeleteIndexFileReference(),
+                callback);
 
         return mergeOp;
     }
 
     @Override
-    public void addMergedComponent(Object newComponent, List<Object> mergedComponents) {
-        diskComponents.removeAll(mergedComponents);
-        diskComponents.addLast(newComponent);
-    }
-
-    @Override
-    public void cleanUpAfterMerge(List<Object> mergedComponents) throws HyracksDataException {
-        for (Object o : mergedComponents) {
-            LSMInvertedIndexComponent component = (LSMInvertedIndexComponent) o;
+    public void cleanUpAfterMerge(List<ILSMComponent> mergedComponents) throws HyracksDataException {
+        for (ILSMComponent c : mergedComponents) {
+            LSMInvertedIndexComponent component = (LSMInvertedIndexComponent) c;
             BTree oldDeletedKeysBTree = component.getDeletedKeysBTree();
             oldDeletedKeysBTree.deactivate();
             oldDeletedKeysBTree.destroy();
@@ -508,16 +454,17 @@
     }
 
     @Override
-    public Object flush(ILSMIOOperation operation) throws HyracksDataException, IndexException {
+    public ILSMComponent flush(ILSMIOOperation operation) throws HyracksDataException, IndexException {
         LSMInvertedIndexFlushOperation flushOp = (LSMInvertedIndexFlushOperation) operation;
 
         // Create an inverted index instance to be bulk loaded.
-        IInvertedIndex diskInvertedIndex = createDiskInvIndex(diskInvIndexFactory, flushOp.getDictBTreeFlushTarget(),
-                true);
+        LSMInvertedIndexComponent component = createDiskInvIndexComponent(componentFactory,
+                flushOp.getDictBTreeFlushTarget(), flushOp.getDeletedKeysBTreeFlushTarget(), true);
+        IInvertedIndex diskInvertedIndex = component.getInvIndex();
 
         // Create a scan cursor on the BTree underlying the in-memory inverted index.
-        InMemoryInvertedIndexAccessor memInvIndexAccessor = (InMemoryInvertedIndexAccessor) memComponent.getInvIndex()
-                .createAccessor(NoOpOperationCallback.INSTANCE, NoOpOperationCallback.INSTANCE);
+        InMemoryInvertedIndexAccessor memInvIndexAccessor = (InMemoryInvertedIndexAccessor) mutableComponent
+                .getInvIndex().createAccessor(NoOpOperationCallback.INSTANCE, NoOpOperationCallback.INSTANCE);
         BTreeAccessor memBTreeAccessor = memInvIndexAccessor.getBTreeAccessor();
         RangePredicate nullPred = new RangePredicate(null, null, true, true, null, null);
         IIndexCursor scanCursor = memBTreeAccessor.createSearchCursor();
@@ -536,11 +483,10 @@
         invIndexBulkLoader.end();
 
         // Create an BTree instance for the deleted keys.
-        BTree diskDeletedKeysBTree = (BTree) createDiskTree(deletedKeysBTreeFactory,
-                flushOp.getDeletedKeysBTreeFlushTarget(), true);
+        BTree diskDeletedKeysBTree = component.getDeletedKeysBTree();
 
         // Create a scan cursor on the deleted keys BTree underlying the in-memory inverted index.
-        IIndexAccessor deletedKeysBTreeAccessor = memComponent.getDeletedKeysBTree().createAccessor(
+        IIndexAccessor deletedKeysBTreeAccessor = mutableComponent.getDeletedKeysBTree().createAccessor(
                 NoOpOperationCallback.INSTANCE, NoOpOperationCallback.INSTANCE);
         IIndexCursor deletedKeysScanCursor = deletedKeysBTreeAccessor.createSearchCursor();
         deletedKeysBTreeAccessor.search(deletedKeysScanCursor, nullPred);
@@ -560,32 +506,28 @@
         return new LSMInvertedIndexComponent(diskInvertedIndex, diskDeletedKeysBTree);
     }
 
-    @Override
-    public void addFlushedComponent(Object index) {
-        diskComponents.addFirst(index);
+    private ILSMComponent createBulkLoadTarget() throws HyracksDataException, IndexException {
+        LSMComponentFileReferences componentFileRefs = fileManager.getRelFlushFileReference();
+        return createDiskInvIndexComponent(componentFactory, componentFileRefs.getInsertIndexFileReference(),
+                componentFileRefs.getDeleteIndexFileReference(), true);
     }
 
     public class LSMInvertedIndexBulkLoader implements IIndexBulkLoader {
-        private final IInvertedIndex invIndex;
-        private final BTree deletedKeysBTree;
+        private final ILSMComponent component;
         private final IIndexBulkLoader invIndexBulkLoader;
 
         public LSMInvertedIndexBulkLoader(float fillFactor, boolean verifyInput) throws IndexException {
             // Note that by using a flush target file name, we state that the
             // new bulk loaded tree is "newer" than any other merged tree.
             try {
-                LSMInvertedIndexFileNameComponent fileNameComponent = (LSMInvertedIndexFileNameComponent) fileManager
-                        .getRelFlushFileName();
-                FileReference dictBTreeFileRef = fileManager.createFlushFile(fileNameComponent.getDictBTreeFileName());
-                invIndex = createDiskInvIndex(diskInvIndexFactory, dictBTreeFileRef, true);
-                // Create an empty deleted-keys BTree.
-                FileReference deletedKeysBTreeFile = fileManager.createFlushFile(fileNameComponent
-                        .getDeletedKeysBTreeFileName());
-                deletedKeysBTree = (BTree) createDiskTree(deletedKeysBTreeFactory, deletedKeysBTreeFile, true);
+                component = createBulkLoadTarget();
             } catch (HyracksDataException e) {
                 throw new TreeIndexException(e);
+            } catch (IndexException e) {
+                throw new TreeIndexException(e);
             }
-            invIndexBulkLoader = invIndex.createBulkLoader(fillFactor, verifyInput);
+            invIndexBulkLoader = ((LSMInvertedIndexComponent) component).getInvIndex().createBulkLoader(fillFactor,
+                    verifyInput);
         }
 
         @Override
@@ -605,69 +547,33 @@
         }
 
         protected void handleException() throws HyracksDataException {
-            invIndex.deactivate();
-            invIndex.destroy();
-            deletedKeysBTree.deactivate();
-            deletedKeysBTree.destroy();
+            ((LSMInvertedIndexComponent) component).getInvIndex().deactivate();
+            ((LSMInvertedIndexComponent) component).getInvIndex().destroy();
+            ((LSMInvertedIndexComponent) component).getDeletedKeysBTree().deactivate();
+            ((LSMInvertedIndexComponent) component).getDeletedKeysBTree().destroy();
         }
 
         @Override
         public void end() throws IndexException, HyracksDataException {
             invIndexBulkLoader.end();
-            LSMInvertedIndexComponent diskComponent = new LSMInvertedIndexComponent(invIndex, deletedKeysBTree);
-            lsmHarness.addBulkLoadedComponent(diskComponent);
+            lsmHarness.addBulkLoadedComponent(component);
         }
     }
 
     @Override
-    public void resetInMemoryComponent() throws HyracksDataException {
+    public void resetMutableComponent() throws HyracksDataException {
         memFreePageManager.reset();
-        memComponent.getInvIndex().clear();
-        memComponent.getDeletedKeysBTree().clear();
+        mutableComponent.getInvIndex().clear();
+        mutableComponent.getDeletedKeysBTree().clear();
     }
 
     @Override
     public long getMemoryAllocationSize() {
-        InMemoryBufferCache memBufferCache = (InMemoryBufferCache) memComponent.getInvIndex().getBufferCache();
+        InMemoryBufferCache memBufferCache = (InMemoryBufferCache) mutableComponent.getInvIndex().getBufferCache();
         return memBufferCache.getNumPages() * memBufferCache.getPageSize();
     }
 
     @Override
-    public IInMemoryFreePageManager getInMemoryFreePageManager() {
-        return memFreePageManager;
-    }
-
-    @Override
-    public List<Object> getDiskComponents() {
-        return diskComponents;
-    }
-
-    @Override
-    public ILSMComponentFinalizer getComponentFinalizer() {
-        return componentFinalizer;
-    }
-
-    @Override
-    public ILSMFlushController getFlushController() {
-        return lsmHarness.getFlushController();
-    }
-
-    @Override
-    public ILSMOperationTracker getOperationTracker() {
-        return lsmHarness.getOperationTracker();
-    }
-
-    @Override
-    public ILSMIOOperationScheduler getIOScheduler() {
-        return lsmHarness.getIOScheduler();
-    }
-
-    @Override
-    public IBufferCache getBufferCache() {
-        return diskBufferCache;
-    }
-
-    @Override
     public IInvertedListCursor createInvertedListCursor() {
         throw new UnsupportedOperationException("Cannot create inverted list cursor on lsm inverted index.");
     }
@@ -701,4 +607,26 @@
     public IBinaryTokenizerFactory getTokenizerFactory() {
         return tokenizerFactory;
     }
+
+    @Override
+    public void markAsValid(ILSMComponent lsmComponent) throws HyracksDataException {
+        LSMInvertedIndexComponent invIndexComponent = (LSMInvertedIndexComponent) lsmComponent;
+        OnDiskInvertedIndex invIndex = (OnDiskInvertedIndex) invIndexComponent.getInvIndex();
+        ITreeIndex treeIndex = invIndex.getBTree();
+        // Flush inverted index first.
+        forceFlushDirtyPages(treeIndex);
+        forceFlushInvListsFileDirtyPages(invIndex);
+        // Flush deleted keys BTree.
+        forceFlushDirtyPages(invIndexComponent.getDeletedKeysBTree());
+        // We use the dictionary BTree for marking the inverted index as valid.
+        markAsValidInternal(treeIndex);
+    }
+
+    protected void forceFlushInvListsFileDirtyPages(OnDiskInvertedIndex invIndex) throws HyracksDataException {
+        int fileId = invIndex.getInvListsFileId();
+        IBufferCache bufferCache = invIndex.getBufferCache();
+        int startPageId = 0;
+        int maxPageId = invIndex.getInvListsMaxPageId();
+        forceFlushDirtyPages(bufferCache, fileId, startPageId, maxPageId);
+    }
 }
diff --git a/hyracks-storage-am-lsm-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexAccessor.java b/hyracks-storage-am-lsm-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexAccessor.java
index cd870b4..5d329a0 100644
--- a/hyracks-storage-am-lsm-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexAccessor.java
+++ b/hyracks-storage-am-lsm-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexAccessor.java
@@ -16,7 +16,6 @@
 package edu.uci.ics.hyracks.storage.am.lsm.invertedindex.impls;
 
 import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
-import edu.uci.ics.hyracks.api.io.FileReference;
 import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
 import edu.uci.ics.hyracks.storage.am.common.api.IIndexCursor;
 import edu.uci.ics.hyracks.storage.am.common.api.ISearchPredicate;
@@ -28,9 +27,9 @@
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIndexAccessor;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIndexFileManager;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIndexOperationContext;
+import edu.uci.ics.hyracks.storage.am.lsm.common.impls.LSMComponentFileReferences;
 import edu.uci.ics.hyracks.storage.am.lsm.invertedindex.api.IInvertedIndexAccessor;
 import edu.uci.ics.hyracks.storage.am.lsm.invertedindex.api.IInvertedListCursor;
-import edu.uci.ics.hyracks.storage.am.lsm.invertedindex.impls.LSMInvertedIndexFileManager.LSMInvertedIndexFileNameComponent;
 
 public class LSMInvertedIndexAccessor implements ILSMIndexAccessor, IInvertedIndexAccessor {
 
@@ -82,12 +81,9 @@
 
     @Override
     public ILSMIOOperation createFlushOperation(ILSMIOOperationCallback callback) {
-        LSMInvertedIndexFileNameComponent fileNameComponent = (LSMInvertedIndexFileNameComponent) fileManager
-                .getRelFlushFileName();
-        FileReference dictBTreeFileRef = fileManager.createFlushFile(fileNameComponent.getDictBTreeFileName());
-        FileReference deletedKeysBTreeFileRef = fileManager.createFlushFile(fileNameComponent
-                .getDeletedKeysBTreeFileName());
-        return new LSMInvertedIndexFlushOperation(lsmHarness.getIndex(), dictBTreeFileRef, deletedKeysBTreeFileRef,
+        LSMComponentFileReferences componentFileRefs = fileManager.getRelFlushFileReference();
+        return new LSMInvertedIndexFlushOperation(lsmHarness.getIndex(),
+                componentFileRefs.getInsertIndexFileReference(), componentFileRefs.getDeleteIndexFileReference(),
                 callback);
     }
 
diff --git a/hyracks-storage-am-lsm-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexComponent.java b/hyracks-storage-am-lsm-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexComponent.java
new file mode 100644
index 0000000..ce82317
--- /dev/null
+++ b/hyracks-storage-am-lsm-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexComponent.java
@@ -0,0 +1,76 @@
+/*
+ * Copyright 2009-2012 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.lsm.invertedindex.impls;
+
+import edu.uci.ics.hyracks.storage.am.btree.impls.BTree;
+import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMComponent;
+import edu.uci.ics.hyracks.storage.am.lsm.common.impls.LSMComponentState;
+import edu.uci.ics.hyracks.storage.am.lsm.invertedindex.api.IInvertedIndex;
+
+public class LSMInvertedIndexComponent implements ILSMComponent {
+
+    private final IInvertedIndex invIndex;
+    private final BTree deletedKeysBTree;
+
+    public LSMInvertedIndexComponent(IInvertedIndex invIndex, BTree deletedKeysBTree) {
+        this.invIndex = invIndex;
+        this.deletedKeysBTree = deletedKeysBTree;
+    }
+
+    @Override
+    public void activate() {
+        // TODO Auto-generated method stub
+
+    }
+
+    @Override
+    public void deactivate() {
+        // TODO Auto-generated method stub
+
+    }
+
+    @Override
+    public void threadEnter() {
+        // TODO Auto-generated method stub
+
+    }
+
+    @Override
+    public void threadExit() {
+        // TODO Auto-generated method stub
+
+    }
+
+    @Override
+    public void setState(LSMComponentState state) {
+        // TODO Auto-generated method stub
+
+    }
+
+    @Override
+    public LSMComponentState getState() {
+        // TODO Auto-generated method stub
+        return null;
+    }
+
+    public IInvertedIndex getInvIndex() {
+        return invIndex;
+    }
+
+    public BTree getDeletedKeysBTree() {
+        return deletedKeysBTree;
+    }
+}
diff --git a/hyracks-storage-am-lsm-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexComponentFactory.java b/hyracks-storage-am-lsm-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexComponentFactory.java
new file mode 100644
index 0000000..23fc0ae
--- /dev/null
+++ b/hyracks-storage-am-lsm-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexComponentFactory.java
@@ -0,0 +1,48 @@
+/*
+ * Copyright 2009-2012 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.lsm.invertedindex.impls;
+
+import edu.uci.ics.hyracks.storage.am.btree.impls.BTree;
+import edu.uci.ics.hyracks.storage.am.common.api.IndexException;
+import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMComponent;
+import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMComponentFactory;
+import edu.uci.ics.hyracks.storage.am.lsm.common.impls.LSMComponentFileReferences;
+import edu.uci.ics.hyracks.storage.am.lsm.common.impls.TreeIndexFactory;
+import edu.uci.ics.hyracks.storage.am.lsm.invertedindex.ondisk.OnDiskInvertedIndexFactory;
+import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
+
+public class LSMInvertedIndexComponentFactory implements ILSMComponentFactory {
+    private final OnDiskInvertedIndexFactory diskInvIndexFactory;
+    private final TreeIndexFactory<BTree> btreeFactory;
+
+    public LSMInvertedIndexComponentFactory(OnDiskInvertedIndexFactory diskInvIndexFactory,
+            TreeIndexFactory<BTree> btreeFactory) {
+        this.diskInvIndexFactory = diskInvIndexFactory;
+        this.btreeFactory = btreeFactory;
+    }
+
+    @Override
+    public ILSMComponent createLSMComponentInstance(LSMComponentFileReferences cfr) throws IndexException {
+        return new LSMInvertedIndexComponent(
+                diskInvIndexFactory.createIndexInstance(cfr.getInsertIndexFileReference()),
+                btreeFactory.createIndexInstance(cfr.getDeleteIndexFileReference()));
+    }
+
+    @Override
+    public IBufferCache getBufferCache() {
+        return diskInvIndexFactory.getBufferCache();
+    }
+}
diff --git a/hyracks-storage-am-lsm-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexComponentFinalizer.java b/hyracks-storage-am-lsm-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexComponentFinalizer.java
deleted file mode 100644
index 40f19e9..0000000
--- a/hyracks-storage-am-lsm-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexComponentFinalizer.java
+++ /dev/null
@@ -1,61 +0,0 @@
-/*
- * 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.lsm.invertedindex.impls;
-
-import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
-import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndex;
-import edu.uci.ics.hyracks.storage.am.lsm.common.impls.TreeIndexComponentFinalizer;
-import edu.uci.ics.hyracks.storage.am.lsm.invertedindex.impls.LSMInvertedIndex.LSMInvertedIndexComponent;
-import edu.uci.ics.hyracks.storage.am.lsm.invertedindex.ondisk.OnDiskInvertedIndex;
-import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
-import edu.uci.ics.hyracks.storage.common.file.IFileMapProvider;
-
-public class LSMInvertedIndexComponentFinalizer extends TreeIndexComponentFinalizer {
-
-    public LSMInvertedIndexComponentFinalizer(IFileMapProvider fileMapProvider) {
-        super(fileMapProvider);
-    }
-
-    @Override
-    public boolean isValid(Object lsmComponent) throws HyracksDataException {
-        LSMInvertedIndexComponent invIndexComponent = (LSMInvertedIndexComponent) lsmComponent;
-        OnDiskInvertedIndex invIndex = (OnDiskInvertedIndex) invIndexComponent.getInvIndex();
-        // Use the dictionary BTree for validation.
-        return super.isValid(invIndex.getBTree());
-    }
-
-    @Override
-    public void finalize(Object lsmComponent) throws HyracksDataException {
-        LSMInvertedIndexComponent invIndexComponent = (LSMInvertedIndexComponent) lsmComponent;
-        OnDiskInvertedIndex invIndex = (OnDiskInvertedIndex) invIndexComponent.getInvIndex();
-        ITreeIndex treeIndex = invIndex.getBTree();
-        // Flush inverted index first.
-        forceFlushDirtyPages(treeIndex);
-        forceFlushInvListsFileDirtyPages(invIndex);
-        // Flush deleted keys BTree.
-        forceFlushDirtyPages(invIndexComponent.getDeletedKeysBTree());
-        // We use the dictionary BTree for marking the inverted index as valid.
-        markAsValid(treeIndex);
-    }
-
-    protected void forceFlushInvListsFileDirtyPages(OnDiskInvertedIndex invIndex) throws HyracksDataException {
-        int fileId = invIndex.getInvListsFileId();
-        IBufferCache bufferCache = invIndex.getBufferCache();
-        int startPageId = 0;
-        int maxPageId = invIndex.getInvListsMaxPageId();
-        forceFlushDirtyPages(bufferCache, fileId, startPageId, maxPageId);
-    }
-}
diff --git a/hyracks-storage-am-lsm-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexFileManager.java b/hyracks-storage-am-lsm-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexFileManager.java
index 436aba2..e4eebeb 100644
--- a/hyracks-storage-am-lsm-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexFileManager.java
+++ b/hyracks-storage-am-lsm-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexFileManager.java
@@ -19,6 +19,7 @@
 import java.io.FilenameFilter;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.Date;
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
@@ -27,8 +28,9 @@
 import edu.uci.ics.hyracks.api.io.FileReference;
 import edu.uci.ics.hyracks.api.io.IIOManager;
 import edu.uci.ics.hyracks.api.io.IODeviceHandle;
-import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMComponentFinalizer;
+import edu.uci.ics.hyracks.storage.am.common.api.IndexException;
 import edu.uci.ics.hyracks.storage.am.lsm.common.impls.BTreeFactory;
+import edu.uci.ics.hyracks.storage.am.lsm.common.impls.LSMComponentFileReferences;
 import edu.uci.ics.hyracks.storage.am.lsm.common.impls.LSMIndexFileManager;
 import edu.uci.ics.hyracks.storage.am.lsm.invertedindex.api.IInvertedIndexFileNameMapper;
 import edu.uci.ics.hyracks.storage.common.file.IFileMapProvider;
@@ -61,39 +63,45 @@
     }
 
     @Override
-    public Object getRelFlushFileName() {
-        String baseName = (String) super.getRelFlushFileName();
-        return new LSMInvertedIndexFileNameComponent(baseName + SPLIT_STRING + DICT_BTREE_SUFFIX, baseName
-                + SPLIT_STRING + DELETED_KEYS_BTREE_SUFFIX);
-
+    public LSMComponentFileReferences getRelFlushFileReference() {
+        Date date = new Date();
+        String ts = formatter.format(date);
+        String baseName = baseDir + ts + SPLIT_STRING + ts;
+        // Begin timestamp and end timestamp are identical since it is a flush
+        return new LSMComponentFileReferences(createFlushFile(baseName + SPLIT_STRING + DICT_BTREE_SUFFIX),
+                createFlushFile(baseName + SPLIT_STRING + DELETED_KEYS_BTREE_SUFFIX));
     }
 
     @Override
-    public Object getRelMergeFileName(String firstFileName, String lastFileName) throws HyracksDataException {
-        String baseName = (String) super.getRelMergeFileName(firstFileName, lastFileName);
-        return new LSMInvertedIndexFileNameComponent(baseName + SPLIT_STRING + DICT_BTREE_SUFFIX, baseName
-                + SPLIT_STRING + DELETED_KEYS_BTREE_SUFFIX);
+    public LSMComponentFileReferences getRelMergeFileReference(String firstFileName, String lastFileName)
+            throws HyracksDataException {
+        String[] firstTimestampRange = firstFileName.split(SPLIT_STRING);
+        String[] lastTimestampRange = lastFileName.split(SPLIT_STRING);
+
+        String baseName = baseDir + firstTimestampRange[0] + SPLIT_STRING + lastTimestampRange[1];
+        // Get the range of timestamps by taking the earliest and the latest timestamps
+        return new LSMComponentFileReferences(createMergeFile(baseName + SPLIT_STRING + DICT_BTREE_SUFFIX),
+                createMergeFile(baseName + SPLIT_STRING + DELETED_KEYS_BTREE_SUFFIX));
     }
 
     @Override
-    public List<Object> cleanupAndGetValidFiles(ILSMComponentFinalizer componentFinalizer) throws HyracksDataException {
-        List<Object> validFiles = new ArrayList<Object>();
+    public List<LSMComponentFileReferences> cleanupAndGetValidFiles() throws HyracksDataException, IndexException {
+        List<LSMComponentFileReferences> validFiles = new ArrayList<LSMComponentFileReferences>();
         ArrayList<ComparableFileName> allDictBTreeFiles = new ArrayList<ComparableFileName>();
         ArrayList<ComparableFileName> allDeletedKeysBTreeFiles = new ArrayList<ComparableFileName>();
 
         // Gather files from all IODeviceHandles.
         for (IODeviceHandle dev : ioManager.getIODevices()) {
-            cleanupAndGetValidFilesInternal(dev, deletedKeysBTreeFilter, btreeFactory, componentFinalizer,
-                    allDeletedKeysBTreeFiles);
+            cleanupAndGetValidFilesInternal(dev, deletedKeysBTreeFilter, btreeFactory, allDeletedKeysBTreeFiles);
             HashSet<String> deletedKeysBTreeFilesSet = new HashSet<String>();
-            for (ComparableFileName cmpFileName : allDictBTreeFiles) {
+            for (ComparableFileName cmpFileName : allDeletedKeysBTreeFiles) {
                 int index = cmpFileName.fileName.lastIndexOf(SPLIT_STRING);
                 deletedKeysBTreeFilesSet.add(cmpFileName.fileName.substring(0, index));
             }
             // We use the dictionary BTree of the inverted index for validation.
             // List of valid dictionary BTree files that may or may not have a deleted-keys BTree buddy. Will check for buddies below.
             ArrayList<ComparableFileName> tmpAllBTreeFiles = new ArrayList<ComparableFileName>();
-            cleanupAndGetValidFilesInternal(dev, dictBTreeFilter, btreeFactory, componentFinalizer, tmpAllBTreeFiles);
+            cleanupAndGetValidFilesInternal(dev, dictBTreeFilter, btreeFactory, tmpAllBTreeFiles);
             // Look for buddy deleted-keys BTrees for all valid dictionary BTrees. 
             // If no buddy is found, delete the file, otherwise add the dictionary BTree to allBTreeFiles. 
             for (ComparableFileName cmpFileName : tmpAllBTreeFiles) {
@@ -123,8 +131,8 @@
         }
 
         if (allDictBTreeFiles.size() == 1 && allDeletedKeysBTreeFiles.size() == 1) {
-            validFiles.add(new LSMInvertedIndexFileNameComponent(allDictBTreeFiles.get(0).fullPath,
-                    allDeletedKeysBTreeFiles.get(0).fullPath));
+            validFiles.add(new LSMComponentFileReferences(allDictBTreeFiles.get(0).fileRef, allDeletedKeysBTreeFiles
+                    .get(0).fileRef));
             return validFiles;
         }
 
@@ -175,8 +183,7 @@
         while (dictBTreeFileIter.hasNext() && deletedKeysBTreeIter.hasNext()) {
             ComparableFileName cmpDictBTreeFile = dictBTreeFileIter.next();
             ComparableFileName cmpDeletedKeysBTreeFile = deletedKeysBTreeIter.next();
-            validFiles.add(new LSMInvertedIndexFileNameComponent(cmpDictBTreeFile.fullPath,
-                    cmpDeletedKeysBTreeFile.fullPath));
+            validFiles.add(new LSMComponentFileReferences(cmpDictBTreeFile.fileRef, cmpDeletedKeysBTreeFile.fileRef));
         }
 
         return validFiles;
@@ -188,26 +195,4 @@
         String file = dictBTreeFilePath.substring(0, index);
         return file + SPLIT_STRING + INVLISTS_SUFFIX;
     }
-
-    /**
-     * No need to store the inverted-lists file name because it can be mapped from the
-     * dictionary BTree file name. 
-     */
-    public class LSMInvertedIndexFileNameComponent {
-        private final String dictBTreeFileName;
-        private final String deletedKeysBTreeFileName;
-
-        LSMInvertedIndexFileNameComponent(String dictBTreeFileName, String deletedKeysBTreeFileName) {
-            this.dictBTreeFileName = dictBTreeFileName;
-            this.deletedKeysBTreeFileName = deletedKeysBTreeFileName;
-        }
-
-        public String getDictBTreeFileName() {
-            return dictBTreeFileName;
-        }
-
-        public String getDeletedKeysBTreeFileName() {
-            return deletedKeysBTreeFileName;
-        }
-    }
 }
diff --git a/hyracks-storage-am-lsm-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexMergeOperation.java b/hyracks-storage-am-lsm-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexMergeOperation.java
index afae347..e6baf9d 100644
--- a/hyracks-storage-am-lsm-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexMergeOperation.java
+++ b/hyracks-storage-am-lsm-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexMergeOperation.java
@@ -24,23 +24,24 @@
 import edu.uci.ics.hyracks.storage.am.common.api.IIndexCursor;
 import edu.uci.ics.hyracks.storage.am.common.api.IndexException;
 import edu.uci.ics.hyracks.storage.am.common.impls.NoOpOperationCallback;
+import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMComponent;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIOOperation;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallback;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIndex;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIndexAccessor;
-import edu.uci.ics.hyracks.storage.am.lsm.invertedindex.impls.LSMInvertedIndex.LSMInvertedIndexComponent;
 import edu.uci.ics.hyracks.storage.am.lsm.invertedindex.ondisk.OnDiskInvertedIndex;
 
 public class LSMInvertedIndexMergeOperation implements ILSMIOOperation {
     private final ILSMIndex index;
-    private final List<Object> mergingComponents;
+    private final List<ILSMComponent> mergingComponents;
     private final IIndexCursor cursor;
     private final FileReference dictBTreeMergeTarget;
     private final FileReference deletedKeysBTreeMergeTarget;
     private final ILSMIOOperationCallback callback;
 
-    public LSMInvertedIndexMergeOperation(ILSMIndex index, List<Object> mergingComponents, IIndexCursor cursor,
-            FileReference dictBTreeMergeTarget, FileReference deletedKeysBTreeMergeTarget, ILSMIOOperationCallback callback) {
+    public LSMInvertedIndexMergeOperation(ILSMIndex index, List<ILSMComponent> mergingComponents, IIndexCursor cursor,
+            FileReference dictBTreeMergeTarget, FileReference deletedKeysBTreeMergeTarget,
+            ILSMIOOperationCallback callback) {
         this.index = index;
         this.mergingComponents = mergingComponents;
         this.cursor = cursor;
@@ -93,7 +94,7 @@
         return cursor;
     }
 
-    public List<Object> getMergingComponents() {
+    public List<ILSMComponent> getMergingComponents() {
         return mergingComponents;
     }
 
diff --git a/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/AbstractLSMRTree.java b/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/AbstractLSMRTree.java
index 57fd2db..f1cab5e 100644
--- a/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/AbstractLSMRTree.java
+++ b/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/AbstractLSMRTree.java
@@ -16,7 +16,6 @@
 package edu.uci.ics.hyracks.storage.am.lsm.rtree.impls;
 
 import java.io.File;
-import java.util.LinkedList;
 import java.util.List;
 
 import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparatorFactory;
@@ -38,56 +37,33 @@
 import edu.uci.ics.hyracks.storage.am.common.impls.NoOpOperationCallback;
 import edu.uci.ics.hyracks.storage.am.common.ophelpers.IndexOperation;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.IInMemoryBufferCache;
-import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMComponentFinalizer;
+import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMComponent;
+import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMComponentFactory;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMFlushController;
-import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMHarness;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIOOperationScheduler;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIndexAccessor;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIndexFileManager;
-import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIndexInternal;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMMergePolicy;
-import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMOperationTracker;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMOperationTrackerFactory;
 import edu.uci.ics.hyracks.storage.am.lsm.common.freepage.InMemoryBufferCache;
+import edu.uci.ics.hyracks.storage.am.lsm.common.impls.AbstractLSMIndex;
 import edu.uci.ics.hyracks.storage.am.lsm.common.impls.BlockingIOOperationCallback;
-import edu.uci.ics.hyracks.storage.am.lsm.common.impls.LSMHarness;
+import edu.uci.ics.hyracks.storage.am.lsm.common.impls.LSMComponentFileReferences;
 import edu.uci.ics.hyracks.storage.am.lsm.common.impls.TreeIndexFactory;
 import edu.uci.ics.hyracks.storage.am.rtree.api.IRTreeInteriorFrame;
 import edu.uci.ics.hyracks.storage.am.rtree.api.IRTreeLeafFrame;
 import edu.uci.ics.hyracks.storage.am.rtree.impls.RTree;
-import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
 import edu.uci.ics.hyracks.storage.common.file.IFileMapProvider;
 
-public abstract class AbstractLSMRTree implements ILSMIndexInternal, ITreeIndex {
-
-    public static class LSMRTreeComponent {
-        private final RTree rtree;
-        private final BTree btree;
-
-        LSMRTreeComponent(RTree rtree, BTree btree) {
-            this.rtree = rtree;
-            this.btree = btree;
-        }
-
-        public RTree getRTree() {
-            return rtree;
-        }
-
-        public BTree getBTree() {
-            return btree;
-        }
-    }
-
-    protected final ILSMHarness lsmHarness;
+public abstract class AbstractLSMRTree extends AbstractLSMIndex implements ITreeIndex {
 
     protected final ILinearizeComparatorFactory linearizer;
     protected final int[] comparatorFields;
     protected final IBinaryComparatorFactory[] linearizerArray;
 
     // In-memory components.
-    protected final LSMRTreeComponent memComponent;
+    protected final LSMRTreeComponent mutableComponent;
     protected final IInMemoryBufferCache memBufferCache;
-    protected final IInMemoryFreePageManager memFreePageManager;
 
     // This is used to estimate number of tuples in the memory RTree and BTree
     // for efficient memory allocation in the sort operation prior to flushing
@@ -96,16 +72,8 @@
     protected TreeTupleSorter rTreeTupleSorter;
 
     // On-disk components.
-    protected final ILSMIndexFileManager fileManager;
-    protected final IBufferCache diskBufferCache;
-    protected final IFileMapProvider diskFileMapProvider;
     // For creating RTree's used in flush and merge.
-    protected final TreeIndexFactory<RTree> diskRTreeFactory;
-    // List of LSMRTreeComponent instances. Using Object for better sharing via
-    // ILSMIndex + LSMHarness.
-    protected final LinkedList<Object> diskComponents;
-    // Helps to guarantees physical consistency of LSM components.
-    protected final ILSMComponentFinalizer componentFinalizer;
+    protected final ILSMComponentFactory componentFactory;
 
     private IBinaryComparatorFactory[] btreeCmpFactories;
     private IBinaryComparatorFactory[] rtreeCmpFactories;
@@ -116,47 +84,38 @@
     protected final ITreeIndexFrameFactory rtreeLeafFrameFactory;
     protected final ITreeIndexFrameFactory btreeLeafFrameFactory;
 
-    private boolean isActivated;
-
     public AbstractLSMRTree(IInMemoryBufferCache memBufferCache, IInMemoryFreePageManager memFreePageManager,
             ITreeIndexFrameFactory rtreeInteriorFrameFactory, ITreeIndexFrameFactory rtreeLeafFrameFactory,
             ITreeIndexFrameFactory btreeInteriorFrameFactory, ITreeIndexFrameFactory btreeLeafFrameFactory,
             ILSMIndexFileManager fileManager, TreeIndexFactory<RTree> diskRTreeFactory,
-            IFileMapProvider diskFileMapProvider, ILSMComponentFinalizer componentFinalizer, int fieldCount,
+            ILSMComponentFactory componentFactory, IFileMapProvider diskFileMapProvider, int fieldCount,
             IBinaryComparatorFactory[] rtreeCmpFactories, IBinaryComparatorFactory[] btreeCmpFactories,
             ILinearizeComparatorFactory linearizer, int[] comparatorFields, IBinaryComparatorFactory[] linearizerArray,
-            ILSMFlushController flushController, ILSMMergePolicy mergePolicy, ILSMOperationTrackerFactory opTrackerFactory,
-            ILSMIOOperationScheduler ioScheduler) {
+            ILSMFlushController flushController, ILSMMergePolicy mergePolicy,
+            ILSMOperationTrackerFactory opTrackerFactory, ILSMIOOperationScheduler ioScheduler) {
+        super(memFreePageManager, diskRTreeFactory.getBufferCache(), fileManager, diskFileMapProvider, flushController,
+                mergePolicy, opTrackerFactory, ioScheduler);
         RTree memRTree = new RTree(memBufferCache, ((InMemoryBufferCache) memBufferCache).getFileMapProvider(),
                 memFreePageManager, rtreeInteriorFrameFactory, rtreeLeafFrameFactory, rtreeCmpFactories, fieldCount,
-                new FileReference(new File("membtree")));
+                new FileReference(new File("memrtree")));
         BTree memBTree = new BTree(memBufferCache, ((InMemoryBufferCache) memBufferCache).getFileMapProvider(),
                 memFreePageManager, btreeInteriorFrameFactory, btreeLeafFrameFactory, btreeCmpFactories, fieldCount,
-                new FileReference(new File("memrtree")));
-        memComponent = new LSMRTreeComponent(memRTree, memBTree);
+                new FileReference(new File("membtree")));
+        mutableComponent = new LSMRTreeComponent(memRTree, memBTree);
         this.memBufferCache = memBufferCache;
-        this.memFreePageManager = memFreePageManager;
-        this.diskBufferCache = diskRTreeFactory.getBufferCache();
-        this.diskFileMapProvider = diskFileMapProvider;
-        this.fileManager = fileManager;
         this.rtreeInteriorFrameFactory = rtreeInteriorFrameFactory;
         this.rtreeLeafFrameFactory = rtreeLeafFrameFactory;
         this.btreeInteriorFrameFactory = btreeInteriorFrameFactory;
         this.btreeLeafFrameFactory = btreeLeafFrameFactory;
-        this.diskRTreeFactory = diskRTreeFactory;
+        this.componentFactory = componentFactory;
         this.btreeCmpFactories = btreeCmpFactories;
         this.rtreeCmpFactories = rtreeCmpFactories;
-        ILSMOperationTracker opTracker = opTrackerFactory.createOperationTracker(this);
-        this.lsmHarness = new LSMHarness(this, flushController, mergePolicy, opTracker, ioScheduler);
-        this.componentFinalizer = componentFinalizer;
         this.linearizer = linearizer;
         this.comparatorFields = comparatorFields;
         this.linearizerArray = linearizerArray;
-        diskComponents = new LinkedList<Object>();
         memRTreeTuples = 0;
         memBTreeTuples = 0;
         rTreeTupleSorter = null;
-        isActivated = false;
     }
 
     @Override
@@ -167,7 +126,7 @@
 
         fileManager.deleteDirs();
         fileManager.createDirs();
-        diskComponents.clear();
+        immutableComponents.clear();
     }
 
     @Override
@@ -176,12 +135,11 @@
             return;
         }
 
-        ((InMemoryBufferCache) memComponent.getRTree().getBufferCache()).open();
-        memComponent.getRTree().create();
-        memComponent.getBTree().create();
-        memComponent.getRTree().activate();
-        memComponent.getBTree().activate();
-        isActivated = true;
+        ((InMemoryBufferCache) mutableComponent.getRTree().getBufferCache()).open();
+        mutableComponent.getRTree().create();
+        mutableComponent.getBTree().create();
+        mutableComponent.getRTree().activate();
+        mutableComponent.getBTree().activate();
     }
 
     @Override
@@ -190,8 +148,6 @@
             return;
         }
 
-        isActivated = false;
-
         BlockingIOOperationCallback cb = new BlockingIOOperationCallback();
         ILSMIndexAccessor accessor = (ILSMIndexAccessor) createAccessor(NoOpOperationCallback.INSTANCE,
                 NoOpOperationCallback.INSTANCE);
@@ -202,11 +158,11 @@
             throw new HyracksDataException(e);
         }
 
-        memComponent.getRTree().deactivate();
-        memComponent.getBTree().deactivate();
-        memComponent.getRTree().destroy();
-        memComponent.getBTree().destroy();
-        ((InMemoryBufferCache) memComponent.getRTree().getBufferCache()).close();
+        mutableComponent.getRTree().deactivate();
+        mutableComponent.getBTree().deactivate();
+        mutableComponent.getRTree().destroy();
+        mutableComponent.getBTree().destroy();
+        ((InMemoryBufferCache) mutableComponent.getRTree().getBufferCache()).close();
     }
 
     @Override
@@ -215,9 +171,8 @@
             throw new HyracksDataException("Failed to destroy the index since it is activated.");
         }
 
-        memComponent.getRTree().deactivate();
-        memComponent.getBTree().deactivate();
-        fileManager.deleteDirs();
+        mutableComponent.getRTree().deactivate();
+        mutableComponent.getBTree().deactivate();
     }
 
     @Override
@@ -226,51 +181,68 @@
             throw new HyracksDataException("Failed to clear the index since it is not activated.");
         }
 
-        memComponent.getRTree().clear();
-        memComponent.getBTree().clear();
+        mutableComponent.getRTree().clear();
+        mutableComponent.getBTree().clear();
     }
 
-    @SuppressWarnings("rawtypes")
-    protected ITreeIndex createDiskTree(TreeIndexFactory diskTreeFactory, FileReference fileRef, boolean createTree)
+    protected LSMComponentFileReferences getMergeTargetFileName(List<ILSMComponent> mergingDiskComponents)
             throws HyracksDataException {
+        RTree firstTree = ((LSMRTreeComponent) mergingDiskComponents.get(0)).getRTree();
+        RTree lastTree = ((LSMRTreeComponent) mergingDiskComponents.get(mergingDiskComponents.size() - 1)).getRTree();
+        FileReference firstFile = diskFileMapProvider.lookupFileName(firstTree.getFileId());
+        FileReference lastFile = diskFileMapProvider.lookupFileName(lastTree.getFileId());
+        LSMComponentFileReferences fileRefs = fileManager.getRelMergeFileReference(firstFile.getFile().getName(),
+                lastFile.getFile().getName());
+        return fileRefs;
+    }
+
+    protected LSMRTreeComponent createDiskComponent(ILSMComponentFactory factory, FileReference insertFileRef,
+            FileReference deleteFileRef, boolean createComponent) throws HyracksDataException, IndexException {
         // Create new tree instance.
-        ITreeIndex diskTree = diskTreeFactory.createIndexInstance(fileRef);
-        if (createTree) {
-            diskTree.create();
+        LSMRTreeComponent component = (LSMRTreeComponent) factory
+                .createLSMComponentInstance(new LSMComponentFileReferences(insertFileRef, deleteFileRef));
+        if (createComponent) {
+            component.getRTree().create();
+            if (component.getBTree() != null) {
+                component.getBTree().create();
+            }
         }
         // Tree will be closed during cleanup of merge().
-        diskTree.activate();
-        return diskTree;
+        component.getRTree().activate();
+        if (component.getBTree() != null) {
+            component.getBTree().activate();
+        }
+        return component;
     }
 
     @Override
     public ITreeIndexFrameFactory getLeafFrameFactory() {
-        return memComponent.getRTree().getLeafFrameFactory();
+        return mutableComponent.getRTree().getLeafFrameFactory();
     }
 
     @Override
     public ITreeIndexFrameFactory getInteriorFrameFactory() {
-        return memComponent.getRTree().getInteriorFrameFactory();
+        return mutableComponent.getRTree().getInteriorFrameFactory();
     }
 
     @Override
     public IFreePageManager getFreePageManager() {
-        return memComponent.getRTree().getFreePageManager();
+        return mutableComponent.getRTree().getFreePageManager();
     }
 
     @Override
     public int getFieldCount() {
-        return memComponent.getRTree().getFieldCount();
+        return mutableComponent.getRTree().getFieldCount();
     }
 
     @Override
     public int getRootPageId() {
-        return memComponent.getRTree().getRootPageId();
+        return mutableComponent.getRTree().getRootPageId();
     }
 
     @Override
     public int getFileId() {
-        return memComponent.getRTree().getFileId();
+        return mutableComponent.getRTree().getFileId();
     }
 
     @Override
@@ -327,41 +299,20 @@
     }
 
     @Override
-    public void addMergedComponent(Object newComponent, List<Object> mergedComponents) {
-        diskComponents.removeAll(mergedComponents);
-        diskComponents.addLast(newComponent);
-    }
-
-    @Override
-    public void addFlushedComponent(Object index) {
-        diskComponents.addFirst(index);
-    }
-
-    @Override
-    public IInMemoryFreePageManager getInMemoryFreePageManager() {
-        return memFreePageManager;
-    }
-
-    @Override
-    public void resetInMemoryComponent() throws HyracksDataException {
+    public void resetMutableComponent() throws HyracksDataException {
         memFreePageManager.reset();
-        memComponent.getRTree().clear();
-        memComponent.getBTree().clear();
+        mutableComponent.getRTree().clear();
+        mutableComponent.getBTree().clear();
         memRTreeTuples = 0;
         memBTreeTuples = 0;
     }
 
-    @Override
-    public List<Object> getDiskComponents() {
-        return diskComponents;
-    }
-
     protected LSMRTreeOpContext createOpContext() {
-        return new LSMRTreeOpContext((RTree.RTreeAccessor) memComponent.getRTree().createAccessor(
+        return new LSMRTreeOpContext((RTree.RTreeAccessor) mutableComponent.getRTree().createAccessor(
                 NoOpOperationCallback.INSTANCE, NoOpOperationCallback.INSTANCE),
                 (IRTreeLeafFrame) rtreeLeafFrameFactory.createFrame(),
                 (IRTreeInteriorFrame) rtreeInteriorFrameFactory.createFrame(), memFreePageManager
-                        .getMetaDataFrameFactory().createFrame(), 8, (BTree.BTreeAccessor) memComponent.getBTree()
+                        .getMetaDataFrameFactory().createFrame(), 4, (BTree.BTreeAccessor) mutableComponent.getBTree()
                         .createAccessor(NoOpOperationCallback.INSTANCE, NoOpOperationCallback.INSTANCE),
                 btreeLeafFrameFactory, btreeInteriorFrameFactory, memFreePageManager.getMetaDataFrameFactory()
                         .createFrame(), rtreeCmpFactories, btreeCmpFactories, null, null);
@@ -372,44 +323,22 @@
         return rtreeCmpFactories;
     }
 
-    @Override
-    public IBufferCache getBufferCache() {
-        return diskBufferCache;
-    }
-
-    @Override
-    public ILSMComponentFinalizer getComponentFinalizer() {
-        return componentFinalizer;
-    }
-
-    @Override
-    public ILSMFlushController getFlushController() {
-        return lsmHarness.getFlushController();
-    }
-
-    @Override
-    public ILSMOperationTracker getOperationTracker() {
-        return lsmHarness.getOperationTracker();
-    }
-
-    @Override
-    public ILSMIOOperationScheduler getIOScheduler() {
-        return lsmHarness.getIOScheduler();
-    }
-
     public boolean isEmptyIndex() throws HyracksDataException {
-        return diskComponents.isEmpty()
-                && memComponent.btree.isEmptyTree(memComponent.btree.getInteriorFrameFactory().createFrame())
-                && memComponent.rtree.isEmptyTree(memComponent.rtree.getInteriorFrameFactory().createFrame());
+        return immutableComponents.isEmpty()
+                && mutableComponent.getBTree().isEmptyTree(
+                        mutableComponent.getBTree().getInteriorFrameFactory().createFrame())
+                && mutableComponent.getRTree().isEmptyTree(
+                        mutableComponent.getRTree().getInteriorFrameFactory().createFrame());
     }
 
+    @Override
     public void validate() throws HyracksDataException {
         throw new UnsupportedOperationException("Validation not implemented for LSM R-Trees.");
     }
 
     @Override
     public long getMemoryAllocationSize() {
-        InMemoryBufferCache memBufferCache = (InMemoryBufferCache) memComponent.rtree.getBufferCache();
+        InMemoryBufferCache memBufferCache = (InMemoryBufferCache) mutableComponent.getRTree().getBufferCache();
         return memBufferCache.getNumPages() * memBufferCache.getPageSize();
     }
 }
diff --git a/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTree.java b/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTree.java
index 62e0d0f..df6bcc8 100644
--- a/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTree.java
+++ b/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTree.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2009-2010 by The Regents of the University of California
+ * Copyright 2009-2012 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
@@ -15,7 +15,6 @@
 
 package edu.uci.ics.hyracks.storage.am.lsm.rtree.impls;
 
-import java.io.File;
 import java.util.List;
 import java.util.ListIterator;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -23,7 +22,6 @@
 import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparatorFactory;
 import edu.uci.ics.hyracks.api.dataflow.value.ILinearizeComparatorFactory;
 import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
-import edu.uci.ics.hyracks.api.io.FileReference;
 import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
 import edu.uci.ics.hyracks.storage.am.btree.impls.BTree;
 import edu.uci.ics.hyracks.storage.am.btree.impls.RangePredicate;
@@ -43,6 +41,7 @@
 import edu.uci.ics.hyracks.storage.am.common.impls.NoOpOperationCallback;
 import edu.uci.ics.hyracks.storage.am.common.ophelpers.MultiComparator;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.IInMemoryBufferCache;
+import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMComponent;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMFlushController;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMHarness;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIOOperation;
@@ -52,36 +51,30 @@
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIndexOperationContext;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMMergePolicy;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMOperationTrackerFactory;
+import edu.uci.ics.hyracks.storage.am.lsm.common.impls.LSMComponentFileReferences;
 import edu.uci.ics.hyracks.storage.am.lsm.common.impls.LSMTreeIndexAccessor;
 import edu.uci.ics.hyracks.storage.am.lsm.common.impls.TreeIndexFactory;
-import edu.uci.ics.hyracks.storage.am.lsm.rtree.impls.LSMRTreeFileManager.LSMRTreeFileNameComponent;
 import edu.uci.ics.hyracks.storage.am.rtree.impls.RTree;
-import edu.uci.ics.hyracks.storage.am.rtree.impls.RTree.RTreeBulkLoader;
 import edu.uci.ics.hyracks.storage.am.rtree.impls.RTreeSearchCursor;
 import edu.uci.ics.hyracks.storage.am.rtree.impls.SearchPredicate;
 import edu.uci.ics.hyracks.storage.common.file.IFileMapProvider;
 
 public class LSMRTree extends AbstractLSMRTree {
 
-    // On-disk components.
-    // For creating BTree's used in flush and merge.
-    private final TreeIndexFactory<BTree> diskBTreeFactory;
-
     public LSMRTree(IInMemoryBufferCache memBufferCache, IInMemoryFreePageManager memFreePageManager,
             ITreeIndexFrameFactory rtreeInteriorFrameFactory, ITreeIndexFrameFactory rtreeLeafFrameFactory,
             ITreeIndexFrameFactory btreeInteriorFrameFactory, ITreeIndexFrameFactory btreeLeafFrameFactory,
-            ILSMIndexFileManager fileManager, TreeIndexFactory<RTree> diskRTreeFactory,
+            ILSMIndexFileManager fileNameManager, TreeIndexFactory<RTree> diskRTreeFactory,
             TreeIndexFactory<BTree> diskBTreeFactory, IFileMapProvider diskFileMapProvider, int fieldCount,
             IBinaryComparatorFactory[] rtreeCmpFactories, IBinaryComparatorFactory[] btreeCmpFactories,
             ILinearizeComparatorFactory linearizer, int[] comparatorFields, IBinaryComparatorFactory[] linearizerArray,
             ILSMFlushController flushController, ILSMMergePolicy mergePolicy,
             ILSMOperationTrackerFactory opTrackerFactory, ILSMIOOperationScheduler ioScheduler) {
         super(memBufferCache, memFreePageManager, rtreeInteriorFrameFactory, rtreeLeafFrameFactory,
-                btreeInteriorFrameFactory, btreeLeafFrameFactory, fileManager, diskRTreeFactory, diskFileMapProvider,
-                new LSMRTreeComponentFinalizer(diskFileMapProvider), fieldCount, rtreeCmpFactories, btreeCmpFactories,
-                linearizer, comparatorFields, linearizerArray, flushController, mergePolicy, opTrackerFactory,
-                ioScheduler);
-        this.diskBTreeFactory = diskBTreeFactory;
+                btreeInteriorFrameFactory, btreeLeafFrameFactory, fileNameManager, diskRTreeFactory,
+                new LSMRTreeComponentFactory(diskRTreeFactory, diskBTreeFactory), diskFileMapProvider, fieldCount,
+                rtreeCmpFactories, btreeCmpFactories, linearizer, comparatorFields, linearizerArray, flushController,
+                mergePolicy, opTrackerFactory, ioScheduler);
     }
 
     /**
@@ -95,122 +88,125 @@
     @Override
     public synchronized void activate() throws HyracksDataException {
         super.activate();
-        List<Object> validFileNames = fileManager.cleanupAndGetValidFiles(componentFinalizer);
-        diskComponents.clear();
-        for (Object o : validFileNames) {
-            LSMRTreeFileNameComponent component = (LSMRTreeFileNameComponent) o;
-            FileReference rtreeFile = new FileReference(new File(component.getRTreeFileName()));
-            FileReference btreeFile = new FileReference(new File(component.getBTreeFileName()));
-            RTree rtree = (RTree) createDiskTree(diskRTreeFactory, rtreeFile, false);
-            BTree btree = (BTree) createDiskTree(diskBTreeFactory, btreeFile, false);
-            LSMRTreeComponent diskComponent = new LSMRTreeComponent(rtree, btree);
-            diskComponents.add(diskComponent);
+        List<LSMComponentFileReferences> validFileReferences;
+        try {
+            validFileReferences = fileManager.cleanupAndGetValidFiles();
+        } catch (IndexException e) {
+            throw new HyracksDataException(e);
         }
+        immutableComponents.clear();
+        for (LSMComponentFileReferences lsmComonentFileReference : validFileReferences) {
+            LSMRTreeComponent component;
+            try {
+                component = createDiskComponent(componentFactory,
+                        lsmComonentFileReference.getInsertIndexFileReference(),
+                        lsmComonentFileReference.getDeleteIndexFileReference(), false);
+            } catch (IndexException e) {
+                throw new HyracksDataException(e);
+            }
+            immutableComponents.add(component);
+        }
+        isActivated = true;
     }
 
     @Override
     public synchronized void deactivate() throws HyracksDataException {
         super.deactivate();
-        for (Object o : diskComponents) {
-            LSMRTreeComponent diskComponent = (LSMRTreeComponent) o;
-            RTree rtree = diskComponent.getRTree();
-            BTree btree = diskComponent.getBTree();
+        for (ILSMComponent c : immutableComponents) {
+            LSMRTreeComponent component = (LSMRTreeComponent) c;
+            RTree rtree = component.getRTree();
+            BTree btree = component.getBTree();
             rtree.deactivate();
             btree.deactivate();
         }
+        isActivated = false;
     }
 
     @Override
     public synchronized void destroy() throws HyracksDataException {
-        for (Object o : diskComponents) {
-            LSMRTreeComponent component = (LSMRTreeComponent) o;
+        super.destroy();
+        for (ILSMComponent c : immutableComponents) {
+            LSMRTreeComponent component = (LSMRTreeComponent) c;
             component.getBTree().destroy();
             component.getRTree().destroy();
         }
-        super.destroy();
+        fileManager.deleteDirs();
     }
 
     @Override
     public synchronized void clear() throws HyracksDataException {
-        for (Object o : diskComponents) {
-            LSMRTreeComponent component = (LSMRTreeComponent) o;
+        super.clear();
+        for (ILSMComponent c : immutableComponents) {
+            LSMRTreeComponent component = (LSMRTreeComponent) c;
             component.getBTree().deactivate();
             component.getRTree().deactivate();
             component.getBTree().destroy();
             component.getRTree().destroy();
         }
-        diskComponents.clear();
-        super.clear();
+        immutableComponents.clear();
     }
 
-    private LSMRTreeFileNameComponent getMergeTargetFileName(List<Object> mergingDiskTrees) throws HyracksDataException {
-        RTree firstTree = ((LSMRTreeComponent) mergingDiskTrees.get(0)).getRTree();
-        RTree lastTree = ((LSMRTreeComponent) mergingDiskTrees.get(mergingDiskTrees.size() - 1)).getRTree();
-        FileReference firstFile = diskFileMapProvider.lookupFileName(firstTree.getFileId());
-        FileReference lastFile = diskFileMapProvider.lookupFileName(lastTree.getFileId());
-        LSMRTreeFileNameComponent component = (LSMRTreeFileNameComponent) ((LSMRTreeFileManager) fileManager)
-                .getRelMergeFileName(firstFile.getFile().getName(), lastFile.getFile().getName());
-        return component;
-    }
-
-    public void search(IIndexCursor cursor, List<Object> diskComponents, ISearchPredicate pred,
-            IIndexOperationContext ictx, boolean includeMemComponent, AtomicInteger searcherRefCount)
+    @Override
+    public void search(IIndexCursor cursor, List<ILSMComponent> immutableComponents, ISearchPredicate pred,
+            IIndexOperationContext ictx, boolean includeMutableComponent, AtomicInteger searcherRefCount)
             throws HyracksDataException, IndexException {
         LSMRTreeOpContext ctx = (LSMRTreeOpContext) ictx;
-        int numDiskTrees = diskComponents.size();
-        int numTrees = (includeMemComponent) ? numDiskTrees + 1 : numDiskTrees;
+        int numDiskComponents = immutableComponents.size();
+        int numTrees = (includeMutableComponent) ? numDiskComponents + 1 : numDiskComponents;
 
         ITreeIndexAccessor[] rTreeAccessors = new ITreeIndexAccessor[numTrees];
         ITreeIndexAccessor[] bTreeAccessors = new ITreeIndexAccessor[numTrees];
-        int diskTreeIx = 0;
-        if (includeMemComponent) {
+        int diskComponentIx = 0;
+        if (includeMutableComponent) {
             rTreeAccessors[0] = ctx.memRTreeAccessor;
             bTreeAccessors[0] = ctx.memBTreeAccessor;
-            diskTreeIx++;
+            diskComponentIx++;
         }
 
-        ListIterator<Object> diskTreesIter = diskComponents.listIterator();
-        while (diskTreesIter.hasNext()) {
-            LSMRTreeComponent component = (LSMRTreeComponent) diskTreesIter.next();
+        ListIterator<ILSMComponent> diskComponentIter = immutableComponents.listIterator();
+        while (diskComponentIter.hasNext()) {
+            LSMRTreeComponent component = (LSMRTreeComponent) diskComponentIter.next();
             RTree diskRTree = component.getRTree();
             BTree diskBTree = component.getBTree();
-            rTreeAccessors[diskTreeIx] = diskRTree.createAccessor(NoOpOperationCallback.INSTANCE,
+            rTreeAccessors[diskComponentIx] = diskRTree.createAccessor(NoOpOperationCallback.INSTANCE,
                     NoOpOperationCallback.INSTANCE);
-            bTreeAccessors[diskTreeIx] = diskBTree.createAccessor(NoOpOperationCallback.INSTANCE,
+            bTreeAccessors[diskComponentIx] = diskBTree.createAccessor(NoOpOperationCallback.INSTANCE,
                     NoOpOperationCallback.INSTANCE);
-            diskTreeIx++;
+            diskComponentIx++;
         }
 
         LSMRTreeCursorInitialState initialState = new LSMRTreeCursorInitialState(numTrees, rtreeLeafFrameFactory,
                 rtreeInteriorFrameFactory, btreeLeafFrameFactory, ctx.getBTreeMultiComparator(), rTreeAccessors,
-                bTreeAccessors, searcherRefCount, includeMemComponent, lsmHarness, comparatorFields, linearizerArray,
-                ctx.searchCallback);
+                bTreeAccessors, searcherRefCount, includeMutableComponent, lsmHarness, comparatorFields,
+                linearizerArray, ctx.searchCallback);
         cursor.open(initialState, pred);
     }
 
     @Override
-    public Object flush(ILSMIOOperation operation) throws HyracksDataException, IndexException {
+    public ILSMComponent flush(ILSMIOOperation operation) throws HyracksDataException, IndexException {
         LSMRTreeFlushOperation flushOp = (LSMRTreeFlushOperation) operation;
         // Renaming order is critical because we use assume ordering when we
         // read the file names when we open the tree.
         // The RTree should be renamed before the BTree.
 
         // scan the memory RTree
-        ITreeIndexAccessor memRTreeAccessor = memComponent.getRTree().createAccessor(NoOpOperationCallback.INSTANCE,
-                NoOpOperationCallback.INSTANCE);
+        ITreeIndexAccessor memRTreeAccessor = mutableComponent.getRTree().createAccessor(
+                NoOpOperationCallback.INSTANCE, NoOpOperationCallback.INSTANCE);
         RTreeSearchCursor rtreeScanCursor = (RTreeSearchCursor) memRTreeAccessor.createSearchCursor();
         SearchPredicate rtreeNullPredicate = new SearchPredicate(null, null);
         memRTreeAccessor.search(rtreeScanCursor, rtreeNullPredicate);
-        RTree diskRTree = (RTree) createDiskTree(diskRTreeFactory, flushOp.getRTreeFlushTarget(), true);
+        LSMRTreeComponent component = createDiskComponent(componentFactory, flushOp.getRTreeFlushTarget(),
+                flushOp.getBTreeFlushTarget(), true);
+        RTree diskRTree = component.getRTree();
         IIndexBulkLoader rTreeBulkloader;
         ITreeIndexCursor cursor;
 
         IBinaryComparatorFactory[] linearizerArray = { linearizer };
 
         if (rTreeTupleSorter == null) {
-            rTreeTupleSorter = new TreeTupleSorter(memRTreeTuples, memComponent.getRTree().getFileId(),
+            rTreeTupleSorter = new TreeTupleSorter(memRTreeTuples, mutableComponent.getRTree().getFileId(),
                     linearizerArray, rtreeLeafFrameFactory.createFrame(), rtreeLeafFrameFactory.createFrame(),
-                    memComponent.getRTree().getBufferCache(), comparatorFields);
+                    mutableComponent.getRTree().getBufferCache(), comparatorFields);
         } else {
             rTreeTupleSorter.reset();
         }
@@ -247,12 +243,12 @@
         rTreeBulkloader.end();
 
         // scan the memory BTree
-        ITreeIndexAccessor memBTreeAccessor = memComponent.getBTree().createAccessor(NoOpOperationCallback.INSTANCE,
-                NoOpOperationCallback.INSTANCE);
+        ITreeIndexAccessor memBTreeAccessor = mutableComponent.getBTree().createAccessor(
+                NoOpOperationCallback.INSTANCE, NoOpOperationCallback.INSTANCE);
         IIndexCursor btreeScanCursor = memBTreeAccessor.createSearchCursor();
         RangePredicate btreeNullPredicate = new RangePredicate(null, null, true, true, null, null);
         memBTreeAccessor.search(btreeScanCursor, btreeNullPredicate);
-        BTree diskBTree = (BTree) createDiskTree(diskBTreeFactory, flushOp.getBTreeFlushTarget(), true);
+        BTree diskBTree = component.getBTree();
 
         // BulkLoad the tuples from the in-memory tree into the new disk BTree.
         IIndexBulkLoader bTreeBulkloader = diskBTree.createBulkLoader(1.0f, false);
@@ -270,8 +266,8 @@
     }
 
     @Override
-    public Object merge(List<Object> mergedComponents, ILSMIOOperation operation) throws HyracksDataException,
-            IndexException {
+    public ILSMComponent merge(List<ILSMComponent> mergedComponents, ILSMIOOperation operation)
+            throws HyracksDataException, IndexException {
         LSMRTreeMergeOperation mergeOp = (LSMRTreeMergeOperation) operation;
         ITreeIndexCursor cursor = mergeOp.getCursor();
         mergedComponents.addAll(mergeOp.getMergingComponents());
@@ -283,8 +279,10 @@
         }
 
         // Bulk load the tuples from all on-disk RTrees into the new RTree.
-        RTree mergedRTree = (RTree) createDiskTree(diskRTreeFactory, mergeOp.getRTreeMergeTarget(), true);
-        BTree mergedBTree = (BTree) createDiskTree(diskBTreeFactory, mergeOp.getBTreeMergeTarget(), true);
+        LSMRTreeComponent component = createDiskComponent(componentFactory, mergeOp.getRTreeMergeTarget(),
+                mergeOp.getBTreeMergeTarget(), true);
+        RTree mergedRTree = component.getRTree();
+        BTree mergedBTree = component.getBTree();
 
         IIndexBulkLoader bulkloader = mergedRTree.createBulkLoader(1.0f, false);
         try {
@@ -305,17 +303,15 @@
     }
 
     @Override
-    public void cleanUpAfterMerge(List<Object> mergedComponents) throws HyracksDataException {
-        for (Object o : mergedComponents) {
-            LSMRTreeComponent component = (LSMRTreeComponent) o;
+    public void cleanUpAfterMerge(List<ILSMComponent> mergedComponents) throws HyracksDataException {
+        for (ILSMComponent c : mergedComponents) {
+            LSMRTreeComponent component = (LSMRTreeComponent) c;
             BTree oldBTree = component.getBTree();
-            FileReference btreeFileRef = diskFileMapProvider.lookupFileName(oldBTree.getFileId());
             oldBTree.deactivate();
-            btreeFileRef.getFile().delete();
+            oldBTree.destroy();
             RTree oldRTree = component.getRTree();
-            FileReference rtreeFileRef = diskFileMapProvider.lookupFileName(oldRTree.getFileId());
             oldRTree.deactivate();
-            rtreeFileRef.getFile().delete();
+            oldRTree.destroy();
         }
     }
 
@@ -329,7 +325,7 @@
         cursor = new LSMRTreeSortedCursor(ctx, linearizer);
         ISearchPredicate rtreeSearchPred = new SearchPredicate(null, null);
         // Scan the RTrees, ignoring the in-memory RTree.
-        List<Object> mergingComponents;
+        List<ILSMComponent> mergingComponents;
         try {
             mergingComponents = lsmHarness.search(cursor, rtreeSearchPred, ctx, false);
         } catch (IndexException e) {
@@ -340,11 +336,9 @@
             cursor.close();
             return null;
         }
-        LSMRTreeFileNameComponent fileNames = getMergeTargetFileName(mergingComponents);
-        FileReference rtreeFile = fileManager.createMergeFile(fileNames.getRTreeFileName());
-        FileReference btreeFile = fileManager.createMergeFile(fileNames.getBTreeFileName());
-
-        return new LSMRTreeMergeOperation(lsmHarness.getIndex(), mergingComponents, cursor, rtreeFile, btreeFile,
+        LSMComponentFileReferences relMergeFileRefs = getMergeTargetFileName(mergingComponents);
+        return new LSMRTreeMergeOperation(lsmHarness.getIndex(), mergingComponents, cursor,
+                relMergeFileRefs.getInsertIndexFileReference(), relMergeFileRefs.getDeleteIndexFileReference(),
                 callback);
     }
 
@@ -371,53 +365,76 @@
 
         @Override
         public ILSMIOOperation createFlushOperation(ILSMIOOperationCallback callback) {
-            LSMRTreeFileNameComponent fileNames = (LSMRTreeFileNameComponent) fileManager.getRelFlushFileName();
-            FileReference rtreeFile = fileManager.createFlushFile(fileNames.getRTreeFileName());
-            FileReference btreeFile = fileManager.createFlushFile(fileNames.getBTreeFileName());
-            return new LSMRTreeFlushOperation(lsmHarness.getIndex(), rtreeFile, btreeFile, callback);
+            LSMComponentFileReferences componentFileRefs = fileManager.getRelFlushFileReference();
+            return new LSMRTreeFlushOperation(lsmHarness.getIndex(), componentFileRefs.getInsertIndexFileReference(),
+                    componentFileRefs.getDeleteIndexFileReference(), callback);
         }
     }
 
+    private ILSMComponent createBulkLoadTarget() throws HyracksDataException, IndexException {
+        LSMComponentFileReferences componentFileRefs = fileManager.getRelFlushFileReference();
+        return createDiskComponent(componentFactory, componentFileRefs.getInsertIndexFileReference(),
+                componentFileRefs.getDeleteIndexFileReference(), true);
+    }
+
     @Override
     public IIndexBulkLoader createBulkLoader(float fillLevel, boolean verifyInput) throws TreeIndexException {
         return new LSMRTreeBulkLoader(fillLevel, verifyInput);
     }
 
     public class LSMRTreeBulkLoader implements IIndexBulkLoader {
-        private final RTree diskRTree;
-        private final BTree diskBTree;
-        private final RTreeBulkLoader bulkLoader;
+        private final ILSMComponent component;
+        private final IIndexBulkLoader bulkLoader;
 
         public LSMRTreeBulkLoader(float fillFactor, boolean verifyInput) throws TreeIndexException {
             // Note that by using a flush target file name, we state that the
             // new bulk loaded tree is "newer" than any other merged tree.
             try {
-                LSMRTreeFileNameComponent fileNames = (LSMRTreeFileNameComponent) fileManager.getRelFlushFileName();
-                FileReference rtreeFile = fileManager.createFlushFile(fileNames.getRTreeFileName());
-                diskRTree = (RTree) createDiskTree(diskRTreeFactory, rtreeFile, true);
-                // For each RTree, we require to have a buddy BTree. thus, we
-                // create an
-                // empty BTree.
-                FileReference btreeFile = fileManager.createFlushFile(fileNames.getBTreeFileName());
-                diskBTree = (BTree) createDiskTree(diskBTreeFactory, btreeFile, true);
+                component = createBulkLoadTarget();
             } catch (HyracksDataException e) {
                 throw new TreeIndexException(e);
+            } catch (IndexException e) {
+                throw new TreeIndexException(e);
             }
-
-            bulkLoader = (RTreeBulkLoader) diskRTree.createBulkLoader(fillFactor, verifyInput);
+            bulkLoader = ((LSMRTreeComponent) component).getRTree().createBulkLoader(fillFactor, verifyInput);
         }
 
         @Override
-        public void add(ITupleReference tuple) throws HyracksDataException {
-            bulkLoader.add(tuple);
+        public void add(ITupleReference tuple) throws HyracksDataException, IndexException {
+            try {
+                bulkLoader.add(tuple);
+            } catch (IndexException e) {
+                handleException();
+                throw e;
+            } catch (HyracksDataException e) {
+                handleException();
+                throw e;
+            } catch (RuntimeException e) {
+                handleException();
+                throw e;
+            }
         }
 
         @Override
         public void end() throws HyracksDataException, IndexException {
             bulkLoader.end();
-            LSMRTreeComponent diskComponent = new LSMRTreeComponent(diskRTree, diskBTree);
-            lsmHarness.addBulkLoadedComponent(diskComponent);
+            lsmHarness.addBulkLoadedComponent(component);
         }
 
+        protected void handleException() throws HyracksDataException {
+            ((LSMRTreeComponent) component).getRTree().deactivate();
+            ((LSMRTreeComponent) component).getRTree().destroy();
+            ((LSMRTreeComponent) component).getBTree().deactivate();
+            ((LSMRTreeComponent) component).getBTree().destroy();
+        }
+    }
+
+    @Override
+    public void markAsValid(ILSMComponent lsmComponent) throws HyracksDataException {
+        LSMRTreeComponent component = (LSMRTreeComponent) lsmComponent;
+        forceFlushDirtyPages(component.getRTree());
+        markAsValidInternal(component.getRTree());
+        forceFlushDirtyPages(component.getBTree());
+        markAsValidInternal(component.getBTree());
     }
 }
diff --git a/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTreeComponent.java b/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTreeComponent.java
new file mode 100644
index 0000000..e2b4db5
--- /dev/null
+++ b/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTreeComponent.java
@@ -0,0 +1,76 @@
+/*
+ * Copyright 2009-2012 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.lsm.rtree.impls;
+
+import edu.uci.ics.hyracks.storage.am.btree.impls.BTree;
+import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMComponent;
+import edu.uci.ics.hyracks.storage.am.lsm.common.impls.LSMComponentState;
+import edu.uci.ics.hyracks.storage.am.rtree.impls.RTree;
+
+public class LSMRTreeComponent implements ILSMComponent {
+
+    private final RTree rtree;
+    private final BTree btree;
+
+    public LSMRTreeComponent(RTree rtree, BTree btree) {
+        this.rtree = rtree;
+        this.btree = btree;
+    }
+
+    @Override
+    public void activate() {
+        // TODO Auto-generated method stub
+
+    }
+
+    @Override
+    public void deactivate() {
+        // TODO Auto-generated method stub
+
+    }
+
+    @Override
+    public void threadEnter() {
+        // TODO Auto-generated method stub
+
+    }
+
+    @Override
+    public void threadExit() {
+        // TODO Auto-generated method stub
+
+    }
+
+    @Override
+    public void setState(LSMComponentState state) {
+        // TODO Auto-generated method stub
+
+    }
+
+    @Override
+    public LSMComponentState getState() {
+        // TODO Auto-generated method stub
+        return null;
+    }
+
+    public RTree getRTree() {
+        return rtree;
+    }
+
+    public BTree getBTree() {
+        return btree;
+    }
+}
diff --git a/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTreeComponentFactory.java b/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTreeComponentFactory.java
new file mode 100644
index 0000000..b875c67
--- /dev/null
+++ b/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTreeComponentFactory.java
@@ -0,0 +1,46 @@
+/*
+ * Copyright 2009-2012 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.lsm.rtree.impls;
+
+import edu.uci.ics.hyracks.storage.am.btree.impls.BTree;
+import edu.uci.ics.hyracks.storage.am.common.api.IndexException;
+import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMComponent;
+import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMComponentFactory;
+import edu.uci.ics.hyracks.storage.am.lsm.common.impls.LSMComponentFileReferences;
+import edu.uci.ics.hyracks.storage.am.lsm.common.impls.TreeIndexFactory;
+import edu.uci.ics.hyracks.storage.am.rtree.impls.RTree;
+import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
+
+public class LSMRTreeComponentFactory implements ILSMComponentFactory {
+    private final TreeIndexFactory<RTree> rtreeFactory;
+    private final TreeIndexFactory<BTree> btreeFactory;
+
+    public LSMRTreeComponentFactory(TreeIndexFactory<RTree> rtreeFactory, TreeIndexFactory<BTree> btreeFactory) {
+        this.rtreeFactory = rtreeFactory;
+        this.btreeFactory = btreeFactory;
+    }
+
+    @Override
+    public ILSMComponent createLSMComponentInstance(LSMComponentFileReferences cfr) throws IndexException {
+        return new LSMRTreeComponent(rtreeFactory.createIndexInstance(cfr.getInsertIndexFileReference()),
+                btreeFactory.createIndexInstance(cfr.getDeleteIndexFileReference()));
+    }
+
+    @Override
+    public IBufferCache getBufferCache() {
+        return rtreeFactory.getBufferCache();
+    }
+}
diff --git a/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTreeComponentFinalizer.java b/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTreeComponentFinalizer.java
deleted file mode 100644
index 1d83e00..0000000
--- a/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTreeComponentFinalizer.java
+++ /dev/null
@@ -1,35 +0,0 @@
-/*
- * 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.lsm.rtree.impls;
-
-import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
-import edu.uci.ics.hyracks.storage.am.lsm.common.impls.TreeIndexComponentFinalizer;
-import edu.uci.ics.hyracks.storage.am.lsm.rtree.impls.AbstractLSMRTree.LSMRTreeComponent;
-import edu.uci.ics.hyracks.storage.common.file.IFileMapProvider;
-
-public class LSMRTreeComponentFinalizer extends TreeIndexComponentFinalizer {
-
-	public LSMRTreeComponentFinalizer(IFileMapProvider fileMapProvider) {
-        super(fileMapProvider);
-    }
-
-    @Override
-	public void finalize(Object lsmComponent) throws HyracksDataException {
-		LSMRTreeComponent component = (LSMRTreeComponent) lsmComponent;
-		super.finalize(component.getRTree());
-		super.finalize(component.getBTree());
-	}
-}
diff --git a/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTreeFileManager.java b/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTreeFileManager.java
index 84ee50f..3f7d873 100644
--- a/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTreeFileManager.java
+++ b/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTreeFileManager.java
@@ -19,6 +19,7 @@
 import java.io.FilenameFilter;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.Date;
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
@@ -28,7 +29,8 @@
 import edu.uci.ics.hyracks.api.io.IIOManager;
 import edu.uci.ics.hyracks.api.io.IODeviceHandle;
 import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndex;
-import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMComponentFinalizer;
+import edu.uci.ics.hyracks.storage.am.common.api.IndexException;
+import edu.uci.ics.hyracks.storage.am.lsm.common.impls.LSMComponentFileReferences;
 import edu.uci.ics.hyracks.storage.am.lsm.common.impls.LSMIndexFileManager;
 import edu.uci.ics.hyracks.storage.am.lsm.common.impls.TreeIndexFactory;
 import edu.uci.ics.hyracks.storage.common.file.IFileMapProvider;
@@ -60,29 +62,36 @@
     }
 
     @Override
-    public Object getRelFlushFileName() {
-        String baseName = (String) super.getRelFlushFileName();
-        return new LSMRTreeFileNameComponent(baseName + SPLIT_STRING + RTREE_STRING, baseName + SPLIT_STRING
-                + BTREE_STRING);
-
+    public LSMComponentFileReferences getRelFlushFileReference() {
+        Date date = new Date();
+        String ts = formatter.format(date);
+        String baseName = baseDir + ts + SPLIT_STRING + ts;
+        // Begin timestamp and end timestamp are identical since it is a flush
+        return new LSMComponentFileReferences(createFlushFile(baseName + SPLIT_STRING + RTREE_STRING),
+                createFlushFile(baseName + SPLIT_STRING + BTREE_STRING));
     }
 
     @Override
-    public Object getRelMergeFileName(String firstFileName, String lastFileName) throws HyracksDataException {
-        String baseName = (String) super.getRelMergeFileName(firstFileName, lastFileName);
-        return new LSMRTreeFileNameComponent(baseName + SPLIT_STRING + RTREE_STRING, baseName + SPLIT_STRING
-                + BTREE_STRING);
+    public LSMComponentFileReferences getRelMergeFileReference(String firstFileName, String lastFileName)
+            throws HyracksDataException {
+        String[] firstTimestampRange = firstFileName.split(SPLIT_STRING);
+        String[] lastTimestampRange = lastFileName.split(SPLIT_STRING);
+
+        String baseName = baseDir + firstTimestampRange[0] + SPLIT_STRING + lastTimestampRange[1];
+        // Get the range of timestamps by taking the earliest and the latest timestamps
+        return new LSMComponentFileReferences(createMergeFile(baseName + SPLIT_STRING + RTREE_STRING),
+                createMergeFile(baseName + SPLIT_STRING + BTREE_STRING));
     }
 
     @Override
-    public List<Object> cleanupAndGetValidFiles(ILSMComponentFinalizer componentFinalizer) throws HyracksDataException {
-        List<Object> validFiles = new ArrayList<Object>();
+    public List<LSMComponentFileReferences> cleanupAndGetValidFiles() throws HyracksDataException, IndexException {
+        List<LSMComponentFileReferences> validFiles = new ArrayList<LSMComponentFileReferences>();
         ArrayList<ComparableFileName> allRTreeFiles = new ArrayList<ComparableFileName>();
         ArrayList<ComparableFileName> allBTreeFiles = new ArrayList<ComparableFileName>();
 
         // Gather files from all IODeviceHandles.
         for (IODeviceHandle dev : ioManager.getIODevices()) {
-            cleanupAndGetValidFilesInternal(dev, btreeFilter, btreeFactory, componentFinalizer, allBTreeFiles);
+            cleanupAndGetValidFilesInternal(dev, btreeFilter, btreeFactory, allBTreeFiles);
             HashSet<String> btreeFilesSet = new HashSet<String>();
             for (ComparableFileName cmpFileName : allBTreeFiles) {
                 int index = cmpFileName.fileName.lastIndexOf(SPLIT_STRING);
@@ -90,7 +99,7 @@
             }
             // List of valid RTree files that may or may not have a BTree buddy. Will check for buddies below.
             ArrayList<ComparableFileName> tmpAllRTreeFiles = new ArrayList<ComparableFileName>();
-            cleanupAndGetValidFilesInternal(dev, rtreeFilter, rtreeFactory, componentFinalizer, tmpAllRTreeFiles);
+            cleanupAndGetValidFilesInternal(dev, rtreeFilter, rtreeFactory, tmpAllRTreeFiles);
             // Look for buddy BTrees for all valid RTrees. 
             // If no buddy is found, delete the file, otherwise add the RTree to allRTreeFiles. 
             for (ComparableFileName cmpFileName : tmpAllRTreeFiles) {
@@ -117,7 +126,7 @@
         }
 
         if (allRTreeFiles.size() == 1 && allBTreeFiles.size() == 1) {
-            validFiles.add(new LSMRTreeFileNameComponent(allRTreeFiles.get(0).fullPath, allBTreeFiles.get(0).fullPath));
+            validFiles.add(new LSMComponentFileReferences(allRTreeFiles.get(0).fileRef, allBTreeFiles.get(0).fileRef));
             return validFiles;
         }
 
@@ -168,27 +177,9 @@
         while (rtreeFileIter.hasNext() && btreeFileIter.hasNext()) {
             ComparableFileName cmpRTreeFileName = rtreeFileIter.next();
             ComparableFileName cmpBTreeFileName = btreeFileIter.next();
-            validFiles.add(new LSMRTreeFileNameComponent(cmpRTreeFileName.fullPath, cmpBTreeFileName.fullPath));
+            validFiles.add(new LSMComponentFileReferences(cmpRTreeFileName.fileRef, cmpBTreeFileName.fileRef));
         }
 
         return validFiles;
     }
-
-    public class LSMRTreeFileNameComponent {
-        private final String rtreeFileName;
-        private final String btreeFileName;
-
-        LSMRTreeFileNameComponent(String rtreeFileName, String btreeFileName) {
-            this.rtreeFileName = rtreeFileName;
-            this.btreeFileName = btreeFileName;
-        }
-
-        public String getRTreeFileName() {
-            return rtreeFileName;
-        }
-
-        public String getBTreeFileName() {
-            return btreeFileName;
-        }
-    }
 }
diff --git a/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTreeMergeOperation.java b/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTreeMergeOperation.java
index e55f2c1..7449340 100644
--- a/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTreeMergeOperation.java
+++ b/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTreeMergeOperation.java
@@ -10,21 +10,21 @@
 import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexCursor;
 import edu.uci.ics.hyracks.storage.am.common.api.IndexException;
 import edu.uci.ics.hyracks.storage.am.common.impls.NoOpOperationCallback;
+import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMComponent;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIOOperation;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallback;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIndex;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIndexAccessor;
-import edu.uci.ics.hyracks.storage.am.lsm.rtree.impls.AbstractLSMRTree.LSMRTreeComponent;
 
 public class LSMRTreeMergeOperation implements ILSMIOOperation {
     private final ILSMIndex index;
-    private final List<Object> mergingComponents;
+    private final List<ILSMComponent> mergingComponents;
     private final ITreeIndexCursor cursor;
     private final FileReference rtreeMergeTarget;
     private final FileReference btreeMergeTarget;
     private final ILSMIOOperationCallback callback;
 
-    public LSMRTreeMergeOperation(ILSMIndex index, List<Object> mergingComponents, ITreeIndexCursor cursor,
+    public LSMRTreeMergeOperation(ILSMIndex index, List<ILSMComponent> mergingComponents, ITreeIndexCursor cursor,
             FileReference rtreeMergeTarget, FileReference btreeMergeTarget, ILSMIOOperationCallback callback) {
         this.index = index;
         this.mergingComponents = mergingComponents;
@@ -37,7 +37,7 @@
     @Override
     public List<IODeviceHandle> getReadDevices() {
         List<IODeviceHandle> devs = new ArrayList<IODeviceHandle>();
-        for (Object o : mergingComponents) {
+        for (ILSMComponent o : mergingComponents) {
             LSMRTreeComponent component = (LSMRTreeComponent) o;
             devs.add(component.getRTree().getFileReference().getDevideHandle());
         }
@@ -73,7 +73,7 @@
         return cursor;
     }
 
-    public List<Object> getMergingComponents() {
+    public List<ILSMComponent> getMergingComponents() {
         return mergingComponents;
     }
 
diff --git a/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTreeWithAntiMatterTuples.java b/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTreeWithAntiMatterTuples.java
index d06d5ca..52eb812 100644
--- a/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTreeWithAntiMatterTuples.java
+++ b/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTreeWithAntiMatterTuples.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2009-2010 by The Regents of the University of California
+ * Copyright 2009-2012 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
@@ -15,7 +15,6 @@
 
 package edu.uci.ics.hyracks.storage.am.lsm.rtree.impls;
 
-import java.io.File;
 import java.util.List;
 import java.util.ListIterator;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -23,7 +22,6 @@
 import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparatorFactory;
 import edu.uci.ics.hyracks.api.dataflow.value.ILinearizeComparatorFactory;
 import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
-import edu.uci.ics.hyracks.api.io.FileReference;
 import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
 import edu.uci.ics.hyracks.storage.am.btree.impls.BTreeRangeSearchCursor;
 import edu.uci.ics.hyracks.storage.am.btree.impls.RangePredicate;
@@ -35,7 +33,6 @@
 import edu.uci.ics.hyracks.storage.am.common.api.IModificationOperationCallback;
 import edu.uci.ics.hyracks.storage.am.common.api.ISearchOperationCallback;
 import edu.uci.ics.hyracks.storage.am.common.api.ISearchPredicate;
-import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndex;
 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;
@@ -44,6 +41,8 @@
 import edu.uci.ics.hyracks.storage.am.common.impls.NoOpOperationCallback;
 import edu.uci.ics.hyracks.storage.am.common.ophelpers.MultiComparator;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.IInMemoryBufferCache;
+import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMComponent;
+import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMComponentFactory;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMFlushController;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMHarness;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIOOperation;
@@ -53,13 +52,11 @@
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIndexOperationContext;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMMergePolicy;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMOperationTrackerFactory;
+import edu.uci.ics.hyracks.storage.am.lsm.common.impls.LSMComponentFileReferences;
 import edu.uci.ics.hyracks.storage.am.lsm.common.impls.LSMFlushOperation;
-import edu.uci.ics.hyracks.storage.am.lsm.common.impls.LSMMergeOperation;
 import edu.uci.ics.hyracks.storage.am.lsm.common.impls.LSMTreeIndexAccessor;
-import edu.uci.ics.hyracks.storage.am.lsm.common.impls.TreeIndexComponentFinalizer;
 import edu.uci.ics.hyracks.storage.am.lsm.common.impls.TreeIndexFactory;
 import edu.uci.ics.hyracks.storage.am.rtree.impls.RTree;
-import edu.uci.ics.hyracks.storage.am.rtree.impls.RTree.RTreeBulkLoader;
 import edu.uci.ics.hyracks.storage.am.rtree.impls.RTreeSearchCursor;
 import edu.uci.ics.hyracks.storage.am.rtree.impls.SearchPredicate;
 import edu.uci.ics.hyracks.storage.common.file.IFileMapProvider;
@@ -71,7 +68,7 @@
     // On-disk components.
     // For creating RTree's used in bulk load. Different from diskRTreeFactory
     // because it should have a different tuple writer in it's leaf frames.
-    private final TreeIndexFactory<RTree> bulkLoadRTreeFactory;
+    private final ILSMComponentFactory bulkLoaComponentFactory;
 
     public LSMRTreeWithAntiMatterTuples(IInMemoryBufferCache memBufferCache,
             IInMemoryFreePageManager memFreePageManager, ITreeIndexFrameFactory rtreeInteriorFrameFactory,
@@ -84,114 +81,129 @@
             ILSMMergePolicy mergePolicy, ILSMOperationTrackerFactory opTrackerFactory,
             ILSMIOOperationScheduler ioScheduler) {
         super(memBufferCache, memFreePageManager, rtreeInteriorFrameFactory, rtreeLeafFrameFactory,
-                btreeInteriorFrameFactory, btreeLeafFrameFactory, fileManager, diskRTreeFactory, diskFileMapProvider,
-                new TreeIndexComponentFinalizer(diskFileMapProvider), fieldCount, rtreeCmpFactories, btreeCmpFactories,
-                linearizer, comparatorFields, linearizerArray, flushController, mergePolicy, opTrackerFactory,
-                ioScheduler);
-        this.bulkLoadRTreeFactory = bulkLoadRTreeFactory;
+                btreeInteriorFrameFactory, btreeLeafFrameFactory, fileManager, diskRTreeFactory,
+                new LSMRTreeWithAntiMatterTuplesComponentFactory(diskRTreeFactory), diskFileMapProvider, fieldCount,
+                rtreeCmpFactories, btreeCmpFactories, linearizer, comparatorFields, linearizerArray, flushController,
+                mergePolicy, opTrackerFactory, ioScheduler);
+        bulkLoaComponentFactory = new LSMRTreeWithAntiMatterTuplesComponentFactory(bulkLoadRTreeFactory);
         this.bTreeTupleSorter = null;
     }
 
     @Override
     public synchronized void activate() throws HyracksDataException {
         super.activate();
-        diskComponents.clear();
-        List<Object> validFileNames = fileManager.cleanupAndGetValidFiles(componentFinalizer);
-        for (Object o : validFileNames) {
-            String fileName = (String) o;
-            FileReference fileRef = new FileReference(new File(fileName));
-            RTree rtree = (RTree) createDiskTree(diskRTreeFactory, fileRef, false);
-            diskComponents.add(rtree);
+        immutableComponents.clear();
+        List<LSMComponentFileReferences> validFileReferences;
+        try {
+            validFileReferences = fileManager.cleanupAndGetValidFiles();
+        } catch (IndexException e) {
+            throw new HyracksDataException(e);
         }
+        for (LSMComponentFileReferences lsmComonentFileReference : validFileReferences) {
+            LSMRTreeComponent component;
+            try {
+                component = createDiskComponent(componentFactory,
+                        lsmComonentFileReference.getInsertIndexFileReference(), null, false);
+            } catch (IndexException e) {
+                throw new HyracksDataException(e);
+            }
+            immutableComponents.add(component);
+        }
+        isActivated = true;
     }
 
     @Override
     public synchronized void deactivate() throws HyracksDataException {
         super.deactivate();
-        for (Object o : diskComponents) {
-            RTree rtree = (RTree) o;
+        for (ILSMComponent c : immutableComponents) {
+            RTree rtree = (RTree) ((LSMRTreeComponent) c).getRTree();
             rtree.deactivate();
         }
+        isActivated = false;
     }
 
     @Override
     public synchronized void destroy() throws HyracksDataException {
-        for (Object o : diskComponents) {
-            RTree rtree = (RTree) o;
+        super.destroy();
+        for (ILSMComponent c : immutableComponents) {
+            RTree rtree = (RTree) ((LSMRTreeComponent) c).getRTree();
             rtree.destroy();
         }
-        super.destroy();
+        fileManager.deleteDirs();
     }
 
     @Override
     public synchronized void clear() throws HyracksDataException {
-        for (Object o : diskComponents) {
-            RTree rtree = (RTree) o;
+        super.clear();
+        for (ILSMComponent c : immutableComponents) {
+            RTree rtree = (RTree) ((LSMRTreeComponent) c).getRTree();
             rtree.deactivate();
             rtree.destroy();
         }
-        diskComponents.clear();
-        super.clear();
+        immutableComponents.clear();
     }
 
-    public void search(IIndexCursor cursor, List<Object> diskComponents, ISearchPredicate pred,
-            IIndexOperationContext ictx, boolean includeMemComponent, AtomicInteger searcherRefCount)
+    @Override
+    public void search(IIndexCursor cursor, List<ILSMComponent> immutableComponents, ISearchPredicate pred,
+            IIndexOperationContext ictx, boolean includemutableComponent, AtomicInteger searcherRefCount)
             throws HyracksDataException, IndexException {
         LSMRTreeOpContext ctx = (LSMRTreeOpContext) ictx;
         LSMRTreeWithAntiMatterTuplesSearchCursor lsmTreeCursor = (LSMRTreeWithAntiMatterTuplesSearchCursor) cursor;
-        int numDiskRTrees = diskComponents.size();
+        int numDiskRComponents = immutableComponents.size();
 
         LSMRTreeCursorInitialState initialState;
         ITreeIndexAccessor[] bTreeAccessors = null;
-        if (includeMemComponent) {
+        if (includemutableComponent) {
             // Only in-memory BTree
             bTreeAccessors = new ITreeIndexAccessor[1];
             bTreeAccessors[0] = ctx.memBTreeAccessor;
         }
 
-        initialState = new LSMRTreeCursorInitialState(numDiskRTrees, rtreeLeafFrameFactory, rtreeInteriorFrameFactory,
-                btreeLeafFrameFactory, ctx.getBTreeMultiComparator(), null, bTreeAccessors, searcherRefCount,
-                includeMemComponent, lsmHarness, comparatorFields, linearizerArray, ctx.searchCallback);
+        initialState = new LSMRTreeCursorInitialState(numDiskRComponents, rtreeLeafFrameFactory,
+                rtreeInteriorFrameFactory, btreeLeafFrameFactory, ctx.getBTreeMultiComparator(), null, bTreeAccessors,
+                searcherRefCount, includemutableComponent, lsmHarness, comparatorFields, linearizerArray,
+                ctx.searchCallback);
 
         lsmTreeCursor.open(initialState, pred);
 
-        if (includeMemComponent) {
+        if (includemutableComponent) {
             // Open cursor of in-memory RTree
             ctx.memRTreeAccessor.search(lsmTreeCursor.getMemRTreeCursor(), pred);
         }
 
         // Open cursors of on-disk RTrees.
-        ITreeIndexAccessor[] diskRTreeAccessors = new ITreeIndexAccessor[numDiskRTrees];
-        int diskRTreeIx = 0;
-        ListIterator<Object> diskRTreesIter = diskComponents.listIterator();
-        while (diskRTreesIter.hasNext()) {
-            RTree diskRTree = (RTree) diskRTreesIter.next();
-            diskRTreeAccessors[diskRTreeIx] = diskRTree.createAccessor(NoOpOperationCallback.INSTANCE,
+        ITreeIndexAccessor[] diskRTreeAccessors = new ITreeIndexAccessor[numDiskRComponents];
+        int diskComponentIx = 0;
+        ListIterator<ILSMComponent> diskComponentsIter = immutableComponents.listIterator();
+        while (diskComponentsIter.hasNext()) {
+            RTree diskRTree = (RTree) ((LSMRTreeComponent) diskComponentsIter.next()).getRTree();
+            diskRTreeAccessors[diskComponentIx] = diskRTree.createAccessor(NoOpOperationCallback.INSTANCE,
                     NoOpOperationCallback.INSTANCE);
-            diskRTreeAccessors[diskRTreeIx].search(lsmTreeCursor.getCursor(diskRTreeIx), pred);
-            diskRTreeIx++;
+            diskRTreeAccessors[diskComponentIx].search(lsmTreeCursor.getCursor(diskComponentIx), pred);
+            diskComponentIx++;
         }
         lsmTreeCursor.initPriorityQueue();
     }
 
     @Override
-    public ITreeIndex flush(ILSMIOOperation operation) throws HyracksDataException, IndexException {
+    public ILSMComponent flush(ILSMIOOperation operation) throws HyracksDataException, IndexException {
         LSMFlushOperation flushOp = (LSMFlushOperation) operation;
         // Renaming order is critical because we use assume ordering when we
         // read the file names when we open the tree.
         // The RTree should be renamed before the BTree.
 
         // scan the memory RTree
-        ITreeIndexAccessor memRTreeAccessor = memComponent.getRTree().createAccessor(NoOpOperationCallback.INSTANCE,
-                NoOpOperationCallback.INSTANCE);
+        ITreeIndexAccessor memRTreeAccessor = mutableComponent.getRTree().createAccessor(
+                NoOpOperationCallback.INSTANCE, NoOpOperationCallback.INSTANCE);
         RTreeSearchCursor rtreeScanCursor = (RTreeSearchCursor) memRTreeAccessor.createSearchCursor();
         SearchPredicate rtreeNullPredicate = new SearchPredicate(null, null);
         memRTreeAccessor.search(rtreeScanCursor, rtreeNullPredicate);
-        RTree diskRTree = (RTree) createDiskTree(diskRTreeFactory, flushOp.getFlushTarget(), true);
+        LSMRTreeComponent component = createDiskComponent(componentFactory, flushOp.getFlushTarget(), null, true);
+        RTree diskRTree = component.getRTree();
 
         // scan the memory BTree
-        ITreeIndexAccessor memBTreeAccessor = memComponent.getBTree().createAccessor(NoOpOperationCallback.INSTANCE,
-                NoOpOperationCallback.INSTANCE);
+        ITreeIndexAccessor memBTreeAccessor = mutableComponent.getBTree().createAccessor(
+                NoOpOperationCallback.INSTANCE, NoOpOperationCallback.INSTANCE);
         BTreeRangeSearchCursor btreeScanCursor = (BTreeRangeSearchCursor) memBTreeAccessor.createSearchCursor();
         RangePredicate btreeNullPredicate = new RangePredicate(null, null, true, true, null, null);
         memBTreeAccessor.search(btreeScanCursor, btreeNullPredicate);
@@ -199,13 +211,13 @@
         // Since the LSM-RTree is used as a secondary assumption, the
         // primary key will be the last comparator in the BTree comparators
         if (rTreeTupleSorter == null) {
-            rTreeTupleSorter = new TreeTupleSorter(memRTreeTuples, memComponent.getRTree().getFileId(),
+            rTreeTupleSorter = new TreeTupleSorter(memRTreeTuples, mutableComponent.getRTree().getFileId(),
                     linearizerArray, rtreeLeafFrameFactory.createFrame(), rtreeLeafFrameFactory.createFrame(),
-                    memComponent.getRTree().getBufferCache(), comparatorFields);
+                    mutableComponent.getRTree().getBufferCache(), comparatorFields);
 
-            bTreeTupleSorter = new TreeTupleSorter(memBTreeTuples, memComponent.getBTree().getFileId(),
+            bTreeTupleSorter = new TreeTupleSorter(memBTreeTuples, mutableComponent.getBTree().getFileId(),
                     linearizerArray, btreeLeafFrameFactory.createFrame(), btreeLeafFrameFactory.createFrame(),
-                    memComponent.getBTree().getBufferCache(), comparatorFields);
+                    mutableComponent.getBTree().getBufferCache(), comparatorFields);
         } else {
             rTreeTupleSorter.reset();
             bTreeTupleSorter.reset();
@@ -263,13 +275,13 @@
 
         rTreeBulkloader.end();
 
-        return diskRTree;
+        return component;
     }
 
     @Override
-    public ITreeIndex merge(List<Object> mergedComponents, ILSMIOOperation operation) throws HyracksDataException,
-            IndexException {
-        LSMMergeOperation mergeOp = (LSMMergeOperation) operation;
+    public ILSMComponent merge(List<ILSMComponent> mergedComponents, ILSMIOOperation operation)
+            throws HyracksDataException, IndexException {
+        LSMRTreeMergeOperation mergeOp = (LSMRTreeMergeOperation) operation;
         ITreeIndexCursor cursor = mergeOp.getCursor();
         mergedComponents.addAll(mergeOp.getMergingComponents());
 
@@ -280,7 +292,8 @@
         }
 
         // Bulk load the tuples from all on-disk RTrees into the new RTree.
-        RTree mergedRTree = (RTree) createDiskTree(diskRTreeFactory, mergeOp.getMergeTarget(), true);
+        LSMRTreeComponent component = createDiskComponent(componentFactory, mergeOp.getRTreeMergeTarget(), null, true);
+        RTree mergedRTree = component.getRTree();
         IIndexBulkLoader bulkloader = mergedRTree.createBulkLoader(1.0f, false);
         try {
             while (cursor.hasNext()) {
@@ -292,16 +305,15 @@
             cursor.close();
         }
         bulkloader.end();
-        return mergedRTree;
+        return component;
     }
 
     @Override
-    public void cleanUpAfterMerge(List<Object> mergedComponents) throws HyracksDataException {
-        for (Object o : mergedComponents) {
-            RTree oldRTree = (RTree) o;
-            FileReference fileRef = diskFileMapProvider.lookupFileName(oldRTree.getFileId());
+    public void cleanUpAfterMerge(List<ILSMComponent> mergedComponents) throws HyracksDataException {
+        for (ILSMComponent c : mergedComponents) {
+            RTree oldRTree = (RTree) ((LSMRTreeComponent) c).getRTree();
             oldRTree.deactivate();
-            fileRef.getFile().delete();
+            oldRTree.destroy();
         }
     }
 
@@ -328,9 +340,9 @@
 
         @Override
         public ILSMIOOperation createFlushOperation(ILSMIOOperationCallback callback) {
-            String relFlushFileName = (String) fileManager.getRelFlushFileName();
-            FileReference fileRef = fileManager.createFlushFile(relFlushFileName);
-            return new LSMFlushOperation(lsmHarness.getIndex(), fileRef, callback);
+            LSMComponentFileReferences relFlushFileRefs = fileManager.getRelFlushFileReference();
+            return new LSMFlushOperation(lsmHarness.getIndex(), relFlushFileRefs.getInsertIndexFileReference(),
+                    callback);
         }
     }
 
@@ -339,36 +351,53 @@
         return new LSMRTreeWithAntiMatterTuplesBulkLoader(fillLevel, verifyInput);
     }
 
-    private RTree createBulkLoadTarget() throws HyracksDataException {
-        String relFlushFileName = (String) fileManager.getRelFlushFileName();
-        FileReference fileRef = fileManager.createFlushFile(relFlushFileName);
-        return (RTree) createDiskTree(bulkLoadRTreeFactory, fileRef, true);
+    private ILSMComponent createBulkLoadTarget() throws HyracksDataException, IndexException {
+        LSMComponentFileReferences relFlushFileRefs = fileManager.getRelFlushFileReference();
+        return createDiskComponent(bulkLoaComponentFactory, relFlushFileRefs.getInsertIndexFileReference(), null, true);
     }
 
     public class LSMRTreeWithAntiMatterTuplesBulkLoader implements IIndexBulkLoader {
-        private final RTree diskRTree;
-        private final RTreeBulkLoader bulkLoader;
+        private final ILSMComponent component;
+        private final IIndexBulkLoader bulkLoader;
 
         public LSMRTreeWithAntiMatterTuplesBulkLoader(float fillFactor, boolean verifyInput) throws TreeIndexException {
             // Note that by using a flush target file name, we state that the
             // new bulk loaded tree is "newer" than any other merged tree.
             try {
-                diskRTree = createBulkLoadTarget();
+                component = createBulkLoadTarget();
             } catch (HyracksDataException e) {
                 throw new TreeIndexException(e);
+            } catch (IndexException e) {
+                throw new TreeIndexException(e);
             }
-            bulkLoader = (RTreeBulkLoader) diskRTree.createBulkLoader(fillFactor, verifyInput);
+            bulkLoader = ((LSMRTreeComponent) component).getRTree().createBulkLoader(fillFactor, verifyInput);
         }
 
         @Override
-        public void add(ITupleReference tuple) throws HyracksDataException {
-            bulkLoader.add(tuple);
+        public void add(ITupleReference tuple) throws HyracksDataException, IndexException {
+            try {
+                bulkLoader.add(tuple);
+            } catch (IndexException e) {
+                handleException();
+                throw e;
+            } catch (HyracksDataException e) {
+                handleException();
+                throw e;
+            } catch (RuntimeException e) {
+                handleException();
+                throw e;
+            }
         }
 
         @Override
         public void end() throws HyracksDataException, IndexException {
             bulkLoader.end();
-            lsmHarness.addBulkLoadedComponent(diskRTree);
+            lsmHarness.addBulkLoadedComponent(component);
+        }
+
+        protected void handleException() throws HyracksDataException {
+            ((LSMRTreeComponent) component).getRTree().deactivate();
+            ((LSMRTreeComponent) component).getRTree().destroy();
         }
 
     }
@@ -381,10 +410,10 @@
         // Ordered scan, ignoring the in-memory RTree.
         // We get back a snapshot of the on-disk RTrees that are going to be
         // merged now, so we can clean them up after the merge has completed.
-        List<Object> mergingDiskRTrees;
+        List<ILSMComponent> mergingComponents;
         try {
-            mergingDiskRTrees = lsmHarness.search(cursor, (SearchPredicate) rtreeSearchPred, ctx, false);
-            if (mergingDiskRTrees.size() <= 1) {
+            mergingComponents = lsmHarness.search(cursor, (SearchPredicate) rtreeSearchPred, ctx, false);
+            if (mergingComponents.size() <= 1) {
                 cursor.close();
                 return null;
             }
@@ -392,13 +421,15 @@
             throw new HyracksDataException(e);
         }
 
-        RTree firstRTree = (RTree) mergingDiskRTrees.get(0);
-        RTree lastRTree = (RTree) mergingDiskRTrees.get(mergingDiskRTrees.size() - 1);
-        FileReference firstFile = diskFileMapProvider.lookupFileName(firstRTree.getFileId());
-        FileReference lastFile = diskFileMapProvider.lookupFileName(lastRTree.getFileId());
-        String relMergeFileName = (String) fileManager.getRelMergeFileName(firstFile.getFile().getName(), lastFile
-                .getFile().getName());
-        FileReference fileRef = fileManager.createMergeFile(relMergeFileName);
-        return new LSMMergeOperation(lsmHarness.getIndex(), mergingDiskRTrees, cursor, fileRef, callback);
+        LSMComponentFileReferences relMergeFileRefs = getMergeTargetFileName(mergingComponents);
+        return new LSMRTreeMergeOperation(lsmHarness.getIndex(), mergingComponents, cursor,
+                relMergeFileRefs.getInsertIndexFileReference(), null, callback);
+    }
+
+    @Override
+    public void markAsValid(ILSMComponent lsmComponent) throws HyracksDataException {
+        RTree rtree = ((LSMRTreeComponent) lsmComponent).getRTree();
+        forceFlushDirtyPages(rtree);
+        markAsValidInternal(rtree);
     }
 }
diff --git a/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTreeWithAntiMatterTuplesComponentFactory.java b/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTreeWithAntiMatterTuplesComponentFactory.java
new file mode 100644
index 0000000..d9573a4
--- /dev/null
+++ b/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTreeWithAntiMatterTuplesComponentFactory.java
@@ -0,0 +1,42 @@
+/*
+ * Copyright 2009-2012 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.lsm.rtree.impls;
+
+import edu.uci.ics.hyracks.storage.am.common.api.IndexException;
+import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMComponent;
+import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMComponentFactory;
+import edu.uci.ics.hyracks.storage.am.lsm.common.impls.LSMComponentFileReferences;
+import edu.uci.ics.hyracks.storage.am.lsm.common.impls.TreeIndexFactory;
+import edu.uci.ics.hyracks.storage.am.rtree.impls.RTree;
+import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
+
+public class LSMRTreeWithAntiMatterTuplesComponentFactory implements ILSMComponentFactory {
+    private final TreeIndexFactory<RTree> rtreeFactory;
+
+    public LSMRTreeWithAntiMatterTuplesComponentFactory(TreeIndexFactory<RTree> rtreeFactory) {
+        this.rtreeFactory = rtreeFactory;
+    }
+
+    @Override
+    public ILSMComponent createLSMComponentInstance(LSMComponentFileReferences cfr) throws IndexException {
+        return new LSMRTreeComponent(rtreeFactory.createIndexInstance(cfr.getInsertIndexFileReference()), null);
+    }
+
+    @Override
+    public IBufferCache getBufferCache() {
+        return rtreeFactory.getBufferCache();
+    }
+}
diff --git a/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/RTreeFactory.java b/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/RTreeFactory.java
index 4adbffb..71e228b 100644
--- a/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/RTreeFactory.java
+++ b/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/RTreeFactory.java
@@ -19,6 +19,7 @@
 import edu.uci.ics.hyracks.api.io.FileReference;
 import edu.uci.ics.hyracks.storage.am.common.api.IFreePageManagerFactory;
 import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexFrameFactory;
+import edu.uci.ics.hyracks.storage.am.common.api.IndexException;
 import edu.uci.ics.hyracks.storage.am.lsm.common.impls.TreeIndexFactory;
 import edu.uci.ics.hyracks.storage.am.rtree.impls.RTree;
 import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
@@ -34,7 +35,7 @@
     }
 
     @Override
-    public RTree createIndexInstance(FileReference file) {
+    public RTree createIndexInstance(FileReference file) throws IndexException {
         return new RTree(bufferCache, fileMapProvider, freePageManagerFactory.createFreePageManager(),
                 interiorFrameFactory, leafFrameFactory, cmpFactories, fieldCount, file);
     }
diff --git a/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/utils/LSMRTreeUtils.java b/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/utils/LSMRTreeUtils.java
index c6562d7..68ec4fe 100644
--- a/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/utils/LSMRTreeUtils.java
+++ b/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/utils/LSMRTreeUtils.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2009-2010 by The Regents of the University of California
+ * Copyright 2009-2012 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
@@ -64,9 +64,9 @@
             IBufferCache diskBufferCache, IFileMapProvider diskFileMapProvider, ITypeTraits[] typeTraits,
             IBinaryComparatorFactory[] rtreeCmpFactories, IBinaryComparatorFactory[] btreeCmpFactories,
             IPrimitiveValueProviderFactory[] valueProviderFactories, RTreePolicyType rtreePolicyType,
-            ILSMFlushController flushController, ILSMMergePolicy mergePolicy, ILSMOperationTrackerFactory opTrackerFactory,
-            ILSMIOOperationScheduler ioScheduler, ILinearizeComparatorFactory linearizeCmpFactory)
-            throws TreeIndexException {
+            ILSMFlushController flushController, ILSMMergePolicy mergePolicy,
+            ILSMOperationTrackerFactory opTrackerFactory, ILSMIOOperationScheduler ioScheduler,
+            ILinearizeComparatorFactory linearizeCmpFactory) throws TreeIndexException {
         LSMTypeAwareTupleWriterFactory rtreeTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(typeTraits, false);
         LSMTypeAwareTupleWriterFactory btreeTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(typeTraits, true);
 
diff --git a/hyracks-storage-am-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/rtree/impls/RTreeSearchCursor.java b/hyracks-storage-am-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/rtree/impls/RTreeSearchCursor.java
index 0b71a02..fd80071 100644
--- a/hyracks-storage-am-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/rtree/impls/RTreeSearchCursor.java
+++ b/hyracks-storage-am-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/rtree/impls/RTreeSearchCursor.java
@@ -234,12 +234,8 @@
     }
 
     @Override
-    public void reset() {
-        try {
-            close();
-        } catch (Exception e) {
-            e.printStackTrace();
-        }
+    public void reset() throws HyracksDataException {
+        close();
     }
 
     @Override
diff --git a/hyracks-tests/hyracks-storage-am-lsm-common-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsm/common/DummyLSMComponentFinalizer.java b/hyracks-tests/hyracks-storage-am-lsm-common-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsm/common/DummyLSMComponentFinalizer.java
deleted file mode 100644
index 5b3a0b5..0000000
--- a/hyracks-tests/hyracks-storage-am-lsm-common-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsm/common/DummyLSMComponentFinalizer.java
+++ /dev/null
@@ -1,33 +0,0 @@
-/*
- * 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.lsm.common;
-
-
-import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
-import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMComponentFinalizer;
-
-public class DummyLSMComponentFinalizer implements ILSMComponentFinalizer {
-
-	@Override
-	public boolean isValid(Object lsmComponent)
-			throws HyracksDataException {
-		return true;
-	}
-
-	@Override
-	public void finalize(Object lsmComponent) throws HyracksDataException {
-	}
-}
diff --git a/hyracks-tests/hyracks-storage-am-lsm-common-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsm/common/DummyLSMIndexFileManager.java b/hyracks-tests/hyracks-storage-am-lsm-common-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsm/common/DummyLSMIndexFileManager.java
new file mode 100644
index 0000000..a09c54d
--- /dev/null
+++ b/hyracks-tests/hyracks-storage-am-lsm-common-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsm/common/DummyLSMIndexFileManager.java
@@ -0,0 +1,50 @@
+/*
+ * Copyright 2009-2012 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.lsm.common;
+
+import java.io.File;
+import java.io.FilenameFilter;
+import java.util.ArrayList;
+
+import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
+import edu.uci.ics.hyracks.api.io.FileReference;
+import edu.uci.ics.hyracks.api.io.IIOManager;
+import edu.uci.ics.hyracks.api.io.IODeviceHandle;
+import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndex;
+import edu.uci.ics.hyracks.storage.am.common.api.IndexException;
+import edu.uci.ics.hyracks.storage.am.lsm.common.impls.AbstractLSMIndexFileManager;
+import edu.uci.ics.hyracks.storage.am.lsm.common.impls.TreeIndexFactory;
+import edu.uci.ics.hyracks.storage.common.file.IFileMapProvider;
+
+public class DummyLSMIndexFileManager extends AbstractLSMIndexFileManager {
+
+    public DummyLSMIndexFileManager(IIOManager ioManager, IFileMapProvider fileMapProvider, FileReference file,
+            TreeIndexFactory<? extends ITreeIndex> treeFactory) {
+        super(ioManager, fileMapProvider, file, treeFactory);
+    }
+
+    protected void cleanupAndGetValidFilesInternal(IODeviceHandle dev, FilenameFilter filter,
+            TreeIndexFactory<? extends ITreeIndex> treeFactory, ArrayList<ComparableFileName> allFiles)
+            throws HyracksDataException, IndexException {
+        File dir = new File(dev.getPath(), baseDir);
+        String[] files = dir.list(filter);
+        for (String fileName : files) {
+            File file = new File(dir.getPath() + File.separator + fileName);
+            FileReference fileRef = new FileReference(file);
+            allFiles.add(new ComparableFileName(fileRef));
+        }
+    }
+}
diff --git a/hyracks-tests/hyracks-storage-am-lsm-common-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsm/common/DummyTreeFactory.java b/hyracks-tests/hyracks-storage-am-lsm-common-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsm/common/DummyTreeFactory.java
index d6494f8..8b22771 100644
--- a/hyracks-tests/hyracks-storage-am-lsm-common-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsm/common/DummyTreeFactory.java
+++ b/hyracks-tests/hyracks-storage-am-lsm-common-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsm/common/DummyTreeFactory.java
@@ -1,7 +1,23 @@
+/*
+ * Copyright 2009-2012 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.lsm.common;
 
 import edu.uci.ics.hyracks.api.io.FileReference;
 import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndex;
+import edu.uci.ics.hyracks.storage.am.common.api.IndexException;
 import edu.uci.ics.hyracks.storage.am.lsm.common.impls.TreeIndexFactory;
 
 public class DummyTreeFactory extends TreeIndexFactory<ITreeIndex> {
@@ -11,7 +27,7 @@
     }
 
     @Override
-    public ITreeIndex createIndexInstance(FileReference file) {
+    public ITreeIndex createIndexInstance(FileReference file) throws IndexException {
         return null;
     }
 
diff --git a/hyracks-tests/hyracks-storage-am-lsm-common-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsm/common/LSMIndexFileManagerTest.java b/hyracks-tests/hyracks-storage-am-lsm-common-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsm/common/LSMIndexFileManagerTest.java
index 85677e4..161f4ce 100644
--- a/hyracks-tests/hyracks-storage-am-lsm-common-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsm/common/LSMIndexFileManagerTest.java
+++ b/hyracks-tests/hyracks-storage-am-lsm-common-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsm/common/LSMIndexFileManagerTest.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2009-2010 by The Regents of the University of California
+ * Copyright 2009-2012 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
@@ -38,9 +38,9 @@
 import edu.uci.ics.hyracks.api.io.FileReference;
 import edu.uci.ics.hyracks.api.io.IODeviceHandle;
 import edu.uci.ics.hyracks.control.nc.io.IOManager;
-import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMComponentFinalizer;
+import edu.uci.ics.hyracks.storage.am.common.api.IndexException;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIndexFileManager;
-import edu.uci.ics.hyracks.storage.am.lsm.common.impls.LSMIndexFileManager;
+import edu.uci.ics.hyracks.storage.am.lsm.common.impls.LSMComponentFileReferences;
 import edu.uci.ics.hyracks.storage.common.file.IFileMapProvider;
 import edu.uci.ics.hyracks.test.support.TestStorageManagerComponentHolder;
 
@@ -54,8 +54,6 @@
     protected IFileMapProvider fileMapProvider;
     protected String baseDir;
     protected FileReference file;
-    // Dummy finalizer that always considers components valid. Just for testing.
-    protected ILSMComponentFinalizer componentFinalizer = new DummyLSMComponentFinalizer();
 
     @Before
     public void setUp() throws HyracksException {
@@ -75,19 +73,22 @@
     }
 
     public void sortOrderTest(boolean testFlushFileName) throws InterruptedException, HyracksDataException {
-        ILSMIndexFileManager fileManager = new LSMIndexFileManager(ioManager, fileMapProvider, file, new DummyTreeFactory());
+        ILSMIndexFileManager fileManager = new DummyLSMIndexFileManager(ioManager, fileMapProvider, file,
+                new DummyTreeFactory());
         LinkedList<String> fileNames = new LinkedList<String>();
 
         int numFileNames = 100;
         long sleepTime = 5;
         for (int i = 0; i < numFileNames; i++) {
-            String flushFileName = (String) fileManager.getRelFlushFileName();
+            String flushFileName = (String) fileManager.getRelFlushFileReference().getInsertIndexFileReference()
+                    .getFile().getName();
             if (testFlushFileName) {
                 fileNames.addFirst(flushFileName);
             }
             Thread.sleep(sleepTime);
             if (!testFlushFileName) {
-                String secondFlushFileName = (String) fileManager.getRelFlushFileName();
+                String secondFlushFileName = (String) fileManager.getRelFlushFileReference()
+                        .getInsertIndexFileReference().getFile().getName();
                 String mergeFileName = getMergeFileName(fileManager, flushFileName, secondFlushFileName);
                 fileNames.addFirst(mergeFileName);
                 Thread.sleep(sleepTime);
@@ -112,8 +113,9 @@
         sortOrderTest(false);
     }
 
-    public void cleanInvalidFilesTest(IOManager ioManager) throws InterruptedException, IOException {
-        ILSMIndexFileManager fileManager = new LSMIndexFileManager(ioManager, fileMapProvider, file, new DummyTreeFactory());
+    public void cleanInvalidFilesTest(IOManager ioManager) throws InterruptedException, IOException, IndexException {
+        ILSMIndexFileManager fileManager = new DummyLSMIndexFileManager(ioManager, fileMapProvider, file,
+                new DummyTreeFactory());
         fileManager.createDirs();
 
         List<FileReference> flushFiles = new ArrayList<FileReference>();
@@ -123,9 +125,8 @@
         long sleepTime = 5;
         // Generate a bunch of flush files.
         for (int i = 0; i < numFileNames; i++) {
-            String relFlushFileName = (String) fileManager.getRelFlushFileName();
-            FileReference flushFile = fileManager.createFlushFile(relFlushFileName);
-            flushFiles.add(flushFile);
+            LSMComponentFileReferences relFlushFileRefs = fileManager.getRelFlushFileReference();
+            flushFiles.add(relFlushFileRefs.getInsertIndexFileReference());
             Thread.sleep(sleepTime);
         }
         allFiles.addAll(flushFiles);
@@ -177,14 +178,13 @@
         Collections.sort(expectedValidFiles, fileManager.getFileNameComparator());
 
         // Pass null and a dummy component finalizer. We don't test for physical consistency in this test.
-        List<Object> validFiles = fileManager.cleanupAndGetValidFiles(componentFinalizer);
+        List<LSMComponentFileReferences> lsmComonentFileReference = fileManager.cleanupAndGetValidFiles();
 
         // Check actual files against expected files.
-        assertEquals(expectedValidFiles.size(), validFiles.size());
+        assertEquals(expectedValidFiles.size(), lsmComonentFileReference.size());
         for (int i = 0; i < expectedValidFiles.size(); i++) {
-            String fileName = (String) validFiles.get(i);
-            File f = new File(fileName);
-            assertEquals(expectedValidFiles.get(i), f.getName());
+            assertEquals(expectedValidFiles.get(i), lsmComonentFileReference.get(i).getInsertIndexFileReference()
+                    .getFile().getName());
         }
 
         // Make sure invalid files were removed from all IODevices.
@@ -212,21 +212,21 @@
     }
 
     @Test
-    public void singleIODeviceTest() throws InterruptedException, IOException {
+    public void singleIODeviceTest() throws InterruptedException, IOException, IndexException {
         IOManager singleDeviceIOManager = createIOManager(1);
         cleanInvalidFilesTest(singleDeviceIOManager);
         cleanDirs(singleDeviceIOManager);
     }
 
     @Test
-    public void twoIODevicesTest() throws InterruptedException, IOException {
+    public void twoIODevicesTest() throws InterruptedException, IOException, IndexException {
         IOManager twoDevicesIOManager = createIOManager(2);
         cleanInvalidFilesTest(twoDevicesIOManager);
         cleanDirs(twoDevicesIOManager);
     }
 
     @Test
-    public void fourIODevicesTest() throws InterruptedException, IOException {
+    public void fourIODevicesTest() throws InterruptedException, IOException, IndexException {
         IOManager fourDevicesIOManager = createIOManager(4);
         cleanInvalidFilesTest(fourDevicesIOManager);
         cleanDirs(fourDevicesIOManager);
@@ -259,16 +259,16 @@
 
     private FileReference simulateMerge(ILSMIndexFileManager fileManager, FileReference a, FileReference b)
             throws HyracksDataException {
-        String relMergeFileName = (String) fileManager
-                .getRelMergeFileName(a.getFile().getName(), b.getFile().getName());
-        FileReference mergeFile = fileManager.createMergeFile(relMergeFileName);
-        return mergeFile;
+        LSMComponentFileReferences relMergeFileRefs = fileManager.getRelMergeFileReference(a.getFile().getName(), b
+                .getFile().getName());
+        return relMergeFileRefs.getInsertIndexFileReference();
     }
 
     private String getMergeFileName(ILSMIndexFileManager fileNameManager, String firstFile, String lastFile)
             throws HyracksDataException {
         File f1 = new File(firstFile);
         File f2 = new File(lastFile);
-        return (String) fileNameManager.getRelMergeFileName(f1.getName(), f2.getName());
+        return (String) fileNameManager.getRelMergeFileReference(f1.getName(), f2.getName())
+                .getInsertIndexFileReference().getFile().getName();
     }
 }