diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/bootstrap/TestNodeController.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/bootstrap/TestNodeController.java
index 90b9a1e..505ef8f 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/bootstrap/TestNodeController.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/bootstrap/TestNodeController.java
@@ -76,6 +76,8 @@
 import org.apache.hyracks.api.context.IHyracksTaskContext;
 import org.apache.hyracks.api.dataflow.ActivityId;
 import org.apache.hyracks.api.dataflow.OperatorDescriptorId;
+import org.apache.hyracks.api.dataflow.TaskAttemptId;
+import org.apache.hyracks.api.dataflow.TaskId;
 import org.apache.hyracks.api.dataflow.value.IRecordDescriptorProvider;
 import org.apache.hyracks.api.dataflow.value.ISerializerDeserializer;
 import org.apache.hyracks.api.dataflow.value.ITypeTraits;
@@ -121,11 +123,10 @@
     public static final int DEFAULT_HYRACKS_CC_CLIENT_PORT = 1098;
     public static final int DEFAULT_HYRACKS_CC_CLUSTER_PORT = 1099;
     public static final int KB32 = 32768;
-    public static final int PARTITION = 0;
     public static final double BLOOM_FILTER_FALSE_POSITIVE_RATE = 0.01;
     public static final TransactionSubsystemProvider TXN_SUBSYSTEM_PROVIDER = TransactionSubsystemProvider.INSTANCE;
     // Mutables
-    private long jobCounter = 0L;
+    private long jobCounter = 100L;
     private final String testConfigFileName;
     private final boolean runHDFS;
 
@@ -155,7 +156,11 @@
     }
 
     public TxnId getTxnJobId(IHyracksTaskContext ctx) {
-        return new TxnId(ctx.getJobletContext().getJobId().getId());
+        return getTxnJobId(ctx.getJobletContext().getJobId());
+    }
+
+    public TxnId getTxnJobId(JobId jobId) {
+        return new TxnId(jobId.getId());
     }
 
     public Pair<LSMInsertDeleteOperatorNodePushable, CommitRuntime> getInsertPipeline(IHyracksTaskContext ctx,
@@ -177,12 +182,14 @@
         IRecordDescriptorProvider recordDescProvider = primaryIndexInfo.getInsertRecordDescriptorProvider();
         IIndexDataflowHelperFactory indexHelperFactory = new IndexDataflowHelperFactory(
                 storageComponentProvider.getStorageManager(), primaryIndexInfo.getFileSplitProvider());
-        LSMInsertDeleteOperatorNodePushable insertOp = new LSMInsertDeleteOperatorNodePushable(ctx, PARTITION,
-                primaryIndexInfo.primaryIndexInsertFieldsPermutations,
-                recordDescProvider.getInputRecordDescriptor(new ActivityId(new OperatorDescriptorId(0), 0), 0), op,
-                true, indexHelperFactory, modOpCallbackFactory, null);
-        CommitRuntime commitOp = new CommitRuntime(ctx, getTxnJobId(ctx), dataset.getDatasetId(),
-                primaryIndexInfo.primaryKeyIndexes, false, true, PARTITION, true);
+        LSMInsertDeleteOperatorNodePushable insertOp =
+                new LSMInsertDeleteOperatorNodePushable(ctx, ctx.getTaskAttemptId().getTaskId().getPartition(),
+                        primaryIndexInfo.primaryIndexInsertFieldsPermutations,
+                        recordDescProvider.getInputRecordDescriptor(new ActivityId(new OperatorDescriptorId(0), 0), 0),
+                        op, true, indexHelperFactory, modOpCallbackFactory, null);
+        CommitRuntime commitOp =
+                new CommitRuntime(ctx, getTxnJobId(ctx), dataset.getDatasetId(), primaryIndexInfo.primaryKeyIndexes,
+                        false, true, ctx.getTaskAttemptId().getTaskId().getPartition(), true);
         insertOp.setOutputFrameWriter(0, commitOp, primaryIndexInfo.rDesc);
         commitOp.setInputRecordDescriptor(0, primaryIndexInfo.rDesc);
         return Pair.of(insertOp, commitOp);
@@ -203,7 +210,8 @@
                 null, null, true, true, indexDataflowHelperFactory, false, false, null,
                 NoOpOperationCallbackFactory.INSTANCE, filterFields, filterFields, false);
         BTreeSearchOperatorNodePushable searchOp =
-                searchOpDesc.createPushRuntime(ctx, primaryIndexInfo.getSearchRecordDescriptorProvider(), PARTITION, 1);
+                searchOpDesc.createPushRuntime(ctx, primaryIndexInfo.getSearchRecordDescriptorProvider(),
+                        ctx.getTaskAttemptId().getTaskId().getPartition(), 1);
         emptyTupleOp.setOutputFrameWriter(0, searchOp,
                 primaryIndexInfo.getSearchRecordDescriptorProvider().getInputRecordDescriptor(null, 0));
         searchOp.setOutputFrameWriter(0, countOp, primaryIndexInfo.rDesc);
@@ -236,7 +244,7 @@
 
     public PrimaryIndexInfo createPrimaryIndex(Dataset dataset, IAType[] primaryKeyTypes, ARecordType recordType,
             ARecordType metaType, int[] filterFields, IStorageComponentProvider storageComponentProvider,
-            int[] primaryKeyIndexes, List<Integer> primaryKeyIndicators)
+            int[] primaryKeyIndexes, List<Integer> primaryKeyIndicators, int partition)
             throws AlgebricksException, HyracksDataException, RemoteException, ACIDException {
         MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
         org.apache.hyracks.algebricks.common.utils.Pair<ILSMMergePolicyFactory, Map<String, String>> mergePolicy =
@@ -254,8 +262,8 @@
             IndexBuilderFactory indexBuilderFactory =
                     new IndexBuilderFactory(storageComponentProvider.getStorageManager(),
                             primaryIndexInfo.getFileSplitProvider(), resourceFactory, !dataset.isTemp());
-            IHyracksTaskContext ctx = createTestContext(false);
-            IIndexBuilder indexBuilder = indexBuilderFactory.create(ctx, 0);
+            IHyracksTaskContext ctx = createTestContext(newJobId(), partition, false);
+            IIndexBuilder indexBuilder = indexBuilderFactory.create(ctx, partition);
             indexBuilder.build();
         } finally {
             mdProvider.getLocks().unlock();
@@ -292,12 +300,12 @@
         return primaryIndexTypeTraits;
     }
 
-    public IHyracksTaskContext createTestContext(boolean withMessaging) throws HyracksDataException {
+    public IHyracksTaskContext createTestContext(JobId jobId, int partition, boolean withMessaging)
+            throws HyracksDataException {
         IHyracksTaskContext ctx = TestUtils.create(KB32);
         if (withMessaging) {
             TaskUtil.put(HyracksConstants.KEY_MESSAGE, new VSizeFrame(ctx), ctx);
         }
-        JobId jobId = newJobId();
         IHyracksJobletContext jobletCtx = Mockito.mock(IHyracksJobletContext.class);
         JobEventListenerFactory factory = new JobEventListenerFactory(new TxnId(jobId.getId()), true);
         Mockito.when(jobletCtx.getJobletEventListenerFactory()).thenReturn(factory);
@@ -306,6 +314,9 @@
         ctx = Mockito.spy(ctx);
         Mockito.when(ctx.getJobletContext()).thenReturn(jobletCtx);
         Mockito.when(ctx.getIoManager()).thenReturn(ExecutionTestUtil.integrationUtil.ncs[0].getIoManager());
+        TaskAttemptId taskId =
+                new TaskAttemptId(new TaskId(new ActivityId(new OperatorDescriptorId(0), 0), partition), 0);
+        Mockito.when(ctx.getTaskAttemptId()).thenReturn(taskId);
         return ctx;
     }
 
@@ -377,7 +388,7 @@
                     (CcApplicationContext) ExecutionTestUtil.integrationUtil.cc.getApplicationContext();
             FileSplit[] splits = SplitsAndConstraintsUtil.getIndexSplits(appCtx.getClusterStateManager(), dataset,
                     index.getIndexName(), nodes);
-            fileSplitProvider = new ConstantFileSplitProvider(Arrays.copyOfRange(splits, 0, 1));
+            fileSplitProvider = new ConstantFileSplitProvider(splits);
         }
 
         public Index getIndex() {
@@ -448,12 +459,13 @@
         IRecordDescriptorProvider recordDescProvider = primaryIndexInfo.getInsertRecordDescriptorProvider();
         IIndexDataflowHelperFactory indexHelperFactory = new IndexDataflowHelperFactory(
                 storageComponentProvider.getStorageManager(), primaryIndexInfo.getFileSplitProvider());
-        LSMPrimaryUpsertOperatorNodePushable insertOp = new LSMPrimaryUpsertOperatorNodePushable(ctx, PARTITION,
-                indexHelperFactory, primaryIndexInfo.primaryIndexInsertFieldsPermutations,
-                recordDescProvider.getInputRecordDescriptor(new ActivityId(new OperatorDescriptorId(0), 0), 0),
-                modificationCallbackFactory, searchCallbackFactory, keyIndexes.length, recordType, -1,
-                frameOpCallbackFactory == null ? dataset.getFrameOpCallbackFactory() : frameOpCallbackFactory,
-                MissingWriterFactory.INSTANCE, hasSecondaries);
+        LSMPrimaryUpsertOperatorNodePushable insertOp =
+                new LSMPrimaryUpsertOperatorNodePushable(ctx, ctx.getTaskAttemptId().getTaskId().getPartition(),
+                        indexHelperFactory, primaryIndexInfo.primaryIndexInsertFieldsPermutations,
+                        recordDescProvider.getInputRecordDescriptor(new ActivityId(new OperatorDescriptorId(0), 0), 0),
+                        modificationCallbackFactory, searchCallbackFactory, keyIndexes.length, recordType, -1,
+                        frameOpCallbackFactory == null ? dataset.getFrameOpCallbackFactory() : frameOpCallbackFactory,
+                        MissingWriterFactory.INSTANCE, hasSecondaries);
         RecordDescriptor upsertOutRecDesc = getUpsertOutRecDesc(primaryIndexInfo.rDesc, dataset,
                 filterFields == null ? 0 : filterFields.length, recordType, metaType);
         // fix pk fields
@@ -463,7 +475,7 @@
             pkFieldsInCommitOp[i] = diff + i;
         }
         CommitRuntime commitOp = new CommitRuntime(ctx, getTxnJobId(ctx), dataset.getDatasetId(), pkFieldsInCommitOp,
-                false, true, PARTITION, true);
+                false, true, ctx.getTaskAttemptId().getTaskId().getPartition(), true);
         insertOp.setOutputFrameWriter(0, commitOp, upsertOutRecDesc);
         commitOp.setInputRecordDescriptor(0, upsertOutRecDesc);
         return Pair.of(insertOp, commitOp);
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/ComponentRollbackTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/ComponentRollbackTest.java
index b09e000..1ac1aa6 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/ComponentRollbackTest.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/ComponentRollbackTest.java
@@ -58,6 +58,7 @@
 import org.apache.hyracks.api.context.IHyracksTaskContext;
 import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
 import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.api.job.JobId;
 import org.apache.hyracks.api.test.CountAnswer;
 import org.apache.hyracks.api.test.FrameWriterTestUtils;
 import org.apache.hyracks.api.test.FrameWriterTestUtils.FrameWriterOperation;
@@ -133,20 +134,21 @@
     public void createIndex() throws Exception {
         List<List<String>> partitioningKeys = new ArrayList<>();
         partitioningKeys.add(Collections.singletonList("key"));
+        int partition = 0;
         dataset = new TestDataset(DATAVERSE_NAME, DATASET_NAME, DATAVERSE_NAME, DATA_TYPE_NAME, NODE_GROUP_NAME,
                 NoMergePolicyFactory.NAME, null, new InternalDatasetDetails(null, PartitioningStrategy.HASH,
                         partitioningKeys, null, null, null, false, null, false),
                 null, DatasetType.INTERNAL, DATASET_ID, 0);
         PrimaryIndexInfo primaryIndexInfo = nc.createPrimaryIndex(dataset, KEY_TYPES, RECORD_TYPE, META_TYPE, null,
-                storageManager, KEY_INDEXES, KEY_INDICATORS_LIST);
+                storageManager, KEY_INDEXES, KEY_INDICATORS_LIST, partition);
         IndexDataflowHelperFactory iHelperFactory =
                 new IndexDataflowHelperFactory(nc.getStorageManager(), primaryIndexInfo.getFileSplitProvider());
-        ctx = nc.createTestContext(false);
-        indexDataflowHelper = iHelperFactory.create(ctx.getJobletContext().getServiceContext(), 0);
+        JobId jobId = nc.newJobId();
+        ctx = nc.createTestContext(jobId, partition, false);
+        indexDataflowHelper = iHelperFactory.create(ctx.getJobletContext().getServiceContext(), partition);
         indexDataflowHelper.open();
         lsmBtree = (TestLsmBtree) indexDataflowHelper.getIndexInstance();
         indexDataflowHelper.close();
-        nc.newJobId();
         txnCtx = nc.getTransactionManager().beginTransaction(nc.getTxnJobId(ctx),
                 new TransactionOptions(ITransactionManager.AtomicityLevel.ENTITY_LEVEL));
         insertOp = nc.getInsertPipeline(ctx, dataset, KEY_TYPES, RECORD_TYPE, META_TYPE, null, KEY_INDEXES,
@@ -158,7 +160,7 @@
         indexDataflowHelper.destroy();
     }
 
-    private void allowAllOps(TestLsmBtree lsmBtree) {
+    static void allowAllOps(TestLsmBtree lsmBtree) {
         lsmBtree.addModifyCallback(sem -> sem.release());
         lsmBtree.addFlushCallback(sem -> sem.release());
         lsmBtree.addSearchCallback(sem -> sem.release());
@@ -197,19 +199,17 @@
             List<ILSMDiskComponent> diskComponents = lsmBtree.getDiskComponents();
             Assert.assertEquals(9, diskComponents.size());
             Assert.assertTrue(memComponents.get(lsmBtree.getCurrentMemoryComponentIndex()).isModified());
-            searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
+            searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
             ILSMIndexAccessor lsmAccessor = lsmBtree.createAccessor(NoOpIndexAccessParameters.INSTANCE);
-            dsLifecycleMgr.getComponentIdGenerator(DATASET_ID).refresh();
             // rollback a memory component
             lsmAccessor.deleteComponents(memoryComponentsPredicate);
-            searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS - RECORDS_PER_COMPONENT);
+            searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS - RECORDS_PER_COMPONENT);
             // rollback the last disk component
             lsmAccessor = lsmBtree.createAccessor(NoOpIndexAccessParameters.INSTANCE);
             long lsn = AbstractLSMIOOperationCallback.getTreeIndexLSN(diskComponents.get(0).getMetadata());
             DiskComponentLsnPredicate pred = new DiskComponentLsnPredicate(lsn);
-            dsLifecycleMgr.getComponentIdGenerator(DATASET_ID).refresh();
             lsmAccessor.deleteComponents(pred);
-            searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS - (2 * RECORDS_PER_COMPONENT));
+            searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS - (2 * RECORDS_PER_COMPONENT));
         } catch (Throwable e) {
             e.printStackTrace();
             Assert.fail(e.getMessage());
@@ -247,12 +247,11 @@
             List<ILSMDiskComponent> diskComponents = lsmBtree.getDiskComponents();
             Assert.assertEquals(9, diskComponents.size());
             Assert.assertTrue(memComponents.get(lsmBtree.getCurrentMemoryComponentIndex()).isModified());
-            searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
+            searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
             ILSMIndexAccessor lsmAccessor = lsmBtree.createAccessor(NoOpIndexAccessParameters.INSTANCE);
-            dsLifecycleMgr.getComponentIdGenerator(DATASET_ID).refresh();
             // rollback a memory component
             lsmAccessor.deleteComponents(memoryComponentsPredicate);
-            searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS - RECORDS_PER_COMPONENT);
+            searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS - RECORDS_PER_COMPONENT);
 
             // insert again
             nc.newJobId();
@@ -270,14 +269,13 @@
             }
             insertOp.close();
             nc.getTransactionManager().commitTransaction(txnCtx.getTxnId());
-            searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
+            searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
             // rollback the last disk component
             lsmAccessor = lsmBtree.createAccessor(NoOpIndexAccessParameters.INSTANCE);
             long lsn = AbstractLSMIOOperationCallback.getTreeIndexLSN(diskComponents.get(0).getMetadata());
             DiskComponentLsnPredicate pred = new DiskComponentLsnPredicate(lsn);
-            dsLifecycleMgr.getComponentIdGenerator(DATASET_ID).refresh();
             lsmAccessor.deleteComponents(pred);
-            searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS - (2 * RECORDS_PER_COMPONENT));
+            searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS - (2 * RECORDS_PER_COMPONENT));
         } catch (Throwable e) {
             e.printStackTrace();
             Assert.fail(e.getMessage());
@@ -317,12 +315,11 @@
             List<ILSMDiskComponent> diskComponents = lsmBtree.getDiskComponents();
             Assert.assertEquals(9, diskComponents.size());
             Assert.assertTrue(memComponents.get(lsmBtree.getCurrentMemoryComponentIndex()).isModified());
-            Searcher firstSearcher = new Searcher(nc, ctx, dataset, storageManager, lsmBtree, TOTAL_NUM_OF_RECORDS);
+            Searcher firstSearcher = new Searcher(nc, 0, dataset, storageManager, lsmBtree, TOTAL_NUM_OF_RECORDS);
             // wait till firstSearcher enter the components
             firstSearcher.waitUntilEntered();
             // now that we enetered, we will rollback
             ILSMIndexAccessor lsmAccessor = lsmBtree.createAccessor(NoOpIndexAccessParameters.INSTANCE);
-            dsLifecycleMgr.getComponentIdGenerator(DATASET_ID).refresh();
             // rollback a memory component
             lsmAccessor.deleteComponents(
                     c -> (c instanceof ILSMMemoryComponent && ((ILSMMemoryComponent) c).isModified()));
@@ -331,24 +328,23 @@
             lsmBtree.allowSearch(1);
             Assert.assertTrue(firstSearcher.result());
             // search now and ensure
-            searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS - RECORDS_PER_COMPONENT);
+            searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS - RECORDS_PER_COMPONENT);
             // rollback the last disk component
             // re-block searches
             lsmBtree.clearSearchCallbacks();
-            Searcher secondSearcher = new Searcher(nc, ctx, dataset, storageManager, lsmBtree,
+            Searcher secondSearcher = new Searcher(nc, 0, dataset, storageManager, lsmBtree,
                     TOTAL_NUM_OF_RECORDS - RECORDS_PER_COMPONENT);
             // wait till firstSearcher enter the components
             secondSearcher.waitUntilEntered();
             lsmAccessor = lsmBtree.createAccessor(NoOpIndexAccessParameters.INSTANCE);
             long lsn = AbstractLSMIOOperationCallback.getTreeIndexLSN(diskComponents.get(0).getMetadata());
             DiskComponentLsnPredicate pred = new DiskComponentLsnPredicate(lsn);
-            dsLifecycleMgr.getComponentIdGenerator(DATASET_ID).refresh();
             lsmAccessor.deleteComponents(pred);
             // now that the rollback has completed, we will unblock the search
             lsmBtree.addSearchCallback(sem -> sem.release());
             lsmBtree.allowSearch(1);
             Assert.assertTrue(secondSearcher.result());
-            searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS - (2 * RECORDS_PER_COMPONENT));
+            searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS - (2 * RECORDS_PER_COMPONENT));
         } catch (Throwable e) {
             e.printStackTrace();
             Assert.fail(e.getMessage());
@@ -386,7 +382,7 @@
             List<ILSMDiskComponent> diskComponents = lsmBtree.getDiskComponents();
             Assert.assertEquals(9, diskComponents.size());
             Assert.assertTrue(memComponents.get(lsmBtree.getCurrentMemoryComponentIndex()).isModified());
-            searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
+            searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
             // disable flushes
             lsmBtree.clearFlushCallbacks();
             Flusher firstFlusher = new Flusher(lsmBtree);
@@ -395,7 +391,7 @@
             // now that we enetered, we will rollback. This will not proceed since it is waiting for the flush to complete
             Rollerback rollerback = new Rollerback(lsmBtree, memoryComponentsPredicate);
             // now that the rollback has completed, we will search
-            searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
+            searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
             //unblock the flush
             lsmBtree.allowFlush(1);
             // ensure rollback completed
@@ -403,7 +399,7 @@
             // ensure current mem component is not modified
             Assert.assertFalse(memComponents.get(lsmBtree.getCurrentMemoryComponentIndex()).isModified());
             // search now and ensure
-            searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
+            searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
         } catch (Throwable e) {
             e.printStackTrace();
             Assert.fail(e.getMessage());
@@ -442,7 +438,7 @@
             List<ILSMDiskComponent> diskComponents = lsmBtree.getDiskComponents();
             Assert.assertEquals(9, diskComponents.size());
             Assert.assertTrue(memComponents.get(lsmBtree.getCurrentMemoryComponentIndex()).isModified());
-            searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
+            searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
             // Now, we will start a full merge
             Merger merger = new Merger(lsmBtree);
             ILSMIndexAccessor mergeAccessor = lsmBtree.createAccessor(NoOpIndexAccessParameters.INSTANCE);
@@ -459,7 +455,7 @@
             Rollerback rollerback = new Rollerback(lsmBtree, new DiskComponentLsnPredicate(lsn));
             // rollback is now waiting for the merge to complete
             // we will search
-            searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
+            searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
             //unblock the merge
             lsmBtree.allowMerge(1);
             // ensure rollback completes
@@ -467,7 +463,7 @@
             // ensure current mem component is not modified
             Assert.assertFalse(memComponents.get(lsmBtree.getCurrentMemoryComponentIndex()).isModified());
             // search now and ensure that we rolled back the merged component
-            searchAndAssertCount(nc, ctx, dataset, storageManager,
+            searchAndAssertCount(nc, 0, dataset, storageManager,
                     TOTAL_NUM_OF_RECORDS - ((numMergedComponents + 1/*memory component*/) * RECORDS_PER_COMPONENT));
         } catch (Throwable e) {
             e.printStackTrace();
@@ -506,7 +502,7 @@
             List<ILSMDiskComponent> diskComponents = lsmBtree.getDiskComponents();
             Assert.assertEquals(9, diskComponents.size());
             Assert.assertTrue(memComponents.get(lsmBtree.getCurrentMemoryComponentIndex()).isModified());
-            searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
+            searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
             // disable flushes
             // disable searches
             lsmBtree.clearFlushCallbacks();
@@ -514,7 +510,7 @@
             Flusher firstFlusher = new Flusher(lsmBtree);
             dsLifecycleMgr.flushDataset(dataset.getDatasetId(), true);
             firstFlusher.waitUntilCount(1);
-            Searcher firstSearcher = new Searcher(nc, ctx, dataset, storageManager, lsmBtree, TOTAL_NUM_OF_RECORDS);
+            Searcher firstSearcher = new Searcher(nc, 0, dataset, storageManager, lsmBtree, TOTAL_NUM_OF_RECORDS);
             // wait till firstSearcher enter the components
             firstSearcher.waitUntilEntered();
             // now that we enetered, we will rollback rollback a memory component
@@ -528,7 +524,7 @@
             rollerback.complete();
             Assert.assertFalse(memComponents.get(lsmBtree.getCurrentMemoryComponentIndex()).isModified());
             // search now and ensure the rollback was no op since it waits for ongoing flushes
-            searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
+            searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
         } catch (Throwable e) {
             e.printStackTrace();
             Assert.fail(e.getMessage());
@@ -566,7 +562,7 @@
             List<ILSMDiskComponent> diskComponents = lsmBtree.getDiskComponents();
             Assert.assertEquals(9, diskComponents.size());
             Assert.assertTrue(memComponents.get(lsmBtree.getCurrentMemoryComponentIndex()).isModified());
-            searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
+            searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
             // disable flushes
             // disable searches
             lsmBtree.clearFlushCallbacks();
@@ -574,7 +570,7 @@
             dsLifecycleMgr.flushDataset(dataset.getDatasetId(), true);
             firstFlusher.waitUntilCount(1);
             lsmBtree.clearSearchCallbacks();
-            Searcher firstSearcher = new Searcher(nc, ctx, dataset, storageManager, lsmBtree, TOTAL_NUM_OF_RECORDS);
+            Searcher firstSearcher = new Searcher(nc, 0, dataset, storageManager, lsmBtree, TOTAL_NUM_OF_RECORDS);
             // wait till firstSearcher enter the components
             firstSearcher.waitUntilEntered();
             // now that we enetered, we will rollback
@@ -589,7 +585,7 @@
             rollerback.complete();
             Assert.assertFalse(memComponents.get(lsmBtree.getCurrentMemoryComponentIndex()).isModified());
             // search now and ensure
-            searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
+            searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
         } catch (Throwable e) {
             e.printStackTrace();
             Assert.fail(e.getMessage());
@@ -628,7 +624,7 @@
             List<ILSMDiskComponent> diskComponents = lsmBtree.getDiskComponents();
             Assert.assertEquals(9, diskComponents.size());
             Assert.assertTrue(memComponents.get(lsmBtree.getCurrentMemoryComponentIndex()).isModified());
-            searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
+            searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
             // Now, we will start a merge
             Merger merger = new Merger(lsmBtree);
             ILSMIndexAccessor mergeAccessor = lsmBtree.createAccessor(NoOpIndexAccessParameters.INSTANCE);
@@ -643,7 +639,7 @@
             merger.waitUntilCount(1);
             // we will block search
             lsmBtree.clearSearchCallbacks();
-            Searcher firstSearcher = new Searcher(nc, ctx, dataset, storageManager, lsmBtree, TOTAL_NUM_OF_RECORDS);
+            Searcher firstSearcher = new Searcher(nc, 0, dataset, storageManager, lsmBtree, TOTAL_NUM_OF_RECORDS);
             // wait till firstSearcher enter the components
             firstSearcher.waitUntilEntered();
             // now that we enetered, we will rollback
@@ -657,7 +653,7 @@
             Assert.assertTrue(firstSearcher.result());
             rollerback.complete();
             // now that the rollback has completed, we will search
-            searchAndAssertCount(nc, ctx, dataset, storageManager,
+            searchAndAssertCount(nc, 0, dataset, storageManager,
                     TOTAL_NUM_OF_RECORDS - ((numMergedComponents + 1/*memory component*/) * RECORDS_PER_COMPONENT));
             // ensure current mem component is not modified
             Assert.assertFalse(memComponents.get(lsmBtree.getCurrentMemoryComponentIndex()).isModified());
@@ -699,7 +695,7 @@
             List<ILSMDiskComponent> diskComponents = lsmBtree.getDiskComponents();
             Assert.assertEquals(9, diskComponents.size());
             Assert.assertTrue(memComponents.get(lsmBtree.getCurrentMemoryComponentIndex()).isModified());
-            searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
+            searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
             // Now, we will start a merge
             Merger merger = new Merger(lsmBtree);
             ILSMIndexAccessor mergeAccessor = lsmBtree.createAccessor(NoOpIndexAccessParameters.INSTANCE);
@@ -714,7 +710,7 @@
             merger.waitUntilCount(1);
             // we will block search
             lsmBtree.clearSearchCallbacks();
-            Searcher firstSearcher = new Searcher(nc, ctx, dataset, storageManager, lsmBtree, TOTAL_NUM_OF_RECORDS);
+            Searcher firstSearcher = new Searcher(nc, 0, dataset, storageManager, lsmBtree, TOTAL_NUM_OF_RECORDS);
             // wait till firstSearcher enter the components
             firstSearcher.waitUntilEntered();
             // now that we enetered, we will rollback
@@ -724,11 +720,11 @@
             lsmBtree.allowSearch(1);
             Assert.assertTrue(firstSearcher.result());
             // even though rollback has been called, it is still waiting for the merge to complete
-            searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
+            searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
             //unblock the merge
             lsmBtree.allowMerge(1);
             rollerBack.complete();
-            searchAndAssertCount(nc, ctx, dataset, storageManager,
+            searchAndAssertCount(nc, 0, dataset, storageManager,
                     TOTAL_NUM_OF_RECORDS - ((numMergedComponents + 1/*memory component*/) * RECORDS_PER_COMPONENT));
             // ensure current mem component is not modified
             Assert.assertFalse(memComponents.get(lsmBtree.getCurrentMemoryComponentIndex()).isModified());
@@ -748,7 +744,6 @@
                 @Override
                 public void run() {
                     ILSMIndexAccessor lsmAccessor = lsmBtree.createAccessor(NoOpIndexAccessParameters.INSTANCE);
-                    dsLifecycleMgr.getComponentIdGenerator(DATASET_ID).refresh();
                     try {
                         lsmAccessor.deleteComponents(predicate);
                     } catch (HyracksDataException e) {
@@ -768,13 +763,13 @@
         }
     }
 
-    private class Searcher {
+    static class Searcher {
         private final ExecutorService executor = Executors.newSingleThreadExecutor();
         private Future<Boolean> task;
         private volatile boolean entered = false;
 
-        public Searcher(TestNodeController nc, IHyracksTaskContext ctx, Dataset dataset,
-                StorageComponentProvider storageManager, TestLsmBtree lsmBtree, int numOfRecords) {
+        public Searcher(TestNodeController nc, int partition, Dataset dataset, StorageComponentProvider storageManager,
+                TestLsmBtree lsmBtree, int numOfRecords) {
             lsmBtree.addSearchCallback(sem -> {
                 synchronized (Searcher.this) {
                     entered = true;
@@ -784,7 +779,7 @@
             Callable<Boolean> callable = new Callable<Boolean>() {
                 @Override
                 public Boolean call() throws Exception {
-                    searchAndAssertCount(nc, ctx, dataset, storageManager, numOfRecords);
+                    searchAndAssertCount(nc, partition, dataset, storageManager, numOfRecords);
                     return true;
                 }
             };
@@ -840,7 +835,7 @@
         }
     }
 
-    private class DiskComponentLsnPredicate implements Predicate<ILSMComponent> {
+    private static class DiskComponentLsnPredicate implements Predicate<ILSMComponent> {
         private final long lsn;
 
         public DiskComponentLsnPredicate(long lsn) {
@@ -860,10 +855,11 @@
         }
     }
 
-    private void searchAndAssertCount(TestNodeController nc, IHyracksTaskContext ctx, Dataset dataset,
+    static void searchAndAssertCount(TestNodeController nc, int partition, Dataset dataset,
             StorageComponentProvider storageManager, int numOfRecords)
             throws HyracksDataException, AlgebricksException {
-        nc.newJobId();
+        JobId jobId = nc.newJobId();
+        IHyracksTaskContext ctx = nc.createTestContext(jobId, partition, false);
         TestTupleCounterFrameWriter countOp = create(nc.getSearchOutputDesc(KEY_TYPES, RECORD_TYPE, META_TYPE),
                 Collections.emptyList(), Collections.emptyList(), false);
         IPushRuntime emptyTupleOp = nc.getFullScanPipeline(countOp, ctx, dataset, KEY_TYPES, RECORD_TYPE, META_TYPE,
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/LogMarkerTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/LogMarkerTest.java
index 1907ed6..aa08c2d 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/LogMarkerTest.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/LogMarkerTest.java
@@ -50,6 +50,7 @@
 import org.apache.hyracks.api.comm.VSizeFrame;
 import org.apache.hyracks.api.context.IHyracksTaskContext;
 import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
+import org.apache.hyracks.api.job.JobId;
 import org.apache.hyracks.api.test.CountAnswer;
 import org.apache.hyracks.api.test.FrameWriterTestUtils;
 import org.apache.hyracks.api.test.FrameWriterTestUtils.FrameWriterOperation;
@@ -117,9 +118,9 @@
                     null, DatasetType.INTERNAL, DATASET_ID, 0);
             try {
                 PrimaryIndexInfo indexInfo = nc.createPrimaryIndex(dataset, KEY_TYPES, RECORD_TYPE, META_TYPE, null,
-                        storageManager, KEY_INDEXES, KEY_INDICATORS_LIST);
-                IHyracksTaskContext ctx = nc.createTestContext(true);
-                nc.newJobId();
+                        storageManager, KEY_INDEXES, KEY_INDICATORS_LIST, 0);
+                JobId jobId = nc.newJobId();
+                IHyracksTaskContext ctx = nc.createTestContext(jobId, 0, true);
                 ITransactionContext txnCtx = nc.getTransactionManager().beginTransaction(nc.getTxnJobId(ctx),
                         new TransactionOptions(ITransactionManager.AtomicityLevel.ENTITY_LEVEL));
                 LSMInsertDeleteOperatorNodePushable insertOp = nc.getInsertPipeline(ctx, dataset, KEY_TYPES,
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/MultiPartitionLSMIndexTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/MultiPartitionLSMIndexTest.java
new file mode 100644
index 0000000..ec9c2f6
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/MultiPartitionLSMIndexTest.java
@@ -0,0 +1,196 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.asterix.test.dataflow;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.asterix.app.bootstrap.TestNodeController;
+import org.apache.asterix.app.bootstrap.TestNodeController.PrimaryIndexInfo;
+import org.apache.asterix.app.data.gen.TupleGenerator;
+import org.apache.asterix.app.data.gen.TupleGenerator.GenerationFunction;
+import org.apache.asterix.app.nc.NCAppRuntimeContext;
+import org.apache.asterix.common.api.IDatasetLifecycleManager;
+import org.apache.asterix.common.config.DatasetConfig.DatasetType;
+import org.apache.asterix.common.dataflow.LSMInsertDeleteOperatorNodePushable;
+import org.apache.asterix.common.transactions.ITransactionContext;
+import org.apache.asterix.common.transactions.ITransactionManager;
+import org.apache.asterix.common.transactions.TransactionOptions;
+import org.apache.asterix.external.util.DataflowUtils;
+import org.apache.asterix.file.StorageComponentProvider;
+import org.apache.asterix.metadata.entities.Dataset;
+import org.apache.asterix.metadata.entities.Index;
+import org.apache.asterix.metadata.entities.InternalDatasetDetails;
+import org.apache.asterix.metadata.entities.InternalDatasetDetails.PartitioningStrategy;
+import org.apache.asterix.om.types.ARecordType;
+import org.apache.asterix.om.types.BuiltinType;
+import org.apache.asterix.om.types.IAType;
+import org.apache.asterix.test.common.TestHelper;
+import org.apache.hyracks.api.comm.VSizeFrame;
+import org.apache.hyracks.api.context.IHyracksTaskContext;
+import org.apache.hyracks.api.job.JobId;
+import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAppender;
+import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
+import org.apache.hyracks.storage.am.common.api.IIndexDataflowHelper;
+import org.apache.hyracks.storage.am.common.dataflow.IndexDataflowHelperFactory;
+import org.apache.hyracks.storage.am.lsm.btree.impl.TestLsmBtree;
+import org.apache.hyracks.storage.am.lsm.common.impls.NoMergePolicyFactory;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+public class MultiPartitionLSMIndexTest {
+    private static final IAType[] KEY_TYPES = { BuiltinType.AINT32 };
+    private static final ARecordType RECORD_TYPE = new ARecordType("TestRecordType", new String[] { "key", "value" },
+            new IAType[] { BuiltinType.AINT32, BuiltinType.AINT64 }, false);
+    private static final GenerationFunction[] RECORD_GEN_FUNCTION =
+            { GenerationFunction.DETERMINISTIC, GenerationFunction.DETERMINISTIC };
+    private static final boolean[] UNIQUE_RECORD_FIELDS = { true, false };
+    private static final ARecordType META_TYPE = null;
+    private static final GenerationFunction[] META_GEN_FUNCTION = null;
+    private static final boolean[] UNIQUE_META_FIELDS = null;
+    private static final int[] KEY_INDEXES = { 0 };
+    private static final int[] KEY_INDICATORS = { Index.RECORD_INDICATOR };
+    private static final List<Integer> KEY_INDICATORS_LIST = Arrays.asList(new Integer[] { Index.RECORD_INDICATOR });
+    private static final int TOTAL_NUM_OF_RECORDS = 10000;
+    private static final int RECORDS_PER_COMPONENT = 1000;
+    private static final int DATASET_ID = 101;
+    private static final String DATAVERSE_NAME = "TestDV";
+    private static final String DATASET_NAME = "TestDS";
+    private static final String DATA_TYPE_NAME = "DUMMY";
+    private static final String NODE_GROUP_NAME = "DEFAULT";
+    private static final StorageComponentProvider storageManager = new StorageComponentProvider();
+    private static final int NUM_PARTITIONS = 2;
+    private static TestNodeController nc;
+    private static NCAppRuntimeContext ncAppCtx;
+    private static IDatasetLifecycleManager dsLifecycleMgr;
+    private static Dataset dataset;
+    private static ITransactionContext txnCtx;
+    private static TestLsmBtree[] primarylsmBtrees;
+    private static IHyracksTaskContext[] taskCtxs;
+    private static IIndexDataflowHelper[] indexDataflowHelpers;
+    private static LSMInsertDeleteOperatorNodePushable[] insertOps;
+
+    @BeforeClass
+    public static void setUp() throws Exception {
+        System.out.println("SetUp: ");
+        TestHelper.deleteExistingInstanceFiles();
+        nc = new TestNodeController(null, false);
+        nc.init();
+        ncAppCtx = nc.getAppRuntimeContext();
+        dsLifecycleMgr = ncAppCtx.getDatasetLifecycleManager();
+    }
+
+    @AfterClass
+    public static void tearDown() throws Exception {
+        System.out.println("TearDown");
+        nc.deInit();
+        TestHelper.deleteExistingInstanceFiles();
+    }
+
+    @Before
+    public void createIndex() throws Exception {
+        List<List<String>> partitioningKeys = new ArrayList<>();
+        partitioningKeys.add(Collections.singletonList("key"));
+        dataset = new TestDataset(DATAVERSE_NAME, DATASET_NAME, DATAVERSE_NAME, DATA_TYPE_NAME, NODE_GROUP_NAME,
+                NoMergePolicyFactory.NAME, null, new InternalDatasetDetails(null, PartitioningStrategy.HASH,
+                        partitioningKeys, null, null, null, false, null, false),
+                null, DatasetType.INTERNAL, DATASET_ID, 0);
+        taskCtxs = new IHyracksTaskContext[NUM_PARTITIONS];
+        indexDataflowHelpers = new IIndexDataflowHelper[NUM_PARTITIONS];
+        primarylsmBtrees = new TestLsmBtree[NUM_PARTITIONS];
+        insertOps = new LSMInsertDeleteOperatorNodePushable[NUM_PARTITIONS];
+        JobId jobId = nc.newJobId();
+        txnCtx = nc.getTransactionManager().beginTransaction(nc.getTxnJobId(jobId),
+                new TransactionOptions(ITransactionManager.AtomicityLevel.ENTITY_LEVEL));
+        for (int i = 0; i < taskCtxs.length; i++) {
+            taskCtxs[i] = nc.createTestContext(jobId, i, false);
+            PrimaryIndexInfo primaryIndexInfo = nc.createPrimaryIndex(dataset, KEY_TYPES, RECORD_TYPE, META_TYPE, null,
+                    storageManager, KEY_INDEXES, KEY_INDICATORS_LIST, i);
+            IndexDataflowHelperFactory iHelperFactory =
+                    new IndexDataflowHelperFactory(nc.getStorageManager(), primaryIndexInfo.getFileSplitProvider());
+            indexDataflowHelpers[i] = iHelperFactory.create(taskCtxs[i].getJobletContext().getServiceContext(), i);
+            indexDataflowHelpers[i].open();
+            primarylsmBtrees[i] = (TestLsmBtree) indexDataflowHelpers[i].getIndexInstance();
+            indexDataflowHelpers[i].close();
+            insertOps[i] = nc.getInsertPipeline(taskCtxs[i], dataset, KEY_TYPES, RECORD_TYPE, META_TYPE, null,
+                    KEY_INDEXES, KEY_INDICATORS_LIST, storageManager).getLeft();
+        }
+    }
+
+    @After
+    public void destroyIndex() throws Exception {
+        for (IIndexDataflowHelper indexDataflowHelper : indexDataflowHelpers) {
+            indexDataflowHelper.destroy();
+        }
+    }
+
+    @Test
+    public void testFlushOneFullOneEmpty() {
+        try {
+            // allow all operations
+            for (int i = 0; i < NUM_PARTITIONS; i++) {
+                ComponentRollbackTest.allowAllOps(primarylsmBtrees[i]);
+            }
+
+            insertOps[0].open();
+            TupleGenerator tupleGenerator = new TupleGenerator(RECORD_TYPE, META_TYPE, KEY_INDEXES, KEY_INDICATORS,
+                    RECORD_GEN_FUNCTION, UNIQUE_RECORD_FIELDS, META_GEN_FUNCTION, UNIQUE_META_FIELDS);
+            VSizeFrame frame = new VSizeFrame(taskCtxs[0]);
+            FrameTupleAppender tupleAppender = new FrameTupleAppender(frame);
+            int numFlushes = 0;
+            for (int j = 0; j < TOTAL_NUM_OF_RECORDS; j++) {
+                // flush every RECORDS_PER_COMPONENT records
+                if (j % RECORDS_PER_COMPONENT == (RECORDS_PER_COMPONENT - 1) && j + 1 != TOTAL_NUM_OF_RECORDS) {
+                    if (tupleAppender.getTupleCount() > 0) {
+                        tupleAppender.write(insertOps[0], true);
+                    }
+                    dsLifecycleMgr.flushDataset(dataset.getDatasetId(), false);
+                    numFlushes++;
+                }
+                ITupleReference tuple = tupleGenerator.next();
+                DataflowUtils.addTupleToFrame(tupleAppender, tuple, insertOps[0]);
+            }
+            if (tupleAppender.getTupleCount() > 0) {
+                tupleAppender.write(insertOps[0], true);
+            }
+            insertOps[0].close();
+            dsLifecycleMgr.flushDataset(dataset.getDatasetId(), false);
+            numFlushes++;
+            nc.getTransactionManager().commitTransaction(txnCtx.getTxnId());
+            // search now and ensure partition 0 has all the records
+            ComponentRollbackTest.searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
+            // and that partition 1 has no records
+            ComponentRollbackTest.searchAndAssertCount(nc, 1, dataset, storageManager, 0);
+            // and that partition 0 has numFlushes disk components
+            Assert.assertEquals(numFlushes, primarylsmBtrees[0].getDiskComponents().size());
+            // and that partition 1 has no disk components
+            Assert.assertEquals(0, primarylsmBtrees[1].getDiskComponents().size());
+        } catch (Throwable e) {
+            e.printStackTrace();
+            Assert.fail(e.getMessage());
+        }
+    }
+
+}
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/SearchCursorComponentSwitchTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/SearchCursorComponentSwitchTest.java
new file mode 100644
index 0000000..77d3795
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/SearchCursorComponentSwitchTest.java
@@ -0,0 +1,266 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.asterix.test.dataflow;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.asterix.app.bootstrap.TestNodeController;
+import org.apache.asterix.app.bootstrap.TestNodeController.PrimaryIndexInfo;
+import org.apache.asterix.app.data.gen.TestTupleCounterFrameWriter;
+import org.apache.asterix.app.data.gen.TupleGenerator;
+import org.apache.asterix.app.data.gen.TupleGenerator.GenerationFunction;
+import org.apache.asterix.app.nc.NCAppRuntimeContext;
+import org.apache.asterix.common.api.IDatasetLifecycleManager;
+import org.apache.asterix.common.config.DatasetConfig.DatasetType;
+import org.apache.asterix.common.dataflow.LSMInsertDeleteOperatorNodePushable;
+import org.apache.asterix.common.transactions.ITransactionContext;
+import org.apache.asterix.common.transactions.ITransactionManager;
+import org.apache.asterix.common.transactions.TransactionOptions;
+import org.apache.asterix.external.util.DataflowUtils;
+import org.apache.asterix.file.StorageComponentProvider;
+import org.apache.asterix.metadata.entities.Dataset;
+import org.apache.asterix.metadata.entities.Index;
+import org.apache.asterix.metadata.entities.InternalDatasetDetails;
+import org.apache.asterix.metadata.entities.InternalDatasetDetails.PartitioningStrategy;
+import org.apache.asterix.om.types.ARecordType;
+import org.apache.asterix.om.types.BuiltinType;
+import org.apache.asterix.om.types.IAType;
+import org.apache.asterix.test.common.TestHelper;
+import org.apache.asterix.test.dataflow.ComponentRollbackTest.Searcher;
+import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
+import org.apache.hyracks.algebricks.runtime.base.IPushRuntime;
+import org.apache.hyracks.api.comm.VSizeFrame;
+import org.apache.hyracks.api.context.IHyracksTaskContext;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.api.job.JobId;
+import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAppender;
+import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
+import org.apache.hyracks.storage.am.common.api.IIndexDataflowHelper;
+import org.apache.hyracks.storage.am.common.dataflow.IndexDataflowHelperFactory;
+import org.apache.hyracks.storage.am.common.impls.NoOpIndexAccessParameters;
+import org.apache.hyracks.storage.am.lsm.btree.impl.TestLsmBtree;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMDiskComponent;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexAccessor;
+import org.apache.hyracks.storage.am.lsm.common.impls.BlockingIOOperationCallbackWrapper;
+import org.apache.hyracks.storage.am.lsm.common.impls.NoMergePolicyFactory;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+public class SearchCursorComponentSwitchTest {
+    private static final IAType[] KEY_TYPES = { BuiltinType.AINT32 };
+    private static final ARecordType RECORD_TYPE = new ARecordType("TestRecordType", new String[] { "key", "value" },
+            new IAType[] { BuiltinType.AINT32, BuiltinType.AINT64 }, false);
+    private static final GenerationFunction[] RECORD_GEN_FUNCTION =
+            { GenerationFunction.DETERMINISTIC, GenerationFunction.DETERMINISTIC };
+    private static final boolean[] UNIQUE_RECORD_FIELDS = { true, false };
+    private static final ARecordType META_TYPE = null;
+    private static final GenerationFunction[] META_GEN_FUNCTION = null;
+    private static final boolean[] UNIQUE_META_FIELDS = null;
+    private static final int[] KEY_INDEXES = { 0 };
+    private static final int[] KEY_INDICATORS = { Index.RECORD_INDICATOR };
+    private static final List<Integer> KEY_INDICATORS_LIST = Arrays.asList(new Integer[] { Index.RECORD_INDICATOR });
+    private static final int TOTAL_NUM_OF_RECORDS = 2000;
+    private static final int RECORDS_PER_COMPONENT = 1000;
+    private static final int DATASET_ID = 101;
+    private static final String DATAVERSE_NAME = "TestDV";
+    private static final String DATASET_NAME = "TestDS";
+    private static final String DATA_TYPE_NAME = "DUMMY";
+    private static final String NODE_GROUP_NAME = "DEFAULT";
+    private static final StorageComponentProvider storageManager = new StorageComponentProvider();
+    private static TestNodeController nc;
+    private static TestLsmBtree lsmBtree;
+    private static NCAppRuntimeContext ncAppCtx;
+    private static IDatasetLifecycleManager dsLifecycleMgr;
+    private static Dataset dataset;
+    private static IHyracksTaskContext ctx;
+    private static IIndexDataflowHelper indexDataflowHelper;
+    private static ITransactionContext txnCtx;
+    private static LSMInsertDeleteOperatorNodePushable insertOp;
+
+    @BeforeClass
+    public static void setUp() throws Exception {
+        System.out.println("SetUp: ");
+        TestHelper.deleteExistingInstanceFiles();
+        nc = new TestNodeController(null, false);
+        nc.init();
+        ncAppCtx = nc.getAppRuntimeContext();
+        dsLifecycleMgr = ncAppCtx.getDatasetLifecycleManager();
+    }
+
+    @AfterClass
+    public static void tearDown() throws Exception {
+        System.out.println("TearDown");
+        nc.deInit();
+        TestHelper.deleteExistingInstanceFiles();
+    }
+
+    @Before
+    public void createIndex() throws Exception {
+        List<List<String>> partitioningKeys = new ArrayList<>();
+        partitioningKeys.add(Collections.singletonList("key"));
+        dataset = new TestDataset(DATAVERSE_NAME, DATASET_NAME, DATAVERSE_NAME, DATA_TYPE_NAME, NODE_GROUP_NAME,
+                NoMergePolicyFactory.NAME, null, new InternalDatasetDetails(null, PartitioningStrategy.HASH,
+                        partitioningKeys, null, null, null, false, null, false),
+                null, DatasetType.INTERNAL, DATASET_ID, 0);
+        PrimaryIndexInfo primaryIndexInfo = nc.createPrimaryIndex(dataset, KEY_TYPES, RECORD_TYPE, META_TYPE, null,
+                storageManager, KEY_INDEXES, KEY_INDICATORS_LIST, 0);
+        IndexDataflowHelperFactory iHelperFactory =
+                new IndexDataflowHelperFactory(nc.getStorageManager(), primaryIndexInfo.getFileSplitProvider());
+        JobId jobId = nc.newJobId();
+        ctx = nc.createTestContext(jobId, 0, false);
+        indexDataflowHelper = iHelperFactory.create(ctx.getJobletContext().getServiceContext(), 0);
+        indexDataflowHelper.open();
+        lsmBtree = (TestLsmBtree) indexDataflowHelper.getIndexInstance();
+        indexDataflowHelper.close();
+        txnCtx = nc.getTransactionManager().beginTransaction(nc.getTxnJobId(ctx),
+                new TransactionOptions(ITransactionManager.AtomicityLevel.ENTITY_LEVEL));
+        insertOp = nc.getInsertPipeline(ctx, dataset, KEY_TYPES, RECORD_TYPE, META_TYPE, null, KEY_INDEXES,
+                KEY_INDICATORS_LIST, storageManager).getLeft();
+    }
+
+    @After
+    public void destroyIndex() throws Exception {
+        indexDataflowHelper.destroy();
+    }
+
+    void unblockSearch(TestLsmBtree lsmBtree) {
+        lsmBtree.addSearchCallback(sem -> sem.release());
+        lsmBtree.allowSearch(1);
+    }
+
+    @Test
+    public void testCursorSwitchSucceed() {
+        try {
+            // allow all operations
+            ComponentRollbackTest.allowAllOps(lsmBtree);
+            // except search
+            lsmBtree.clearSearchCallbacks();
+            insertOp.open();
+            TupleGenerator tupleGenerator = new TupleGenerator(RECORD_TYPE, META_TYPE, KEY_INDEXES, KEY_INDICATORS,
+                    RECORD_GEN_FUNCTION, UNIQUE_RECORD_FIELDS, META_GEN_FUNCTION, UNIQUE_META_FIELDS);
+            VSizeFrame frame = new VSizeFrame(ctx);
+            FrameTupleAppender tupleAppender = new FrameTupleAppender(frame);
+            Searcher firstSearcher = null;
+            for (int j = 0; j < TOTAL_NUM_OF_RECORDS; j++) {
+                // flush every RECORDS_PER_COMPONENT records
+                if (j % RECORDS_PER_COMPONENT == (RECORDS_PER_COMPONENT - 1) && j + 1 != TOTAL_NUM_OF_RECORDS) {
+                    if (tupleAppender.getTupleCount() > 0) {
+                        tupleAppender.write(insertOp, true);
+                    }
+                    dsLifecycleMgr.flushDataset(dataset.getDatasetId(), false);
+                }
+                ITupleReference tuple = tupleGenerator.next();
+                DataflowUtils.addTupleToFrame(tupleAppender, tuple, insertOp);
+            }
+            if (tupleAppender.getTupleCount() > 0) {
+                tupleAppender.write(insertOp, true);
+            }
+            insertOp.close();
+            // start the search
+            firstSearcher = new Searcher(nc, 0, dataset, storageManager, lsmBtree, TOTAL_NUM_OF_RECORDS);
+            // wait till firstSearcher enter the components
+            firstSearcher.waitUntilEntered();
+            dsLifecycleMgr.flushDataset(dataset.getDatasetId(), false);
+            nc.getTransactionManager().commitTransaction(txnCtx.getTxnId());
+            // unblock the search
+            unblockSearch(lsmBtree);
+            // ensure the search got the correct number
+            Assert.assertTrue(firstSearcher.result());
+            // search now and ensure
+            searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
+        } catch (Throwable e) {
+            e.printStackTrace();
+            Assert.fail(e.getMessage());
+        }
+    }
+
+    @Test
+    public void testCursorSwitchFails() {
+        try {
+            // allow all operations
+            ComponentRollbackTest.allowAllOps(lsmBtree);
+            // except search
+            lsmBtree.clearSearchCallbacks();
+            insertOp.open();
+            TupleGenerator tupleGenerator = new TupleGenerator(RECORD_TYPE, META_TYPE, KEY_INDEXES, KEY_INDICATORS,
+                    RECORD_GEN_FUNCTION, UNIQUE_RECORD_FIELDS, META_GEN_FUNCTION, UNIQUE_META_FIELDS);
+            VSizeFrame frame = new VSizeFrame(ctx);
+            FrameTupleAppender tupleAppender = new FrameTupleAppender(frame);
+            Searcher firstSearcher = null;
+            for (int j = 0; j < TOTAL_NUM_OF_RECORDS; j++) {
+                // flush every RECORDS_PER_COMPONENT records
+                if (j % RECORDS_PER_COMPONENT == (RECORDS_PER_COMPONENT - 1) && j + 1 != TOTAL_NUM_OF_RECORDS) {
+                    if (tupleAppender.getTupleCount() > 0) {
+                        tupleAppender.write(insertOp, true);
+                    }
+                    dsLifecycleMgr.flushDataset(dataset.getDatasetId(), false);
+                }
+                ITupleReference tuple = tupleGenerator.next();
+                DataflowUtils.addTupleToFrame(tupleAppender, tuple, insertOp);
+            }
+            if (tupleAppender.getTupleCount() > 0) {
+                tupleAppender.write(insertOp, true);
+            }
+            insertOp.close();
+            // start the search
+            firstSearcher = new Searcher(nc, 0, dataset, storageManager, lsmBtree, TOTAL_NUM_OF_RECORDS);
+            // wait till firstSearcher enter the components
+            firstSearcher.waitUntilEntered();
+            dsLifecycleMgr.flushDataset(dataset.getDatasetId(), false);
+            nc.getTransactionManager().commitTransaction(txnCtx.getTxnId());
+            // merge all components
+            ILSMIndexAccessor mergeAccessor = lsmBtree.createAccessor(NoOpIndexAccessParameters.INSTANCE);
+            List<ILSMDiskComponent> mergedComponents = new ArrayList<>(lsmBtree.getDiskComponents());
+            BlockingIOOperationCallbackWrapper ioCallback =
+                    new BlockingIOOperationCallbackWrapper(lsmBtree.getIOOperationCallback());
+            mergeAccessor.scheduleMerge(ioCallback, mergedComponents);
+            ioCallback.waitForIO();
+            // unblock the search
+            unblockSearch(lsmBtree);
+            // ensure the search got the correct number
+            Assert.assertTrue(firstSearcher.result());
+            // search now and ensure
+            searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
+        } catch (Throwable e) {
+            e.printStackTrace();
+            Assert.fail(e.getMessage());
+        }
+    }
+
+    private void searchAndAssertCount(TestNodeController nc, IHyracksTaskContext ctx, Dataset dataset,
+            StorageComponentProvider storageManager, int numOfRecords)
+            throws HyracksDataException, AlgebricksException {
+        nc.newJobId();
+        TestTupleCounterFrameWriter countOp =
+                ComponentRollbackTest.create(nc.getSearchOutputDesc(KEY_TYPES, RECORD_TYPE, META_TYPE),
+                        Collections.emptyList(), Collections.emptyList(), false);
+        IPushRuntime emptyTupleOp = nc.getFullScanPipeline(countOp, ctx, dataset, KEY_TYPES, RECORD_TYPE, META_TYPE,
+                new NoMergePolicyFactory(), null, null, KEY_INDEXES, KEY_INDICATORS_LIST, storageManager);
+        emptyTupleOp.open();
+        emptyTupleOp.close();
+        Assert.assertEquals(numOfRecords, countOp.getCount());
+    }
+}
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestDataset.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestDataset.java
index e0502de..0c4983a 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestDataset.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestDataset.java
@@ -21,7 +21,6 @@
 import java.util.Map;
 
 import org.apache.asterix.common.config.DatasetConfig.DatasetType;
-import org.apache.asterix.common.context.DatasetLSMComponentIdGeneratorFactory;
 import org.apache.asterix.metadata.IDatasetDetails;
 import org.apache.asterix.metadata.declared.MetadataProvider;
 import org.apache.asterix.metadata.entities.Dataset;
@@ -63,6 +62,6 @@
 
     @Override
     public ILSMIOOperationCallbackFactory getIoOperationCallbackFactory(Index index) throws AlgebricksException {
-        return new TestLsmBtreeIoOpCallbackFactory(new DatasetLSMComponentIdGeneratorFactory(getDatasetId()));
+        return new TestLsmBtreeIoOpCallbackFactory(getComponentIdGeneratorFactory());
     }
 }
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestLsmBtreeIoOpCallbackFactory.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestLsmBtreeIoOpCallbackFactory.java
index fa37c20..44967e3 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestLsmBtreeIoOpCallbackFactory.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestLsmBtreeIoOpCallbackFactory.java
@@ -18,8 +18,9 @@
  */
 package org.apache.asterix.test.dataflow;
 
-import org.apache.asterix.common.ioopcallbacks.AbstractLSMIndexIOOperationCallbackFactory;
 import org.apache.asterix.common.ioopcallbacks.LSMBTreeIOOperationCallback;
+import org.apache.asterix.common.ioopcallbacks.LSMBTreeIOOperationCallbackFactory;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentIdGenerator;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentIdGeneratorFactory;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMDiskComponent;
@@ -28,7 +29,7 @@
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
 import org.apache.hyracks.storage.am.lsm.common.impls.EmptyComponent;
 
-public class TestLsmBtreeIoOpCallbackFactory extends AbstractLSMIndexIOOperationCallbackFactory {
+public class TestLsmBtreeIoOpCallbackFactory extends LSMBTreeIOOperationCallbackFactory {
 
     private static final long serialVersionUID = 1L;
 
@@ -97,7 +98,8 @@
         }
 
         @Override
-        public void afterFinalize(LSMIOOperationType opType, ILSMDiskComponent newComponent) {
+        public void afterFinalize(LSMIOOperationType opType, ILSMDiskComponent newComponent)
+                throws HyracksDataException {
             super.afterFinalize(opType, newComponent);
             synchronized (TestLsmBtreeIoOpCallbackFactory.this) {
                 if (newComponent != null) {
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/logging/CheckpointingTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/logging/CheckpointingTest.java
index 1cc24d5..1be1d26 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/logging/CheckpointingTest.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/logging/CheckpointingTest.java
@@ -54,6 +54,7 @@
 import org.apache.asterix.transaction.management.service.recovery.AbstractCheckpointManager;
 import org.apache.hyracks.api.comm.VSizeFrame;
 import org.apache.hyracks.api.context.IHyracksTaskContext;
+import org.apache.hyracks.api.job.JobId;
 import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAppender;
 import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
 import org.apache.hyracks.storage.am.lsm.common.impls.NoMergePolicyFactory;
@@ -126,9 +127,9 @@
                     null, DatasetType.INTERNAL, DATASET_ID, 0);
             try {
                 nc.createPrimaryIndex(dataset, KEY_TYPES, RECORD_TYPE, META_TYPE, null, storageManager, KEY_INDEXES,
-                        KEY_INDICATOR_LIST);
-                IHyracksTaskContext ctx = nc.createTestContext(false);
-                nc.newJobId();
+                        KEY_INDICATOR_LIST, 0);
+                JobId jobId = nc.newJobId();
+                IHyracksTaskContext ctx = nc.createTestContext(jobId, 0, false);
                 ITransactionContext txnCtx = nc.getTransactionManager().beginTransaction(nc.getTxnJobId(ctx),
                         new TransactionOptions(ITransactionManager.AtomicityLevel.ENTITY_LEVEL));
                 // Prepare insert operation
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/DeallocatableTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/DeallocatableTest.java
index 5ee0e9f..cfd251b 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/DeallocatableTest.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/DeallocatableTest.java
@@ -28,6 +28,7 @@
 import org.apache.hyracks.api.context.IHyracksTaskContext;
 import org.apache.hyracks.api.dataflow.ConnectorDescriptorId;
 import org.apache.hyracks.api.dataflow.TaskAttemptId;
+import org.apache.hyracks.api.job.JobId;
 import org.apache.hyracks.api.partitions.PartitionId;
 import org.apache.hyracks.comm.channels.NetworkOutputChannel;
 import org.apache.hyracks.control.nc.NodeControllerService;
@@ -58,7 +59,8 @@
             final NodeControllerService ncs =
                     (NodeControllerService) nc.getAppRuntimeContext().getServiceContext().getControllerService();
             final TaskAttemptId taId = Mockito.mock(TaskAttemptId.class);
-            final IHyracksTaskContext ctx = nc.createTestContext(true);
+            JobId jobId = nc.newJobId();
+            final IHyracksTaskContext ctx = nc.createTestContext(jobId, 0, true);
             final ConnectorDescriptorId codId = new ConnectorDescriptorId(1);
             final PartitionId pid = new PartitionId(ctx.getJobletContext().getJobId(), codId, 1, 1);
             final ChannelControlBlock ccb = ncs.getNetworkManager()
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/DiskIsFullTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/DiskIsFullTest.java
index cb83d56..8e8b3e9 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/DiskIsFullTest.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/DiskIsFullTest.java
@@ -33,7 +33,6 @@
 import org.apache.asterix.common.config.DatasetConfig.DatasetType;
 import org.apache.asterix.common.dataflow.LSMInsertDeleteOperatorNodePushable;
 import org.apache.asterix.common.exceptions.ExceptionUtils;
-import org.apache.asterix.common.transactions.ITransactionContext;
 import org.apache.asterix.external.util.DataflowUtils;
 import org.apache.asterix.file.StorageComponentProvider;
 import org.apache.asterix.metadata.entities.Dataset;
@@ -49,6 +48,7 @@
 import org.apache.hyracks.api.context.IHyracksTaskContext;
 import org.apache.hyracks.api.exceptions.ErrorCode;
 import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.api.job.JobId;
 import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAppender;
 import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
 import org.apache.hyracks.storage.am.lsm.common.impls.NoMergePolicyFactory;
@@ -125,9 +125,9 @@
                     null, DatasetType.INTERNAL, DATASET_ID, 0);
             try {
                 nc.createPrimaryIndex(dataset, KEY_TYPES, RECORD_TYPE, META_TYPE, null, storageManager, KEY_INDEXES,
-                        KEY_INDICATOR_LIST);
-                IHyracksTaskContext ctx = nc.createTestContext(false);
-                nc.newJobId();
+                        KEY_INDICATOR_LIST, 0);
+                JobId jobId = nc.newJobId();
+                IHyracksTaskContext ctx = nc.createTestContext(jobId, 0, false);
                 // Prepare insert operation
                 LSMInsertDeleteOperatorNodePushable insertOp = nc.getInsertPipeline(ctx, dataset, KEY_TYPES,
                         RECORD_TYPE, META_TYPE, null, KEY_INDEXES, KEY_INDICATOR_LIST, storageManager).getLeft();
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/IndexDropOperatorNodePushableTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/IndexDropOperatorNodePushableTest.java
index 7ced87d..cc6c0f7 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/IndexDropOperatorNodePushableTest.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/IndexDropOperatorNodePushableTest.java
@@ -18,8 +18,6 @@
  */
 package org.apache.asterix.test.storage;
 
-import static org.apache.hyracks.storage.am.common.dataflow.IndexDropOperatorDescriptor.DropOption;
-
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -51,9 +49,11 @@
 import org.apache.hyracks.api.exceptions.ErrorCode;
 import org.apache.hyracks.api.exceptions.HyracksDataException;
 import org.apache.hyracks.api.io.FileSplit;
+import org.apache.hyracks.api.job.JobId;
 import org.apache.hyracks.dataflow.std.file.ConstantFileSplitProvider;
 import org.apache.hyracks.storage.am.common.api.IIndexDataflowHelper;
 import org.apache.hyracks.storage.am.common.dataflow.IndexDataflowHelperFactory;
+import org.apache.hyracks.storage.am.common.dataflow.IndexDropOperatorDescriptor.DropOption;
 import org.apache.hyracks.storage.am.common.dataflow.IndexDropOperatorNodePushable;
 import org.apache.hyracks.storage.am.lsm.common.impls.NoMergePolicyFactory;
 import org.junit.After;
@@ -106,15 +106,15 @@
             Dataset dataset = new Dataset(DATAVERSE_NAME, DATASET_NAME, DATAVERSE_NAME, DATA_TYPE_NAME, NODE_GROUP_NAME,
                     NoMergePolicyFactory.NAME, null,
                     new InternalDatasetDetails(null, InternalDatasetDetails.PartitioningStrategy.HASH, partitioningKeys,
-                            null, null, null, false, null, false), null, DatasetConfig.DatasetType.INTERNAL, DATASET_ID,
-                    0);
+                            null, null, null, false, null, false),
+                    null, DatasetConfig.DatasetType.INTERNAL, DATASET_ID, 0);
             // create dataset
-            TestNodeController.PrimaryIndexInfo indexInfo =
-                    nc.createPrimaryIndex(dataset, KEY_TYPES, RECORD_TYPE, META_TYPE, null, storageManager, KEY_INDEXES,
-                            KEY_INDICATORS_LIST);
+            TestNodeController.PrimaryIndexInfo indexInfo = nc.createPrimaryIndex(dataset, KEY_TYPES, RECORD_TYPE,
+                    META_TYPE, null, storageManager, KEY_INDEXES, KEY_INDICATORS_LIST, 0);
             IndexDataflowHelperFactory helperFactory =
                     new IndexDataflowHelperFactory(nc.getStorageManager(), indexInfo.getFileSplitProvider());
-            IHyracksTaskContext ctx = nc.createTestContext(true);
+            JobId jobId = nc.newJobId();
+            IHyracksTaskContext ctx = nc.createTestContext(jobId, 0, true);
             IIndexDataflowHelper dataflowHelper = helperFactory.create(ctx.getJobletContext().getServiceContext(), 0);
             dropInUse(ctx, helperFactory, dataflowHelper);
             dropInUseWithWait(ctx, helperFactory, dataflowHelper);
@@ -144,21 +144,21 @@
             testExecutor.executeSqlppUpdateOrDdl("CREATE DATASET " + datasetName + "(KeyType) PRIMARY KEY id;", format);
             testExecutor.executeSqlppUpdateOrDdl("CREATE INDEX " + indexName + " on " + datasetName + "(foo)", format);
             final MetadataTransactionContext mdTxn = MetadataManager.INSTANCE.beginTransaction();
-            ICcApplicationContext appCtx =
-                    (ICcApplicationContext) ExecutionTestUtil.integrationUtil.getClusterControllerService()
-                            .getApplicationContext();
+            ICcApplicationContext appCtx = (ICcApplicationContext) ExecutionTestUtil.integrationUtil
+                    .getClusterControllerService().getApplicationContext();
             MetadataProvider metadataProver = new MetadataProvider(appCtx, null);
             metadataProver.setMetadataTxnContext(mdTxn);
             final String defaultDv = MetadataBuiltinEntities.DEFAULT_DATAVERSE.getDataverseName();
             final Dataset dataset = MetadataManager.INSTANCE.getDataset(mdTxn, defaultDv, datasetName);
             MetadataManager.INSTANCE.commitTransaction(mdTxn);
-            FileSplit[] splits = SplitsAndConstraintsUtil
-                    .getIndexSplits(appCtx.getClusterStateManager(), dataset, indexName, Arrays.asList("asterix_nc1"));
+            FileSplit[] splits = SplitsAndConstraintsUtil.getIndexSplits(appCtx.getClusterStateManager(), dataset,
+                    indexName, Arrays.asList("asterix_nc1"));
             final ConstantFileSplitProvider constantFileSplitProvider =
                     new ConstantFileSplitProvider(Arrays.copyOfRange(splits, 0, 1));
             IndexDataflowHelperFactory helperFactory =
                     new IndexDataflowHelperFactory(nc.getStorageManager(), constantFileSplitProvider);
-            IHyracksTaskContext ctx = nc.createTestContext(true);
+            JobId jobId = nc.newJobId();
+            IHyracksTaskContext ctx = nc.createTestContext(jobId, 0, true);
             IIndexDataflowHelper dataflowHelper = helperFactory.create(ctx.getJobletContext().getServiceContext(), 0);
             dropInUse(ctx, helperFactory, dataflowHelper);
         } finally {
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/AbstractLSMIOOperationCallback.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/AbstractLSMIOOperationCallback.java
index e445fe4..c33e2d1 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/AbstractLSMIOOperationCallback.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/AbstractLSMIOOperationCallback.java
@@ -34,6 +34,7 @@
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMMemoryComponent;
 import org.apache.hyracks.storage.am.lsm.common.impls.DiskComponentMetadata;
+import org.apache.hyracks.storage.am.lsm.common.impls.EmptyComponent;
 import org.apache.hyracks.storage.am.lsm.common.util.ComponentUtils;
 import org.apache.hyracks.storage.am.lsm.common.util.LSMComponentIdUtils;
 
@@ -88,7 +89,7 @@
     }
 
     @Override
-    public void afterFinalize(LSMIOOperationType opType, ILSMDiskComponent newComponent) {
+    public void afterFinalize(LSMIOOperationType opType, ILSMDiskComponent newComponent) throws HyracksDataException {
         // The operation was complete and the next I/O operation for the LSM index didn't start yet
         if (opType == LSMIOOperationType.FLUSH && newComponent != null) {
             synchronized (this) {
@@ -100,6 +101,13 @@
                 }
                 readIndex = (readIndex + 1) % mutableLastLSNs.length;
             }
+            if (newComponent == EmptyComponent.INSTANCE) {
+                // This component was just deleted, we refresh the component id, when it gets recycled, it will get
+                // the new id from the component id generator.
+                // It is assumed that the component delete caller will ensure that corresponding components in secondary
+                // indexes are deleted as well
+                idGenerator.refresh();
+            }
         }
     }
 
diff --git a/hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/base/AbstractOneInputOneOutputOneFramePushRuntime.java b/hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/base/AbstractOneInputOneOutputOneFramePushRuntime.java
index 2d8eaed..a7468a7 100644
--- a/hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/base/AbstractOneInputOneOutputOneFramePushRuntime.java
+++ b/hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/base/AbstractOneInputOneOutputOneFramePushRuntime.java
@@ -51,13 +51,25 @@
 
     @Override
     public void close() throws HyracksDataException {
+        HyracksDataException closeException = null;
         try {
             flushIfNotFailed();
         } catch (Exception e) {
+            closeException = HyracksDataException.create(e);
             writer.fail();
-            throw e;
         } finally {
-            writer.close();
+            try {
+                writer.close();
+            } catch (Exception e) {
+                if (closeException == null) {
+                    closeException = HyracksDataException.create(e);
+                } else {
+                    closeException.addSuppressed(e);
+                }
+            }
+        }
+        if (closeException != null) {
+            throw closeException;
         }
     }
 
diff --git a/hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/meta/AlgebricksMetaOperatorDescriptor.java b/hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/meta/AlgebricksMetaOperatorDescriptor.java
index f6ebf19..25a4aea 100644
--- a/hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/meta/AlgebricksMetaOperatorDescriptor.java
+++ b/hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/meta/AlgebricksMetaOperatorDescriptor.java
@@ -78,33 +78,48 @@
     public IOperatorNodePushable createPushRuntime(final IHyracksTaskContext ctx,
             final IRecordDescriptorProvider recordDescProvider, int partition, int nPartitions) {
         if (inputArity == 0) {
-            return createSourceInputPushRuntime(ctx);
+            return new SourcePushRuntime(ctx);
         } else {
             return createOneInputOneOutputPushRuntime(ctx, recordDescProvider);
         }
     }
 
-    private IOperatorNodePushable createSourceInputPushRuntime(final IHyracksTaskContext ctx) {
-        return new AbstractUnaryOutputSourceOperatorNodePushable() {
+    private class SourcePushRuntime extends AbstractUnaryOutputSourceOperatorNodePushable {
+        private final IHyracksTaskContext ctx;
 
-            @Override
-            public void initialize() throws HyracksDataException {
-                IFrameWriter startOfPipeline;
-                RecordDescriptor pipelineOutputRecordDescriptor =
-                        outputArity > 0 ? AlgebricksMetaOperatorDescriptor.this.outRecDescs[0] : null;
-                PipelineAssembler pa =
-                        new PipelineAssembler(pipeline, inputArity, outputArity, null, pipelineOutputRecordDescriptor);
-                startOfPipeline = pa.assemblePipeline(writer, ctx);
+        public SourcePushRuntime(IHyracksTaskContext ctx) {
+            this.ctx = ctx;
+        }
+
+        @Override
+        public void initialize() throws HyracksDataException {
+            IFrameWriter startOfPipeline;
+            RecordDescriptor pipelineOutputRecordDescriptor =
+                    outputArity > 0 ? AlgebricksMetaOperatorDescriptor.this.outRecDescs[0] : null;
+            PipelineAssembler pa =
+                    new PipelineAssembler(pipeline, inputArity, outputArity, null, pipelineOutputRecordDescriptor);
+            startOfPipeline = pa.assemblePipeline(writer, ctx);
+            HyracksDataException exception = null;
+            try {
+                startOfPipeline.open();
+            } catch (Exception e) {
+                startOfPipeline.fail();
+                exception = HyracksDataException.create(e);
+            } finally {
                 try {
-                    startOfPipeline.open();
-                } catch (Exception e) {
-                    startOfPipeline.fail();
-                    throw e;
-                } finally {
                     startOfPipeline.close();
+                } catch (Exception e) {
+                    if (exception == null) {
+                        exception = HyracksDataException.create(e);
+                    } else {
+                        exception.addSuppressed(e);
+                    }
                 }
             }
-        };
+            if (exception != null) {
+                throw exception;
+            }
+        }
     }
 
     private IOperatorNodePushable createOneInputOneOutputPushRuntime(final IHyracksTaskContext ctx,
diff --git a/hyracks-fullstack/hyracks/hyracks-http/src/main/java/org/apache/hyracks/http/server/HttpServer.java b/hyracks-fullstack/hyracks/hyracks-http/src/main/java/org/apache/hyracks/http/server/HttpServer.java
index d64858b..71ddbc0 100644
--- a/hyracks-fullstack/hyracks/hyracks-http/src/main/java/org/apache/hyracks/http/server/HttpServer.java
+++ b/hyracks-fullstack/hyracks/hyracks-http/src/main/java/org/apache/hyracks/http/server/HttpServer.java
@@ -49,8 +49,8 @@
     // Constants
     private static final int LOW_WRITE_BUFFER_WATER_MARK = 8 * 1024;
     private static final int HIGH_WRITE_BUFFER_WATER_MARK = 32 * 1024;
-    protected static final WriteBufferWaterMark WRITE_BUFFER_WATER_MARK = new WriteBufferWaterMark(
-            LOW_WRITE_BUFFER_WATER_MARK, HIGH_WRITE_BUFFER_WATER_MARK);
+    protected static final WriteBufferWaterMark WRITE_BUFFER_WATER_MARK =
+            new WriteBufferWaterMark(LOW_WRITE_BUFFER_WATER_MARK, HIGH_WRITE_BUFFER_WATER_MARK);
     protected static final int RECEIVE_BUFFER_SIZE = 4096;
     protected static final int DEFAULT_NUM_EXECUTOR_THREADS = 16;
     protected static final int DEFAULT_REQUEST_QUEUE_SIZE = 256;
@@ -92,8 +92,8 @@
         long directMemoryBudget = numExecutorThreads * (long) HIGH_WRITE_BUFFER_WATER_MARK
                 + numExecutorThreads * HttpServerInitializer.RESPONSE_CHUNK_SIZE;
         LOGGER.log(Level.INFO, "The output direct memory budget for this server is " + directMemoryBudget + " bytes");
-        long inputBudgetEstimate = (long) HttpServerInitializer.MAX_REQUEST_INITIAL_LINE_LENGTH
-                * (requestQueueSize + numExecutorThreads);
+        long inputBudgetEstimate =
+                (long) HttpServerInitializer.MAX_REQUEST_INITIAL_LINE_LENGTH * (requestQueueSize + numExecutorThreads);
         inputBudgetEstimate = inputBudgetEstimate * 2;
         LOGGER.log(Level.INFO,
                 "The \"estimated\" input direct memory budget for this server is " + inputBudgetEstimate + " bytes");
@@ -111,7 +111,7 @@
                 doStart();
                 setStarted();
             } catch (Throwable e) { // NOSONAR
-                LOGGER.log(Level.SEVERE, "Failure starting an Http Server", e);
+                LOGGER.log(Level.SEVERE, "Failure starting an Http Server with port: " + port, e);
                 setFailed(e);
                 throw e;
             }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/dataflow/ExternalBTreeLocalResource.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/dataflow/ExternalBTreeLocalResource.java
index a859f68..673bd3b 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/dataflow/ExternalBTreeLocalResource.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/dataflow/ExternalBTreeLocalResource.java
@@ -62,6 +62,6 @@
                 cmpFactories, bloomFilterKeyFields, bloomFilterFalsePositiveRate,
                 mergePolicyFactory.createMergePolicy(mergePolicyProperties, serviceCtx),
                 opTrackerProvider.getOperationTracker(serviceCtx), ioSchedulerProvider.getIoScheduler(serviceCtx),
-                ioOpCallbackFactory, durable, metadataPageManagerFactory);
+                ioOpCallbackFactory, durable, metadataPageManagerFactory, serviceCtx.getTracer());
     }
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/dataflow/ExternalBTreeWithBuddyLocalResource.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/dataflow/ExternalBTreeWithBuddyLocalResource.java
index 9422253..7e44c63 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/dataflow/ExternalBTreeWithBuddyLocalResource.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/dataflow/ExternalBTreeWithBuddyLocalResource.java
@@ -65,6 +65,6 @@
                 typeTraits, cmpFactories, bloomFilterFalsePositiveRate,
                 mergePolicyFactory.createMergePolicy(mergePolicyProperties, serviceCtx),
                 opTrackerProvider.getOperationTracker(serviceCtx), ioSchedulerProvider.getIoScheduler(serviceCtx),
-                ioOpCallbackFactory, bloomFilterKeyFields, durable, metadataPageManagerFactory);
+                ioOpCallbackFactory, bloomFilterKeyFields, durable, metadataPageManagerFactory, serviceCtx.getTracer());
     }
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/ExternalBTree.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/ExternalBTree.java
index 7dc5939..5b6ff9e 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/ExternalBTree.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/ExternalBTree.java
@@ -62,6 +62,7 @@
 import org.apache.hyracks.storage.common.ISearchOperationCallback;
 import org.apache.hyracks.storage.common.ISearchPredicate;
 import org.apache.hyracks.storage.common.buffercache.IBufferCache;
+import org.apache.hyracks.util.trace.ITracer;
 
 /**
  * This is an lsm b-tree that does not have memory component and is modified
@@ -94,18 +95,18 @@
             ILSMDiskComponentFactory bulkLoadComponentFactory, ILSMDiskComponentFactory transactionComponentFactory,
             double bloomFilterFalsePositiveRate, IBinaryComparatorFactory[] cmpFactories, ILSMMergePolicy mergePolicy,
             ILSMOperationTracker opTracker, ILSMIOOperationScheduler ioScheduler,
-            ILSMIOOperationCallbackFactory ioOpCallbackFactory, boolean durable) {
+            ILSMIOOperationCallbackFactory ioOpCallbackFactory, boolean durable, ITracer tracer) {
         super(ioManager, insertLeafFrameFactory, deleteLeafFrameFactory, bufferCache, fileManager, componentFactory,
                 bulkLoadComponentFactory, bloomFilterFalsePositiveRate, cmpFactories, mergePolicy, opTracker,
-                ioScheduler, ioOpCallbackFactory, false, durable);
+                ioScheduler, ioOpCallbackFactory, false, durable, tracer);
         this.transactionComponentFactory = transactionComponentFactory;
         this.secondDiskComponents = new LinkedList<>();
         this.interiorFrameFactory = interiorFrameFactory;
     }
 
     @Override
-    public ExternalIndexHarness getLsmHarness() {
-        return (ExternalIndexHarness) super.getLsmHarness();
+    public ExternalIndexHarness getHarness() {
+        return (ExternalIndexHarness) super.getHarness();
     }
 
     // The subsume merged components is overridden to account for:
@@ -162,9 +163,9 @@
 
     // This method creates the appropriate opContext for the targeted version
     public ExternalBTreeOpContext createOpContext(ISearchOperationCallback searchCallback, int targetVersion) {
-        return new ExternalBTreeOpContext(insertLeafFrameFactory, deleteLeafFrameFactory, searchCallback,
+        return new ExternalBTreeOpContext(this, insertLeafFrameFactory, deleteLeafFrameFactory, searchCallback,
                 ((LSMBTreeWithBloomFilterDiskComponentFactory) componentFactory).getBloomFilterKeyFields().length,
-                cmpFactories, targetVersion, getLsmHarness());
+                cmpFactories, targetVersion, getHarness(), tracer);
     }
 
     // The only reason to override the following method is that it uses a different context object
@@ -194,7 +195,7 @@
         FileReference lastFile = lastBTree.getFileReference();
         LSMComponentFileReferences relMergeFileRefs =
                 fileManager.getRelMergeFileReference(firstFile.getFile().getName(), lastFile.getFile().getName());
-        ILSMIndexAccessor accessor = new LSMTreeIndexAccessor(getLsmHarness(), opCtx, cursorFactory);
+        ILSMIndexAccessor accessor = new LSMTreeIndexAccessor(getHarness(), opCtx, cursorFactory);
         ioScheduler.scheduleOperation(new LSMBTreeMergeOperation(accessor, cursor,
                 relMergeFileRefs.getInsertIndexFileReference(), relMergeFileRefs.getBloomFilterFileReference(),
                 callback, fileManager.getBaseDir().getAbsolutePath()));
@@ -259,7 +260,7 @@
                 diskComponents.add(component);
                 secondDiskComponents.add(component);
             }
-            getLsmHarness().indexFirstTimeActivated();
+            getHarness().indexFirstTimeActivated();
         } else {
             // This index has been opened before
             for (ILSMDiskComponent c : diskComponents) {
@@ -309,7 +310,7 @@
         if (!isActive) {
             throw new HyracksDataException("Failed to clear the index since it is not activated.");
         }
-        getLsmHarness().indexClear();
+        getHarness().indexClear();
 
         for (ILSMDiskComponent c : diskComponents) {
             c.deactivateAndDestroy();
@@ -451,7 +452,7 @@
                     component.markAsValid(durable);
                     component.deactivate();
                 } else {
-                    getLsmHarness().addBulkLoadedComponent(component);
+                    getHarness().addBulkLoadedComponent(component);
                 }
             }
         }
@@ -490,14 +491,14 @@
     @Override
     public ILSMIndexAccessor createAccessor(IIndexAccessParameters iap) {
         ExternalBTreeOpContext opCtx = createOpContext(iap.getSearchOperationCallback(), version);
-        return new LSMTreeIndexAccessor(getLsmHarness(), opCtx, cursorFactory);
+        return new LSMTreeIndexAccessor(getHarness(), opCtx, cursorFactory);
     }
 
     @Override
     public ILSMIndexAccessor createAccessor(ISearchOperationCallback searchCallback, int targetIndexVersion)
             throws HyracksDataException {
         ExternalBTreeOpContext opCtx = createOpContext(searchCallback, targetIndexVersion);
-        return new LSMTreeIndexAccessor(getLsmHarness(), opCtx, cursorFactory);
+        return new LSMTreeIndexAccessor(getHarness(), opCtx, cursorFactory);
     }
 
     @Override
@@ -553,7 +554,7 @@
             component = createDiskComponent(componentFactory, componentFileReferences.getInsertIndexFileReference(),
                     null, componentFileReferences.getBloomFilterFileReference(), false);
         }
-        getLsmHarness().addTransactionComponents(component);
+        getHarness().addTransactionComponents(component);
     }
 
     @Override
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/ExternalBTreeOpContext.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/ExternalBTreeOpContext.java
index f94c38a..b0c31ad 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/ExternalBTreeOpContext.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/ExternalBTreeOpContext.java
@@ -22,9 +22,11 @@
 import org.apache.hyracks.storage.am.btree.api.IBTreeLeafFrame;
 import org.apache.hyracks.storage.am.common.api.ITreeIndexFrameFactory;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMHarness;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
 import org.apache.hyracks.storage.am.lsm.common.impls.AbstractLSMIndexOperationContext;
 import org.apache.hyracks.storage.common.ISearchOperationCallback;
 import org.apache.hyracks.storage.common.MultiComparator;
+import org.apache.hyracks.util.trace.ITracer;
 
 public class ExternalBTreeOpContext extends AbstractLSMIndexOperationContext {
     private IBTreeLeafFrame insertLeafFrame;
@@ -34,11 +36,11 @@
     private final int targetIndexVersion;
     private LSMBTreeCursorInitialState searchInitialState;
 
-    public ExternalBTreeOpContext(ITreeIndexFrameFactory insertLeafFrameFactory,
+    public ExternalBTreeOpContext(ILSMIndex index, ITreeIndexFrameFactory insertLeafFrameFactory,
             ITreeIndexFrameFactory deleteLeafFrameFactory, ISearchOperationCallback searchCallback,
             int numBloomFilterKeyFields, IBinaryComparatorFactory[] cmpFactories, int targetIndexVersion,
-            ILSMHarness lsmHarness) {
-        super(null, null, null, searchCallback, null);
+            ILSMHarness lsmHarness, ITracer tracer) {
+        super(index, null, null, null, searchCallback, null, tracer);
         if (cmpFactories != null) {
             this.cmp = MultiComparator.create(cmpFactories);
         } else {
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/ExternalBTreeWithBuddy.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/ExternalBTreeWithBuddy.java
index c495b69..c6ce339 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/ExternalBTreeWithBuddy.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/ExternalBTreeWithBuddy.java
@@ -66,6 +66,7 @@
 import org.apache.hyracks.storage.common.ISearchOperationCallback;
 import org.apache.hyracks.storage.common.ISearchPredicate;
 import org.apache.hyracks.storage.common.buffercache.IBufferCache;
+import org.apache.hyracks.util.trace.ITracer;
 
 public class ExternalBTreeWithBuddy extends AbstractLSMIndex implements ITreeIndex, ITwoPCIndex {
 
@@ -89,9 +90,10 @@
             ILSMDiskComponentFactory bulkLoadComponentFactory, double bloomFilterFalsePositiveRate,
             ILSMMergePolicy mergePolicy, ILSMOperationTracker opTracker, ILSMIOOperationScheduler ioScheduler,
             ILSMIOOperationCallbackFactory ioOpCallbackFactory, IBinaryComparatorFactory[] btreeCmpFactories,
-            IBinaryComparatorFactory[] buddyBtreeCmpFactories, int[] buddyBTreeFields, boolean durable) {
+            IBinaryComparatorFactory[] buddyBtreeCmpFactories, int[] buddyBTreeFields, boolean durable,
+            ITracer tracer) {
         super(ioManager, diskBufferCache, fileManager, bloomFilterFalsePositiveRate, mergePolicy, opTracker,
-                ioScheduler, ioOpCallbackFactory, componentFactory, bulkLoadComponentFactory, durable);
+                ioScheduler, ioOpCallbackFactory, componentFactory, bulkLoadComponentFactory, durable, tracer);
         this.btreeCmpFactories = btreeCmpFactories;
         this.buddyBtreeCmpFactories = buddyBtreeCmpFactories;
         this.buddyBTreeFields = buddyBTreeFields;
@@ -125,7 +127,7 @@
                 diskComponents.add(component);
                 secondDiskComponents.add(component);
             }
-            ((ExternalIndexHarness) getLsmHarness()).indexFirstTimeActivated();
+            ((ExternalIndexHarness) getHarness()).indexFirstTimeActivated();
         } else {
             // This index has been opened before or is brand new with no
             // components. It should also maintain the version pointer
@@ -147,7 +149,7 @@
         if (!isActive) {
             throw new HyracksDataException("Failed to clear the index since it is not activated.");
         }
-        ((ExternalIndexHarness) getLsmHarness()).indexClear();
+        ((ExternalIndexHarness) getHarness()).indexClear();
         for (ILSMDiskComponent c : diskComponents) {
             c.deactivateAndDestroy();
             // Remove from second list to avoid destroying twice
@@ -182,7 +184,7 @@
 
     @Override
     public ILSMIndexAccessor createAccessor(IIndexAccessParameters iap) throws HyracksDataException {
-        return new LSMTreeIndexAccessor(getLsmHarness(), createOpContext(iap.getSearchOperationCallback(), version),
+        return new LSMTreeIndexAccessor(getHarness(), createOpContext(iap.getSearchOperationCallback(), version),
                 ctx -> new LSMBTreeWithBuddySearchCursor(ctx, buddyBTreeFields));
     }
 
@@ -273,7 +275,7 @@
         List<ILSMComponent> mergingComponents = ctx.getComponentHolder();
         ITreeIndexCursor cursor = new LSMBTreeWithBuddySortedCursor(bctx, buddyBTreeFields);
         LSMComponentFileReferences relMergeFileRefs = getMergeTargetFileName(mergingComponents);
-        ILSMIndexAccessor accessor = new LSMTreeIndexAccessor(getLsmHarness(), bctx,
+        ILSMIndexAccessor accessor = new LSMTreeIndexAccessor(getHarness(), bctx,
                 opCtx -> new LSMBTreeWithBuddySearchCursor(opCtx, buddyBTreeFields));
 
         // Since we have two lists of components, to tell whether we need to
@@ -296,9 +298,9 @@
 
     // This method creates the appropriate opContext for the targeted version
     public ExternalBTreeWithBuddyOpContext createOpContext(ISearchOperationCallback searchCallback, int targetVersion) {
-        return new ExternalBTreeWithBuddyOpContext(btreeCmpFactories, buddyBtreeCmpFactories, searchCallback,
-                targetVersion, getLsmHarness(), btreeInteriorFrameFactory, btreeLeafFrameFactory,
-                buddyBtreeLeafFrameFactory);
+        return new ExternalBTreeWithBuddyOpContext(this, btreeCmpFactories, buddyBtreeCmpFactories, searchCallback,
+                targetVersion, getHarness(), btreeInteriorFrameFactory, btreeLeafFrameFactory,
+                buddyBtreeLeafFrameFactory, tracer);
     }
 
     @Override
@@ -528,7 +530,7 @@
                     component.markAsValid(durable);
                     component.deactivate();
                 } else {
-                    getLsmHarness().addBulkLoadedComponent(component);
+                    getHarness().addBulkLoadedComponent(component);
                 }
             }
         }
@@ -564,7 +566,7 @@
     @Override
     public ILSMIndexAccessor createAccessor(ISearchOperationCallback searchCallback, int targetIndexVersion)
             throws HyracksDataException {
-        return new LSMTreeIndexAccessor(getLsmHarness(), createOpContext(searchCallback, targetIndexVersion),
+        return new LSMTreeIndexAccessor(getHarness(), createOpContext(searchCallback, targetIndexVersion),
                 ctx -> new LSMBTreeWithBuddySearchCursor(ctx, buddyBTreeFields));
     }
 
@@ -609,7 +611,7 @@
                     componentFileRefrences.getDeleteIndexFileReference(),
                     componentFileRefrences.getBloomFilterFileReference(), false);
         }
-        ((ExternalIndexHarness) getLsmHarness()).addTransactionComponents(component);
+        ((ExternalIndexHarness) getHarness()).addTransactionComponents(component);
     }
 
     @Override
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/ExternalBTreeWithBuddyOpContext.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/ExternalBTreeWithBuddyOpContext.java
index d5cd2e2..85d4ab2 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/ExternalBTreeWithBuddyOpContext.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/ExternalBTreeWithBuddyOpContext.java
@@ -22,10 +22,12 @@
 import org.apache.hyracks.storage.am.common.api.ITreeIndexFrameFactory;
 import org.apache.hyracks.storage.am.common.impls.NoOpOperationCallback;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMHarness;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
 import org.apache.hyracks.storage.am.lsm.common.impls.AbstractLSMIndexOperationContext;
 import org.apache.hyracks.storage.common.IModificationOperationCallback;
 import org.apache.hyracks.storage.common.ISearchOperationCallback;
 import org.apache.hyracks.storage.common.MultiComparator;
+import org.apache.hyracks.util.trace.ITracer;
 
 public class ExternalBTreeWithBuddyOpContext extends AbstractLSMIndexOperationContext {
     private MultiComparator bTreeCmp;
@@ -33,11 +35,12 @@
     private final int targetIndexVersion;
     private LSMBTreeWithBuddyCursorInitialState searchInitialState;
 
-    public ExternalBTreeWithBuddyOpContext(IBinaryComparatorFactory[] btreeCmpFactories,
+    public ExternalBTreeWithBuddyOpContext(ILSMIndex index, IBinaryComparatorFactory[] btreeCmpFactories,
             IBinaryComparatorFactory[] buddyBtreeCmpFactories, ISearchOperationCallback searchCallback,
             int targetIndexVersion, ILSMHarness lsmHarness, ITreeIndexFrameFactory btreeInteriorFrameFactory,
-            ITreeIndexFrameFactory btreeLeafFrameFactory, ITreeIndexFrameFactory buddyBtreeLeafFrameFactory) {
-        super(null, null, null, searchCallback, null);
+            ITreeIndexFrameFactory btreeLeafFrameFactory, ITreeIndexFrameFactory buddyBtreeLeafFrameFactory,
+            ITracer tracer) {
+        super(index, null, null, null, searchCallback, null, tracer);
         this.targetIndexVersion = targetIndexVersion;
         this.bTreeCmp = MultiComparator.create(btreeCmpFactories);
         this.buddyBTreeCmp = MultiComparator.create(buddyBtreeCmpFactories);
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTree.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTree.java
index e69aae1..6929530 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTree.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTree.java
@@ -125,9 +125,10 @@
             ILSMDiskComponentFactory componentFactory, ILSMDiskComponentFactory bulkLoadComponentFactory,
             double bloomFilterFalsePositiveRate, IBinaryComparatorFactory[] cmpFactories, ILSMMergePolicy mergePolicy,
             ILSMOperationTracker opTracker, ILSMIOOperationScheduler ioScheduler,
-            ILSMIOOperationCallbackFactory ioOpCallbackFactory, boolean needKeyDupCheck, boolean durable) {
+            ILSMIOOperationCallbackFactory ioOpCallbackFactory, boolean needKeyDupCheck, boolean durable,
+            ITracer tracer) {
         super(ioManager, bufferCache, fileManager, bloomFilterFalsePositiveRate, mergePolicy, opTracker, ioScheduler,
-                ioOpCallbackFactory, componentFactory, bulkLoadComponentFactory, durable);
+                ioOpCallbackFactory, componentFactory, bulkLoadComponentFactory, durable, tracer);
         this.insertLeafFrameFactory = insertLeafFrameFactory;
         this.deleteLeafFrameFactory = deleteLeafFrameFactory;
         this.cmpFactories = cmpFactories;
@@ -364,10 +365,11 @@
     public LSMBTreeOpContext createOpContext(IModificationOperationCallback modificationCallback,
             ISearchOperationCallback searchCallback) {
         int numBloomFilterKeyFields = hasBloomFilter
-                ? ((LSMBTreeWithBloomFilterDiskComponentFactory) componentFactory).getBloomFilterKeyFields().length : 0;
-        return new LSMBTreeOpContext(memoryComponents, insertLeafFrameFactory, deleteLeafFrameFactory,
+                ? ((LSMBTreeWithBloomFilterDiskComponentFactory) componentFactory).getBloomFilterKeyFields().length
+                : 0;
+        return new LSMBTreeOpContext(this, memoryComponents, insertLeafFrameFactory, deleteLeafFrameFactory,
                 modificationCallback, searchCallback, numBloomFilterKeyFields, getTreeFields(), getFilterFields(),
-                getLsmHarness(), getFilterCmpFactories());
+                getHarness(), getFilterCmpFactories(), tracer);
     }
 
     @Override
@@ -376,7 +378,7 @@
     }
 
     public ILSMIndexAccessor createAccessor(AbstractLSMIndexOperationContext opCtx) {
-        return new LSMTreeIndexAccessor(getLsmHarness(), opCtx, cursorFactory);
+        return new LSMTreeIndexAccessor(getHarness(), opCtx, cursorFactory);
     }
 
     @Override
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeOpContext.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeOpContext.java
index 951ec81..9cab94e 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeOpContext.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeOpContext.java
@@ -32,12 +32,14 @@
 import org.apache.hyracks.storage.am.common.impls.NoOpOperationCallback;
 import org.apache.hyracks.storage.am.common.ophelpers.IndexOperation;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMHarness;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMMemoryComponent;
 import org.apache.hyracks.storage.am.lsm.common.impls.AbstractLSMIndexOperationContext;
 import org.apache.hyracks.storage.common.IIndexAccessParameters;
 import org.apache.hyracks.storage.common.IModificationOperationCallback;
 import org.apache.hyracks.storage.common.ISearchOperationCallback;
 import org.apache.hyracks.storage.common.MultiComparator;
+import org.apache.hyracks.util.trace.ITracer;
 
 public final class LSMBTreeOpContext extends AbstractLSMIndexOperationContext {
 
@@ -62,11 +64,12 @@
     private BTree.BTreeAccessor currentMutableBTreeAccessor;
     private BTreeOpContext currentMutableBTreeOpCtx;
 
-    public LSMBTreeOpContext(List<ILSMMemoryComponent> mutableComponents, ITreeIndexFrameFactory insertLeafFrameFactory,
-            ITreeIndexFrameFactory deleteLeafFrameFactory, IModificationOperationCallback modificationCallback,
-            ISearchOperationCallback searchCallback, int numBloomFilterKeyFields, int[] btreeFields, int[] filterFields,
-            ILSMHarness lsmHarness, IBinaryComparatorFactory[] filterCmpFactories) {
-        super(btreeFields, filterFields, filterCmpFactories, searchCallback, modificationCallback);
+    public LSMBTreeOpContext(ILSMIndex index, List<ILSMMemoryComponent> mutableComponents,
+            ITreeIndexFrameFactory insertLeafFrameFactory, ITreeIndexFrameFactory deleteLeafFrameFactory,
+            IModificationOperationCallback modificationCallback, ISearchOperationCallback searchCallback,
+            int numBloomFilterKeyFields, int[] btreeFields, int[] filterFields, ILSMHarness lsmHarness,
+            IBinaryComparatorFactory[] filterCmpFactories, ITracer tracer) {
+        super(index, btreeFields, filterFields, filterCmpFactories, searchCallback, modificationCallback, tracer);
         LSMBTreeMemoryComponent c = (LSMBTreeMemoryComponent) mutableComponents.get(0);
         IBinaryComparatorFactory cmpFactories[] = c.getIndex().getComparatorFactories();
         if (cmpFactories[0] != null) {
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeRangeSearchCursor.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeRangeSearchCursor.java
index 6cc8fbb..3e14fb9 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeRangeSearchCursor.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeRangeSearchCursor.java
@@ -34,8 +34,10 @@
 import org.apache.hyracks.storage.am.common.impls.NoOpIndexAccessParameters;
 import org.apache.hyracks.storage.am.common.impls.NoOpOperationCallback;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponent;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponent.ComponentState;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponent.LSMComponentType;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexOperationContext;
+import org.apache.hyracks.storage.am.lsm.common.impls.AbstractLSMMemoryComponent;
 import org.apache.hyracks.storage.am.lsm.common.impls.LSMIndexSearchCursor;
 import org.apache.hyracks.storage.common.ICursorInitialState;
 import org.apache.hyracks.storage.common.IIndexCursor;
@@ -47,10 +49,11 @@
     private final RangePredicate reusablePred;
 
     private ISearchOperationCallback searchCallback;
-    private RangePredicate predicate;
+
     private BTreeAccessor[] btreeAccessors;
     private ArrayTupleBuilder tupleBuilder;
     private boolean canCallProceed = true;
+    private int tupleFromMemoryComponentCount = 0;
 
     public LSMBTreeRangeSearchCursor(ILSMIndexOperationContext opCtx) {
         this(opCtx, false);
@@ -73,6 +76,9 @@
         outputElement = outputPriorityQueue.poll();
         needPushElementIntoQueue = true;
         canCallProceed = false;
+        if (outputElement.getCursorIndex() == 0) {
+            tupleFromMemoryComponentCount++;
+        }
     }
 
     /**
@@ -87,6 +93,15 @@
      */
     @Override
     protected void checkPriorityQueue() throws HyracksDataException {
+        // Every SWITCH_COMPONENT_CYCLE calls, check if memory components need to be swapped with disk components
+        // We should do this regardless of the value of includeMutableComponent. This is because if the cursor
+        // of the memory component has gone past the end of the in memory component, then the includeMutableComponent
+        // will be set to false. Still, when that happens, we want to exit the memory component to allow it to be
+        // recycled and used for modifications.
+        if (hasNextCallCount >= SWITCH_COMPONENT_CYCLE) {
+            replaceMemoryComponentWithDiskComponentIfNeeded();
+            hasNextCallCount = 0;
+        }
         while (!outputPriorityQueue.isEmpty() || needPushElementIntoQueue) {
             if (!outputPriorityQueue.isEmpty()) {
                 PriorityQueueElement queueHead = outputPriorityQueue.peek();
@@ -97,7 +112,7 @@
                         if (!searchCallback.proceed(queueHead.getTuple())) {
                             // In case proceed() fails and there is an in-memory component,
                             // we can't simply use this element since there might be a change.
-                            PriorityQueueElement mutableElement = removeMutable(outputPriorityQueue);
+                            PriorityQueueElement mutableElement = remove(outputPriorityQueue, 0);
                             if (mutableElement != null) {
                                 // Copies the current queue head
                                 if (tupleBuilder == null) {
@@ -166,10 +181,7 @@
                     } else {
                         // If the previous tuple and the head tuple are different
                         // the info of previous tuple is useless
-                        if (needPushElementIntoQueue == true) {
-                            pushIntoQueueFromCursorAndReplaceThisElement(outputElement);
-                            needPushElementIntoQueue = false;
-                        }
+                        pushOutputElementIntoQueueIfNeeded();
                         canCallProceed = true;
                         outputElement = null;
                     }
@@ -185,16 +197,128 @@
 
     }
 
-    private PriorityQueueElement removeMutable(PriorityQueue<PriorityQueueElement> outputPriorityQueue) {
-        // Scans the PQ for the mutable component's element and delete it
-        // since it can be changed.
-        // (i.e. we can't ensure that the element is the most current one.)
+    private void pushOutputElementIntoQueueIfNeeded() throws HyracksDataException {
+        if (needPushElementIntoQueue) {
+            pushIntoQueueFromCursorAndReplaceThisElement(outputElement);
+            needPushElementIntoQueue = false;
+        }
+    }
+
+    private void replaceMemoryComponentWithDiskComponentIfNeeded() throws HyracksDataException {
+        int replaceFrom = replaceFrom();
+        if (replaceFrom < 0) {
+            // no switch is needed, check if we need to re-do the search on the memory component.
+            // searches and modifications compete on the pages of the memory component
+            // if the cursor on the memory component is not advancing, we re-do the operation in order
+            // to release the latches and allow modifications to proceed
+            redoMemoryComponentSearchIfNeeded();
+            return;
+        }
+        opCtx.getIndex().getHarness().replaceMemoryComponentsWithDiskComponents(getOpCtx(), replaceFrom);
+        // redo the search on the new component
+        for (int i = replaceFrom; i < switchRequest.length; i++) {
+            if (switchRequest[i] && switchedElements[i] != null) {
+                copyTuple.reset(switchComponentTupleBuilders[i].getFieldEndOffsets(),
+                        switchComponentTupleBuilders[i].getByteArray());
+                reusablePred.setLowKey(copyTuple, true);
+                rangeCursors[i].reset();
+                ILSMComponent component = operationalComponents.get(i);
+                BTree btree = (BTree) component.getIndex();
+                if (i == 0 && component.getType() != LSMComponentType.MEMORY) {
+                    includeMutableComponent = false;
+                }
+                btreeAccessors[i].reset(btree, NoOpOperationCallback.INSTANCE, NoOpOperationCallback.INSTANCE);
+                btreeAccessors[i].search(rangeCursors[i], reusablePred);
+                pushIntoQueueFromCursorAndReplaceThisElement(switchedElements[i]);
+            }
+            switchRequest[i] = false;
+            // any failed switch makes further switches pointless
+            switchPossible = switchPossible && operationalComponents.get(i).getType() == LSMComponentType.DISK;
+        }
+    }
+
+    private int replaceFrom() throws HyracksDataException {
+        int replaceFrom = -1;
+        if (!switchPossible) {
+            return replaceFrom;
+        }
+        for (int i = 0; i < operationalComponents.size(); i++) {
+            ILSMComponent next = operationalComponents.get(i);
+            if (next.getType() == LSMComponentType.DISK) {
+                if (i == 0) {
+                    // if the first component is a disk component, then switch is not possible
+                    switchPossible = false;
+                }
+                break;
+            } else if (next.getState() == ComponentState.UNREADABLE_UNWRITABLE) {
+                // if the component is UNREADABLE_UNWRITABLE, then it means that the flush has been completed while
+                // the search cursor is inside the component, a switch candidate
+                if (replaceFrom < 0) {
+                    replaceFrom = i;
+                }
+                // we return the outputElement to the priority queue if it came from this component
+                if (outputElement != null && outputElement.getCursorIndex() == i) {
+                    pushIntoQueueFromCursorAndReplaceThisElement(outputElement);
+                    needPushElementIntoQueue = false;
+                    outputElement = null;
+                    canCallProceed = true;
+                }
+                PriorityQueueElement element = remove(outputPriorityQueue, i);
+                // if this cursor is still active (has an element)
+                // then we copy the search key to restart the operation after
+                // replacing the component
+                if (element != null) {
+                    if (switchComponentTupleBuilders[i] == null) {
+                        switchComponentTupleBuilders[i] = new ArrayTupleBuilder(cmp.getKeyFieldCount());
+                    }
+                    TupleUtils.copyTuple(switchComponentTupleBuilders[i], element.getTuple(), cmp.getKeyFieldCount());
+                }
+                rangeCursors[i].reset();
+                rangeCursors[i].close();
+                switchRequest[i] = true;
+                switchedElements[i] = element;
+            }
+        }
+        return replaceFrom;
+    }
+
+    private void redoMemoryComponentSearchIfNeeded() throws HyracksDataException {
+        if (!includeMutableComponent) {
+            return;
+        }
+        // if the last n records, none were from memory and there are writers inside the component,
+        // we need to re-do the search so the cursor doesn't block modifications due to latches over page
+        if (tupleFromMemoryComponentCount == 0
+                && ((AbstractLSMMemoryComponent) operationalComponents.get(0)).getWriterCount() > 0) {
+            // When we reach here, we know that the mutable component element is not the outputElement
+            // since if it was the output element, the tupleFromMemoryComponentCount would be at least 1
+            PriorityQueueElement mutableElement = remove(outputPriorityQueue, 0);
+            if (mutableElement != null) {
+                // if the element is null, then there is nothing to do since no latches are held
+                if (tupleBuilder == null) {
+                    tupleBuilder = new ArrayTupleBuilder(cmp.getKeyFieldCount());
+                }
+                TupleUtils.copyTuple(tupleBuilder, mutableElement.getTuple(), cmp.getKeyFieldCount());
+                copyTuple.reset(tupleBuilder.getFieldEndOffsets(), tupleBuilder.getByteArray());
+                // Unlatches/unpins the leaf page of the index.
+                rangeCursors[0].reset();
+                // Re-traverses the index.
+                reusablePred.setLowKey(copyTuple, true);
+                btreeAccessors[0].search(rangeCursors[0], reusablePred);
+                includeMutableComponent = pushIntoQueueFromCursorAndReplaceThisElement(mutableElement);
+            }
+        }
+        tupleFromMemoryComponentCount = 0;
+    }
+
+    private PriorityQueueElement remove(PriorityQueue<PriorityQueueElement> outputPriorityQueue, int cursorIndex) {
+        // Scans the PQ for the component's element and delete it
         Iterator<PriorityQueueElement> it = outputPriorityQueue.iterator();
         while (it.hasNext()) {
-            PriorityQueueElement mutableElement = it.next();
-            if (mutableElement.getCursorIndex() == 0) {
+            PriorityQueueElement e = it.next();
+            if (e.getCursorIndex() == cursorIndex) {
                 it.remove();
-                return mutableElement;
+                return e;
             }
         }
         return null;
@@ -207,7 +331,7 @@
         operationalComponents = lsmInitialState.getOperationalComponents();
         lsmHarness = lsmInitialState.getLSMHarness();
         searchCallback = lsmInitialState.getSearchOperationCallback();
-        predicate = (RangePredicate) lsmInitialState.getSearchPredicate();
+        RangePredicate predicate = (RangePredicate) lsmInitialState.getSearchPredicate();
         reusablePred.setLowKeyComparator(cmp);
         reusablePred.setHighKey(predicate.getHighKey(), predicate.isHighKeyInclusive());
         reusablePred.setHighKeyComparator(predicate.getHighKeyComparator());
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/utils/LSMBTreeUtil.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/utils/LSMBTreeUtil.java
index a0b1905..08e5af0 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/utils/LSMBTreeUtil.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/utils/LSMBTreeUtil.java
@@ -132,7 +132,7 @@
             int[] bloomFilterKeyFields, double bloomFilterFalsePositiveRate, ILSMMergePolicy mergePolicy,
             ILSMOperationTracker opTracker, ILSMIOOperationScheduler ioScheduler,
             ILSMIOOperationCallbackFactory ioOpCallbackFactory, boolean durable,
-            IMetadataPageManagerFactory freePageManagerFactory) {
+            IMetadataPageManagerFactory freePageManagerFactory, ITracer tracer) {
         LSMBTreeTupleWriterFactory insertTupleWriterFactory =
                 new LSMBTreeTupleWriterFactory(typeTraits, cmpFactories.length, false, false);
         LSMBTreeTupleWriterFactory deleteTupleWriterFactory =
@@ -176,14 +176,15 @@
         return new ExternalBTree(ioManager, interiorFrameFactory, insertLeafFrameFactory, deleteLeafFrameFactory,
                 diskBufferCache, fileNameManager, componentFactory, bulkLoadComponentFactory,
                 transactionComponentFactory, bloomFilterFalsePositiveRate, cmpFactories, mergePolicy, opTracker,
-                ioScheduler, ioOpCallbackFactory, durable);
+                ioScheduler, ioOpCallbackFactory, durable, tracer);
     }
 
     public static ExternalBTreeWithBuddy createExternalBTreeWithBuddy(IIOManager ioManager, FileReference file,
             IBufferCache diskBufferCache, ITypeTraits[] typeTraits, IBinaryComparatorFactory[] cmpFactories,
             double bloomFilterFalsePositiveRate, ILSMMergePolicy mergePolicy, ILSMOperationTracker opTracker,
             ILSMIOOperationScheduler ioScheduler, ILSMIOOperationCallbackFactory ioOpCallbackFactory,
-            int[] buddyBTreeFields, boolean durable, IMetadataPageManagerFactory freePageManagerFactory) {
+            int[] buddyBTreeFields, boolean durable, IMetadataPageManagerFactory freePageManagerFactory,
+            ITracer tracer) {
         ITypeTraits[] buddyBtreeTypeTraits = new ITypeTraits[buddyBTreeFields.length];
         IBinaryComparatorFactory[] buddyBtreeCmpFactories = new IBinaryComparatorFactory[buddyBTreeFields.length];
         for (int i = 0; i < buddyBtreeTypeTraits.length; i++) {
@@ -232,6 +233,6 @@
         return new ExternalBTreeWithBuddy(ioManager, interiorFrameFactory, insertLeafFrameFactory,
                 buddyBtreeLeafFrameFactory, diskBufferCache, fileNameManager, componentFactory,
                 bulkLoadComponentFactory, bloomFilterFalsePositiveRate, mergePolicy, opTracker, ioScheduler,
-                ioOpCallbackFactory, cmpFactories, buddyBtreeCmpFactories, buddyBTreeFields, durable);
+                ioOpCallbackFactory, cmpFactories, buddyBtreeCmpFactories, buddyBTreeFields, durable, tracer);
     }
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMComponent.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMComponent.java
index ab8e899..fc55ce5 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMComponent.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMComponent.java
@@ -129,6 +129,11 @@
     IIndex getIndex();
 
     /**
+     * @return the {@link ILSMIndex} this component belong to
+     */
+    ILSMIndex getLsmIndex();
+
+    /**
      *
      * @return id of the component
      * @throws HyracksDataException
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMHarness.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMHarness.java
index 89c8cb9..b32dd0f 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMHarness.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMHarness.java
@@ -253,4 +253,17 @@
      */
     void deleteComponents(ILSMIndexOperationContext ctx, Predicate<ILSMComponent> predicate)
             throws HyracksDataException;
+
+    /**
+     * Replace the memory components in this operation context with their corresponding disk
+     * components if possible
+     *
+     * @param ctx
+     *            the operation context
+     * @param startIndex
+     *            the index of the first component to switch
+     * @throws HyracksDataException
+     */
+    void replaceMemoryComponentsWithDiskComponents(ILSMIndexOperationContext ctx, int startIndex)
+            throws HyracksDataException;
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIndex.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIndex.java
index 52f9f06..62493f4 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIndex.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIndex.java
@@ -143,4 +143,9 @@
      * @return The number of all memory components (active and inactive)
      */
     int getNumberOfAllMemoryComponents();
+
+    /**
+     * @return the {@link ILSMHarness} of the index
+     */
+    ILSMHarness getHarness();
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIndexOperationContext.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIndexOperationContext.java
index 5b0378a..ec9124d 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIndexOperationContext.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIndexOperationContext.java
@@ -56,4 +56,30 @@
     PermutingTupleReference getFilterTuple();
 
     MultiComparator getFilterCmp();
+
+    /**
+     * @return the {@link ILSMIndex} of the component
+     */
+    ILSMIndex getIndex();
+
+    /**
+     * Performance tracing method. Logs the accumulated counters for number of tuples
+     *
+     * @param tupleCount
+     *            the number of tuples represented by the counters
+     */
+    void logPerformanceCounters(int tupleCount);
+
+    /**
+     * Increment the time taken for entering and exiting components
+     *
+     * @param increment
+     *            the time increment in nanoseconds
+     */
+    void incrementEnterExitTime(long increment);
+
+    /**
+     * @return true if performance tracing is enabled, false otherwise
+     */
+    boolean isTracingEnabled();
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMComponent.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMComponent.java
index ac124ba..84d2fe5 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMComponent.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMComponent.java
@@ -28,7 +28,7 @@
     protected final ILSMComponentFilter filter;
     protected final AbstractLSMIndex lsmIndex;
     // Mutables
-    protected ComponentState state;
+    protected volatile ComponentState state;
     protected int readerCount;
 
     public AbstractLSMComponent(AbstractLSMIndex lsmIndex, ILSMComponentFilter filter) {
@@ -46,4 +46,9 @@
     public ILSMComponentFilter getLSMComponentFilter() {
         return filter;
     }
+
+    @Override
+    public final AbstractLSMIndex getLsmIndex() {
+        return lsmIndex;
+    }
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMDiskComponent.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMDiskComponent.java
index b664102..bb27236 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMDiskComponent.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMDiskComponent.java
@@ -18,6 +18,7 @@
  */
 package org.apache.hyracks.storage.am.lsm.common.impls;
 
+import java.util.logging.Level;
 import java.util.logging.Logger;
 
 import org.apache.hyracks.api.exceptions.HyracksDataException;
@@ -49,11 +50,6 @@
     }
 
     @Override
-    public AbstractLSMIndex getLsmIndex() {
-        return lsmIndex;
-    }
-
-    @Override
     public boolean threadEnter(LSMOperationType opType, boolean isMutableComponent) {
         if (state == ComponentState.INACTIVE) {
             throw new IllegalStateException("Trying to enter an inactive disk component");
@@ -148,6 +144,9 @@
     @Override
     public void markAsValid(boolean persist) throws HyracksDataException {
         ComponentUtils.markAsValid(getMetadataHolder(), persist);
+        if (LOGGER.isLoggable(Level.INFO)) {
+            LOGGER.log(Level.INFO, "Marked as valid component with id: " + getId());
+        }
     }
 
     @Override
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMIndex.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMIndex.java
index b0cc318..dc808ad 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMIndex.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMIndex.java
@@ -27,6 +27,8 @@
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.logging.Level;
+import java.util.logging.Logger;
 
 import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
 import org.apache.hyracks.api.exceptions.ErrorCode;
@@ -71,6 +73,7 @@
 import org.apache.hyracks.util.trace.ITracer;
 
 public abstract class AbstractLSMIndex implements ILSMIndex {
+    private static final Logger LOGGER = Logger.getLogger(AbstractLSMIndex.class.getName());
     protected final ILSMHarness lsmHarness;
     protected final IIOManager ioManager;
     protected final ILSMIOOperationScheduler ioScheduler;
@@ -142,7 +145,7 @@
             double bloomFilterFalsePositiveRate, ILSMMergePolicy mergePolicy, ILSMOperationTracker opTracker,
             ILSMIOOperationScheduler ioScheduler, ILSMIOOperationCallbackFactory ioOpCallbackFactory,
             ILSMDiskComponentFactory componentFactory, ILSMDiskComponentFactory bulkLoadComponentFactory,
-            boolean durable) {
+            boolean durable, ITracer tracer) {
         this.ioManager = ioManager;
         this.diskBufferCache = diskBufferCache;
         this.fileManager = fileManager;
@@ -152,6 +155,7 @@
         this.componentFactory = componentFactory;
         this.bulkLoadComponentFactory = bulkLoadComponentFactory;
         this.durable = durable;
+        this.tracer = tracer;
         lsmHarness = new ExternalIndexHarness(this, mergePolicy, opTracker, diskBufferCache.isReplicationEnabled());
         isActive = false;
         diskComponents = new LinkedList<>();
@@ -303,12 +307,12 @@
                 operationalComponents.add(memoryComponents.get(cmc));
                 break;
             case INSERT:
-                addOperationalMutableComponents(operationalComponents);
+                addOperationalMutableComponents(operationalComponents, true);
                 operationalComponents.addAll(immutableComponents);
                 break;
             case SEARCH:
                 if (memoryComponentsAllocated) {
-                    addOperationalMutableComponents(operationalComponents);
+                    addOperationalMutableComponents(operationalComponents, false);
                 }
                 if (filterManager != null) {
                     for (ILSMComponent c : immutableComponents) {
@@ -375,18 +379,23 @@
         ioScheduler.scheduleOperation(TracedIOOperation.wrap(mergeOp, tracer));
     }
 
-    private void addOperationalMutableComponents(List<ILSMComponent> operationalComponents) {
+    private void addOperationalMutableComponents(List<ILSMComponent> operationalComponents, boolean modification) {
         int cmc = currentMutableComponentId.get();
         int numMutableComponents = memoryComponents.size();
         for (int i = 0; i < numMutableComponents - 1; i++) {
             ILSMMemoryComponent c = memoryComponents.get((cmc + i + 1) % numMutableComponents);
             if (c.isReadable()) {
-                // Make sure newest components are added first
+                // Make sure newest components are added first if readable
                 operationalComponents.add(0, c);
             }
         }
-        // The current mutable component is always added
-        operationalComponents.add(0, memoryComponents.get(cmc));
+        // The current mutable component is added if modification operation or if readable
+        // This ensures that activation of new component only happens in case of modifications
+        // and allow for controlling that without stopping search operations
+        ILSMMemoryComponent c = memoryComponents.get(cmc);
+        if (modification || c.isReadable()) {
+            operationalComponents.add(0, c);
+        }
     }
 
     @Override
@@ -421,7 +430,7 @@
     }
 
     @Override
-    public final synchronized void allocateMemoryComponents() throws HyracksDataException {
+    public synchronized void allocateMemoryComponents() throws HyracksDataException {
         if (!isActive) {
             throw HyracksDataException.create(ErrorCode.CANNOT_ALLOCATE_MEMORY_FOR_INACTIVE_INDEX);
         }
@@ -652,7 +661,8 @@
         return filterManager;
     }
 
-    public ILSMHarness getLsmHarness() {
+    @Override
+    public ILSMHarness getHarness() {
         return lsmHarness;
     }
 
@@ -681,8 +691,15 @@
     public final ILSMDiskComponent flush(ILSMIOOperation operation) throws HyracksDataException {
         ILSMIndexAccessor accessor = operation.getAccessor();
         ILSMIndexOperationContext opCtx = accessor.getOpContext();
-        return opCtx.getOperation() == IndexOperation.DELETE_MEMORY_COMPONENT ? EmptyComponent.INSTANCE
-                : doFlush(operation);
+        if (opCtx.getOperation() == IndexOperation.DELETE_MEMORY_COMPONENT) {
+            return EmptyComponent.INSTANCE;
+        } else {
+            if (LOGGER.isLoggable(Level.INFO)) {
+                FlushOperation flushOp = (FlushOperation) operation;
+                LOGGER.log(Level.INFO, "Flushing component with id: " + flushOp.getFlushingComponent().getId());
+            }
+            return doFlush(operation);
+        }
     }
 
     @Override
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMIndexOperationContext.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMIndexOperationContext.java
index 065d465..0818b08 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMIndexOperationContext.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMIndexOperationContext.java
@@ -27,14 +27,18 @@
 import org.apache.hyracks.storage.am.common.tuples.PermutingTupleReference;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponent;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMDiskComponent;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexOperationContext;
 import org.apache.hyracks.storage.common.IModificationOperationCallback;
 import org.apache.hyracks.storage.common.ISearchOperationCallback;
 import org.apache.hyracks.storage.common.ISearchPredicate;
 import org.apache.hyracks.storage.common.MultiComparator;
+import org.apache.hyracks.util.trace.ITracer;
+import org.apache.hyracks.util.trace.ITracer.Scope;
 
 public abstract class AbstractLSMIndexOperationContext implements ILSMIndexOperationContext {
 
+    protected final ILSMIndex index;
     protected final PermutingTupleReference indexTuple;
     protected final MultiComparator filterCmp;
     protected final PermutingTupleReference filterTuple;
@@ -47,10 +51,14 @@
     protected IndexOperation op;
     protected boolean accessingComponents = false;
     protected ISearchPredicate searchPredicate;
+    protected final ITracer tracer;
+    protected final long traceCategory;
+    private long enterExitTime = 0L;
 
-    public AbstractLSMIndexOperationContext(int[] treeFields, int[] filterFields,
+    public AbstractLSMIndexOperationContext(ILSMIndex index, int[] treeFields, int[] filterFields,
             IBinaryComparatorFactory[] filterCmpFactories, ISearchOperationCallback searchCallback,
-            IModificationOperationCallback modificationCallback) {
+            IModificationOperationCallback modificationCallback, ITracer tracer) {
+        this.index = index;
         this.searchCallback = searchCallback;
         this.modificationCallback = modificationCallback;
         this.componentHolder = new LinkedList<>();
@@ -73,6 +81,8 @@
             filterTuple = null;
             allFields = null;
         }
+        this.tracer = tracer;
+        this.traceCategory = tracer.getRegistry().get("op-ctx");
     }
 
     @Override
@@ -153,4 +163,32 @@
     public ISearchPredicate getSearchPredicate() {
         return searchPredicate;
     }
+
+    @Override
+    public final boolean isTracingEnabled() {
+        return tracer.isEnabled(traceCategory);
+    }
+
+    @Override
+    public void logPerformanceCounters(int tupleCount) {
+        if (isTracingEnabled()) {
+            tracer.instant("store-counters", traceCategory, Scope.t,
+                    "{\"count\":" + tupleCount + ",\"enter-exit-duration-ns\":" + enterExitTime + "}");
+            resetCounters();
+        }
+    }
+
+    public void resetCounters() {
+        enterExitTime = 0L;
+    }
+
+    @Override
+    public void incrementEnterExitTime(long increment) {
+        enterExitTime += increment;
+    }
+
+    @Override
+    public ILSMIndex getIndex() {
+        return index;
+    }
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMMemoryComponent.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMMemoryComponent.java
index 6a186dc..17dadcb 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMMemoryComponent.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMMemoryComponent.java
@@ -19,6 +19,8 @@
 package org.apache.hyracks.storage.am.lsm.common.impls;
 
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.logging.Level;
+import java.util.logging.Logger;
 
 import org.apache.hyracks.api.exceptions.HyracksDataException;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentFilter;
@@ -32,6 +34,7 @@
 
 public abstract class AbstractLSMMemoryComponent extends AbstractLSMComponent implements ILSMMemoryComponent {
 
+    private static final Logger LOGGER = Logger.getLogger(AbstractLSMMemoryComponent.class.getName());
     private final IVirtualBufferCache vbc;
     private final AtomicBoolean isModified;
     private int writerCount;
@@ -277,6 +280,9 @@
             throw new IllegalStateException(
                     "LSM memory component receives illegal id. Old id " + this.componentId + ", new id " + componentId);
         }
+        if (LOGGER.isLoggable(Level.INFO)) {
+            LOGGER.log(Level.INFO, "Component Id was reset from " + this.componentId + " to " + componentId);
+        }
         this.componentId = componentId;
         LSMComponentIdUtils.persist(this.componentId, metadata);
     }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/ComponentReplacementContext.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/ComponentReplacementContext.java
new file mode 100644
index 0000000..98c1560
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/ComponentReplacementContext.java
@@ -0,0 +1,215 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.hyracks.storage.am.lsm.common.impls;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.storage.am.common.ophelpers.IndexOperation;
+import org.apache.hyracks.storage.am.common.tuples.PermutingTupleReference;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponent;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponent.LSMComponentType;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentId;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMDiskComponent;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexOperationContext;
+import org.apache.hyracks.storage.common.IModificationOperationCallback;
+import org.apache.hyracks.storage.common.ISearchOperationCallback;
+import org.apache.hyracks.storage.common.ISearchPredicate;
+import org.apache.hyracks.storage.common.MultiComparator;
+
+public class ComponentReplacementContext implements ILSMIndexOperationContext {
+    private static final Logger LOGGER = Logger.getLogger(ComponentReplacementContext.class.getName());
+    private final List<ILSMComponent> components;
+    private final List<ILSMComponent> diskComponents;
+    private final List<ILSMComponentId> replacedComponentIds;
+    private final int[] swapIndexes;
+    private int count = 0;
+    boolean accessingComponent = true;
+
+    public ComponentReplacementContext(ILSMIndex lsmIndex) {
+        components = new ArrayList<>(lsmIndex.getNumberOfAllMemoryComponents());
+        replacedComponentIds = new ArrayList<>(lsmIndex.getNumberOfAllMemoryComponents());
+        swapIndexes = new int[lsmIndex.getNumberOfAllMemoryComponents()];
+        diskComponents = new ArrayList<>(lsmIndex.getNumberOfAllMemoryComponents());
+    }
+
+    @Override
+    public void setOperation(IndexOperation newOp) throws HyracksDataException {
+        // Do nothing
+    }
+
+    @Override
+    public IndexOperation getOperation() {
+        return IndexOperation.SEARCH;
+    }
+
+    @Override
+    public void reset() {
+        accessingComponent = true;
+        components.clear();
+        diskComponents.clear();
+        replacedComponentIds.clear();
+        count = 0;
+    }
+
+    @Override
+    public List<ILSMComponent> getComponentHolder() {
+        return components;
+    }
+
+    @Override
+    public List<ILSMDiskComponent> getComponentsToBeMerged() {
+        return Collections.emptyList();
+    }
+
+    @Override
+    public ISearchOperationCallback getSearchOperationCallback() {
+        return null;
+    }
+
+    @Override
+    public IModificationOperationCallback getModificationCallback() {
+        return null;
+    }
+
+    @Override
+    public void setCurrentMutableComponentId(int currentMutableComponentId) {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public void setSearchPredicate(ISearchPredicate searchPredicate) {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public ISearchPredicate getSearchPredicate() {
+        return null;
+    }
+
+    @Override
+    public List<ILSMDiskComponent> getComponentsToBeReplicated() {
+        return Collections.emptyList();
+    }
+
+    @Override
+    public boolean isAccessingComponents() {
+        return accessingComponent;
+    }
+
+    @Override
+    public void setAccessingComponents(boolean accessingComponents) {
+        // Ignore since this is only used for component replacement
+    }
+
+    @Override
+    public PermutingTupleReference getIndexTuple() {
+        return null;
+    }
+
+    @Override
+    public PermutingTupleReference getFilterTuple() {
+        return null;
+    }
+
+    @Override
+    public MultiComparator getFilterCmp() {
+        return null;
+    }
+
+    @Override
+    public void logPerformanceCounters(int tupleCount) {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public void incrementEnterExitTime(long increment) {
+        // Ignore since this is only used for component replacement
+    }
+
+    public boolean proceed(List<ILSMDiskComponent> allDiskComponents) throws HyracksDataException {
+        for (int i = 0; i < components.size(); i++) {
+            replacedComponentIds.add(components.get(i).getId());
+            // ensure that disk component exists
+            boolean found = false;
+            LOGGER.log(Level.INFO, "Looking for a component with the id: " + replacedComponentIds.get(i));
+            for (int j = 0; j < allDiskComponents.size(); j++) {
+                ILSMDiskComponent dc = allDiskComponents.get(j);
+                ILSMComponentId diskComponentId = dc.getId();
+                LOGGER.log(Level.INFO, "Next disk component id: " + diskComponentId);
+                if (diskComponentId.equals(replacedComponentIds.get(i))) {
+                    found = true;
+                    diskComponents.add(dc);
+                    break;
+                }
+            }
+            if (!found) {
+                // component has been merged?
+                LOGGER.log(Level.WARNING, "Memory Component with id = " + replacedComponentIds.get(i)
+                        + " was flushed and merged before search cursor replaces it");
+                return false;
+            }
+        }
+        return true;
+    }
+
+    public void swapIndex(int i) {
+        swapIndexes[count] = i;
+        count++;
+    }
+
+    public void prepareToEnter() {
+        components.clear();
+        components.addAll(diskComponents);
+        accessingComponent = false;
+    }
+
+    public void replace(ILSMIndexOperationContext ctx) {
+        // Called after exit and enter has been completed
+        try {
+            for (int i = 0; i < count; i++) {
+                ILSMComponent removed = ctx.getComponentHolder().remove(swapIndexes[i]);
+                if (removed.getType() == LSMComponentType.MEMORY) {
+                    LOGGER.log(Level.INFO, "Removed a memory component from the search operation");
+                } else {
+                    throw new IllegalStateException("Disk components can't be removed from the search operation");
+                }
+                ctx.getComponentHolder().add(swapIndexes[i], diskComponents.get(i));
+            }
+        } catch (Exception e) {
+            LOGGER.log(Level.WARNING, "Failure replacing memory components with disk components", e);
+            throw e;
+        }
+    }
+
+    @Override
+    public ILSMIndex getIndex() {
+        return null;
+    }
+
+    @Override
+    public boolean isTracingEnabled() {
+        return false;
+    }
+}
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMComponentId.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMComponentId.java
index dd86f65..442af56 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMComponentId.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMComponentId.java
@@ -78,13 +78,7 @@
             return false;
         }
         LSMComponentId other = (LSMComponentId) obj;
-        if (maxId != other.maxId) {
-            return false;
-        }
-        if (minId != other.minId) {
-            return false;
-        }
-        return true;
+        return maxId == other.maxId && minId == other.minId;
     }
 
     @Override
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMHarness.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMHarness.java
index b0abeb1..e1d5114 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMHarness.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMHarness.java
@@ -63,6 +63,7 @@
     private static final Logger LOGGER = Logger.getLogger(LSMHarness.class.getName());
 
     protected final ILSMIndex lsmIndex;
+    protected final ComponentReplacementContext componentReplacementCtx;
     protected final ILSMMergePolicy mergePolicy;
     protected final ILSMOperationTracker opTracker;
     protected final AtomicBoolean fullMergeIsRequested;
@@ -84,83 +85,97 @@
         if (replicationEnabled) {
             this.componentsToBeReplicated = new ArrayList<>();
         }
+        componentReplacementCtx = new ComponentReplacementContext(lsmIndex);
     }
 
     protected boolean getAndEnterComponents(ILSMIndexOperationContext ctx, LSMOperationType opType,
             boolean isTryOperation) throws HyracksDataException {
-        validateOperationEnterComponentsState(ctx);
-        synchronized (opTracker) {
-            while (true) {
-                lsmIndex.getOperationalComponents(ctx);
-                // Before entering the components, prune those corner cases that indeed should not proceed.
-                switch (opType) {
-                    case FLUSH:
-                        // if the lsm index does not have memory components allocated, then nothing to flush
-                        if (!lsmIndex.isMemoryComponentsAllocated()) {
-                            return false;
-                        }
-                        ILSMMemoryComponent flushingComponent = (ILSMMemoryComponent) ctx.getComponentHolder().get(0);
-                        if (!flushingComponent.isModified()) {
-                            if (flushingComponent.getState() == ComponentState.READABLE_UNWRITABLE) {
-                                //The mutable component has not been modified by any writer. There is nothing to flush.
-                                //since the component is empty, set its state back to READABLE_WRITABLE only when it's
-                                //state has been set to READABLE_UNWRITABLE
-                                flushingComponent.setState(ComponentState.READABLE_WRITABLE);
-                                opTracker.notifyAll();
-
-                                // Call recycled only when we change it's state is reset back to READABLE_WRITABLE
-                                // Otherwise, if the component is in other state, e.g., INACTIVE, or
-                                // READABLE_UNWRITABLE_FLUSHING, it's not considered as being recycled here.
-                                lsmIndex.getIOOperationCallback().recycled(flushingComponent);
+        long before = 0L;
+        if (ctx.isTracingEnabled()) {
+            before = System.nanoTime();
+        }
+        try {
+            validateOperationEnterComponentsState(ctx);
+            synchronized (opTracker) {
+                while (true) {
+                    lsmIndex.getOperationalComponents(ctx);
+                    // Before entering the components, prune those corner cases that indeed should not proceed.
+                    switch (opType) {
+                        case FLUSH:
+                            // if the lsm index does not have memory components allocated, then nothing to flush
+                            if (!lsmIndex.isMemoryComponentsAllocated()) {
+                                return false;
                             }
-                            return false;
-                        }
-                        if (flushingComponent.getWriterCount() > 0) {
-                            /*
-                             * This case is a case where even though FLUSH log was flushed to disk and scheduleFlush is triggered,
-                             * the current in-memory component (whose state was changed to READABLE_WRITABLE (RW)
-                             * from READABLE_UNWRITABLE(RU) before FLUSH log was written to log tail (which is memory buffer of log file)
-                             * and then the state was changed back to RW (as shown in the following scenario)) can have writers
-                             * based on the current code base/design.
-                             * Thus, the writer count of the component may be greater than 0.
-                             * if this happens, intead of throwing exception, scheduleFlush() deal with this situation by not flushing
-                             * the component.
-                             * Please see issue 884 for more detail information:
-                             * https://code.google.com/p/asterixdb/issues/detail?id=884&q=owner%3Akisskys%40gmail.com&colspec=ID%20Type%20Status%20Priority%20Milestone%20Owner%20Summary%20ETA%20Severity
-                             *
-                             */
-                            return false;
-                        }
-                        break;
-                    case MERGE:
-                        if (ctx.getComponentHolder().size() < 2
-                                && ctx.getOperation() != IndexOperation.DELETE_DISK_COMPONENTS) {
-                            // There is only a single component. There is nothing to merge.
-                            return false;
-                        }
-                        break;
-                    default:
-                        break;
-                }
-                if (enterComponents(ctx, opType)) {
-                    return true;
-                } else if (isTryOperation) {
-                    return false;
-                }
-                try {
-                    // Flush and merge operations should never reach this wait call, because they are always try operations.
-                    // If they fail to enter the components, then it means that there are an ongoing flush/merge operation on
-                    // the same components, so they should not proceed.
-                    if (opType == LSMOperationType.MODIFICATION) {
-                        // before waiting, make sure the index is in a modifiable state to avoid waiting forever.
-                        ensureIndexModifiable();
+                            ILSMMemoryComponent flushingComponent =
+                                    (ILSMMemoryComponent) ctx.getComponentHolder().get(0);
+                            if (!flushingComponent.isModified()) {
+                                recycle(flushingComponent);
+                                return false;
+                            }
+                            if (flushingComponent.getWriterCount() > 0) {
+                                /*
+                                 * This case is a case where even though FLUSH log was flushed to disk
+                                 * and scheduleFlush is triggered, the current in-memory component (whose state was
+                                 * changed to READABLE_WRITABLE (RW) from READABLE_UNWRITABLE(RU) before FLUSH log
+                                 * was written to log tail (which is memory buffer of log file) and then the state was
+                                 * changed back to RW (as shown in the following scenario)) can have writers based on
+                                 * the current code base/design. Thus, the writer count of the component may be greater
+                                 * than 0. if this happens, intead of throwing exception, scheduleFlush() deal with
+                                 * this situation by not flushing the component.
+                                 * for more detailed information: ASTERIXDB-1027
+                                 */
+                                return false;
+                            }
+                            break;
+                        case MERGE:
+                            if (ctx.getComponentHolder().size() < 2
+                                    && ctx.getOperation() != IndexOperation.DELETE_DISK_COMPONENTS) {
+                                // There is only a single component. There is nothing to merge.
+                                return false;
+                            }
+                            break;
+                        default:
+                            break;
                     }
-                    opTracker.wait();
-                } catch (InterruptedException e) {
-                    Thread.currentThread().interrupt();
-                    throw HyracksDataException.create(e);
+                    if (enterComponents(ctx, opType)) {
+                        return true;
+                    } else if (isTryOperation) {
+                        return false;
+                    }
+                    try {
+                        // Flush and merge operations should never reach this wait call,
+                        // because they are always try operations. If they fail to enter the components,
+                        // then it means that there are an ongoing flush/merge operation on
+                        // the same components, so they should not proceed.
+                        if (opType == LSMOperationType.MODIFICATION) {
+                            // before waiting, make sure the index is in a modifiable state to avoid waiting forever.
+                            ensureIndexModifiable();
+                        }
+                        opTracker.wait();
+                    } catch (InterruptedException e) {
+                        Thread.currentThread().interrupt();
+                        throw HyracksDataException.create(e);
+                    }
                 }
             }
+        } finally {
+            if (ctx.isTracingEnabled()) {
+                ctx.incrementEnterExitTime(System.nanoTime() - before);
+            }
+        }
+    }
+
+    private void recycle(ILSMMemoryComponent flushingComponent) throws HyracksDataException {
+        if (flushingComponent.getState() == ComponentState.READABLE_UNWRITABLE) {
+            //The mutable component has not been modified by any writer.
+            // There is nothing to flush. Since the component is empty, set its state back
+            // to READABLE_WRITABLE only when it's state has been set to READABLE_UNWRITABLE
+            flushingComponent.setState(ComponentState.READABLE_WRITABLE);
+            opTracker.notifyAll(); // NOSONAR: Always synchronized from caller
+            // Call recycled only when we change it's state is reset back to READABLE_WRITABLE
+            // Otherwise, if the component is in other state, e.g., INACTIVE, or
+            // READABLE_UNWRITABLE_FLUSHING, it's not considered as being recycled here.
+            lsmIndex.getIOOperationCallback().recycled(flushingComponent);
         }
     }
 
@@ -179,7 +194,7 @@
                 numEntered++;
             }
             entranceSuccessful = numEntered == components.size();
-        } catch (Throwable e) {
+        } catch (Throwable e) { // NOSONAR: Log and re-throw
             if (LOGGER.isLoggable(Level.SEVERE)) {
                 LOGGER.log(Level.SEVERE, opType.name() + " failed to enter components on " + lsmIndex, e);
             }
@@ -196,10 +211,12 @@
                     i++;
                     numEntered--;
                 }
-                return false;
             }
-            ctx.setAccessingComponents(true);
         }
+        if (!entranceSuccessful) {
+            return false;
+        }
+        ctx.setAccessingComponents(true);
         // Check if there is any action that is needed to be taken based on the operation type
         switch (opType) {
             case FLUSH:
@@ -209,7 +226,7 @@
                 lsmIndex.changeMutableComponent();
                 // Notify all waiting threads whenever a flush has been scheduled since they will check
                 // again if they can grab and enter the mutable component.
-                opTracker.notifyAll();
+                opTracker.notifyAll(); // NOSONAR: Always called from a synchronized block
                 break;
             case MERGE:
                 lsmIndex.getIOOperationCallback().beforeOperation(LSMIOOperationType.MERGE);
@@ -221,8 +238,8 @@
         return true;
     }
 
-    private void exitComponents(ILSMIndexOperationContext ctx, LSMOperationType opType, ILSMDiskComponent newComponent,
-            boolean failedOperation) throws HyracksDataException {
+    private void doExitComponents(ILSMIndexOperationContext ctx, LSMOperationType opType,
+            ILSMDiskComponent newComponent, boolean failedOperation) throws HyracksDataException {
         /**
          * FLUSH and MERGE operations should always exit the components
          * to notify waiting threads.
@@ -249,75 +266,10 @@
                     } else if (opType == LSMOperationType.MERGE) {
                         opTracker.notifyAll();
                     }
-
-                    int i = 0;
-                    // First check if there is any action that is needed to be taken based on the state of each component.
-                    for (ILSMComponent c : ctx.getComponentHolder()) {
-                        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:
-                                    tracer.instant(c.toString(), traceCategory, Scope.p, lsmIndex.toString());
-                                    ((AbstractLSMMemoryComponent) c).reset();
-                                    // Notify all waiting threads whenever the mutable component's state has changed to
-                                    // inactive. This is important because even though we switched the mutable
-                                    // components, it is possible that the component that we just switched to is still
-                                    // busy flushing its data to disk. Thus, the notification that was issued upon
-                                    // scheduling the flush is not enough.
-                                    opTracker.notifyAll();
-                                    break;
-                                default:
-                                    break;
-                            }
-                        } else {
-                            switch (c.getState()) {
-                                case INACTIVE:
-                                    lsmIndex.addInactiveDiskComponent((AbstractLSMDiskComponent) c);
-                                    break;
-                                default:
-                                    break;
-                            }
-                        }
-                        i++;
-                    }
+                    exitOperationalComponents(ctx, opType, failedOperation);
                     ctx.setAccessingComponents(false);
-                    // Then, perform any action that is needed to be taken based on the operation type.
-                    switch (opType) {
-                        case FLUSH:
-                            // newComponent is null if the flush op. was not performed.
-                            if (!failedOperation && newComponent != null) {
-                                lsmIndex.addDiskComponent(newComponent);
-                                if (replicationEnabled) {
-                                    componentsToBeReplicated.clear();
-                                    componentsToBeReplicated.add(newComponent);
-                                    triggerReplication(componentsToBeReplicated, false, opType);
-                                }
-                                mergePolicy.diskComponentAdded(lsmIndex, false);
-                            }
-                            break;
-                        case MERGE:
-                            // newComponent is null if the merge op. was not performed.
-                            if (!failedOperation && newComponent != null) {
-                                lsmIndex.subsumeMergedComponents(newComponent, ctx.getComponentHolder());
-                                if (replicationEnabled) {
-                                    componentsToBeReplicated.clear();
-                                    componentsToBeReplicated.add(newComponent);
-                                    triggerReplication(componentsToBeReplicated, false, opType);
-                                }
-                                mergePolicy.diskComponentAdded(lsmIndex, fullMergeIsRequested.get());
-                            }
-                            break;
-                        default:
-                            break;
-                    }
-                } catch (Throwable e) {
+                    exitOperation(ctx, opType, newComponent, failedOperation);
+                } catch (Throwable e) { // NOSONAR: Log and re-throw
                     if (LOGGER.isLoggable(Level.SEVERE)) {
                         LOGGER.log(Level.SEVERE, e.getMessage(), e);
                     }
@@ -326,7 +278,8 @@
                     if (failedOperation && (opType == LSMOperationType.MODIFICATION
                             || opType == LSMOperationType.FORCE_MODIFICATION)) {
                         //When the operation failed, completeOperation() method must be called
-                        //in order to decrement active operation count which was incremented in beforeOperation() method.
+                        //in order to decrement active operation count which was incremented
+                        // in beforeOperation() method.
                         opTracker.completeOperation(lsmIndex, opType, ctx.getSearchOperationCallback(),
                                 ctx.getModificationCallback());
                     } else {
@@ -344,9 +297,9 @@
                     if (!inactiveDiskComponents.isEmpty()) {
                         for (ILSMDiskComponent inactiveComp : inactiveDiskComponents) {
                             if (inactiveComp.getFileReferenceCount() == 1) {
-                                if (inactiveDiskComponentsToBeDeleted == null) {
-                                    inactiveDiskComponentsToBeDeleted = new LinkedList<>();
-                                }
+                                inactiveDiskComponentsToBeDeleted =
+                                        inactiveDiskComponentsToBeDeleted == null ? new LinkedList<>()
+                                                : inactiveDiskComponentsToBeDeleted;
                                 inactiveDiskComponentsToBeDeleted.add(inactiveComp);
                             }
                         }
@@ -370,15 +323,97 @@
                     for (ILSMDiskComponent c : inactiveDiskComponentsToBeDeleted) {
                         c.deactivateAndDestroy();
                     }
-                } catch (Throwable e) {
+                } catch (Throwable e) { // NOSONAR Log and re-throw
                     if (LOGGER.isLoggable(Level.WARNING)) {
                         LOGGER.log(Level.WARNING, "Failure scheduling replication or destroying merged component", e);
                     }
-                    throw e;
+                    throw e; // NOSONAR: The last call in the finally clause
                 }
             }
         }
+    }
 
+    private void exitOperation(ILSMIndexOperationContext ctx, LSMOperationType opType, ILSMDiskComponent newComponent,
+            boolean failedOperation) throws HyracksDataException {
+        // Then, perform any action that is needed to be taken based on the operation type.
+        switch (opType) {
+            case FLUSH:
+                // newComponent is null if the flush op. was not performed.
+                if (!failedOperation && newComponent != null) {
+                    lsmIndex.addDiskComponent(newComponent);
+                    if (replicationEnabled) {
+                        componentsToBeReplicated.clear();
+                        componentsToBeReplicated.add(newComponent);
+                        triggerReplication(componentsToBeReplicated, false, opType);
+                    }
+                    mergePolicy.diskComponentAdded(lsmIndex, false);
+                }
+                break;
+            case MERGE:
+                // newComponent is null if the merge op. was not performed.
+                if (!failedOperation && newComponent != null) {
+                    lsmIndex.subsumeMergedComponents(newComponent, ctx.getComponentHolder());
+                    if (replicationEnabled) {
+                        componentsToBeReplicated.clear();
+                        componentsToBeReplicated.add(newComponent);
+                        triggerReplication(componentsToBeReplicated, false, opType);
+                    }
+                    mergePolicy.diskComponentAdded(lsmIndex, fullMergeIsRequested.get());
+                }
+                break;
+            default:
+                break;
+        }
+    }
+
+    private void exitOperationalComponents(ILSMIndexOperationContext ctx, LSMOperationType opType,
+            boolean failedOperation) throws HyracksDataException {
+        // First check if there is any action that is needed to be taken
+        // based on the state of each component.
+        for (int i = 0; i < ctx.getComponentHolder().size(); i++) {
+            ILSMComponent c = ctx.getComponentHolder().get(i);
+            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:
+                        tracer.instant(c.toString(), traceCategory, Scope.p, lsmIndex.toString());
+                        ((AbstractLSMMemoryComponent) c).reset();
+                        // Notify all waiting threads whenever the mutable component's state
+                        // has changed to inactive. This is important because even though we switched
+                        // the mutable components, it is possible that the component that we just
+                        // switched to is still busy flushing its data to disk. Thus, the notification
+                        // that was issued upon scheduling the flush is not enough.
+                        opTracker.notifyAll(); // NOSONAR: Always called inside synchronized block
+                        break;
+                    default:
+                        break;
+                }
+            } else if (c.getState() == ComponentState.INACTIVE) {
+                lsmIndex.addInactiveDiskComponent((AbstractLSMDiskComponent) c);
+            }
+        }
+    }
+
+    private void exitComponents(ILSMIndexOperationContext ctx, LSMOperationType opType, ILSMDiskComponent newComponent,
+            boolean failedOperation) throws HyracksDataException {
+        long before = 0L;
+        if (ctx.isTracingEnabled()) {
+            before = System.nanoTime();
+        }
+        try {
+            doExitComponents(ctx, opType, newComponent, failedOperation);
+        } finally {
+            if (ctx.isTracingEnabled()) {
+                ctx.incrementEnterExitTime(System.nanoTime() - before);
+            }
+        }
     }
 
     @Override
@@ -532,7 +567,7 @@
             newComponent = lsmIndex.flush(operation);
             operation.getCallback().afterOperation(LSMIOOperationType.FLUSH, null, newComponent);
             newComponent.markAsValid(lsmIndex.isDurable());
-        } catch (Throwable e) {
+        } catch (Throwable e) { // NOSONAR Log and re-throw
             failedOperation = true;
             if (LOGGER.isLoggable(Level.SEVERE)) {
                 LOGGER.log(Level.SEVERE, "Flush failed on " + lsmIndex, e);
@@ -562,8 +597,9 @@
             throws HyracksDataException {
         fullMergeIsRequested.set(true);
         if (!getAndEnterComponents(ctx, LSMOperationType.MERGE, true)) {
-            // If the merge cannot be scheduled because there is already an ongoing merge on subset/all of the components, then
-            // whenever the current merge has finished, it will schedule the full merge again.
+            // If the merge cannot be scheduled because there is already an ongoing merge on
+            // subset/all of the components, then whenever the current merge has finished,
+            // it will schedule the full merge again.
             callback.afterFinalize(LSMIOOperationType.MERGE, null);
             return;
         }
@@ -583,7 +619,7 @@
             newComponent = lsmIndex.merge(operation);
             operation.getCallback().afterOperation(LSMIOOperationType.MERGE, ctx.getComponentHolder(), newComponent);
             newComponent.markAsValid(lsmIndex.isDurable());
-        } catch (Throwable e) {
+        } catch (Throwable e) { // NOSONAR: Log and re-throw
             failedOperation = true;
             if (LOGGER.isLoggable(Level.SEVERE)) {
                 LOGGER.log(Level.SEVERE, "Failed merge operation on " + lsmIndex, e);
@@ -707,6 +743,7 @@
             throw e;
         } finally {
             exit(ctx);
+            ctx.logPerformanceCounters(accessor.getTupleCount());
         }
     }
 
@@ -878,4 +915,36 @@
     public String toString() {
         return getClass().getSimpleName() + ":" + lsmIndex;
     }
+
+    @Override
+    public void replaceMemoryComponentsWithDiskComponents(ILSMIndexOperationContext ctx, int startIndex)
+            throws HyracksDataException {
+        synchronized (opTracker) {
+            componentReplacementCtx.reset();
+            for (int i = 0; i < ctx.getComponentHolder().size(); i++) {
+                if (i >= startIndex) {
+                    ILSMComponent next = ctx.getComponentHolder().get(i);
+                    if (next.getType() == LSMComponentType.MEMORY
+                            && next.getState() == ComponentState.UNREADABLE_UNWRITABLE) {
+                        componentReplacementCtx.getComponentHolder().add(next);
+                        componentReplacementCtx.swapIndex(i);
+                    }
+                }
+            }
+            if (componentReplacementCtx.getComponentHolder().isEmpty()) {
+                throw new IllegalStateException(
+                        "replaceMemoryComponentsWithDiskComponents called with no potential components");
+            }
+            // before we exit, we should keep the replaced component ids
+            // we should also ensure that exact disk component replacement exist
+            if (componentReplacementCtx.proceed(lsmIndex.getDiskComponents())) {
+                // exit old component
+                exitComponents(componentReplacementCtx, LSMOperationType.SEARCH, null, false);
+                // enter new component
+                componentReplacementCtx.prepareToEnter();
+                enterComponents(componentReplacementCtx, LSMOperationType.SEARCH);
+                componentReplacementCtx.replace(ctx);
+            }
+        }
+    }
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMIndexDiskComponentBulkLoader.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMIndexDiskComponentBulkLoader.java
index 000d5cf..7bc0660 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMIndexDiskComponentBulkLoader.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMIndexDiskComponentBulkLoader.java
@@ -62,7 +62,7 @@
                 //then after operation should be called from harness as well
                 //https://issues.apache.org/jira/browse/ASTERIXDB-1764
                 lsmIndex.getIOOperationCallback().afterOperation(LSMIOOperationType.LOAD, null, component);
-                lsmIndex.getLsmHarness().addBulkLoadedComponent(component);
+                lsmIndex.getHarness().addBulkLoadedComponent(component);
             }
         } finally {
             lsmIndex.getIOOperationCallback().afterFinalize(LSMIOOperationType.LOAD, component);
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMIndexReplicationJob.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMIndexReplicationJob.java
index a35d00e..1c28ef1 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMIndexReplicationJob.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMIndexReplicationJob.java
@@ -45,7 +45,7 @@
     @Override
     public void endReplication() throws HyracksDataException {
         if (operationContext != null) {
-            ((AbstractLSMIndex) (lsmIndex)).getLsmHarness().endReplication(operationContext);
+            ((AbstractLSMIndex) (lsmIndex)).getHarness().endReplication(operationContext);
         }
     }
 
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMIndexSearchCursor.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMIndexSearchCursor.java
index 724a909..17c681c 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMIndexSearchCursor.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMIndexSearchCursor.java
@@ -24,6 +24,7 @@
 import java.util.PriorityQueue;
 
 import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
 import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
 import org.apache.hyracks.storage.am.common.api.ITreeIndexCursor;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponent;
@@ -36,9 +37,13 @@
 import org.apache.hyracks.storage.common.buffercache.IBufferCache;
 
 public abstract class LSMIndexSearchCursor implements ITreeIndexCursor {
+    protected static final int SWITCH_COMPONENT_CYCLE = 100;
     protected final ILSMIndexOperationContext opCtx;
     protected final boolean returnDeletedTuples;
     protected PriorityQueueElement outputElement;
+    protected final ArrayTupleBuilder[] switchComponentTupleBuilders;
+    protected final boolean[] switchRequest;
+    protected final PriorityQueueElement[] switchedElements;
     protected IIndexCursor[] rangeCursors;
     protected PriorityQueueElement[] pqes;
     protected PriorityQueue<PriorityQueueElement> outputPriorityQueue;
@@ -47,6 +52,8 @@
     protected boolean needPushElementIntoQueue;
     protected boolean includeMutableComponent;
     protected ILSMHarness lsmHarness;
+    protected boolean switchPossible = true;
+    protected int hasNextCallCount = 0;
 
     protected List<ILSMComponent> operationalComponents;
 
@@ -55,6 +62,9 @@
         this.returnDeletedTuples = returnDeletedTuples;
         outputElement = null;
         needPushElementIntoQueue = false;
+        switchComponentTupleBuilders = new ArrayTupleBuilder[opCtx.getIndex().getNumberOfAllMemoryComponents()];
+        switchRequest = new boolean[switchComponentTupleBuilders.length];
+        switchedElements = new PriorityQueueElement[switchComponentTupleBuilders.length];
     }
 
     public ILSMIndexOperationContext getOpCtx() {
@@ -98,9 +108,13 @@
 
     @Override
     public void reset() throws HyracksDataException {
+        hasNextCallCount = 0;
+        switchPossible = true;
         outputElement = null;
         needPushElementIntoQueue = false;
-
+        for (int i = 0; i < switchRequest.length; i++) {
+            switchRequest[i] = false;
+        }
         try {
             if (outputPriorityQueue != null) {
                 outputPriorityQueue.clear();
@@ -121,6 +135,7 @@
 
     @Override
     public boolean hasNext() throws HyracksDataException {
+        hasNextCallCount++;
         checkPriorityQueue();
         return !outputPriorityQueue.isEmpty();
     }
@@ -196,7 +211,7 @@
     }
 
     protected void checkPriorityQueue() throws HyracksDataException {
-        while (!outputPriorityQueue.isEmpty() || (needPushElementIntoQueue == true)) {
+        while (!outputPriorityQueue.isEmpty() || needPushElementIntoQueue) {
             if (!outputPriorityQueue.isEmpty()) {
                 PriorityQueueElement checkElement = outputPriorityQueue.peek();
                 // If there is no previous tuple or the previous tuple can be ignored
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndex.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndex.java
index 9e9f78a..030e4fd 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndex.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndex.java
@@ -237,7 +237,7 @@
                     deletedKeysBTreeAccessors,
                     ((LSMInvertedIndexMemoryComponent) memoryComponents.get(currentMutableComponentId.get()))
                             .getBuddyIndex().getLeafFrameFactory(),
-                    ictx, includeMutableComponent, getLsmHarness(), operationalComponents);
+                    ictx, includeMutableComponent, getHarness(), operationalComponents);
         } else {
             LSMInvertedIndexMemoryComponent mutableComponent =
                     (LSMInvertedIndexMemoryComponent) memoryComponents.get(currentMutableComponentId.get());
@@ -246,7 +246,7 @@
             initState = new LSMInvertedIndexRangeSearchCursorInitialState(tokensAndKeysCmp, keyCmp, keysOnlyTuple,
                     ((LSMInvertedIndexMemoryComponent) memoryComponents.get(currentMutableComponentId.get()))
                             .getBuddyIndex().getLeafFrameFactory(),
-                    includeMutableComponent, getLsmHarness(), indexAccessors, deletedKeysBTreeAccessors, pred,
+                    includeMutableComponent, getHarness(), indexAccessors, deletedKeysBTreeAccessors, pred,
                     operationalComponents);
         }
         return initState;
@@ -427,15 +427,15 @@
 
     @Override
     public ILSMIndexAccessor createAccessor(IIndexAccessParameters iap) throws HyracksDataException {
-        return new LSMInvertedIndexAccessor(getLsmHarness(),
+        return new LSMInvertedIndexAccessor(getHarness(),
                 createOpContext(iap.getModificationCallback(), iap.getSearchOperationCallback()));
     }
 
     @Override
     protected LSMInvertedIndexOpContext createOpContext(IModificationOperationCallback modificationCallback,
             ISearchOperationCallback searchCallback) throws HyracksDataException {
-        return new LSMInvertedIndexOpContext(memoryComponents, modificationCallback, searchCallback,
-                invertedIndexFieldsForNonBulkLoadOps, filterFieldsForNonBulkLoadOps, getFilterCmpFactories());
+        return new LSMInvertedIndexOpContext(this, memoryComponents, modificationCallback, searchCallback,
+                invertedIndexFieldsForNonBulkLoadOps, filterFieldsForNonBulkLoadOps, getFilterCmpFactories(), tracer);
     }
 
     @Override
@@ -481,7 +481,7 @@
     protected ILSMIOOperation createFlushOperation(AbstractLSMIndexOperationContext opCtx,
             LSMComponentFileReferences componentFileRefs, ILSMIOOperationCallback callback)
             throws HyracksDataException {
-        return new LSMInvertedIndexFlushOperation(new LSMInvertedIndexAccessor(getLsmHarness(), opCtx),
+        return new LSMInvertedIndexFlushOperation(new LSMInvertedIndexAccessor(getHarness(), opCtx),
                 componentFileRefs.getInsertIndexFileReference(), componentFileRefs.getDeleteIndexFileReference(),
                 componentFileRefs.getBloomFilterFileReference(), callback, fileManager.getBaseDir().getAbsolutePath());
     }
@@ -489,7 +489,7 @@
     @Override
     protected ILSMIOOperation createMergeOperation(AbstractLSMIndexOperationContext opCtx,
             LSMComponentFileReferences mergeFileRefs, ILSMIOOperationCallback callback) throws HyracksDataException {
-        ILSMIndexAccessor accessor = new LSMInvertedIndexAccessor(getLsmHarness(), opCtx);
+        ILSMIndexAccessor accessor = new LSMInvertedIndexAccessor(getHarness(), opCtx);
         IIndexCursor cursor = new LSMInvertedIndexRangeSearchCursor(opCtx);
         return new LSMInvertedIndexMergeOperation(accessor, cursor, mergeFileRefs.getInsertIndexFileReference(),
                 mergeFileRefs.getDeleteIndexFileReference(), mergeFileRefs.getBloomFilterFileReference(), callback,
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexOpContext.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexOpContext.java
index 12341da..55da252 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexOpContext.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexOpContext.java
@@ -25,12 +25,14 @@
 import org.apache.hyracks.api.exceptions.HyracksDataException;
 import org.apache.hyracks.storage.am.common.impls.NoOpIndexAccessParameters;
 import org.apache.hyracks.storage.am.common.tuples.PermutingTupleReference;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMMemoryComponent;
 import org.apache.hyracks.storage.am.lsm.common.impls.AbstractLSMIndexOperationContext;
 import org.apache.hyracks.storage.am.lsm.invertedindex.api.IInvertedIndexAccessor;
 import org.apache.hyracks.storage.common.IIndexAccessor;
 import org.apache.hyracks.storage.common.IModificationOperationCallback;
 import org.apache.hyracks.storage.common.ISearchOperationCallback;
+import org.apache.hyracks.util.trace.ITracer;
 
 public class LSMInvertedIndexOpContext extends AbstractLSMIndexOperationContext {
 
@@ -45,11 +47,12 @@
     private IInvertedIndexAccessor currentMutableInvIndexAccessors;
     private IIndexAccessor currentDeletedKeysBTreeAccessors;
 
-    public LSMInvertedIndexOpContext(List<ILSMMemoryComponent> mutableComponents,
+    public LSMInvertedIndexOpContext(ILSMIndex index, List<ILSMMemoryComponent> mutableComponents,
             IModificationOperationCallback modificationCallback, ISearchOperationCallback searchCallback,
-            int[] invertedIndexFields, int[] filterFields, IBinaryComparatorFactory[] filterComparatorFactories)
-            throws HyracksDataException {
-        super(invertedIndexFields, filterFields, filterComparatorFactories, searchCallback, modificationCallback);
+            int[] invertedIndexFields, int[] filterFields, IBinaryComparatorFactory[] filterComparatorFactories,
+            ITracer tracer) throws HyracksDataException {
+        super(index, invertedIndexFields, filterFields, filterComparatorFactories, searchCallback, modificationCallback,
+                tracer);
         mutableInvIndexAccessors = new IInvertedIndexAccessor[mutableComponents.size()];
         deletedKeysBTreeAccessors = new IIndexAccessor[mutableComponents.size()];
         for (int i = 0; i < mutableComponents.size(); i++) {
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/ondisk/OnDiskInvertedIndexOpContext.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/ondisk/OnDiskInvertedIndexOpContext.java
index fda3775..623fc74 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/ondisk/OnDiskInvertedIndexOpContext.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/ondisk/OnDiskInvertedIndexOpContext.java
@@ -81,4 +81,5 @@
     public MultiComparator getPrefixSearchCmp() {
         return prefixSearchCmp;
     }
+
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/dataflow/ExternalRTreeLocalResource.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/dataflow/ExternalRTreeLocalResource.java
index 6918b19..9960590 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/dataflow/ExternalRTreeLocalResource.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/dataflow/ExternalRTreeLocalResource.java
@@ -72,7 +72,7 @@
                 bloomFilterFalsePositiveRate, mergePolicyFactory.createMergePolicy(mergePolicyProperties, ncServiceCtx),
                 opTrackerProvider.getOperationTracker(ncServiceCtx), ioSchedulerProvider.getIoScheduler(ncServiceCtx),
                 ioOpCallbackFactory, linearizeCmpFactory, buddyBTreeFields, durable, isPointMBR,
-                metadataPageManagerFactory);
+                metadataPageManagerFactory, ncServiceCtx.getTracer());
 
     }
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/AbstractLSMRTree.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/AbstractLSMRTree.java
index 152b0a8..aac8905 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/AbstractLSMRTree.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/AbstractLSMRTree.java
@@ -127,9 +127,9 @@
             ILinearizeComparatorFactory linearizer, int[] comparatorFields, IBinaryComparatorFactory[] linearizerArray,
             double bloomFilterFalsePositiveRate, ILSMMergePolicy mergePolicy, ILSMOperationTracker opTracker,
             ILSMIOOperationScheduler ioScheduler, ILSMIOOperationCallbackFactory ioOpCallbackFactory, boolean durable,
-            boolean isPointMBR) {
+            boolean isPointMBR, ITracer tracer) {
         super(ioManager, diskBufferCache, fileManager, bloomFilterFalsePositiveRate, mergePolicy, opTracker,
-                ioScheduler, ioOpCallbackFactory, componentFactory, componentFactory, durable);
+                ioScheduler, ioOpCallbackFactory, componentFactory, componentFactory, durable, tracer);
         this.rtreeInteriorFrameFactory = rtreeInteriorFrameFactory;
         this.rtreeLeafFrameFactory = rtreeLeafFrameFactory;
         this.btreeInteriorFrameFactory = btreeInteriorFrameFactory;
@@ -231,9 +231,9 @@
     @Override
     protected LSMRTreeOpContext createOpContext(IModificationOperationCallback modCallback,
             ISearchOperationCallback searchCallback) {
-        return new LSMRTreeOpContext(memoryComponents, rtreeLeafFrameFactory, rtreeInteriorFrameFactory,
-                btreeLeafFrameFactory, modCallback, searchCallback, getTreeFields(), getFilterFields(), getLsmHarness(),
-                comparatorFields, linearizerArray, getFilterCmpFactories());
+        return new LSMRTreeOpContext(this, memoryComponents, rtreeLeafFrameFactory, rtreeInteriorFrameFactory,
+                btreeLeafFrameFactory, modCallback, searchCallback, getTreeFields(), getFilterFields(), getHarness(),
+                comparatorFields, linearizerArray, getFilterCmpFactories(), tracer);
     }
 
     @Override
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/ExternalRTree.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/ExternalRTree.java
index ca24b13..6f61935 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/ExternalRTree.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/ExternalRTree.java
@@ -61,6 +61,7 @@
 import org.apache.hyracks.storage.common.ISearchOperationCallback;
 import org.apache.hyracks.storage.common.ISearchPredicate;
 import org.apache.hyracks.storage.common.buffercache.IBufferCache;
+import org.apache.hyracks.util.trace.ITracer;
 
 /**
  * This is an lsm r-tree that does not have memory component and is modified
@@ -89,18 +90,18 @@
             int[] comparatorFields, IBinaryComparatorFactory[] linearizerArray, ILSMMergePolicy mergePolicy,
             ILSMOperationTracker opTracker, ILSMIOOperationScheduler ioScheduler,
             ILSMIOOperationCallbackFactory ioOpCallbackFactory, int[] buddyBTreeFields, boolean durable,
-            boolean isPointMBR) {
+            boolean isPointMBR, ITracer tracer) {
         super(ioManager, rtreeInteriorFrameFactory, rtreeLeafFrameFactory, btreeInteriorFrameFactory,
                 btreeLeafFrameFactory, diskBufferCache, fileNameManager, componentFactory, bloomFilterFalsePositiveRate,
                 rtreeCmpFactories, btreeCmpFactories, linearizer, comparatorFields, linearizerArray, mergePolicy,
-                opTracker, ioScheduler, ioOpCallbackFactory, buddyBTreeFields, durable, isPointMBR);
+                opTracker, ioScheduler, ioOpCallbackFactory, buddyBTreeFields, durable, isPointMBR, tracer);
         this.secondDiskComponents = new LinkedList<>();
         this.fieldCount = fieldCount;
     }
 
     @Override
-    public ExternalIndexHarness getLsmHarness() {
-        return (ExternalIndexHarness) super.getLsmHarness();
+    public ExternalIndexHarness getHarness() {
+        return (ExternalIndexHarness) super.getHarness();
     }
 
     // The subsume merged components is overridden to account for:
@@ -207,7 +208,7 @@
                 diskComponents.add(component);
                 secondDiskComponents.add(component);
             }
-            getLsmHarness().indexFirstTimeActivated();
+            getHarness().indexFirstTimeActivated();
         } else {
             // This index has been opened before or is brand new with no components
             // components. It should also maintain the version pointer
@@ -349,7 +350,7 @@
         if (!isActive) {
             throw new HyracksDataException("Failed to clear the index since it is not activated.");
         }
-        getLsmHarness().indexClear();
+        getHarness().indexClear();
 
         for (ILSMDiskComponent c : diskComponents) {
             c.deactivateAndDestroy();
@@ -532,7 +533,7 @@
                     component.markAsValid(durable);
                     component.deactivate();
                 } else {
-                    getLsmHarness().addBulkLoadedComponent(component);
+                    getHarness().addBulkLoadedComponent(component);
                 }
             }
         }
@@ -587,7 +588,7 @@
         LSMComponentFileReferences relMergeFileRefs =
                 getMergeFileReferences((ILSMDiskComponent) mergingComponents.get(0),
                         (ILSMDiskComponent) mergingComponents.get(mergingComponents.size() - 1));
-        ILSMIndexAccessor accessor = new LSMRTreeAccessor(getLsmHarness(), rctx, buddyBTreeFields);
+        ILSMIndexAccessor accessor = new LSMRTreeAccessor(getHarness(), rctx, buddyBTreeFields);
         // create the merge operation.
         LSMRTreeMergeOperation mergeOp = new LSMRTreeMergeOperation(accessor, cursor,
                 relMergeFileRefs.getInsertIndexFileReference(), relMergeFileRefs.getDeleteIndexFileReference(),
@@ -598,22 +599,22 @@
     @Override
     public ILSMIndexAccessor createAccessor(ISearchOperationCallback searchCallback, int targetIndexVersion)
             throws HyracksDataException {
-        return new LSMRTreeAccessor(getLsmHarness(), createOpContext(searchCallback, targetIndexVersion),
+        return new LSMRTreeAccessor(getHarness(), createOpContext(searchCallback, targetIndexVersion),
                 buddyBTreeFields);
     }
 
     // This method creates the appropriate opContext for the targeted version
     public ExternalRTreeOpContext createOpContext(ISearchOperationCallback searchCallback, int targetVersion) {
-        return new ExternalRTreeOpContext(rtreeCmpFactories, btreeCmpFactories, searchCallback, targetVersion,
-                getLsmHarness(), comparatorFields, linearizerArray, rtreeLeafFrameFactory, rtreeInteriorFrameFactory,
-                btreeLeafFrameFactory);
+        return new ExternalRTreeOpContext(this, rtreeCmpFactories, btreeCmpFactories, searchCallback, targetVersion,
+                getHarness(), comparatorFields, linearizerArray, rtreeLeafFrameFactory, rtreeInteriorFrameFactory,
+                btreeLeafFrameFactory, tracer);
     }
 
     // The accessor for disk only indexes don't use modification callback and
     // always carry the target index version with them
     @Override
     public ILSMIndexAccessor createAccessor(IIndexAccessParameters iap) {
-        return new LSMRTreeAccessor(getLsmHarness(), createOpContext(iap.getSearchOperationCallback(), version),
+        return new LSMRTreeAccessor(getHarness(), createOpContext(iap.getSearchOperationCallback(), version),
                 buddyBTreeFields);
     }
 
@@ -646,7 +647,7 @@
                     componentFileRefrences.getDeleteIndexFileReference(),
                     componentFileRefrences.getBloomFilterFileReference(), false);
         }
-        getLsmHarness().addTransactionComponents(component);
+        getHarness().addTransactionComponents(component);
     }
 
     @Override
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/ExternalRTreeOpContext.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/ExternalRTreeOpContext.java
index 73be537..09237a3 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/ExternalRTreeOpContext.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/ExternalRTreeOpContext.java
@@ -21,9 +21,11 @@
 import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
 import org.apache.hyracks.storage.am.common.api.ITreeIndexFrameFactory;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMHarness;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
 import org.apache.hyracks.storage.am.lsm.common.impls.AbstractLSMIndexOperationContext;
 import org.apache.hyracks.storage.common.ISearchOperationCallback;
 import org.apache.hyracks.storage.common.MultiComparator;
+import org.apache.hyracks.util.trace.ITracer;
 
 public class ExternalRTreeOpContext extends AbstractLSMIndexOperationContext {
     private MultiComparator bTreeCmp;
@@ -31,12 +33,13 @@
     private final int targetIndexVersion;
     private LSMRTreeCursorInitialState initialState;
 
-    public ExternalRTreeOpContext(IBinaryComparatorFactory[] rtreeCmpFactories,
+    public ExternalRTreeOpContext(ILSMIndex index, IBinaryComparatorFactory[] rtreeCmpFactories,
             IBinaryComparatorFactory[] btreeCmpFactories, ISearchOperationCallback searchCallback,
             int targetIndexVersion, ILSMHarness lsmHarness, int[] comparatorFields,
             IBinaryComparatorFactory[] linearizerArray, ITreeIndexFrameFactory rtreeLeafFrameFactory,
-            ITreeIndexFrameFactory rtreeInteriorFrameFactory, ITreeIndexFrameFactory btreeLeafFrameFactory) {
-        super(null, null, null, searchCallback, null);
+            ITreeIndexFrameFactory rtreeInteriorFrameFactory, ITreeIndexFrameFactory btreeLeafFrameFactory,
+            ITracer tracer) {
+        super(index, null, null, null, searchCallback, null, tracer);
         this.targetIndexVersion = targetIndexVersion;
         this.bTreeCmp = MultiComparator.create(btreeCmpFactories);
         this.rTreeCmp = MultiComparator.create(rtreeCmpFactories);
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/LSMRTree.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/LSMRTree.java
index 73cb206..9712b7c 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/LSMRTree.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/LSMRTree.java
@@ -63,6 +63,7 @@
 import org.apache.hyracks.storage.common.IIndexCursor;
 import org.apache.hyracks.storage.common.ISearchPredicate;
 import org.apache.hyracks.storage.common.buffercache.IBufferCache;
+import org.apache.hyracks.util.trace.ITracer;
 
 public class LSMRTree extends AbstractLSMRTree {
     protected final int[] buddyBTreeFields;
@@ -99,11 +100,11 @@
             int[] comparatorFields, IBinaryComparatorFactory[] linearizerArray, ILSMMergePolicy mergePolicy,
             ILSMOperationTracker opTracker, ILSMIOOperationScheduler ioScheduler,
             ILSMIOOperationCallbackFactory ioOpCallbackFactory, int[] buddyBTreeFields, boolean durable,
-            boolean isPointMBR) {
+            boolean isPointMBR, ITracer tracer) {
         super(ioManager, rtreeInteriorFrameFactory, rtreeLeafFrameFactory, btreeInteriorFrameFactory,
                 btreeLeafFrameFactory, diskBufferCache, fileNameManager, componentFactory, rtreeCmpFactories,
                 btreeCmpFactories, linearizer, comparatorFields, linearizerArray, bloomFilterFalsePositiveRate,
-                mergePolicy, opTracker, ioScheduler, ioOpCallbackFactory, durable, isPointMBR);
+                mergePolicy, opTracker, ioScheduler, ioOpCallbackFactory, durable, isPointMBR, tracer);
         this.buddyBTreeFields = buddyBTreeFields;
     }
 
@@ -276,7 +277,7 @@
 
     @Override
     public ILSMIndexAccessor createAccessor(IIndexAccessParameters iap) {
-        return new LSMRTreeAccessor(getLsmHarness(),
+        return new LSMRTreeAccessor(getHarness(),
                 createOpContext(iap.getModificationCallback(), iap.getSearchOperationCallback()), buddyBTreeFields);
     }
 
@@ -321,7 +322,7 @@
     protected ILSMIOOperation createFlushOperation(AbstractLSMIndexOperationContext opCtx,
             LSMComponentFileReferences componentFileRefs, ILSMIOOperationCallback callback)
             throws HyracksDataException {
-        LSMRTreeAccessor accessor = new LSMRTreeAccessor(getLsmHarness(), opCtx, buddyBTreeFields);
+        LSMRTreeAccessor accessor = new LSMRTreeAccessor(getHarness(), opCtx, buddyBTreeFields);
         return new LSMRTreeFlushOperation(accessor, componentFileRefs.getInsertIndexFileReference(),
                 componentFileRefs.getDeleteIndexFileReference(), componentFileRefs.getBloomFilterFileReference(),
                 callback, fileManager.getBaseDir().getAbsolutePath());
@@ -331,7 +332,7 @@
     protected ILSMIOOperation createMergeOperation(AbstractLSMIndexOperationContext opCtx,
             LSMComponentFileReferences mergeFileRefs, ILSMIOOperationCallback callback) throws HyracksDataException {
         ITreeIndexCursor cursor = new LSMRTreeSortedCursor(opCtx, linearizer, buddyBTreeFields);
-        ILSMIndexAccessor accessor = new LSMRTreeAccessor(getLsmHarness(), opCtx, buddyBTreeFields);
+        ILSMIndexAccessor accessor = new LSMRTreeAccessor(getHarness(), opCtx, buddyBTreeFields);
         return new LSMRTreeMergeOperation(accessor, cursor, mergeFileRefs.getInsertIndexFileReference(),
                 mergeFileRefs.getDeleteIndexFileReference(), mergeFileRefs.getBloomFilterFileReference(), callback,
                 fileManager.getBaseDir().getAbsolutePath());
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/LSMRTreeOpContext.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/LSMRTreeOpContext.java
index f54f8b9..50f1961 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/LSMRTreeOpContext.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/LSMRTreeOpContext.java
@@ -29,6 +29,7 @@
 import org.apache.hyracks.storage.am.common.impls.NoOpOperationCallback;
 import org.apache.hyracks.storage.am.common.ophelpers.IndexOperation;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMHarness;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMMemoryComponent;
 import org.apache.hyracks.storage.am.lsm.common.impls.AbstractLSMIndexOperationContext;
 import org.apache.hyracks.storage.am.rtree.impls.RTree;
@@ -37,6 +38,7 @@
 import org.apache.hyracks.storage.common.ISearchOperationCallback;
 import org.apache.hyracks.storage.common.ISearchPredicate;
 import org.apache.hyracks.storage.common.MultiComparator;
+import org.apache.hyracks.util.trace.ITracer;
 
 public final class LSMRTreeOpContext extends AbstractLSMIndexOperationContext {
 
@@ -50,12 +52,14 @@
     private BTreeOpContext currentBTreeOpContext;
     private LSMRTreeCursorInitialState searchInitialState;
 
-    public LSMRTreeOpContext(List<ILSMMemoryComponent> mutableComponents, ITreeIndexFrameFactory rtreeLeafFrameFactory,
-            ITreeIndexFrameFactory rtreeInteriorFrameFactory, ITreeIndexFrameFactory btreeLeafFrameFactory,
-            IModificationOperationCallback modificationCallback, ISearchOperationCallback searchCallback,
-            int[] rtreeFields, int[] filterFields, ILSMHarness lsmHarness, int[] comparatorFields,
-            IBinaryComparatorFactory[] linearizerArray, IBinaryComparatorFactory[] filterComparatorFactories) {
-        super(rtreeFields, filterFields, filterComparatorFactories, searchCallback, modificationCallback);
+    public LSMRTreeOpContext(ILSMIndex index, List<ILSMMemoryComponent> mutableComponents,
+            ITreeIndexFrameFactory rtreeLeafFrameFactory, ITreeIndexFrameFactory rtreeInteriorFrameFactory,
+            ITreeIndexFrameFactory btreeLeafFrameFactory, IModificationOperationCallback modificationCallback,
+            ISearchOperationCallback searchCallback, int[] rtreeFields, int[] filterFields, ILSMHarness lsmHarness,
+            int[] comparatorFields, IBinaryComparatorFactory[] linearizerArray,
+            IBinaryComparatorFactory[] filterComparatorFactories, ITracer tracer) {
+        super(index, rtreeFields, filterFields, filterComparatorFactories, searchCallback, modificationCallback,
+                tracer);
         mutableRTreeAccessors = new RTree.RTreeAccessor[mutableComponents.size()];
         mutableBTreeAccessors = new BTree.BTreeAccessor[mutableComponents.size()];
         rtreeOpContexts = new RTreeOpContext[mutableComponents.size()];
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/LSMRTreeWithAntiMatterTuples.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/LSMRTreeWithAntiMatterTuples.java
index a20eeb7..41e9b92 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/LSMRTreeWithAntiMatterTuples.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/LSMRTreeWithAntiMatterTuples.java
@@ -214,14 +214,14 @@
     @Override
     public ILSMIndexAccessor createAccessor(IIndexAccessParameters iap) {
         LSMRTreeOpContext opCtx = createOpContext(iap.getModificationCallback(), iap.getSearchOperationCallback());
-        return new LSMTreeIndexAccessor(getLsmHarness(), opCtx, cursorFactory);
+        return new LSMTreeIndexAccessor(getHarness(), opCtx, cursorFactory);
     }
 
     @Override
     protected ILSMIOOperation createFlushOperation(AbstractLSMIndexOperationContext opCtx,
             LSMComponentFileReferences componentFileRefs, ILSMIOOperationCallback callback)
             throws HyracksDataException {
-        ILSMIndexAccessor accessor = new LSMTreeIndexAccessor(getLsmHarness(), opCtx, cursorFactory);
+        ILSMIndexAccessor accessor = new LSMTreeIndexAccessor(getHarness(), opCtx, cursorFactory);
         return new LSMRTreeFlushOperation(accessor, componentFileRefs.getInsertIndexFileReference(), null, null,
                 callback, fileManager.getBaseDir().getAbsolutePath());
     }
@@ -235,7 +235,7 @@
             returnDeletedTuples = true;
         }
         ITreeIndexCursor cursor = new LSMRTreeWithAntiMatterTuplesSearchCursor(opCtx, returnDeletedTuples);
-        ILSMIndexAccessor accessor = new LSMTreeIndexAccessor(getLsmHarness(), opCtx, cursorFactory);
+        ILSMIndexAccessor accessor = new LSMTreeIndexAccessor(getHarness(), opCtx, cursorFactory);
         return new LSMRTreeMergeOperation(accessor, cursor, mergeFileRefs.getInsertIndexFileReference(), null, null,
                 callback, fileManager.getBaseDir().getAbsolutePath());
     }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/utils/LSMRTreeUtils.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/utils/LSMRTreeUtils.java
index c9df11f..451b122 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/utils/LSMRTreeUtils.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/utils/LSMRTreeUtils.java
@@ -72,6 +72,7 @@
 import org.apache.hyracks.storage.am.rtree.linearize.ZCurveIntComparatorFactory;
 import org.apache.hyracks.storage.am.rtree.tuples.RTreeTypeAwareTupleWriterFactory;
 import org.apache.hyracks.storage.common.buffercache.IBufferCache;
+import org.apache.hyracks.util.trace.ITracer;
 
 public class LSMRTreeUtils {
     public static LSMRTree createLSMTree(IIOManager ioManager, List<IVirtualBufferCache> virtualBufferCaches,
@@ -243,7 +244,7 @@
             ILSMOperationTracker opTracker, ILSMIOOperationScheduler ioScheduler,
             ILSMIOOperationCallbackFactory ioOpCallbackFactory, ILinearizeComparatorFactory linearizeCmpFactory,
             int[] buddyBTreeFields, boolean durable, boolean isPointMBR,
-            IMetadataPageManagerFactory freePageManagerFactory) throws HyracksDataException {
+            IMetadataPageManagerFactory freePageManagerFactory, ITracer tracer) throws HyracksDataException {
 
         int keyFieldCount = rtreeCmpFactories.length;
         int valueFieldCount = typeTraits.length - keyFieldCount;
@@ -290,7 +291,7 @@
                 btreeLeafFrameFactory, diskBufferCache, fileNameManager, componentFactory, bloomFilterFalsePositiveRate,
                 typeTraits.length, rtreeCmpFactories, btreeCmpFactories, linearizeCmpFactory, comparatorFields,
                 linearizerArray, mergePolicy, opTracker, ioScheduler, ioOpCallbackFactory, buddyBTreeFields, durable,
-                isPointMBR);
+                isPointMBR, tracer);
     }
 
     public static ILinearizeComparatorFactory proposeBestLinearizer(ITypeTraits[] typeTraits, int numKeyFields)
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/TestLsmBtree.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/TestLsmBtree.java
index 70a29db..8782565 100644
--- a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/TestLsmBtree.java
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/TestLsmBtree.java
@@ -60,6 +60,7 @@
     private final List<ITestOpCallback> searchCallbacks = new ArrayList<>();
     private final List<ITestOpCallback> flushCallbacks = new ArrayList<>();
     private final List<ITestOpCallback> mergeCallbacks = new ArrayList<>();
+    private final List<ITestOpCallback> allocateComponentCallbacks = new ArrayList<>();
 
     private volatile int numScheduledFlushes;
     private volatile int numStartedFlushes;
@@ -171,7 +172,7 @@
 
     @Override
     public ILSMIndexAccessor createAccessor(AbstractLSMIndexOperationContext opCtx) {
-        return new LSMTreeIndexAccessor(getLsmHarness(), opCtx, ctx -> new TestLsmBtreeSearchCursor(ctx, this));
+        return new LSMTreeIndexAccessor(getHarness(), opCtx, ctx -> new TestLsmBtreeSearchCursor(ctx, this));
     }
 
     public int getNumScheduledFlushes() {
@@ -257,4 +258,26 @@
     public Semaphore getSearchSemaphore() {
         return searchSemaphore;
     }
+
+    public void addAllocateCallback(ITestOpCallback callback) {
+        synchronized (allocateComponentCallbacks) {
+            allocateComponentCallbacks.add(callback);
+        }
+    }
+
+    public void clearAllocateCallbacks() {
+        synchronized (allocateComponentCallbacks) {
+            allocateComponentCallbacks.clear();
+        }
+    }
+
+    @Override
+    public void allocateMemoryComponents() throws HyracksDataException {
+        super.allocateMemoryComponents();
+        synchronized (allocateComponentCallbacks) {
+            for (ITestOpCallback callback : allocateComponentCallbacks) {
+                callback(callback, null);
+            }
+        }
+    }
 }