diff --git a/hyracks/hyracks-storage-am-common/src/main/java/edu/uci/ics/hyracks/storage/am/common/dataflow/IndexDataflowHelper.java b/hyracks/hyracks-storage-am-common/src/main/java/edu/uci/ics/hyracks/storage/am/common/dataflow/IndexDataflowHelper.java
index 5e07e0c..d051678 100644
--- a/hyracks/hyracks-storage-am-common/src/main/java/edu/uci/ics/hyracks/storage/am/common/dataflow/IndexDataflowHelper.java
+++ b/hyracks/hyracks-storage-am-common/src/main/java/edu/uci/ics/hyracks/storage/am/common/dataflow/IndexDataflowHelper.java
@@ -62,74 +62,66 @@
     }
 
     public void create() throws HyracksDataException {
-        synchronized (lcManager) {
-            long resourceID = getResourceID();
-            index = lcManager.getIndex(resourceID);
-            if (index != null) {
-                lcManager.unregister(resourceID);
-            } else {
-                index = createIndexInstance();
-            }
-
-            // The previous resource ID needs to be removed since calling IIndex.create() may possibly destroy 
-            // any physical artifact that the LocalResourceRepository is managing (e.g. a file containing the resource ID). 
-            // Once the index has been created, a new resource ID can be generated.
-            if (resourceID != -1) {
-                localResourceRepository.deleteResourceByName(file.getFile().getPath());
-            }
-            index.create();
-            try {
-                //TODO Create LocalResource through LocalResourceFactory interface
-                resourceID = resourceIdFactory.createId();
-                ILocalResourceFactory localResourceFactory = opDesc.getLocalResourceFactoryProvider()
-                        .getLocalResourceFactory();
-                localResourceRepository.insert(localResourceFactory.createLocalResource(resourceID, file.getFile()
-                        .getPath(), partition));
-            } catch (IOException e) {
-                throw new HyracksDataException(e);
-            }
-            lcManager.register(resourceID, index);
+        long resourceID = getResourceID();
+        index = lcManager.getIndex(resourceID);
+        if (index != null) {
+            lcManager.unregister(resourceID);
+        } else {
+            index = createIndexInstance();
         }
+
+        // The previous resource ID needs to be removed since calling IIndex.create() may possibly destroy 
+        // any physical artifact that the LocalResourceRepository is managing (e.g. a file containing the resource ID). 
+        // Once the index has been created, a new resource ID can be generated.
+        if (resourceID != -1) {
+            localResourceRepository.deleteResourceByName(file.getFile().getPath());
+        }
+        index.create();
+        try {
+            //TODO Create LocalResource through LocalResourceFactory interface
+            resourceID = resourceIdFactory.createId();
+            ILocalResourceFactory localResourceFactory = opDesc.getLocalResourceFactoryProvider()
+                    .getLocalResourceFactory();
+            localResourceRepository.insert(localResourceFactory.createLocalResource(resourceID, file.getFile()
+                    .getPath(), partition));
+        } catch (IOException e) {
+            throw new HyracksDataException(e);
+        }
+        lcManager.register(resourceID, index);
     }
 
     public void open() throws HyracksDataException {
-        synchronized (lcManager) {
-            long resourceID = getResourceID();
+        long resourceID = getResourceID();
 
-            if (resourceID == -1) {
-                throw new HyracksDataException("Index does not have a valid resource ID. Has it been created yet?");
-            }
-
-            index = lcManager.getIndex(resourceID);
-            if (index == null) {
-                index = createIndexInstance();
-                lcManager.register(resourceID, index);
-            }
-            lcManager.open(resourceID);
+        if (resourceID == -1) {
+            throw new HyracksDataException("Index does not have a valid resource ID. Has it been created yet?");
         }
+
+        index = lcManager.getIndex(resourceID);
+        if (index == null) {
+            index = createIndexInstance();
+            lcManager.register(resourceID, index);
+        }
+        lcManager.open(resourceID);
     }
 
     public void close() throws HyracksDataException {
-        synchronized (lcManager) {
-            lcManager.close(getResourceID());
-        }
+        lcManager.close(getResourceID());
     }
 
     public void destroy() throws HyracksDataException {
-        synchronized (lcManager) {
-            long resourceID = getResourceID();
-            index = lcManager.getIndex(resourceID);
-            if (index != null) {
-                lcManager.unregister(resourceID);
-            } else {
-                index = createIndexInstance();
-            }
-
-            if (resourceID != -1) {
-                localResourceRepository.deleteResourceByName(file.getFile().getPath());
-            }
-            index.destroy();
+        long resourceID = getResourceID();
+        index = lcManager.getIndex(resourceID);
+        if (index != null) {
+            lcManager.unregister(resourceID);
+        } else {
+            index = createIndexInstance();
         }
+
+        if (resourceID != -1) {
+            localResourceRepository.deleteResourceByName(file.getFile().getPath());
+        }
+        index.destroy();
     }
 
     public FileReference getFileReference() {
diff --git a/hyracks/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/btree/impls/LSMBTree.java b/hyracks/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/btree/impls/LSMBTree.java
index 660d5ea..f03de4b 100644
--- a/hyracks/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/btree/impls/LSMBTree.java
+++ b/hyracks/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.ArrayList;
 import java.util.List;
 
 import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparatorFactory;
@@ -63,7 +62,6 @@
 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.IMutableComponentSwitcherCallback;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.IVirtualBufferCache;
 import edu.uci.ics.hyracks.storage.am.lsm.common.freepage.VirtualFreePageManager;
 import edu.uci.ics.hyracks.storage.am.lsm.common.impls.AbstractLSMIndex;
@@ -77,9 +75,6 @@
 
 public class LSMBTree extends AbstractLSMIndex implements ITreeIndex {
 
-    // In-memory components.
-    private final List<LSMBTreeMutableComponent> mutableComponents;
-
     // For creating BTree's used in flush and merge.
     private final LSMBTreeImmutableComponentFactory componentFactory;
     // For creating BTree's used in bulk load. Different from diskBTreeFactory
@@ -100,8 +95,6 @@
             ILSMIOOperationScheduler ioScheduler, ILSMIOOperationCallbackProvider ioOpCallbackProvider) {
         super(virtualBufferCaches, diskBTreeFactory.getBufferCache(), fileManager, diskFileMapProvider,
                 bloomFilterFalsePositiveRate, mergePolicy, opTracker, ioScheduler, ioOpCallbackProvider);
-        mutableComponents = new ArrayList<LSMBTreeMutableComponent>();
-        final int numMutableComponents = virtualBufferCaches.size();
         int i = 0;
         for (IVirtualBufferCache virtualBufferCache : virtualBufferCaches) {
             LSMBTreeMutableComponent mutableComponent = new LSMBTreeMutableComponent(new BTree(virtualBufferCache,
@@ -109,21 +102,6 @@
                             virtualBufferCache.getNumPages()), interiorFrameFactory, insertLeafFrameFactory,
                     cmpFactories, fieldCount, new FileReference(new File(fileManager.getBaseDir() + "_virtual_" + i))),
                     virtualBufferCache, i == 0 ? true : false);
-
-            mutableComponent.registerOnFlushCallback(new IMutableComponentSwitcherCallback() {
-
-                @Override
-                public void switchComponents() throws HyracksDataException {
-                    currentMutableComponentId.set((currentMutableComponentId.get() + 1) % numMutableComponents);
-                    mutableComponents.get(currentMutableComponentId.get()).setActive();
-                }
-
-                @Override
-                public void requestFlush(boolean isFlushNeeded) {
-                    flushRequests[currentMutableComponentId.get()].set(isFlushNeeded);
-                }
-            });
-
             mutableComponents.add(mutableComponent);
             ++i;
         }
@@ -143,7 +121,7 @@
 
         fileManager.deleteDirs();
         fileManager.createDirs();
-        componentsRef.get().clear();
+        diskComponents.clear();
     }
 
     @Override
@@ -151,12 +129,13 @@
         if (isActivated) {
             throw new HyracksDataException("Failed to activate the index since it is already activated.");
         }
-        for (LSMBTreeMutableComponent mutableComponent : mutableComponents) {
+        for (ILSMComponent c : mutableComponents) {
+            LSMBTreeMutableComponent mutableComponent = (LSMBTreeMutableComponent) c;
             ((IVirtualBufferCache) mutableComponent.getBTree().getBufferCache()).open();
             mutableComponent.getBTree().create();
             mutableComponent.getBTree().activate();
         }
-        List<ILSMComponent> immutableComponents = componentsRef.get();
+        List<ILSMComponent> immutableComponents = diskComponents;
         immutableComponents.clear();
         List<LSMComponentFileReferences> validFileReferences;
         try {
@@ -196,7 +175,7 @@
             }
         }
 
-        List<ILSMComponent> immutableComponents = componentsRef.get();
+        List<ILSMComponent> immutableComponents = diskComponents;
         for (ILSMComponent c : immutableComponents) {
             LSMBTreeImmutableComponent component = (LSMBTreeImmutableComponent) c;
             BTree btree = component.getBTree();
@@ -204,7 +183,8 @@
             btree.deactivate();
             bloomFilter.deactivate();
         }
-        for (LSMBTreeMutableComponent mutableComponent : mutableComponents) {
+        for (ILSMComponent c : mutableComponents) {
+            LSMBTreeMutableComponent mutableComponent = (LSMBTreeMutableComponent) c;
             mutableComponent.getBTree().deactivate();
             mutableComponent.getBTree().destroy();
             ((IVirtualBufferCache) mutableComponent.getBTree().getBufferCache()).close();
@@ -223,13 +203,14 @@
             throw new HyracksDataException("Failed to destroy the index since it is activated.");
         }
 
-        List<ILSMComponent> immutableComponents = componentsRef.get();
+        List<ILSMComponent> immutableComponents = diskComponents;
         for (ILSMComponent c : immutableComponents) {
             LSMBTreeImmutableComponent component = (LSMBTreeImmutableComponent) c;
             component.getBTree().destroy();
             component.getBloomFilter().destroy();
         }
-        for (LSMBTreeMutableComponent mutableComponent : mutableComponents) {
+        for (ILSMComponent c : mutableComponents) {
+            LSMBTreeMutableComponent mutableComponent = (LSMBTreeMutableComponent) c;
             mutableComponent.getBTree().destroy();
         }
         fileManager.deleteDirs();
@@ -241,8 +222,9 @@
             throw new HyracksDataException("Failed to clear the index since it is not activated.");
         }
 
-        List<ILSMComponent> immutableComponents = componentsRef.get();
-        for (LSMBTreeMutableComponent mutableComponent : mutableComponents) {
+        List<ILSMComponent> immutableComponents = diskComponents;
+        for (ILSMComponent c : mutableComponents) {
+            LSMBTreeMutableComponent mutableComponent = (LSMBTreeMutableComponent) c;
             mutableComponent.getBTree().clear();
             mutableComponent.reset();
         }
@@ -258,7 +240,7 @@
 
     @Override
     public void getOperationalComponents(ILSMIndexOperationContext ctx) {
-        List<ILSMComponent> immutableComponents = componentsRef.get();
+        List<ILSMComponent> immutableComponents = diskComponents;
         List<ILSMComponent> operationalComponents = ctx.getComponentHolder();
         operationalComponents.clear();
         int cmc = currentMutableComponentId.get();
@@ -275,8 +257,8 @@
             case SEARCH:
             case INSERT:
                 for (int i = 0; i < numMutableComponents - 1; i++) {
-                    LSMBTreeMutableComponent mutableComponent = mutableComponents.get((cmc + i + 1)
-                            % numMutableComponents);
+                    ILSMComponent c = mutableComponents.get((cmc + i + 1) % numMutableComponents);
+                    LSMBTreeMutableComponent mutableComponent = (LSMBTreeMutableComponent) c;
                     if (mutableComponent.isReadable()) {
                         // Make sure newest components are added first
                         operationalComponents.add(0, mutableComponent);
@@ -308,12 +290,13 @@
                 ctx.currentMutableBTreeAccessor.upsert(tuple);
                 break;
         }
-        mutableComponents.get(currentMutableComponentId.get()).setIsModified();
     }
 
     private boolean insert(ITupleReference tuple, LSMBTreeOpContext ctx) throws HyracksDataException, IndexException {
-        MultiComparator comparator = MultiComparator.createIgnoreFieldLength(mutableComponents
-                .get(currentMutableComponentId.get()).getBTree().getComparatorFactories());
+        ILSMComponent c = ctx.getComponentHolder().get(0);
+        LSMBTreeMutableComponent mutableComponent = (LSMBTreeMutableComponent) c;
+        MultiComparator comparator = MultiComparator.createIgnoreFieldLength(mutableComponent.getBTree()
+                .getComparatorFactories());
         LSMBTreePointSearchCursor searchCursor = new LSMBTreePointSearchCursor(ctx);
         IIndexCursor memCursor = new BTreeRangeSearchCursor(ctx.currentMutableBTreeOpCtx.leafFrame, false);
         RangePredicate predicate = new RangePredicate(tuple, tuple, true, true, comparator, comparator);
@@ -345,7 +328,7 @@
         // components
 
         // This is a hack to avoid searching the current active mutable component twice. It is critical to add it back once the search is over.
-        ILSMComponent c = ctx.getComponentHolder().remove(0);
+        ILSMComponent firstComponent = ctx.getComponentHolder().remove(0);
         search(ctx, searchCursor, predicate);
         try {
             if (searchCursor.hasNext()) {
@@ -354,7 +337,7 @@
         } finally {
             searchCursor.close();
             // Add the current active mutable component back
-            ctx.getComponentHolder().add(0, c);
+            ctx.getComponentHolder().add(0, firstComponent);
         }
 
         ctx.currentMutableBTreeAccessor.upsertIfConditionElseInsert(tuple, AntimatterAwareTupleAcceptor.INSTANCE);
@@ -654,33 +637,44 @@
 
     @Override
     public ITreeIndexFrameFactory getInteriorFrameFactory() {
-        return mutableComponents.get(currentMutableComponentId.get()).getBTree().getInteriorFrameFactory();
+        LSMBTreeMutableComponent mutableComponent = (LSMBTreeMutableComponent) mutableComponents
+                .get(currentMutableComponentId.get());
+        return mutableComponent.getBTree().getInteriorFrameFactory();
     }
 
     @Override
     public int getFieldCount() {
-        return mutableComponents.get(currentMutableComponentId.get()).getBTree().getFieldCount();
+        LSMBTreeMutableComponent mutableComponent = (LSMBTreeMutableComponent) mutableComponents
+                .get(currentMutableComponentId.get());
+        return mutableComponent.getBTree().getFieldCount();
     }
 
     @Override
     public int getFileId() {
-        return mutableComponents.get(currentMutableComponentId.get()).getBTree().getFileId();
+        LSMBTreeMutableComponent mutableComponent = (LSMBTreeMutableComponent) mutableComponents
+                .get(currentMutableComponentId.get());
+        return mutableComponent.getBTree().getFileId();
     }
 
     @Override
     public IFreePageManager getFreePageManager() {
-        return mutableComponents.get(currentMutableComponentId.get()).getBTree().getFreePageManager();
+        LSMBTreeMutableComponent mutableComponent = (LSMBTreeMutableComponent) mutableComponents
+                .get(currentMutableComponentId.get());
+        return mutableComponent.getBTree().getFreePageManager();
     }
 
     @Override
     public ITreeIndexFrameFactory getLeafFrameFactory() {
-        return mutableComponents.get(currentMutableComponentId.get()).getBTree().getLeafFrameFactory();
+        LSMBTreeMutableComponent mutableComponent = (LSMBTreeMutableComponent) mutableComponents
+                .get(currentMutableComponentId.get());
+        return mutableComponent.getBTree().getLeafFrameFactory();
     }
 
     @Override
     public long getMemoryAllocationSize() {
         long size = 0;
-        for (LSMBTreeMutableComponent mutableComponent : mutableComponents) {
+        for (ILSMComponent c : mutableComponents) {
+            LSMBTreeMutableComponent mutableComponent = (LSMBTreeMutableComponent) c;
             IBufferCache virtualBufferCache = mutableComponent.getBTree().getBufferCache();
             size += virtualBufferCache.getNumPages() * virtualBufferCache.getPageSize();
         }
@@ -689,26 +683,18 @@
 
     @Override
     public int getRootPageId() {
-        return mutableComponents.get(currentMutableComponentId.get()).getBTree().getRootPageId();
-    }
-
-    public boolean isEmptyIndex() throws HyracksDataException {
-        boolean isModified = false;
-        for (LSMBTreeMutableComponent mutableComponent : mutableComponents) {
-            if (mutableComponent.isModified()) {
-                isModified = true;
-                break;
-            }
-        }
-        return componentsRef.get().isEmpty() && !isModified;
+        LSMBTreeMutableComponent mutableComponent = (LSMBTreeMutableComponent) mutableComponents
+                .get(currentMutableComponentId.get());
+        return mutableComponent.getBTree().getRootPageId();
     }
 
     @Override
     public void validate() throws HyracksDataException {
-        for (LSMBTreeMutableComponent mutableComponent : mutableComponents) {
+        for (ILSMComponent c : mutableComponents) {
+            LSMBTreeMutableComponent mutableComponent = (LSMBTreeMutableComponent) c;
             mutableComponent.getBTree().validate();
         }
-        List<ILSMComponent> immutableComponents = componentsRef.get();
+        List<ILSMComponent> immutableComponents = diskComponents;
         for (ILSMComponent c : immutableComponents) {
             BTree btree = (BTree) ((LSMBTreeImmutableComponent) c).getBTree();
             btree.validate();
diff --git a/hyracks/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/btree/impls/LSMBTreeImmutableComponent.java b/hyracks/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/btree/impls/LSMBTreeImmutableComponent.java
index affdef8..add1f63 100644
--- a/hyracks/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/btree/impls/LSMBTreeImmutableComponent.java
+++ b/hyracks/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/btree/impls/LSMBTreeImmutableComponent.java
@@ -17,9 +17,9 @@
 import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
 import edu.uci.ics.hyracks.storage.am.bloomfilter.impls.BloomFilter;
 import edu.uci.ics.hyracks.storage.am.btree.impls.BTree;
-import edu.uci.ics.hyracks.storage.am.lsm.common.impls.AbstractImmutableLSMComponent;
+import edu.uci.ics.hyracks.storage.am.lsm.common.impls.AbstractDiskLSMComponent;
 
-public class LSMBTreeImmutableComponent extends AbstractImmutableLSMComponent {
+public class LSMBTreeImmutableComponent extends AbstractDiskLSMComponent {
     private final BTree btree;
     private final BloomFilter bloomFilter;
 
diff --git a/hyracks/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/btree/impls/LSMBTreeMutableComponent.java b/hyracks/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/btree/impls/LSMBTreeMutableComponent.java
index c0605e5..63dd0ae 100644
--- a/hyracks/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/btree/impls/LSMBTreeMutableComponent.java
+++ b/hyracks/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/btree/impls/LSMBTreeMutableComponent.java
@@ -18,9 +18,9 @@
 import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
 import edu.uci.ics.hyracks.storage.am.btree.impls.BTree;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.IVirtualBufferCache;
-import edu.uci.ics.hyracks.storage.am.lsm.common.impls.AbstractMutableLSMComponent;
+import edu.uci.ics.hyracks.storage.am.lsm.common.impls.AbstractMemoryLSMComponent;
 
-public class LSMBTreeMutableComponent extends AbstractMutableLSMComponent {
+public class LSMBTreeMutableComponent extends AbstractMemoryLSMComponent {
 
     private final BTree btree;
 
diff --git a/hyracks/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/btree/impls/LSMBTreeOpContext.java b/hyracks/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/btree/impls/LSMBTreeOpContext.java
index 036c306..d36f45d 100644
--- a/hyracks/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/btree/impls/LSMBTreeOpContext.java
+++ b/hyracks/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/btree/impls/LSMBTreeOpContext.java
@@ -49,25 +49,25 @@
     public final ISearchOperationCallback searchCallback;
     private final List<ILSMComponent> componentHolder;
 
-    public LSMBTreeOpContext(List<LSMBTreeMutableComponent> mutableComponents,
-            ITreeIndexFrameFactory insertLeafFrameFactory, ITreeIndexFrameFactory deleteLeafFrameFactory,
-            IModificationOperationCallback modificationCallback, ISearchOperationCallback searchCallback,
-            int numBloomFilterKeyFields) {
-        IBinaryComparatorFactory cmpFactories[] = mutableComponents.get(0).getBTree().getComparatorFactories();
+    public LSMBTreeOpContext(List<ILSMComponent> mutableComponents, ITreeIndexFrameFactory insertLeafFrameFactory,
+            ITreeIndexFrameFactory deleteLeafFrameFactory, IModificationOperationCallback modificationCallback,
+            ISearchOperationCallback searchCallback, int numBloomFilterKeyFields) {
+        LSMBTreeMutableComponent c = (LSMBTreeMutableComponent) mutableComponents.get(0);
+        IBinaryComparatorFactory cmpFactories[] = c.getBTree().getComparatorFactories();
         if (cmpFactories[0] != null) {
-            this.cmp = MultiComparator.create(mutableComponents.get(0).getBTree().getComparatorFactories());
+            this.cmp = MultiComparator.create(c.getBTree().getComparatorFactories());
         } else {
             this.cmp = null;
         }
 
-        bloomFilterCmp = MultiComparator.create(mutableComponents.get(0).getBTree().getComparatorFactories(), 0,
-                numBloomFilterKeyFields);
+        bloomFilterCmp = MultiComparator.create(c.getBTree().getComparatorFactories(), 0, numBloomFilterKeyFields);
 
         mutableBTrees = new BTree[mutableComponents.size()];
         mutableBTreeAccessors = new BTree.BTreeAccessor[mutableComponents.size()];
         mutableBTreeOpCtxs = new BTreeOpContext[mutableComponents.size()];
         for (int i = 0; i < mutableComponents.size(); i++) {
-            mutableBTrees[i] = mutableComponents.get(i).getBTree();
+            LSMBTreeMutableComponent mutableComponent = (LSMBTreeMutableComponent) mutableComponents.get(i);
+            mutableBTrees[i] = mutableComponent.getBTree();
             mutableBTreeAccessors[i] = (BTree.BTreeAccessor) mutableBTrees[i].createAccessor(modificationCallback,
                     NoOpOperationCallback.INSTANCE);
             mutableBTreeOpCtxs[i] = mutableBTreeAccessors[i].getOpContext();
diff --git a/hyracks/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/api/ILSMComponent.java b/hyracks/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/api/ILSMComponent.java
index 14b86c9..afbabe0 100644
--- a/hyracks/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/api/ILSMComponent.java
+++ b/hyracks/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/api/ILSMComponent.java
@@ -24,10 +24,21 @@
         DISK
     }
 
-    public boolean threadEnter(LSMOperationType opType, boolean firstComponent) throws HyracksDataException;
+    enum ComponentState {
+        INACTIVE,
+        READABLE_WRITABLE,
+        READABLE_UNWRITABLE,
+        READABLE_UNWRITABLE_FLUSHING,
+        UNREADABLE_UNWRITABLE,
+        READABLE_MERGING
+    }
 
-    public void threadExit(LSMOperationType opType, boolean failedOperation, boolean firstComponent)
+    public boolean threadEnter(LSMOperationType opType, boolean isMutableComponent) throws HyracksDataException;
+
+    public void threadExit(LSMOperationType opType, boolean failedOperation, boolean isMutableComponent)
             throws HyracksDataException;
 
     public LSMComponentType getType();
+
+    public ComponentState getState();
 }
diff --git a/hyracks/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/api/ILSMIndex.java b/hyracks/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/api/ILSMIndex.java
index a92cb42..416a15b 100644
--- a/hyracks/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/api/ILSMIndex.java
+++ b/hyracks/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/api/ILSMIndex.java
@@ -38,13 +38,9 @@
     public ILSMIndexAccessor createAccessor(IModificationOperationCallback modificationCallback,
             ISearchOperationCallback searchCallback) throws HyracksDataException;
 
-    public boolean getFlushStatus();
-
     public ILSMOperationTracker getOperationTracker();
 
     public ILSMIOOperationScheduler getIOScheduler();
 
     public List<ILSMComponent> getImmutableComponents();
-
-    public int getCurrentMutableComponentId();
 }
diff --git a/hyracks/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/api/ILSMIndexInternal.java b/hyracks/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/api/ILSMIndexInternal.java
index 9d6cec4..dddee9d 100644
--- a/hyracks/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/api/ILSMIndexInternal.java
+++ b/hyracks/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/api/ILSMIndexInternal.java
@@ -50,6 +50,12 @@
 
     public void subsumeMergedComponents(ILSMComponent newComponent, List<ILSMComponent> mergedComponents);
 
+    public void changeMutableComponent();
+
+    public void changeFlushStatusForCurrentMutableCompoent(boolean needsFlush);
+
+    public boolean hasFlushRequestForCurrentMutableComponent();
+
     /**
      * Populates the context's component holder with a snapshot of the components involved in the operation.
      * 
diff --git a/hyracks/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/api/IMutableComponentSwitcherCallback.java b/hyracks/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/api/IMutableComponentSwitcherCallback.java
deleted file mode 100644
index 497a634..0000000
--- a/hyracks/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/api/IMutableComponentSwitcherCallback.java
+++ /dev/null
@@ -1,24 +0,0 @@
-/*
- * Copyright 2009-2013 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 IMutableComponentSwitcherCallback {
-
-    public void switchComponents() throws HyracksDataException;
-
-    public void requestFlush(boolean isFlushNeeded);
-}
\ No newline at end of file
diff --git a/hyracks/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/AbstractImmutableLSMComponent.java b/hyracks/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/AbstractImmutableLSMComponent.java
deleted file mode 100644
index f4091de..0000000
--- a/hyracks/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/AbstractImmutableLSMComponent.java
+++ /dev/null
@@ -1,97 +0,0 @@
-/*
- * Copyright 2009-2013 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.lsm.common.api.ILSMComponent;
-
-public abstract class AbstractImmutableLSMComponent implements ILSMComponent {
-
-    private ComponentState state;
-    private int readerCount;
-
-    private enum ComponentState {
-        READABLE,
-        READABLE_MERGING,
-        KILLED
-    }
-
-    public AbstractImmutableLSMComponent() {
-        state = ComponentState.READABLE;
-        readerCount = 0;
-    }
-
-    @Override
-    public boolean threadEnter(LSMOperationType opType, boolean firstComponent) {
-        if (state == ComponentState.KILLED) {
-            return false;
-        }
-
-        switch (opType) {
-            case FORCE_MODIFICATION:
-            case MODIFICATION:
-            case SEARCH:
-                readerCount++;
-                break;
-            case MERGE:
-                if (state == ComponentState.READABLE_MERGING) {
-                    return false;
-                }
-                state = ComponentState.READABLE_MERGING;
-                readerCount++;
-                break;
-            default:
-                throw new UnsupportedOperationException("Unsupported operation " + opType);
-        }
-        return true;
-    }
-
-    @Override
-    public void threadExit(LSMOperationType opType, boolean failedOperation, boolean firstComponent)
-            throws HyracksDataException {
-        switch (opType) {
-            case MERGE:
-                if (failedOperation) {
-                    state = ComponentState.READABLE;
-                }
-            case FORCE_MODIFICATION:
-            case MODIFICATION:
-            case SEARCH:
-                readerCount--;
-                if (readerCount == 0 && state == ComponentState.READABLE_MERGING) {
-                    destroy();
-                    state = ComponentState.KILLED;
-                }
-                break;
-            default:
-                throw new UnsupportedOperationException("Unsupported operation " + opType);
-        }
-    }
-
-    public boolean isMergable() {
-        if (state == ComponentState.READABLE) {
-            return true;
-        }
-        return false;
-    }
-
-    @Override
-    public LSMComponentType getType() {
-        return LSMComponentType.DISK;
-    }
-
-    protected abstract void destroy() throws HyracksDataException;
-
-}
diff --git a/hyracks/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/AbstractLSMIndex.java b/hyracks/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/AbstractLSMIndex.java
index 02f9c45..8b675f2 100644
--- a/hyracks/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/AbstractLSMIndex.java
+++ b/hyracks/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/AbstractLSMIndex.java
@@ -20,7 +20,6 @@
 import java.util.List;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicReference;
 
 import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
 import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndex;
@@ -46,6 +45,7 @@
     protected final ILSMIOOperationCallbackProvider ioOpCallbackProvider;
 
     // In-memory components.   
+    protected final List<ILSMComponent> mutableComponents;
     protected final List<IVirtualBufferCache> virtualBufferCaches;
     protected AtomicInteger currentMutableComponentId;
 
@@ -53,7 +53,7 @@
     protected final IBufferCache diskBufferCache;
     protected final ILSMIndexFileManager fileManager;
     protected final IFileMapProvider diskFileMapProvider;
-    protected final AtomicReference<List<ILSMComponent>> componentsRef;
+    protected final List<ILSMComponent> diskComponents;
     protected final double bloomFilterFalsePositiveRate;
 
     protected boolean isActivated;
@@ -73,8 +73,8 @@
         this.ioOpCallbackProvider = ioOpCallbackProvider;
         lsmHarness = new LSMHarness(this, mergePolicy, opTracker);
         isActivated = false;
-        componentsRef = new AtomicReference<List<ILSMComponent>>();
-        componentsRef.set(new LinkedList<ILSMComponent>());
+        diskComponents = new LinkedList<ILSMComponent>();
+        mutableComponents = new ArrayList<ILSMComponent>();
         currentMutableComponentId = new AtomicInteger();
         flushRequests = new AtomicBoolean[virtualBufferCaches.size()];
         for (int i = 0; i < virtualBufferCaches.size(); i++) {
@@ -137,38 +137,34 @@
 
     @Override
     public void addComponent(ILSMComponent c) {
-        List<ILSMComponent> oldList = componentsRef.get();
-        List<ILSMComponent> newList = new ArrayList<ILSMComponent>();
-        newList.add(c);
-        for (ILSMComponent oc : oldList) {
-            newList.add(oc);
-        }
-        componentsRef.set(newList);
+        diskComponents.add(0, c);
     }
 
     @Override
     public void subsumeMergedComponents(ILSMComponent newComponent, List<ILSMComponent> mergedComponents) {
-        List<ILSMComponent> oldList = componentsRef.get();
-        List<ILSMComponent> newList = new ArrayList<ILSMComponent>();
-        int swapIndex = oldList.indexOf(mergedComponents.get(0));
-        int swapSize = mergedComponents.size();
-        for (int i = 0; i < oldList.size(); i++) {
-            if (i < swapIndex || i >= swapIndex + swapSize) {
-                newList.add(oldList.get(i));
-            } else if (i == swapIndex) {
-                newList.add(newComponent);
-            }
-        }
-        componentsRef.set(newList);
+        int swapIndex = diskComponents.indexOf(mergedComponents.get(0));
+        diskComponents.removeAll(mergedComponents);
+        diskComponents.add(swapIndex, newComponent);
+    }
+
+    @Override
+    public void changeMutableComponent() {
+        currentMutableComponentId.set((currentMutableComponentId.get() + 1) % mutableComponents.size());
+        ((AbstractMemoryLSMComponent) mutableComponents.get(currentMutableComponentId.get())).setActive();
     }
 
     @Override
     public List<ILSMComponent> getImmutableComponents() {
-        return componentsRef.get();
+        return diskComponents;
     }
 
     @Override
-    public boolean getFlushStatus() {
+    public void changeFlushStatusForCurrentMutableCompoent(boolean needsFlush) {
+        flushRequests[currentMutableComponentId.get()].set(needsFlush);
+    }
+
+    @Override
+    public boolean hasFlushRequestForCurrentMutableComponent() {
         return flushRequests[currentMutableComponentId.get()].get();
     }
 
@@ -187,9 +183,16 @@
         return diskBufferCache;
     }
 
-    @Override
-    public int getCurrentMutableComponentId() {
-        return currentMutableComponentId.get();
+    public boolean isEmptyIndex() throws HyracksDataException {
+        boolean isModified = false;
+        for (ILSMComponent c : mutableComponents) {
+            AbstractMemoryLSMComponent mutableComponent = (AbstractMemoryLSMComponent) c;
+            if (mutableComponent.isModified()) {
+                isModified = true;
+                break;
+            }
+        }
+        return diskComponents.isEmpty() && !isModified;
     }
 
     @Override
diff --git a/hyracks/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/AbstractMutableLSMComponent.java b/hyracks/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/AbstractMutableLSMComponent.java
deleted file mode 100644
index 497bb3f..0000000
--- a/hyracks/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/AbstractMutableLSMComponent.java
+++ /dev/null
@@ -1,197 +0,0 @@
-/*
- * Copyright 2009-2013 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.concurrent.atomic.AtomicBoolean;
-
-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.IMutableComponentSwitcherCallback;
-import edu.uci.ics.hyracks.storage.am.lsm.common.api.IVirtualBufferCache;
-
-public abstract class AbstractMutableLSMComponent implements ILSMComponent {
-
-    private int readerCount;
-    private int writerCount;
-    private ComponentState state;
-    private final IVirtualBufferCache vbc;
-
-    private IMutableComponentSwitcherCallback switcherCallback;
-
-    private final AtomicBoolean isModified;
-    private boolean requestedToBeActive;
-
-    private enum ComponentState {
-        READABLE_WRITABLE,
-        READABLE_UNWRITABLE,
-        READABLE_UNWRITABLE_FLUSHING,
-        UNREADABLE_UNWRITABLE,
-        INACTIVE
-    }
-
-    public AbstractMutableLSMComponent(IVirtualBufferCache vbc, boolean isActive) {
-        this.vbc = vbc;
-        readerCount = 0;
-        writerCount = 0;
-        if (isActive) {
-            state = ComponentState.READABLE_WRITABLE;
-        } else {
-            state = ComponentState.INACTIVE;
-        }
-        isModified = new AtomicBoolean();
-    }
-
-    @Override
-    public boolean threadEnter(LSMOperationType opType, boolean firstComponent) throws HyracksDataException {
-        if (state == ComponentState.INACTIVE && requestedToBeActive) {
-            state = ComponentState.READABLE_WRITABLE;
-            requestedToBeActive = false;
-        }
-        switch (opType) {
-            case FORCE_MODIFICATION:
-                if (firstComponent) {
-                    if (state == ComponentState.READABLE_WRITABLE || state == ComponentState.READABLE_UNWRITABLE) {
-                        writerCount++;
-                    } else {
-                        return false;
-                    }
-                } else {
-                    if (state == ComponentState.READABLE_UNWRITABLE
-                            || state == ComponentState.READABLE_UNWRITABLE_FLUSHING) {
-                        readerCount++;
-                    } else {
-                        return false;
-                    }
-                }
-                break;
-            case MODIFICATION:
-                if (firstComponent) {
-                    if (state == ComponentState.READABLE_WRITABLE) {
-                        writerCount++;
-                    } else {
-                        return false;
-                    }
-                } else {
-                    if (state == ComponentState.READABLE_UNWRITABLE
-                            || state == ComponentState.READABLE_UNWRITABLE_FLUSHING) {
-                        readerCount++;
-                    } else {
-                        return false;
-                    }
-                }
-                break;
-            case SEARCH:
-                if (state == ComponentState.READABLE_WRITABLE || state == ComponentState.READABLE_UNWRITABLE
-                        || state == ComponentState.READABLE_UNWRITABLE_FLUSHING) {
-                    readerCount++;
-                } else {
-                    return false;
-                }
-                break;
-            case FLUSH:
-                if (state == ComponentState.READABLE_WRITABLE || state == ComponentState.READABLE_UNWRITABLE) {
-                    assert writerCount == 0;
-                    state = ComponentState.READABLE_UNWRITABLE_FLUSHING;
-                    switcherCallback.requestFlush(false);
-                    switcherCallback.switchComponents();
-                    readerCount++;
-                } else {
-                    return false;
-                }
-                break;
-            default:
-                throw new UnsupportedOperationException("Unsupported operation " + opType);
-        }
-        return true;
-    }
-
-    @Override
-    public void threadExit(LSMOperationType opType, boolean failedOperation, boolean firstComponent)
-            throws HyracksDataException {
-        switch (opType) {
-            case FORCE_MODIFICATION:
-            case MODIFICATION:
-                if (firstComponent) {
-                    writerCount--;
-                    if (state == ComponentState.READABLE_WRITABLE && isFull() && !failedOperation) {
-                        state = ComponentState.READABLE_UNWRITABLE;
-                        switcherCallback.requestFlush(true);
-                    }
-                } else {
-                    readerCount--;
-                    if (state == ComponentState.UNREADABLE_UNWRITABLE && readerCount == 0) {
-                        reset();
-                        state = ComponentState.INACTIVE;
-                    }
-                }
-                break;
-            case SEARCH:
-                readerCount--;
-                if (state == ComponentState.UNREADABLE_UNWRITABLE && readerCount == 0) {
-                    reset();
-                    state = ComponentState.INACTIVE;
-                }
-                break;
-            case FLUSH:
-                assert state == ComponentState.READABLE_UNWRITABLE_FLUSHING;
-                readerCount--;
-                if (readerCount == 0) {
-                    reset();
-                    state = ComponentState.INACTIVE;
-                } else {
-                    state = ComponentState.UNREADABLE_UNWRITABLE;
-                }
-                break;
-            default:
-                throw new UnsupportedOperationException("Unsupported operation " + opType);
-        }
-    }
-
-    public boolean isReadable() {
-        if (state == ComponentState.INACTIVE || state == ComponentState.UNREADABLE_UNWRITABLE) {
-            return false;
-        }
-        return true;
-    }
-
-    @Override
-    public LSMComponentType getType() {
-        return LSMComponentType.MEMORY;
-    }
-
-    public void setActive() {
-        requestedToBeActive = true;
-    }
-
-    public void registerOnFlushCallback(IMutableComponentSwitcherCallback switcherCallback) {
-        this.switcherCallback = switcherCallback;
-    }
-
-    public void setIsModified() {
-        isModified.set(true);
-    }
-
-    public boolean isModified() {
-        return isModified.get();
-    }
-
-    public boolean isFull() {
-        return vbc.isFull();
-    }
-
-    protected void reset() throws HyracksDataException {
-        isModified.set(false);
-    }
-}
diff --git a/hyracks/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/LSMHarness.java b/hyracks/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/LSMHarness.java
index 220dbf0..a633758 100644
--- a/hyracks/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/LSMHarness.java
+++ b/hyracks/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.IndexException;
 import edu.uci.ics.hyracks.storage.am.common.ophelpers.IndexOperation;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMComponent;
+import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMComponent.LSMComponentType;
 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;
@@ -63,7 +64,9 @@
                 List<ILSMComponent> components = ctx.getComponentHolder();
                 try {
                     for (ILSMComponent c : components) {
-                        if (!c.threadEnter(opType, numEntered == 0 ? true : false)) {
+                        boolean isMutableComponent = numEntered == 0 && c.getType() == LSMComponentType.MEMORY ? true
+                                : false;
+                        if (!c.threadEnter(opType, isMutableComponent)) {
                             break;
                         }
                         numEntered++;
@@ -80,7 +83,9 @@
                             if (numEntered == 0) {
                                 break;
                             }
-                            c.threadExit(opType, true, i == 0 ? true : false);
+                            boolean isMutableComponent = i == 0 && c.getType() == LSMComponentType.MEMORY ? true
+                                    : false;
+                            c.threadExit(opType, true, isMutableComponent);
                             i++;
                             numEntered--;
                         }
@@ -100,7 +105,32 @@
             try {
                 int i = 0;
                 for (ILSMComponent c : ctx.getComponentHolder()) {
-                    c.threadExit(opType, failedOperation, i == 0 ? true : false);
+                    boolean isMutableComponent = i == 0 && c.getType() == LSMComponentType.MEMORY ? true : false;
+                    c.threadExit(opType, failedOperation, isMutableComponent);
+
+                    if (c.getType() == LSMComponentType.MEMORY) {
+                        switch (c.getState()) {
+                            case READABLE_UNWRITABLE:
+                                if (isMutableComponent
+                                        && (opType == LSMOperationType.MODIFICATION || opType == LSMOperationType.FORCE_MODIFICATION)) {
+                                    lsmIndex.changeFlushStatusForCurrentMutableCompoent(true);
+                                }
+                                break;
+                            case INACTIVE:
+                                ((AbstractMemoryLSMComponent) c).reset();
+                                break;
+                            default:
+                                break;
+                        }
+                    } else {
+                        switch (c.getState()) {
+                            case INACTIVE:
+                                ((AbstractDiskLSMComponent) c).destroy();
+                                break;
+                            default:
+                                break;
+                        }
+                    }
                     i++;
                 }
             } finally {
@@ -130,6 +160,9 @@
         }
         try {
             lsmIndex.modify(ctx, tuple);
+            // The mutable component is always in the first index.
+            AbstractMemoryLSMComponent mutableComponent = (AbstractMemoryLSMComponent) ctx.getComponentHolder().get(0);
+            mutableComponent.setIsModified();
         } finally {
             exitComponents(ctx, opType, false);
         }
@@ -164,7 +197,7 @@
             return;
         }
         ILSMComponent flushingComponent = ctx.getComponentHolder().get(0);
-        if (!((AbstractMutableLSMComponent) flushingComponent).isModified()) {
+        if (!((AbstractMemoryLSMComponent) flushingComponent).isModified()) {
             callback.beforeOperation();
             callback.afterOperation(null, null);
             exitComponents(ctx, opType, false);
@@ -172,6 +205,9 @@
         } else {
             lsmIndex.scheduleFlush(ctx, callback);
         }
+        // Changing the flush status should *always* precede changing the mutable component.
+        lsmIndex.changeFlushStatusForCurrentMutableCompoent(false);
+        lsmIndex.changeMutableComponent();
     }
 
     @Override
diff --git a/hyracks/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/ThreadCountingTracker.java b/hyracks/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/ThreadCountingTracker.java
index 19ee4b8..db75d7b 100644
--- a/hyracks/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/ThreadCountingTracker.java
+++ b/hyracks/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/impls/ThreadCountingTracker.java
@@ -22,6 +22,7 @@
 import edu.uci.ics.hyracks.storage.am.common.impls.NoOpOperationCallback;
 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.ILSMIndexInternal;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMOperationTracker;
 
 public class ThreadCountingTracker implements ILSMOperationTracker {
@@ -51,7 +52,8 @@
             IModificationOperationCallback modificationCallback) throws HyracksDataException {
         // Flush will only be handled by last exiting thread.
         if (opType == LSMOperationType.MODIFICATION) {
-            if (threadRefCount.decrementAndGet() == 0 && index.getFlushStatus()) {
+            if (threadRefCount.decrementAndGet() == 0
+                    && ((ILSMIndexInternal) index).hasFlushRequestForCurrentMutableComponent()) {
                 ILSMIndexAccessor accessor = (ILSMIndexAccessor) index.createAccessor(NoOpOperationCallback.INSTANCE,
                         NoOpOperationCallback.INSTANCE);
                 accessor.scheduleFlush(NoOpIOOperationCallback.INSTANCE);
diff --git a/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndex.java b/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndex.java
index 5d4b88e..d83bc1d 100644
--- a/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndex.java
+++ b/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndex.java
@@ -60,7 +60,6 @@
 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.IMutableComponentSwitcherCallback;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.IVirtualBufferCache;
 import edu.uci.ics.hyracks.storage.am.lsm.common.freepage.VirtualFreePageManager;
 import edu.uci.ics.hyracks.storage.am.lsm.common.impls.AbstractLSMIndex;
@@ -82,8 +81,6 @@
 
 public class LSMInvertedIndex extends AbstractLSMIndex implements IInvertedIndex {
 
-    // In-memory components.
-    private final List<LSMInvertedIndexMutableComponent> mutableComponents;
     protected final IBinaryTokenizerFactory tokenizerFactory;
 
     // On-disk components.
@@ -116,8 +113,6 @@
         componentFactory = new LSMInvertedIndexComponentFactory(diskInvIndexFactory, deletedKeysBTreeFactory,
                 bloomFilterFactory);
 
-        mutableComponents = new ArrayList<LSMInvertedIndexMutableComponent>();
-        final int numMutableComponents = virtualBufferCaches.size();
         int i = 0;
         for (IVirtualBufferCache virtualBufferCache : virtualBufferCaches) {
             InMemoryInvertedIndex memInvIndex = createInMemoryInvertedIndex(virtualBufferCache,
@@ -128,21 +123,6 @@
                             fileManager.getBaseDir() + "_virtual_del_" + i)));
             LSMInvertedIndexMutableComponent mutableComponent = new LSMInvertedIndexMutableComponent(memInvIndex,
                     deleteKeysBTree, virtualBufferCache, i == 0 ? true : false);
-
-            mutableComponent.registerOnFlushCallback(new IMutableComponentSwitcherCallback() {
-
-                @Override
-                public void switchComponents() throws HyracksDataException {
-                    currentMutableComponentId.set((currentMutableComponentId.get() + 1) % numMutableComponents);
-                    mutableComponents.get(currentMutableComponentId.get()).setActive();
-                }
-
-                @Override
-                public void requestFlush(boolean isFlushNeeded) {
-                    flushRequests[currentMutableComponentId.get()].set(isFlushNeeded);
-                }
-            });
-
             mutableComponents.add(mutableComponent);
             ++i;
         }
@@ -156,7 +136,7 @@
 
         fileManager.deleteDirs();
         fileManager.createDirs();
-        componentsRef.get().clear();
+        diskComponents.clear();
     }
 
     @Override
@@ -166,8 +146,9 @@
         }
 
         try {
-            List<ILSMComponent> immutableComponents = componentsRef.get();
-            for (LSMInvertedIndexMutableComponent mutableComponent : mutableComponents) {
+            List<ILSMComponent> immutableComponents = diskComponents;
+            for (ILSMComponent c : mutableComponents) {
+                LSMInvertedIndexMutableComponent mutableComponent = (LSMInvertedIndexMutableComponent) c;
                 ((IVirtualBufferCache) mutableComponent.getInvIndex().getBufferCache()).open();
                 mutableComponent.getInvIndex().create();
                 mutableComponent.getInvIndex().activate();
@@ -201,12 +182,14 @@
         if (!isActivated) {
             throw new HyracksDataException("Failed to clear the index since it is not activated.");
         }
-        List<ILSMComponent> immutableComponents = componentsRef.get();
-        for (LSMInvertedIndexMutableComponent mutableComponent : mutableComponents) {
+
+        for (ILSMComponent c : mutableComponents) {
+            LSMInvertedIndexMutableComponent mutableComponent = (LSMInvertedIndexMutableComponent) c;
             mutableComponent.getInvIndex().clear();
             mutableComponent.getDeletedKeysBTree().clear();
             mutableComponent.reset();
         }
+        List<ILSMComponent> immutableComponents = diskComponents;
         for (ILSMComponent c : immutableComponents) {
             LSMInvertedIndexImmutableComponent component = (LSMInvertedIndexImmutableComponent) c;
             component.getBloomFilter().deactivate();
@@ -238,14 +221,15 @@
             }
         }
 
-        List<ILSMComponent> immutableComponents = componentsRef.get();
+        List<ILSMComponent> immutableComponents = diskComponents;
         for (ILSMComponent c : immutableComponents) {
             LSMInvertedIndexImmutableComponent component = (LSMInvertedIndexImmutableComponent) c;
             component.getBloomFilter().deactivate();
             component.getInvIndex().deactivate();
             component.getDeletedKeysBTree().deactivate();
         }
-        for (LSMInvertedIndexMutableComponent mutableComponent : mutableComponents) {
+        for (ILSMComponent c : mutableComponents) {
+            LSMInvertedIndexMutableComponent mutableComponent = (LSMInvertedIndexMutableComponent) c;
             mutableComponent.getInvIndex().deactivate();
             mutableComponent.getDeletedKeysBTree().deactivate();
             mutableComponent.getInvIndex().destroy();
@@ -265,11 +249,12 @@
             throw new HyracksDataException("Failed to destroy the index since it is activated.");
         }
 
-        for (LSMInvertedIndexMutableComponent mutableComponent : mutableComponents) {
+        for (ILSMComponent c : mutableComponents) {
+            LSMInvertedIndexMutableComponent mutableComponent = (LSMInvertedIndexMutableComponent) c;
             mutableComponent.getInvIndex().destroy();
             mutableComponent.getDeletedKeysBTree().destroy();
         }
-        List<ILSMComponent> immutableComponents = componentsRef.get();
+        List<ILSMComponent> immutableComponents = diskComponents;
         for (ILSMComponent c : immutableComponents) {
             LSMInvertedIndexImmutableComponent component = (LSMInvertedIndexImmutableComponent) c;
             component.getInvIndex().destroy();
@@ -281,7 +266,7 @@
 
     @Override
     public void getOperationalComponents(ILSMIndexOperationContext ctx) {
-        List<ILSMComponent> immutableComponents = componentsRef.get();
+        List<ILSMComponent> immutableComponents = diskComponents;
         List<ILSMComponent> operationalComponents = ctx.getComponentHolder();
         operationalComponents.clear();
         int cmc = currentMutableComponentId.get();
@@ -295,8 +280,8 @@
                 break;
             case SEARCH:
                 for (int i = 0; i < numMutableComponents - 1; i++) {
-                    LSMInvertedIndexMutableComponent mutableComponent = mutableComponents.get((cmc + i + 1)
-                            % numMutableComponents);
+                    ILSMComponent c = mutableComponents.get((cmc + i + 1) % numMutableComponents);
+                    LSMInvertedIndexMutableComponent mutableComponent = (LSMInvertedIndexMutableComponent) c;
                     if (mutableComponent.isReadable()) {
                         // Make sure newest components are added first
                         operationalComponents.add(0, mutableComponent);
@@ -361,7 +346,6 @@
                 throw new UnsupportedOperationException("Operation " + ctx.getOperation() + " not supported.");
             }
         }
-        mutableComponents.get(currentMutableComponentId.get()).setIsModified();
     }
 
     @Override
@@ -408,21 +392,21 @@
         PermutingTupleReference keysOnlyTuple = createKeysOnlyTupleReference();
         MultiComparator keyCmp = MultiComparator.createIgnoreFieldLength(invListCmpFactories);
 
+        LSMInvertedIndexOpContext ctx = (LSMInvertedIndexOpContext) ictx;
+        ILSMComponent c = ctx.getComponentHolder().get(0);
+        LSMInvertedIndexMutableComponent mutableComponent = (LSMInvertedIndexMutableComponent) c;
         // TODO: This check is not pretty, but it does the job. Come up with something more OO in the future.
         // Distinguish between regular searches and range searches (mostly used in merges).
         if (pred instanceof InvertedIndexSearchPredicate) {
             initState = new LSMInvertedIndexSearchCursorInitialState(keyCmp, keysOnlyTuple, indexAccessors,
-                    deletedKeysBTreeAccessors, mutableComponents.get(currentMutableComponentId.get())
-                            .getDeletedKeysBTree().getLeafFrameFactory(), ictx, includeMutableComponent, lsmHarness,
-                    operationalComponents);
+                    deletedKeysBTreeAccessors, mutableComponent.getDeletedKeysBTree().getLeafFrameFactory(), ictx,
+                    includeMutableComponent, lsmHarness, operationalComponents);
         } else {
-            InMemoryInvertedIndex memInvIndex = (InMemoryInvertedIndex) mutableComponents.get(
-                    currentMutableComponentId.get()).getInvIndex();
+            InMemoryInvertedIndex memInvIndex = (InMemoryInvertedIndex) mutableComponent.getInvIndex();
             MultiComparator tokensAndKeysCmp = MultiComparator.create(memInvIndex.getBTree().getComparatorFactories());
             initState = new LSMInvertedIndexRangeSearchCursorInitialState(tokensAndKeysCmp, keyCmp, keysOnlyTuple,
-                    mutableComponents.get(currentMutableComponentId.get()).getDeletedKeysBTree().getLeafFrameFactory(),
-                    includeMutableComponent, lsmHarness, indexAccessors, deletedKeysBTreeAccessors, pred,
-                    operationalComponents);
+                    mutableComponent.getDeletedKeysBTree().getLeafFrameFactory(), includeMutableComponent, lsmHarness,
+                    indexAccessors, deletedKeysBTreeAccessors, pred, operationalComponents);
         }
         return initState;
     }
@@ -614,17 +598,6 @@
         }
     }
 
-    public boolean isEmptyIndex() throws HyracksDataException {
-        boolean isModified = false;
-        for (LSMInvertedIndexMutableComponent mutableComponent : mutableComponents) {
-            if (mutableComponent.isModified()) {
-                isModified = true;
-                break;
-            }
-        }
-        return componentsRef.get().isEmpty() && !isModified;
-    }
-
     public class LSMInvertedIndexBulkLoader implements IIndexBulkLoader {
         private final ILSMComponent component;
         private final IIndexBulkLoader invIndexBulkLoader;
@@ -745,7 +718,8 @@
     @Override
     public long getMemoryAllocationSize() {
         long size = 0;
-        for (LSMInvertedIndexMutableComponent mutableComponent : mutableComponents) {
+        for (ILSMComponent c : mutableComponents) {
+            LSMInvertedIndexMutableComponent mutableComponent = (LSMInvertedIndexMutableComponent) c;
             IBufferCache virtualBufferCache = mutableComponent.getInvIndex().getBufferCache();
             size += virtualBufferCache.getNumPages() * virtualBufferCache.getPageSize();
         }
@@ -797,11 +771,12 @@
 
     @Override
     public void validate() throws HyracksDataException {
-        for (LSMInvertedIndexMutableComponent mutableComponent : mutableComponents) {
+        for (ILSMComponent c : mutableComponents) {
+            LSMInvertedIndexMutableComponent mutableComponent = (LSMInvertedIndexMutableComponent) c;
             mutableComponent.getInvIndex().validate();
             mutableComponent.getDeletedKeysBTree().validate();
         }
-        List<ILSMComponent> immutableComponents = componentsRef.get();
+        List<ILSMComponent> immutableComponents = diskComponents;
         for (ILSMComponent c : immutableComponents) {
             LSMInvertedIndexImmutableComponent component = (LSMInvertedIndexImmutableComponent) c;
             component.getInvIndex().validate();
diff --git a/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexImmutableComponent.java b/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexImmutableComponent.java
index 829d5d4..ba48880 100644
--- a/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexImmutableComponent.java
+++ b/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexImmutableComponent.java
@@ -17,10 +17,10 @@
 import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
 import edu.uci.ics.hyracks.storage.am.bloomfilter.impls.BloomFilter;
 import edu.uci.ics.hyracks.storage.am.btree.impls.BTree;
-import edu.uci.ics.hyracks.storage.am.lsm.common.impls.AbstractImmutableLSMComponent;
+import edu.uci.ics.hyracks.storage.am.lsm.common.impls.AbstractDiskLSMComponent;
 import edu.uci.ics.hyracks.storage.am.lsm.invertedindex.api.IInvertedIndex;
 
-public class LSMInvertedIndexImmutableComponent extends AbstractImmutableLSMComponent {
+public class LSMInvertedIndexImmutableComponent extends AbstractDiskLSMComponent {
 
     private final IInvertedIndex invIndex;
     private final BTree deletedKeysBTree;
diff --git a/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexMutableComponent.java b/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexMutableComponent.java
index 72fc3ea..9864f93 100644
--- a/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexMutableComponent.java
+++ b/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexMutableComponent.java
@@ -18,10 +18,10 @@
 import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
 import edu.uci.ics.hyracks.storage.am.btree.impls.BTree;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.IVirtualBufferCache;
-import edu.uci.ics.hyracks.storage.am.lsm.common.impls.AbstractMutableLSMComponent;
+import edu.uci.ics.hyracks.storage.am.lsm.common.impls.AbstractMemoryLSMComponent;
 import edu.uci.ics.hyracks.storage.am.lsm.invertedindex.api.IInvertedIndex;
 
-public class LSMInvertedIndexMutableComponent extends AbstractMutableLSMComponent {
+public class LSMInvertedIndexMutableComponent extends AbstractMemoryLSMComponent {
 
     private final IInvertedIndex invIndex;
     private final BTree deletedKeysBTree;
diff --git a/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexOpContext.java b/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexOpContext.java
index 95df583..119b9b7 100644
--- a/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexOpContext.java
+++ b/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexOpContext.java
@@ -50,7 +50,7 @@
     public IInvertedIndexAccessor currentMutableInvIndexAccessors;
     public IIndexAccessor currentDeletedKeysBTreeAccessors;
 
-    public LSMInvertedIndexOpContext(List<LSMInvertedIndexMutableComponent> mutableComponents,
+    public LSMInvertedIndexOpContext(List<ILSMComponent> mutableComponents,
             IModificationOperationCallback modificationCallback, ISearchOperationCallback searchCallback)
             throws HyracksDataException {
         this.componentHolder = new LinkedList<ILSMComponent>();
@@ -61,16 +61,19 @@
         deletedKeysBTreeAccessors = new IIndexAccessor[mutableComponents.size()];
 
         for (int i = 0; i < mutableComponents.size(); i++) {
-            mutableInvIndexAccessors[i] = (IInvertedIndexAccessor) mutableComponents.get(i).getInvIndex()
-                    .createAccessor(NoOpOperationCallback.INSTANCE, NoOpOperationCallback.INSTANCE);
-            deletedKeysBTreeAccessors[i] = mutableComponents.get(i).getDeletedKeysBTree()
-                    .createAccessor(NoOpOperationCallback.INSTANCE, NoOpOperationCallback.INSTANCE);
+            LSMInvertedIndexMutableComponent mutableComponent = (LSMInvertedIndexMutableComponent) mutableComponents
+                    .get(i);
+            mutableInvIndexAccessors[i] = (IInvertedIndexAccessor) mutableComponent.getInvIndex().createAccessor(
+                    NoOpOperationCallback.INSTANCE, NoOpOperationCallback.INSTANCE);
+            deletedKeysBTreeAccessors[i] = mutableComponent.getDeletedKeysBTree().createAccessor(
+                    NoOpOperationCallback.INSTANCE, NoOpOperationCallback.INSTANCE);
         }
 
         assert mutableComponents.size() > 0;
 
         // Project away the document fields, leaving only the key fields.
-        int numKeyFields = mutableComponents.get(0).getInvIndex().getInvListTypeTraits().length;
+        LSMInvertedIndexMutableComponent c = (LSMInvertedIndexMutableComponent) mutableComponents.get(0);
+        int numKeyFields = c.getInvIndex().getInvListTypeTraits().length;
         int[] keyFieldPermutation = new int[numKeyFields];
         for (int i = 0; i < numKeyFields; i++) {
             keyFieldPermutation[i] = NUM_DOCUMENT_FIELDS + i;
diff --git a/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/AbstractLSMRTree.java b/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/AbstractLSMRTree.java
index b253594..a66ac76 100644
--- a/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/AbstractLSMRTree.java
+++ b/hyracks/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.ArrayList;
 import java.util.List;
 
 import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparatorFactory;
@@ -47,7 +46,6 @@
 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.IMutableComponentSwitcherCallback;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.IVirtualBufferCache;
 import edu.uci.ics.hyracks.storage.am.lsm.common.freepage.VirtualFreePageManager;
 import edu.uci.ics.hyracks.storage.am.lsm.common.impls.AbstractLSMIndex;
@@ -66,9 +64,6 @@
     protected final int[] comparatorFields;
     protected final IBinaryComparatorFactory[] linearizerArray;
 
-    // In-memory components.
-    protected final List<LSMRTreeMutableComponent> mutableComponents;
-
     protected TreeTupleSorter rTreeTupleSorter;
 
     // On-disk components.
@@ -95,9 +90,6 @@
             ILSMIOOperationScheduler ioScheduler, ILSMIOOperationCallbackProvider ioOpCallbackProvider) {
         super(virtualBufferCaches, diskRTreeFactory.getBufferCache(), fileManager, diskFileMapProvider,
                 bloomFilterFalsePositiveRate, mergePolicy, opTracker, ioScheduler, ioOpCallbackProvider);
-
-        mutableComponents = new ArrayList<LSMRTreeMutableComponent>();
-        final int numMutableComponents = virtualBufferCaches.size();
         int i = 0;
         for (IVirtualBufferCache virtualBufferCache : virtualBufferCaches) {
             RTree memRTree = new RTree(virtualBufferCache,
@@ -112,21 +104,6 @@
                             + i)));
             LSMRTreeMutableComponent mutableComponent = new LSMRTreeMutableComponent(memRTree, memBTree,
                     virtualBufferCache, i == 0 ? true : false);
-
-            mutableComponent.registerOnFlushCallback(new IMutableComponentSwitcherCallback() {
-
-                @Override
-                public void switchComponents() throws HyracksDataException {
-                    currentMutableComponentId.set((currentMutableComponentId.get() + 1) % numMutableComponents);
-                    mutableComponents.get(currentMutableComponentId.get()).setActive();
-                }
-
-                @Override
-                public void requestFlush(boolean isFlushNeeded) {
-                    flushRequests[currentMutableComponentId.get()].set(isFlushNeeded);
-                }
-            });
-
             mutableComponents.add(mutableComponent);
             ++i;
         }
@@ -152,7 +129,7 @@
 
         fileManager.deleteDirs();
         fileManager.createDirs();
-        componentsRef.get().clear();
+        diskComponents.clear();
     }
 
     @Override
@@ -161,7 +138,8 @@
             throw new HyracksDataException("Failed to activate the index since it is already activated.");
         }
 
-        for (LSMRTreeMutableComponent mutableComponent : mutableComponents) {
+        for (ILSMComponent c : mutableComponents) {
+            LSMRTreeMutableComponent mutableComponent = (LSMRTreeMutableComponent) c;
             ((IVirtualBufferCache) mutableComponent.getRTree().getBufferCache()).open();
             mutableComponent.getRTree().create();
             mutableComponent.getBTree().create();
@@ -188,7 +166,8 @@
             }
         }
 
-        for (LSMRTreeMutableComponent mutableComponent : mutableComponents) {
+        for (ILSMComponent c : mutableComponents) {
+            LSMRTreeMutableComponent mutableComponent = (LSMRTreeMutableComponent) c;
             mutableComponent.getRTree().deactivate();
             mutableComponent.getBTree().deactivate();
             mutableComponent.getRTree().destroy();
@@ -210,7 +189,8 @@
             throw new HyracksDataException("Failed to clear the index since it is not activated.");
         }
 
-        for (LSMRTreeMutableComponent mutableComponent : mutableComponents) {
+        for (ILSMComponent c : mutableComponents) {
+            LSMRTreeMutableComponent mutableComponent = (LSMRTreeMutableComponent) c;
             mutableComponent.getRTree().clear();
             mutableComponent.getBTree().clear();
             mutableComponent.reset();
@@ -221,7 +201,7 @@
     public void getOperationalComponents(ILSMIndexOperationContext ctx) {
         List<ILSMComponent> operationalComponents = ctx.getComponentHolder();
         operationalComponents.clear();
-        List<ILSMComponent> immutableComponents = componentsRef.get();
+        List<ILSMComponent> immutableComponents = diskComponents;
         int cmc = currentMutableComponentId.get();
         ctx.setCurrentMutableComponentId(cmc);
         int numMutableComponents = mutableComponents.size();
@@ -233,8 +213,8 @@
                 break;
             case SEARCH:
                 for (int i = 0; i < numMutableComponents - 1; i++) {
-                    LSMRTreeMutableComponent mutableComponent = mutableComponents.get((cmc + i + 1)
-                            % numMutableComponents);
+                    ILSMComponent c = mutableComponents.get((cmc + i + 1) % numMutableComponents);
+                    LSMRTreeMutableComponent mutableComponent = (LSMRTreeMutableComponent) c;
                     if (mutableComponent.isReadable()) {
                         // Make sure newest components are added first
                         operationalComponents.add(0, mutableComponent);
@@ -302,32 +282,44 @@
 
     @Override
     public ITreeIndexFrameFactory getLeafFrameFactory() {
-        return mutableComponents.get(currentMutableComponentId.get()).getRTree().getLeafFrameFactory();
+        LSMRTreeMutableComponent mutableComponent = (LSMRTreeMutableComponent) mutableComponents
+                .get(currentMutableComponentId.get());
+        return mutableComponent.getRTree().getLeafFrameFactory();
     }
 
     @Override
     public ITreeIndexFrameFactory getInteriorFrameFactory() {
-        return mutableComponents.get(currentMutableComponentId.get()).getRTree().getInteriorFrameFactory();
+        LSMRTreeMutableComponent mutableComponent = (LSMRTreeMutableComponent) mutableComponents
+                .get(currentMutableComponentId.get());
+        return mutableComponent.getRTree().getInteriorFrameFactory();
     }
 
     @Override
     public IFreePageManager getFreePageManager() {
-        return mutableComponents.get(currentMutableComponentId.get()).getRTree().getFreePageManager();
+        LSMRTreeMutableComponent mutableComponent = (LSMRTreeMutableComponent) mutableComponents
+                .get(currentMutableComponentId.get());
+        return mutableComponent.getRTree().getFreePageManager();
     }
 
     @Override
     public int getFieldCount() {
-        return mutableComponents.get(currentMutableComponentId.get()).getRTree().getFieldCount();
+        LSMRTreeMutableComponent mutableComponent = (LSMRTreeMutableComponent) mutableComponents
+                .get(currentMutableComponentId.get());
+        return mutableComponent.getRTree().getFieldCount();
     }
 
     @Override
     public int getRootPageId() {
-        return mutableComponents.get(currentMutableComponentId.get()).getRTree().getRootPageId();
+        LSMRTreeMutableComponent mutableComponent = (LSMRTreeMutableComponent) mutableComponents
+                .get(currentMutableComponentId.get());
+        return mutableComponent.getRTree().getRootPageId();
     }
 
     @Override
     public int getFileId() {
-        return mutableComponents.get(currentMutableComponentId.get()).getRTree().getFileId();
+        LSMRTreeMutableComponent mutableComponent = (LSMRTreeMutableComponent) mutableComponents
+                .get(currentMutableComponentId.get());
+        return mutableComponent.getRTree().getFileId();
     }
 
     @Override
@@ -379,7 +371,6 @@
                 // that all the corresponding insert tuples are deleted
             }
         }
-        mutableComponents.get(currentMutableComponentId.get()).setIsModified();
     }
 
     protected LSMRTreeOpContext createOpContext(IModificationOperationCallback modCallback) {
@@ -394,17 +385,6 @@
         return rtreeCmpFactories;
     }
 
-    public boolean isEmptyIndex() throws HyracksDataException {
-        boolean isModified = false;
-        for (LSMRTreeMutableComponent mutableComponent : mutableComponents) {
-            if (mutableComponent.isModified()) {
-                isModified = true;
-                break;
-            }
-        }
-        return componentsRef.get().isEmpty() && !isModified;
-    }
-
     @Override
     public void validate() throws HyracksDataException {
         throw new UnsupportedOperationException("Validation not implemented for LSM R-Trees.");
@@ -413,7 +393,8 @@
     @Override
     public long getMemoryAllocationSize() {
         long size = 0;
-        for (LSMRTreeMutableComponent mutableComponent : mutableComponents) {
+        for (ILSMComponent c : mutableComponents) {
+            LSMRTreeMutableComponent mutableComponent = (LSMRTreeMutableComponent) c;
             IBufferCache virtualBufferCache = mutableComponent.getRTree().getBufferCache();
             size += virtualBufferCache.getNumPages() * virtualBufferCache.getPageSize();
         }
diff --git a/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTree.java b/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTree.java
index cb6e166..c603e1c 100644
--- a/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTree.java
+++ b/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTree.java
@@ -55,7 +55,6 @@
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMOperationTracker;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.IVirtualBufferCache;
 import edu.uci.ics.hyracks.storage.am.lsm.common.impls.LSMComponentFileReferences;
-import edu.uci.ics.hyracks.storage.am.lsm.common.impls.LSMIndexSearchCursor;
 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.rtree.impls.RTree;
@@ -94,7 +93,7 @@
     @Override
     public synchronized void activate() throws HyracksDataException {
         super.activate();
-        List<ILSMComponent> immutableComponents = componentsRef.get();
+        List<ILSMComponent> immutableComponents = diskComponents;
         List<LSMComponentFileReferences> validFileReferences;
         try {
             validFileReferences = fileManager.cleanupAndGetValidFiles();
@@ -120,7 +119,7 @@
     @Override
     public synchronized void deactivate(boolean flushOnExit) throws HyracksDataException {
         super.deactivate(flushOnExit);
-        List<ILSMComponent> immutableComponents = componentsRef.get();
+        List<ILSMComponent> immutableComponents = diskComponents;
         for (ILSMComponent c : immutableComponents) {
             LSMRTreeImmutableComponent component = (LSMRTreeImmutableComponent) c;
             RTree rtree = component.getRTree();
@@ -141,7 +140,7 @@
     @Override
     public synchronized void destroy() throws HyracksDataException {
         super.destroy();
-        List<ILSMComponent> immutableComponents = componentsRef.get();
+        List<ILSMComponent> immutableComponents = diskComponents;
         for (ILSMComponent c : immutableComponents) {
             LSMRTreeImmutableComponent component = (LSMRTreeImmutableComponent) c;
             component.getBTree().destroy();
@@ -154,7 +153,7 @@
     @Override
     public synchronized void clear() throws HyracksDataException {
         super.clear();
-        List<ILSMComponent> immutableComponents = componentsRef.get();
+        List<ILSMComponent> immutableComponents = diskComponents;
         for (ILSMComponent c : immutableComponents) {
             LSMRTreeImmutableComponent component = (LSMRTreeImmutableComponent) c;
             component.getBTree().deactivate();
diff --git a/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTreeImmutableComponent.java b/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTreeImmutableComponent.java
index cb290c7..f6eea3f 100644
--- a/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTreeImmutableComponent.java
+++ b/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTreeImmutableComponent.java
@@ -17,10 +17,10 @@
 import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
 import edu.uci.ics.hyracks.storage.am.bloomfilter.impls.BloomFilter;
 import edu.uci.ics.hyracks.storage.am.btree.impls.BTree;
-import edu.uci.ics.hyracks.storage.am.lsm.common.impls.AbstractImmutableLSMComponent;
+import edu.uci.ics.hyracks.storage.am.lsm.common.impls.AbstractDiskLSMComponent;
 import edu.uci.ics.hyracks.storage.am.rtree.impls.RTree;
 
-public class LSMRTreeImmutableComponent extends AbstractImmutableLSMComponent {
+public class LSMRTreeImmutableComponent extends AbstractDiskLSMComponent {
     private final RTree rtree;
     private final BTree btree;
     private final BloomFilter bloomFilter;
diff --git a/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTreeMutableComponent.java b/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTreeMutableComponent.java
index e960ee0..0c7107b 100644
--- a/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTreeMutableComponent.java
+++ b/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTreeMutableComponent.java
@@ -18,10 +18,10 @@
 import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
 import edu.uci.ics.hyracks.storage.am.btree.impls.BTree;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.IVirtualBufferCache;
-import edu.uci.ics.hyracks.storage.am.lsm.common.impls.AbstractMutableLSMComponent;
+import edu.uci.ics.hyracks.storage.am.lsm.common.impls.AbstractMemoryLSMComponent;
 import edu.uci.ics.hyracks.storage.am.rtree.impls.RTree;
 
-public class LSMRTreeMutableComponent extends AbstractMutableLSMComponent {
+public class LSMRTreeMutableComponent extends AbstractMemoryLSMComponent {
 
     private final RTree rtree;
     private final BTree btree;
diff --git a/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTreeOpContext.java b/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTreeOpContext.java
index 28b1572..69fc2d9 100644
--- a/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTreeOpContext.java
+++ b/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTreeOpContext.java
@@ -51,7 +51,7 @@
     public final IModificationOperationCallback modificationCallback;
     public final ISearchOperationCallback searchCallback;
 
-    public LSMRTreeOpContext(List<LSMRTreeMutableComponent> mutableComponents, IRTreeLeafFrame rtreeLeafFrame,
+    public LSMRTreeOpContext(List<ILSMComponent> mutableComponents, IRTreeLeafFrame rtreeLeafFrame,
             IRTreeInteriorFrame rtreeInteriorFrame, ITreeIndexFrameFactory btreeLeafFrameFactory,
             ITreeIndexFrameFactory btreeInteriorFrameFactory, IBinaryComparatorFactory[] rtreeCmpFactories,
             IBinaryComparatorFactory[] btreeCmpFactories, IModificationOperationCallback modificationCallback,
@@ -62,10 +62,11 @@
         btreeOpContexts = new BTreeOpContext[mutableComponents.size()];
 
         for (int i = 0; i < mutableComponents.size(); i++) {
-            mutableRTreeAccessors[i] = (RTree.RTreeAccessor) mutableComponents.get(i).getRTree()
-                    .createAccessor(NoOpOperationCallback.INSTANCE, NoOpOperationCallback.INSTANCE);
-            mutableBTreeAccessors[i] = (BTree.BTreeAccessor) mutableComponents.get(i).getBTree()
-                    .createAccessor(NoOpOperationCallback.INSTANCE, NoOpOperationCallback.INSTANCE);
+            LSMRTreeMutableComponent mutableComponent = (LSMRTreeMutableComponent) mutableComponents.get(i);
+            mutableRTreeAccessors[i] = (RTree.RTreeAccessor) mutableComponent.getRTree().createAccessor(
+                    NoOpOperationCallback.INSTANCE, NoOpOperationCallback.INSTANCE);
+            mutableBTreeAccessors[i] = (BTree.BTreeAccessor) mutableComponent.getBTree().createAccessor(
+                    NoOpOperationCallback.INSTANCE, NoOpOperationCallback.INSTANCE);
 
             rtreeOpContexts[i] = mutableRTreeAccessors[i].getOpContext();
             btreeOpContexts[i] = mutableBTreeAccessors[i].getOpContext();
diff --git a/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTreeWithAntiMatterTuples.java b/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTreeWithAntiMatterTuples.java
index 965f39b..5340efd 100644
--- a/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTreeWithAntiMatterTuples.java
+++ b/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/rtree/impls/LSMRTreeWithAntiMatterTuples.java
@@ -87,7 +87,7 @@
     @Override
     public synchronized void activate() throws HyracksDataException {
         super.activate();
-        List<ILSMComponent> immutableComponents = componentsRef.get();
+        List<ILSMComponent> immutableComponents = diskComponents;
         immutableComponents.clear();
         List<LSMComponentFileReferences> validFileReferences;
         try {
@@ -111,7 +111,7 @@
     @Override
     public synchronized void deactivate(boolean flushOnExit) throws HyracksDataException {
         super.deactivate(flushOnExit);
-        List<ILSMComponent> immutableComponents = componentsRef.get();
+        List<ILSMComponent> immutableComponents = diskComponents;
         for (ILSMComponent c : immutableComponents) {
             RTree rtree = (RTree) ((LSMRTreeImmutableComponent) c).getRTree();
             rtree.deactivate();
@@ -127,7 +127,7 @@
     @Override
     public synchronized void destroy() throws HyracksDataException {
         super.destroy();
-        List<ILSMComponent> immutableComponents = componentsRef.get();
+        List<ILSMComponent> immutableComponents = diskComponents;
         for (ILSMComponent c : immutableComponents) {
             RTree rtree = (RTree) ((LSMRTreeImmutableComponent) c).getRTree();
             rtree.destroy();
@@ -138,7 +138,7 @@
     @Override
     public synchronized void clear() throws HyracksDataException {
         super.clear();
-        List<ILSMComponent> immutableComponents = componentsRef.get();
+        List<ILSMComponent> immutableComponents = diskComponents;
         for (ILSMComponent c : immutableComponents) {
             RTree rtree = (RTree) ((LSMRTreeImmutableComponent) c).getRTree();
             rtree.deactivate();