diff --git a/asterixdb/asterix-app/pom.xml b/asterixdb/asterix-app/pom.xml
index 35344ab..ab938d8 100644
--- a/asterixdb/asterix-app/pom.xml
+++ b/asterixdb/asterix-app/pom.xml
@@ -486,7 +486,7 @@
           **/SqlppExecutionWithCancellationTest.java,**/DmlTest.java,**/RepeatedTest.java,**/SqlppExecutionTest.java,
           **/SqlppExecutionColumnTest.java,**/*StaticPartitioning*Test.java,**/*Ssl*Test.java,**/Podman*.java,
           **/*AnalyzedExecutionTest.java,**/SqlppProfiledExecutionTest.java,**/CloudPythonTest.java,
-          **/CloudStorageAzTest.java,**/AzureBlobStorageExternalDatasetTest.java,
+          **/CloudStorageAzTest.java,**/AzureBlobStorageExternalDatasetTest.java,**/GCSCloudStorageUnstableTest,
           **/AzureBlobStorageExternalDatasetOnePartitionTest.java,**/SqlppSinglePointLookupExecutionTest.java,
           **/Atomic*.java, **/AwsS3*.java, **/*SqlppHdfs*.java, **/*RQGTest.java, **/*RQJTest.java
         </test.excludes>
@@ -553,7 +553,7 @@
       <id>asterix-gerrit-cloud-nons3-tests</id>
       <properties>
         <test.includes>
-          **/CloudStorageGCSTest.java, **/CloudStorageAzTest.java,**/AzureBlobStorageExternalDatasetTest.java,
+          **/GCSCloudStorageUnstableTest.java,**/CloudStorageAzTest.java,**/AzureBlobStorageExternalDatasetTest.java,
           **/AzureBlobStorageExternalDatasetOnePartitionTest.java,**/CloudStorageUnstableTest.java, **/*SqlppHdfs*.java
         </test.includes>
         <failIfNoTests>false</failIfNoTests>
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/HaltCallback.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/HaltCallback.java
index 466b617..f66f097 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/HaltCallback.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/HaltCallback.java
@@ -47,12 +47,9 @@
     }
 
     private boolean haltOnFailure(ILSMIOOperation operation) {
-        switch (operation.getIOOperationType()) {
-            case CLEANUP:
-            case REPLICATE:
-                return false;
-            default:
-                return true;
-        }
+        return switch (operation.getIOOperationType()) {
+            case CLEANUP, REPLICATE, MERGE -> false;
+            default -> true;
+        };
     }
 }
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
index f7a6dc1..fde71f8 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
@@ -481,7 +481,7 @@
                         if (stats.getProfileType() == Stats.ProfileType.FULL) {
                             this.jobFlags.add(JobFlag.PROFILE_RUNTIME);
                         }
-                        handleLoadStatement(metadataProvider, stmt, hcc);
+                        handleLoadStatement(metadataProvider, stmt, hcc, requestParameters);
                         break;
                     case COPY_FROM:
                         if (stats.getProfileType() == Stats.ProfileType.FULL) {
@@ -4025,8 +4025,8 @@
         }
     }
 
-    protected void handleLoadStatement(MetadataProvider metadataProvider, Statement stmt, IHyracksClientConnection hcc)
-            throws Exception {
+    protected void handleLoadStatement(MetadataProvider metadataProvider, Statement stmt, IHyracksClientConnection hcc,
+            IRequestParameters requestParameters) throws Exception {
         LoadStatement loadStmt = (LoadStatement) stmt;
         String datasetName = loadStmt.getDatasetName();
         metadataProvider.validateDatabaseObjectName(loadStmt.getNamespace(), datasetName, loadStmt.getSourceLocation());
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/CloudStorageGCSTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/CloudStorageGCSTest.java
index 65b5adf..8c071f0 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/CloudStorageGCSTest.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/CloudStorageGCSTest.java
@@ -40,6 +40,7 @@
 import org.junit.Assume;
 import org.junit.BeforeClass;
 import org.junit.FixMethodOrder;
+import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.MethodSorters;
@@ -57,6 +58,7 @@
  * Run tests in cloud deployment environment
  */
 @RunWith(Parameterized.class)
+@Ignore
 @FixMethodOrder(MethodSorters.NAME_ASCENDING)
 public class CloudStorageGCSTest {
 
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/GCSCloudStorageUnstableTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/GCSCloudStorageUnstableTest.java
new file mode 100644
index 0000000..c97e459
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/GCSCloudStorageUnstableTest.java
@@ -0,0 +1,149 @@
+/*
+ * 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.cloud_storage;
+
+import static org.apache.asterix.api.common.LocalCloudUtil.CLOUD_STORAGE_BUCKET;
+import static org.apache.asterix.api.common.LocalCloudUtil.MOCK_SERVER_REGION;
+
+import java.util.Collection;
+import java.util.List;
+
+import org.apache.asterix.api.common.LocalCloudUtilAdobeMock;
+import org.apache.asterix.common.api.IDatasetLifecycleManager;
+import org.apache.asterix.common.api.INcApplicationContext;
+import org.apache.asterix.common.config.GlobalConfig;
+import org.apache.asterix.common.storage.StorageIOStats;
+import org.apache.asterix.test.common.TestExecutor;
+import org.apache.asterix.test.runtime.ExecutionTestUtil;
+import org.apache.asterix.test.runtime.LangExecutionUtil;
+import org.apache.asterix.testframework.context.TestCaseContext;
+import org.apache.asterix.testframework.xml.Description;
+import org.apache.asterix.testframework.xml.TestCase;
+import org.apache.hyracks.cloud.util.CloudRetryableRequestUtil;
+import org.apache.hyracks.control.nc.NodeControllerService;
+import org.apache.hyracks.storage.common.buffercache.BufferCache;
+import org.apache.hyracks.storage.common.buffercache.IBufferCache;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.Assume;
+import org.junit.BeforeClass;
+import org.junit.FixMethodOrder;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.MethodSorters;
+import org.junit.runners.Parameterized;
+
+import com.google.cloud.NoCredentials;
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.StorageClass;
+import com.google.cloud.storage.StorageOptions;
+
+/**
+ * Run tests in cloud deployment environment
+ */
+@RunWith(Parameterized.class)
+@FixMethodOrder(MethodSorters.NAME_ASCENDING)
+public class GCSCloudStorageUnstableTest {
+
+    private static final Logger LOGGER = LogManager.getLogger();
+
+    private final TestCaseContext tcCtx;
+    private static final String SUITE_TESTS = "testsuite_cloud_storage.xml";
+    private static final String ONLY_TESTS = "testsuite_cloud_storage_only.xml";
+    private static final String CONFIG_FILE_NAME = "src/test/resources/cc-cloud-storage-gcs.conf";
+    private static final String DELTA_RESULT_PATH = "results_cloud";
+    private static final String EXCLUDED_TESTS = "MP";
+    public static final String MOCK_SERVER_HOSTNAME = "http://127.0.0.1:24443";
+    private static final String MOCK_SERVER_PROJECT_ID = "asterixdb-gcs-test-project-id";
+
+    public GCSCloudStorageUnstableTest(TestCaseContext tcCtx) {
+        this.tcCtx = tcCtx;
+    }
+
+    @BeforeClass
+    public static void setUp() throws Exception {
+        LocalCloudUtilAdobeMock.startS3CloudEnvironment(true, true);
+        System.setProperty(CloudRetryableRequestUtil.CLOUD_UNSTABLE_MODE, "true");
+        Storage storage = StorageOptions.newBuilder().setHost(MOCK_SERVER_HOSTNAME)
+                .setCredentials(NoCredentials.getInstance()).setProjectId(MOCK_SERVER_PROJECT_ID).build().getService();
+        cleanup(storage);
+        initialize(storage);
+        storage.close();
+        TestExecutor testExecutor = new TestExecutor(DELTA_RESULT_PATH);
+        testExecutor.executorId = "cloud";
+        testExecutor.stripSubstring = "//DB:";
+        LangExecutionUtil.setUp(CONFIG_FILE_NAME, testExecutor);
+        System.setProperty(GlobalConfig.CONFIG_FILE_PROPERTY, CONFIG_FILE_NAME);
+    }
+
+    @AfterClass
+    public static void tearDown() throws Exception {
+        System.clearProperty(CloudRetryableRequestUtil.CLOUD_UNSTABLE_MODE);
+        LangExecutionUtil.tearDown();
+    }
+
+    @Parameterized.Parameters(name = "GCSCloudStorageUnstableTest {index}: {0}")
+    public static Collection<Object[]> tests() throws Exception {
+        return LangExecutionUtil.tests(ONLY_TESTS, SUITE_TESTS);
+    }
+
+    @Test
+    public void test() throws Exception {
+        List<TestCase.CompilationUnit> cu = tcCtx.getTestCase().getCompilationUnit();
+        Assume.assumeTrue(cu.size() > 1 || !EXCLUDED_TESTS.equals(getText(cu.get(0).getDescription())));
+        LangExecutionUtil.test(tcCtx);
+        for (NodeControllerService nc : ExecutionTestUtil.integrationUtil.ncs) {
+            IDatasetLifecycleManager lifecycleManager =
+                    ((INcApplicationContext) nc.getApplicationContext()).getDatasetLifecycleManager();
+            StorageIOStats stats = lifecycleManager.getDatasetsIOStats();
+            while (stats.getPendingFlushes() != 0 || stats.getPendingMerges() != 0) {
+                stats = lifecycleManager.getDatasetsIOStats();
+            }
+        }
+        IBufferCache bufferCache;
+        for (NodeControllerService nc : ExecutionTestUtil.integrationUtil.ncs) {
+            bufferCache = ((INcApplicationContext) nc.getApplicationContext()).getBufferCache();
+            Assert.assertTrue(((BufferCache) bufferCache).isClean());
+        }
+    }
+
+    private static String getText(Description description) {
+        return description == null ? "" : description.getValue();
+    }
+
+    private static void cleanup(Storage storage) {
+        try {
+            Iterable<Blob> blobs = storage.list(CLOUD_STORAGE_BUCKET).iterateAll();
+            blobs.forEach(Blob::delete);
+            storage.delete(CLOUD_STORAGE_BUCKET);
+        } catch (Exception ex) {
+            // ignore
+        }
+    }
+
+    private static void initialize(Storage storage) {
+        storage.create(BucketInfo.newBuilder(CLOUD_STORAGE_BUCKET).setStorageClass(StorageClass.STANDARD)
+                .setLocation(MOCK_SERVER_REGION).build());
+    }
+}
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/RetryingQueryTranslator.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/RetryingQueryTranslator.java
new file mode 100644
index 0000000..af580b2
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/RetryingQueryTranslator.java
@@ -0,0 +1,157 @@
+/*
+ * 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.cloud_storage;
+
+import static org.apache.asterix.cloud.clients.UnstableCloudClient.ERROR_RATE;
+
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+
+import org.apache.asterix.app.translator.QueryTranslator;
+import org.apache.asterix.common.api.IResponsePrinter;
+import org.apache.asterix.common.dataflow.ICcApplicationContext;
+import org.apache.asterix.compiler.provider.ILangCompilationProvider;
+import org.apache.asterix.lang.common.base.Statement;
+import org.apache.asterix.lang.common.statement.LoadStatement;
+import org.apache.asterix.lang.common.statement.TruncateDatasetStatement;
+import org.apache.asterix.lang.common.struct.Identifier;
+import org.apache.asterix.metadata.declared.MetadataProvider;
+import org.apache.asterix.metadata.utils.Creator;
+import org.apache.asterix.translator.IRequestParameters;
+import org.apache.asterix.translator.SessionOutput;
+import org.apache.hyracks.api.client.IHyracksClientConnection;
+import org.apache.hyracks.api.exceptions.ErrorCode;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.api.util.ExceptionUtils;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+public class RetryingQueryTranslator extends QueryTranslator {
+
+    private static final Logger LOGGER = LogManager.getLogger();
+
+    public RetryingQueryTranslator(ICcApplicationContext appCtx, List<Statement> statements, SessionOutput output,
+            ILangCompilationProvider compilationProvider, ExecutorService executorService,
+            IResponsePrinter responsePrinter) {
+        super(appCtx, statements, output, compilationProvider, executorService, responsePrinter);
+    }
+
+    @Override
+    public void handleCreateIndexStatement(MetadataProvider metadataProvider, Statement stmt,
+            IHyracksClientConnection hcc, IRequestParameters requestParameters, Creator creator) throws Exception {
+        int times = 100;
+        Exception ex = null;
+        double initialErrorRate = ERROR_RATE.get();
+        try {
+            while (times-- > 0) {
+                try {
+                    super.handleCreateIndexStatement(metadataProvider, stmt, hcc, requestParameters, creator);
+                    ex = null;
+                    break;
+                } catch (Exception e) {
+                    ERROR_RATE.set(Double.max(ERROR_RATE.get() - 0.01d, 0.01d));
+                    if (retryOnFailure(e)) {
+                        LOGGER.error("Attempt: {}, Failed to create index", 100 - times, e);
+                        metadataProvider.getLocks().reset();
+                        ex = e;
+                    } else {
+                        throw e;
+                    }
+                }
+            }
+            if (ex != null) {
+                throw ex;
+            }
+        } finally {
+            ERROR_RATE.set(initialErrorRate);
+        }
+    }
+
+    @Override
+    public void handleAnalyzeStatement(MetadataProvider metadataProvider, Statement stmt, IHyracksClientConnection hcc,
+            IRequestParameters requestParameters) throws Exception {
+        int times = 100;
+        Exception ex = null;
+        while (times-- > 0) {
+            try {
+                super.handleAnalyzeStatement(metadataProvider, stmt, hcc, requestParameters);
+                ex = null;
+                break;
+            } catch (Exception e) {
+                if (retryOnFailure(e)) {
+                    LOGGER.error("Attempt: {}, Failed to create index", 100 - times, e);
+                    metadataProvider.getLocks().reset();
+                    ex = e;
+                } else {
+                    throw e;
+                }
+            }
+        }
+        if (ex != null) {
+            throw ex;
+        }
+    }
+
+    @Override
+    public void handleLoadStatement(MetadataProvider metadataProvider, Statement stmt, IHyracksClientConnection hcc,
+            IRequestParameters requestParameters) throws Exception {
+        int times = 100;
+        Exception ex = null;
+        double initialErrorRate = ERROR_RATE.get();
+        try {
+            while (times-- > 0) {
+                try {
+                    super.handleLoadStatement(metadataProvider, stmt, hcc, requestParameters);
+                    ex = null;
+                    break;
+                } catch (Exception e) {
+                    ERROR_RATE.set(Double.max(ERROR_RATE.get() - 0.01d, 0.01d));
+                    if (retryOnFailure(e)) {
+                        LOGGER.error("Attempt: {}, Failed to load", 100 - times, e);
+                        metadataProvider.getLocks().reset();
+                        ex = e;
+                        LoadStatement loadStmt = (LoadStatement) stmt;
+                        TruncateDatasetStatement truncateDatasetStatement = new TruncateDatasetStatement(
+                                loadStmt.getNamespace(), new Identifier(loadStmt.getDatasetName()), true);
+                        super.handleDatasetTruncateStatement(metadataProvider, truncateDatasetStatement,
+                                requestParameters);
+                        metadataProvider.getLocks().reset();
+
+                    } else {
+                        throw e;
+                    }
+                }
+            }
+            if (ex != null) {
+                throw ex;
+            }
+        } finally {
+            ERROR_RATE.set(initialErrorRate);
+        }
+    }
+
+    private boolean retryOnFailure(Exception e) {
+        if (e instanceof HyracksDataException) {
+            return ((HyracksDataException) e).getErrorCode() == ErrorCode.FAILED_IO_OPERATION.intValue()
+                    && ExceptionUtils.getRootCause(e).getMessage().contains("Simulated error");
+        }
+        return false;
+    }
+}
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/UnstableStatementExecutorExtension.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/UnstableStatementExecutorExtension.java
new file mode 100644
index 0000000..bf9aecb
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/UnstableStatementExecutorExtension.java
@@ -0,0 +1,55 @@
+/*
+ * 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.cloud_storage;
+
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+
+import org.apache.asterix.app.cc.IStatementExecutorExtension;
+import org.apache.asterix.common.api.ExtensionId;
+import org.apache.asterix.translator.IStatementExecutorFactory;
+import org.apache.hyracks.algebricks.common.utils.Pair;
+import org.apache.hyracks.api.application.IServiceContext;
+
+public class UnstableStatementExecutorExtension implements IStatementExecutorExtension {
+
+    public static final ExtensionId RETRYING_QUERY_TRANSLATOR_EXTENSION_ID =
+            new ExtensionId(UnstableStatementExecutorExtension.class.getSimpleName(), 0);
+
+    @Override
+    public ExtensionId getId() {
+        return RETRYING_QUERY_TRANSLATOR_EXTENSION_ID;
+    }
+
+    @Override
+    public void configure(List<Pair<String, String>> args, IServiceContext serviceCtx) {
+
+    }
+
+    @Override
+    public IStatementExecutorFactory getQueryTranslatorFactory() {
+        return null;
+    }
+
+    @Override
+    public IStatementExecutorFactory getStatementExecutorFactory(ExecutorService executorService) {
+        return new UnstableStatementExecutorFactory(executorService);
+    }
+}
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/UnstableStatementExecutorFactory.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/UnstableStatementExecutorFactory.java
new file mode 100644
index 0000000..f65276b
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/UnstableStatementExecutorFactory.java
@@ -0,0 +1,46 @@
+/*
+ * 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.cloud_storage;
+
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+
+import org.apache.asterix.app.translator.DefaultStatementExecutorFactory;
+import org.apache.asterix.app.translator.QueryTranslator;
+import org.apache.asterix.common.api.IResponsePrinter;
+import org.apache.asterix.common.context.IStorageComponentProvider;
+import org.apache.asterix.common.dataflow.ICcApplicationContext;
+import org.apache.asterix.compiler.provider.ILangCompilationProvider;
+import org.apache.asterix.lang.common.base.Statement;
+import org.apache.asterix.translator.SessionOutput;
+
+public class UnstableStatementExecutorFactory extends DefaultStatementExecutorFactory {
+    public UnstableStatementExecutorFactory(ExecutorService executorService) {
+        super(executorService);
+    }
+
+    @Override
+    public QueryTranslator create(ICcApplicationContext appCtx, List<Statement> statements, SessionOutput output,
+            ILangCompilationProvider compilationProvider, IStorageComponentProvider storageComponentProvider,
+            IResponsePrinter responsePrinter) {
+        return new RetryingQueryTranslator(appCtx, statements, output, compilationProvider, executorService,
+                responsePrinter);
+    }
+}
diff --git a/asterixdb/asterix-app/src/test/resources/cc-cloud-storage-gcs.conf b/asterixdb/asterix-app/src/test/resources/cc-cloud-storage-gcs.conf
index 9b6f547..376252e 100644
--- a/asterixdb/asterix-app/src/test/resources/cc-cloud-storage-gcs.conf
+++ b/asterixdb/asterix-app/src/test/resources/cc-cloud-storage-gcs.conf
@@ -73,3 +73,5 @@
 cloud.storage.cache.policy=selective
 cloud.max.write.requests.per.second=1000
 cloud.max.read.requests.per.second=5000
+
+[extension/org.apache.asterix.test.cloud_storage.UnstableStatementExecutorExtension]
diff --git a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/CloudResettableInputStream.java b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/CloudResettableInputStream.java
index a233ca5..9b832a2 100644
--- a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/CloudResettableInputStream.java
+++ b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/CloudResettableInputStream.java
@@ -24,6 +24,7 @@
 
 import org.apache.asterix.cloud.clients.ICloudBufferedWriter;
 import org.apache.asterix.cloud.clients.ICloudWriter;
+import org.apache.hyracks.api.exceptions.ErrorCode;
 import org.apache.hyracks.api.exceptions.HyracksDataException;
 import org.apache.hyracks.cloud.io.request.ICloudBeforeRetryRequest;
 import org.apache.hyracks.cloud.io.request.ICloudRequest;
@@ -158,7 +159,7 @@
                     CloudRetryableRequestUtil.runWithNoRetryOnInterruption(request, retry);
                 } catch (Exception e) {
                     LOGGER.error(e);
-                    throw HyracksDataException.create(e);
+                    throw HyracksDataException.create(ErrorCode.FAILED_IO_OPERATION, e);
                 }
             }
             bufferedWriter.finish();
@@ -190,7 +191,7 @@
         try {
             close();
         } catch (IOException e) {
-            throw HyracksDataException.create(e);
+            throw HyracksDataException.create(ErrorCode.FAILED_IO_OPERATION, e);
         }
     }
 
@@ -203,7 +204,7 @@
             CloudRetryableRequestUtil.runWithNoRetryOnInterruption(request, retry);
         } catch (Exception e) {
             LOGGER.error(e);
-            throw HyracksDataException.create(e);
+            throw HyracksDataException.create(ErrorCode.FAILED_IO_OPERATION, e);
         }
 
         writeBuffer.clear();
diff --git a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/UnstableCloudClient.java b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/UnstableCloudClient.java
index 28fa53e..1b39251 100644
--- a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/UnstableCloudClient.java
+++ b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/UnstableCloudClient.java
@@ -25,11 +25,15 @@
 import java.util.Collection;
 import java.util.Random;
 import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
 
 import org.apache.asterix.cloud.CloudResettableInputStream;
 import org.apache.asterix.cloud.IWriteBufferProvider;
 import org.apache.asterix.cloud.clients.aws.s3.S3CloudClient;
+import org.apache.asterix.cloud.clients.google.gcs.GCSCloudClient;
+import org.apache.asterix.cloud.clients.google.gcs.GCSWriter;
 import org.apache.asterix.cloud.clients.profiler.IRequestProfilerLimiter;
+import org.apache.hyracks.api.exceptions.ErrorCode;
 import org.apache.hyracks.api.exceptions.HyracksDataException;
 import org.apache.hyracks.api.io.FileReference;
 import org.apache.hyracks.control.nc.io.IOManager;
@@ -39,7 +43,7 @@
 
 public class UnstableCloudClient implements ICloudClient {
     // 10% error rate
-    private static final double ERROR_RATE = 0.1d;
+    public static final AtomicReference<Double> ERROR_RATE = new AtomicReference<>(0.11d);
     private static final Random RANDOM = new Random(0);
     private final ICloudClient cloudClient;
 
@@ -61,6 +65,9 @@
     public ICloudWriter createWriter(String bucket, String path, IWriteBufferProvider bufferProvider) {
         if (cloudClient instanceof S3CloudClient) {
             return createUnstableWriter((S3CloudClient) cloudClient, bucket, path, bufferProvider);
+        } else if (cloudClient instanceof GCSCloudClient) {
+            return new UnstableGCSCloudWriter(cloudClient.createWriter(bucket, path, bufferProvider),
+                    cloudClient.getWriteBufferSize());
         }
         return cloudClient.createWriter(bucket, path, bufferProvider);
     }
@@ -138,8 +145,8 @@
 
     private static void fail() throws HyracksDataException {
         double prob = RANDOM.nextInt(100) / 100.0d;
-        if (prob <= ERROR_RATE) {
-            throw HyracksDataException.create(new IOException("Simulated error"));
+        if (prob < ERROR_RATE.get()) {
+            throw HyracksDataException.create(ErrorCode.FAILED_IO_OPERATION, new IOException("Simulated error"));
         }
     }
 
@@ -150,6 +157,63 @@
         return new CloudResettableInputStream(bufferedWriter, bufferProvider);
     }
 
+    /**
+     * An unstable cloud writer that mimics the functionality of {@link GCSWriter}
+     */
+    private static class UnstableGCSCloudWriter implements ICloudWriter {
+        private final ICloudWriter writer;
+        private final int writeBufferSize;
+
+        UnstableGCSCloudWriter(ICloudWriter writer, int writeBufferSize) {
+            this.writer = writer;
+            this.writeBufferSize = writeBufferSize;
+        }
+
+        @Override
+        public int write(ByteBuffer header, ByteBuffer page) throws HyracksDataException {
+            return write(header) + write(page);
+        }
+
+        @Override
+        public int write(ByteBuffer page) throws HyracksDataException {
+            if (position() == 0) {
+                fail();
+            }
+            long uploadsToBeTriggered =
+                    ((position() + page.remaining()) / writeBufferSize) - (position() / writeBufferSize);
+            while (uploadsToBeTriggered-- > 0) {
+                fail();
+            }
+            return writer.write(page);
+        }
+
+        @Override
+        public void write(int b) throws HyracksDataException {
+            write(ByteBuffer.wrap(new byte[] { (byte) b }));
+        }
+
+        @Override
+        public int write(byte[] b, int off, int len) throws HyracksDataException {
+            return write(ByteBuffer.wrap(b, off, len));
+        }
+
+        @Override
+        public long position() {
+            return writer.position();
+        }
+
+        @Override
+        public void finish() throws HyracksDataException {
+            fail();
+            writer.finish();
+        }
+
+        @Override
+        public void abort() throws HyracksDataException {
+            writer.abort();
+        }
+    }
+
     private static class UnstableCloudBufferedWriter implements ICloudBufferedWriter {
         private final ICloudBufferedWriter bufferedWriter;
 
diff --git a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/aws/s3/S3BufferedWriter.java b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/aws/s3/S3BufferedWriter.java
index 53d6546..7bd7ad5 100644
--- a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/aws/s3/S3BufferedWriter.java
+++ b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/aws/s3/S3BufferedWriter.java
@@ -26,6 +26,7 @@
 import org.apache.asterix.cloud.clients.ICloudBufferedWriter;
 import org.apache.asterix.cloud.clients.ICloudGuardian;
 import org.apache.asterix.cloud.clients.profiler.IRequestProfilerLimiter;
+import org.apache.hyracks.api.exceptions.ErrorCode;
 import org.apache.hyracks.api.exceptions.HyracksDataException;
 import org.apache.hyracks.cloud.util.CloudRetryableRequestUtil;
 import org.apache.logging.log4j.LogManager;
@@ -131,7 +132,7 @@
         try {
             s3Client.completeMultipartUpload(request);
         } catch (Exception e) {
-            throw HyracksDataException.create(e);
+            throw HyracksDataException.create(ErrorCode.FAILED_IO_OPERATION, e);
         }
     }
 
diff --git a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSWriter.java b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSWriter.java
index 89d8fd5..3a83786 100644
--- a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSWriter.java
+++ b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSWriter.java
@@ -24,6 +24,7 @@
 import org.apache.asterix.cloud.clients.ICloudGuardian;
 import org.apache.asterix.cloud.clients.ICloudWriter;
 import org.apache.asterix.cloud.clients.profiler.IRequestProfilerLimiter;
+import org.apache.hyracks.api.exceptions.ErrorCode;
 import org.apache.hyracks.api.exceptions.HyracksDataException;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
@@ -71,15 +72,15 @@
         while (uploadsToBeTriggered-- > 0) {
             profiler.objectMultipartUpload();
         }
-        setUploadId();
 
         int written = 0;
         try {
+            setUploadId();
             while (page.hasRemaining()) {
                 written += writer.write(page);
             }
         } catch (IOException | RuntimeException e) {
-            throw HyracksDataException.create(e);
+            throw HyracksDataException.create(ErrorCode.FAILED_IO_OPERATION, e);
         }
 
         writtenBytes += written;
@@ -104,13 +105,14 @@
     @Override
     public void finish() throws HyracksDataException {
         guardian.checkWriteAccess(bucket, path);
-        setUploadId();
         profiler.objectMultipartUpload();
         try {
+            setUploadId();
+
             writer.close();
             writer = null;
         } catch (IOException | RuntimeException e) {
-            throw HyracksDataException.create(e);
+            throw HyracksDataException.create(ErrorCode.FAILED_IO_OPERATION, e);
         }
         log("FINISHED");
     }
diff --git a/asterixdb/asterix-column/src/main/java/org/apache/asterix/column/metadata/schema/ObjectSchemaNode.java b/asterixdb/asterix-column/src/main/java/org/apache/asterix/column/metadata/schema/ObjectSchemaNode.java
index 1f74fb3..4cf2d23 100644
--- a/asterixdb/asterix-column/src/main/java/org/apache/asterix/column/metadata/schema/ObjectSchemaNode.java
+++ b/asterixdb/asterix-column/src/main/java/org/apache/asterix/column/metadata/schema/ObjectSchemaNode.java
@@ -164,7 +164,10 @@
 
     public void abort(DataInputStream input, Map<AbstractSchemaNestedNode, RunLengthIntArray> definitionLevels)
             throws IOException {
-        definitionLevels.put(this, new RunLengthIntArray());
+        input.readByte(); // Skip the type tag, see ObjectSchemaNode#serialize
+        if (definitionLevels != null) {
+            definitionLevels.put(this, new RunLengthIntArray());
+        }
 
         int numberOfChildren = input.readInt();
 
diff --git a/asterixdb/asterix-column/src/main/java/org/apache/asterix/column/operation/lsm/flush/FlushColumnMetadata.java b/asterixdb/asterix-column/src/main/java/org/apache/asterix/column/operation/lsm/flush/FlushColumnMetadata.java
index 5aca3a4..9c58247 100644
--- a/asterixdb/asterix-column/src/main/java/org/apache/asterix/column/operation/lsm/flush/FlushColumnMetadata.java
+++ b/asterixdb/asterix-column/src/main/java/org/apache/asterix/column/operation/lsm/flush/FlushColumnMetadata.java
@@ -271,7 +271,8 @@
 
     @Override
     public void abort() throws HyracksDataException {
-        DataInputStream input = new DataInputStream(new ByteArrayInputStream(serializedMetadata.getByteArray()));
+        DataInputStream input = new DataInputStream(new ByteArrayInputStream(serializedMetadata.getByteArray(),
+                serializedMetadata.getStartOffset(), serializedMetadata.getLength()));
         try {
             abort(input);
         } catch (IOException e) {
@@ -280,6 +281,7 @@
     }
 
     private void abort(DataInputStream input) throws IOException {
+        input.skipBytes(OFFSETS_SIZE);
         level = -1;
         repeated = 0;
         changed = false;
@@ -290,6 +292,9 @@
         fieldNamesDictionary.abort(input);
         definitionLevels.clear();
         root.abort(input, definitionLevels);
+        if (metaRoot != null) {
+            metaRoot.abort(input, definitionLevels);
+        }
     }
 
     public static void deserializeWriters(DataInput input, List<IColumnValuesWriter> writers,
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMIOOperationCallback.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMIOOperationCallback.java
index 87aa3bd..52af329 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMIOOperationCallback.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMIOOperationCallback.java
@@ -141,6 +141,17 @@
         }
     }
 
+    @Override
+    public void afterFailure(ILSMIOOperation operation) {
+        if (isMerge(operation)) {
+            try {
+                ioManager.delete(getOperationMaskFilePath(operation));
+            } catch (HyracksDataException e) {
+                operation.getFailure().addSuppressed(e);
+            }
+        }
+    }
+
     protected void addComponentToCheckpoint(ILSMIOOperation operation) throws HyracksDataException {
         // will always update the checkpoint file even if no new component was created
         FileReference target = operation.getTarget();
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMSecondaryIndexBulkLoadNodePushable.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMSecondaryIndexBulkLoadNodePushable.java
index 21c3b27..17c5445 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMSecondaryIndexBulkLoadNodePushable.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMSecondaryIndexBulkLoadNodePushable.java
@@ -57,6 +57,7 @@
 
     private LSMIndexDiskComponentBulkLoader componentBulkLoader;
     private int currentComponentPos = -1;
+    private boolean failed = false;
 
     public LSMSecondaryIndexBulkLoadNodePushable(IHyracksTaskContext ctx, int partition, RecordDescriptor inputRecDesc,
             IIndexDataflowHelperFactory primaryIndexHelperFactory,
@@ -98,7 +99,7 @@
     public void close() throws HyracksDataException {
         HyracksDataException closeException = null;
         try {
-            endCurrentComponent();
+            endOrAbortCurrentComponent();
         } catch (HyracksDataException e) {
             closeException = e;
         }
@@ -151,6 +152,7 @@
     @Override
     public void fail() throws HyracksDataException {
         writer.fail();
+        failed = true;
     }
 
     @Override
@@ -177,15 +179,22 @@
         }
     }
 
-    private void endCurrentComponent() throws HyracksDataException {
+    private void endOrAbortCurrentComponent() throws HyracksDataException {
         if (componentBulkLoader != null) {
-            componentBulkLoader.end();
-            componentBulkLoader = null;
+            try {
+                if (!failed) {
+                    componentBulkLoader.end();
+                } else {
+                    componentBulkLoader.abort();
+                }
+            } finally {
+                componentBulkLoader = null;
+            }
         }
     }
 
     private void loadNewComponent(int componentPos) throws HyracksDataException {
-        endCurrentComponent();
+        endOrAbortCurrentComponent(); // This should never call componentBulkLoader.abort()
         int numTuples = getNumDeletedTuples(componentPos);
         ILSMDiskComponent primaryComponent = primaryIndex.getDiskComponents().get(componentPos);
         Map<String, Object> parameters = new HashMap<>();
diff --git a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/exceptions/ErrorCode.java b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/exceptions/ErrorCode.java
index 7cb107d..d7abf4c 100644
--- a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/exceptions/ErrorCode.java
+++ b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/exceptions/ErrorCode.java
@@ -162,6 +162,7 @@
     EXTRA_FIELD_IN_RESULT_NOT_FOUND_IN_SCHEMA(132),
     EMPTY_TYPE_INFERRED(133),
     SCHEMA_LIMIT_EXCEEDED(134),
+    FAILED_IO_OPERATION(135),
 
     // Compilation error codes.
     RULECOLLECTION_NOT_INSTANCE_OF_LIST(10000),
diff --git a/hyracks-fullstack/hyracks/hyracks-api/src/main/resources/errormsg/en.properties b/hyracks-fullstack/hyracks/hyracks-api/src/main/resources/errormsg/en.properties
index e1fbe30..e9569be 100644
--- a/hyracks-fullstack/hyracks/hyracks-api/src/main/resources/errormsg/en.properties
+++ b/hyracks-fullstack/hyracks/hyracks-api/src/main/resources/errormsg/en.properties
@@ -152,6 +152,7 @@
 132 = Extra field in the result, field '%1$s' does not exist at '%2$s' in the schema
 133 = Schema could not be inferred, empty types found in the result
 134 = Schema Limit exceeded, maximum number of heterogeneous schemas allowed : '%1$s'
+135 = An IO Operation has failed
 
 10000 = The given rule collection %1$s is not an instance of the List class.
 10001 = Cannot compose partition constraint %1$s with %2$s
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-bloomfilter/src/main/java/org/apache/hyracks/storage/am/bloomfilter/impls/BloomFilter.java b/hyracks-fullstack/hyracks/hyracks-storage-am-bloomfilter/src/main/java/org/apache/hyracks/storage/am/bloomfilter/impls/BloomFilter.java
index aa17474..abb03eb 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-bloomfilter/src/main/java/org/apache/hyracks/storage/am/bloomfilter/impls/BloomFilter.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-bloomfilter/src/main/java/org/apache/hyracks/storage/am/bloomfilter/impls/BloomFilter.java
@@ -386,13 +386,15 @@
 
         @Override
         public void end() throws HyracksDataException {
-            allocateAndInitMetaDataPage();
-            pageWriter.write(metaDataPage);
-            for (ICachedPage p : pages) {
-                pageWriter.write(p);
-            }
-            if (hasFailed()) {
-                throw HyracksDataException.create(getFailure());
+            try {
+                allocateAndInitMetaDataPage();
+                pageWriter.write(metaDataPage);
+                for (ICachedPage p : pages) {
+                    pageWriter.write(p);
+                }
+            } catch (HyracksDataException e) {
+                handleException();
+                throw e;
             }
             BloomFilter.this.numBits = numBits;
             BloomFilter.this.numHashes = numHashes;
@@ -401,8 +403,7 @@
             BloomFilter.this.version = BLOCKED_BLOOM_FILTER_VERSION;
         }
 
-        @Override
-        public void abort() throws HyracksDataException {
+        private void handleException() {
             for (ICachedPage p : pages) {
                 if (p != null) {
                     bufferCache.returnPage(p, false);
@@ -414,6 +415,11 @@
         }
 
         @Override
+        public void abort() throws HyracksDataException {
+            handleException();
+        }
+
+        @Override
         public void force() throws HyracksDataException {
             bufferCache.force(fileId, false);
         }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/IPageManager.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/IPageManager.java
index 208d3b7..ff28464 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/IPageManager.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/IPageManager.java
@@ -49,6 +49,12 @@
     void close(IPageWriteFailureCallback failureCallback) throws HyracksDataException;
 
     /**
+     * Return all the pages from the page manager
+     */
+    default void returnAllPages() {
+    }
+
+    /**
      * Create a metadata frame to be used for reading and writing to metadata pages
      *
      * @return a new metadata frame
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/IndexBulkLoadOperatorNodePushable.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/IndexBulkLoadOperatorNodePushable.java
index 0206f20..d606a5c 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/IndexBulkLoadOperatorNodePushable.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/IndexBulkLoadOperatorNodePushable.java
@@ -61,6 +61,7 @@
     protected final IIndexBulkLoader[] bulkLoaders;
     protected ITupleFilter tupleFilter;
     protected FrameTupleReference frameTuple;
+    private boolean failed = false;
 
     public IndexBulkLoadOperatorNodePushable(IIndexDataflowHelperFactory indexHelperFactory, IHyracksTaskContext ctx,
             int partition, int[] fieldPermutation, float fillFactor, boolean verifyInput, long numElementsHint,
@@ -151,6 +152,7 @@
     @Override
     public void fail() throws HyracksDataException {
         writer.fail();
+        failed = true;
     }
 
     protected void initializeBulkLoader(IIndex index, int indexId) throws HyracksDataException {
@@ -159,12 +161,28 @@
     }
 
     private void closeBulkLoaders() throws HyracksDataException {
+        HyracksDataException failure = null;
         for (IIndexBulkLoader bulkLoader : bulkLoaders) {
             // bulkloader can be null if an exception is thrown before it is initialized.
-            if (bulkLoader != null) {
-                bulkLoader.end();
+            try {
+                if (bulkLoader != null) {
+                    if (failure == null && !failed) {
+                        bulkLoader.end();
+                    } else {
+                        bulkLoader.abort();
+                    }
+                }
+            } catch (HyracksDataException e) {
+                if (failure == null) {
+                    failure = e;
+                } else {
+                    failure.addSuppressed(e);
+                }
             }
         }
+        if (failure != null) {
+            throw failure;
+        }
     }
 
     protected static void closeIndexes(IIndex[] indexes, IIndexDataflowHelper[] indexHelpers,
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/freepage/AppendOnlyLinkedMetadataPageManager.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/freepage/AppendOnlyLinkedMetadataPageManager.java
index dae01bf..9c51d92 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/freepage/AppendOnlyLinkedMetadataPageManager.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/freepage/AppendOnlyLinkedMetadataPageManager.java
@@ -152,6 +152,13 @@
         metadataPages.clear();
     }
 
+    @Override
+    public void returnAllPages() {
+        for (ICachedPage page : metadataPages) {
+            bufferCache.returnPage(page, false);
+        }
+    }
+
     /**
      * For storage on append-only media (such as HDFS), the meta data page has to be written last.
      * However, some implementations still write the meta data to the front. To deal with this as well
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/impls/AbstractTreeIndexBulkLoader.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/impls/AbstractTreeIndexBulkLoader.java
index d5f836c..5ef3aab 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/impls/AbstractTreeIndexBulkLoader.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/impls/AbstractTreeIndexBulkLoader.java
@@ -130,6 +130,7 @@
             }
         }
         releasedLatches = true;
+        freePageManager.returnAllPages();
     }
 
     @Override
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 05e78dd..3526d2f 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
@@ -238,73 +238,83 @@
         LSMBTreeFlushOperation flushOp = (LSMBTreeFlushOperation) operation;
         LSMBTreeMemoryComponent flushingComponent = (LSMBTreeMemoryComponent) flushOp.getFlushingComponent();
         IIndexAccessor accessor = flushingComponent.getIndex().createAccessor(NoOpIndexAccessParameters.INSTANCE);
-        ILSMDiskComponent component;
-        ILSMDiskComponentBulkLoader componentBulkLoader;
+        ILSMDiskComponent component = null;
+        ILSMDiskComponentBulkLoader componentBulkLoader = null;
         try {
-            RangePredicate nullPred = new RangePredicate(null, null, true, true, null, null);
-            long numElements = 0L;
-            if (hasBloomFilter) {
-                //count elements in btree for creating Bloomfilter
-                IIndexCursor countingCursor = ((BTreeAccessor) accessor).createCountingSearchCursor();
-                accessor.search(countingCursor, nullPred);
+            try {
+                RangePredicate nullPred = new RangePredicate(null, null, true, true, null, null);
+                long numElements = 0L;
+                if (hasBloomFilter) {
+                    //count elements in btree for creating Bloomfilter
+                    IIndexCursor countingCursor = ((BTreeAccessor) accessor).createCountingSearchCursor();
+                    accessor.search(countingCursor, nullPred);
+                    try {
+                        while (countingCursor.hasNext()) {
+                            countingCursor.next();
+                            ITupleReference countTuple = countingCursor.getTuple();
+                            numElements = IntegerPointable.getInteger(countTuple.getFieldData(0),
+                                    countTuple.getFieldStart(0));
+                        }
+                    } finally {
+                        try {
+                            countingCursor.close();
+                        } finally {
+                            countingCursor.destroy();
+                        }
+                    }
+                }
+                component = createDiskComponent(componentFactory, flushOp.getTarget(), null,
+                        flushOp.getBloomFilterTarget(), true);
+                componentBulkLoader = component.createBulkLoader(operation, 1.0f, false, numElements, false, false,
+                        false, pageWriteCallbackFactory.createPageWriteCallback());
+                IIndexCursor scanCursor = accessor.createSearchCursor(false);
+                accessor.search(scanCursor, nullPred);
                 try {
-                    while (countingCursor.hasNext()) {
-                        countingCursor.next();
-                        ITupleReference countTuple = countingCursor.getTuple();
-                        numElements =
-                                IntegerPointable.getInteger(countTuple.getFieldData(0), countTuple.getFieldStart(0));
+                    while (scanCursor.hasNext()) {
+                        scanCursor.next();
+                        // we can safely throw away updated tuples in secondary BTree components, because they correspond to
+                        // deleted tuples
+                        if (updateAware && ((LSMBTreeTupleReference) scanCursor.getTuple()).isUpdated()) {
+                            continue;
+                        }
+                        componentBulkLoader.add(scanCursor.getTuple());
                     }
                 } finally {
                     try {
-                        countingCursor.close();
+                        scanCursor.close();
                     } finally {
-                        countingCursor.destroy();
+                        scanCursor.destroy();
                     }
                 }
-            }
-            component = createDiskComponent(componentFactory, flushOp.getTarget(), null, flushOp.getBloomFilterTarget(),
-                    true);
-            componentBulkLoader = component.createBulkLoader(operation, 1.0f, false, numElements, false, false, false,
-                    pageWriteCallbackFactory.createPageWriteCallback());
-            IIndexCursor scanCursor = accessor.createSearchCursor(false);
-            accessor.search(scanCursor, nullPred);
-            try {
-                while (scanCursor.hasNext()) {
-                    scanCursor.next();
-                    // we can safely throw away updated tuples in secondary BTree components, because they correspond to
-                    // deleted tuples
-                    if (updateAware && ((LSMBTreeTupleReference) scanCursor.getTuple()).isUpdated()) {
-                        continue;
-                    }
-                    componentBulkLoader.add(scanCursor.getTuple());
-                }
             } finally {
-                try {
-                    scanCursor.close();
-                } finally {
-                    scanCursor.destroy();
-                }
+                accessor.destroy();
             }
-        } finally {
-            accessor.destroy();
+            if (component.getLSMComponentFilter() != null) {
+                List<ITupleReference> filterTuples = new ArrayList<>();
+                filterTuples.add(flushingComponent.getLSMComponentFilter().getMinTuple());
+                filterTuples.add(flushingComponent.getLSMComponentFilter().getMaxTuple());
+                getFilterManager().updateFilter(component.getLSMComponentFilter(), filterTuples,
+                        NoOpOperationCallback.INSTANCE);
+                getFilterManager().writeFilter(component.getLSMComponentFilter(), component.getMetadataHolder());
+            }
+            // Write metadata from memory component to disk
+            // Q. what about the merge operation? how do we resolve conflicts
+            // A. Through providing an appropriate ILSMIOOperationCallback
+            // Must not reset the metadata before the flush is completed
+            // Use the copy of the metadata in the opContext
+            // TODO This code should be in the callback and not in the index
+            flushingComponent.getMetadata().copy(component.getMetadata());
+            componentBulkLoader.end();
+        } catch (Throwable e) {
+            try {
+                if (componentBulkLoader != null) {
+                    componentBulkLoader.abort();
+                }
+            } catch (Throwable th) {
+                e.addSuppressed(th);
+            }
+            throw e;
         }
-        if (component.getLSMComponentFilter() != null) {
-            List<ITupleReference> filterTuples = new ArrayList<>();
-            filterTuples.add(flushingComponent.getLSMComponentFilter().getMinTuple());
-            filterTuples.add(flushingComponent.getLSMComponentFilter().getMaxTuple());
-            getFilterManager().updateFilter(component.getLSMComponentFilter(), filterTuples,
-                    NoOpOperationCallback.INSTANCE);
-            getFilterManager().writeFilter(component.getLSMComponentFilter(), component.getMetadataHolder());
-        }
-        // Write metadata from memory component to disk
-        // Q. what about the merge operation? how do we resolve conflicts
-        // A. Through providing an appropriate ILSMIOOperationCallback
-        // Must not reset the metadata before the flush is completed
-        // Use the copy of the metadata in the opContext
-        // TODO This code should be in the callback and not in the index
-        flushingComponent.getMetadata().copy(component.getMetadata());
-
-        componentBulkLoader.end();
 
         return component;
     }
@@ -313,7 +323,7 @@
     public ILSMDiskComponent doMerge(ILSMIOOperation operation) throws HyracksDataException {
         LSMBTreeMergeOperation mergeOp = (LSMBTreeMergeOperation) operation;
         IIndexCursor cursor = mergeOp.getCursor();
-        ILSMDiskComponent mergedComponent;
+        ILSMDiskComponent mergedComponent = null;
         ILSMDiskComponentBulkLoader componentBulkLoader = null;
         try {
             try {
@@ -349,6 +359,7 @@
                 getFilterManager().writeFilter(mergedComponent.getLSMComponentFilter(),
                         mergedComponent.getMetadataHolder());
             }
+            componentBulkLoader.end();
         } catch (Throwable e) { // NOSONAR.. As per the contract, we should either abort or end
             try {
                 if (componentBulkLoader != null) {
@@ -359,7 +370,6 @@
             }
             throw e;
         }
-        componentBulkLoader.end();
         return mergedComponent;
     }
 
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/AbstractLSMWithBloomFilterDiskComponent.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/AbstractLSMWithBloomFilterDiskComponent.java
index db36c72..c77d7ff 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/AbstractLSMWithBloomFilterDiskComponent.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/AbstractLSMWithBloomFilterDiskComponent.java
@@ -47,8 +47,13 @@
     public void markAsValid(boolean persist, IPageWriteFailureCallback callback) throws HyracksDataException {
         // The order of forcing the dirty page to be flushed is critical. The
         // bloom filter must be always done first.
-        ComponentUtils.markAsValid(getBloomFilterBufferCache(), getBloomFilter(), persist);
-        super.markAsValid(persist, callback);
+        try {
+            ComponentUtils.markAsValid(getBloomFilterBufferCache(), getBloomFilter(), persist);
+            super.markAsValid(persist, callback);
+        } catch (HyracksDataException ex) {
+            returnPages();
+            throw ex;
+        }
     }
 
     @Override
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/AbstractLSMWithBuddyDiskComponent.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/AbstractLSMWithBuddyDiskComponent.java
index fda48f0..158a3b8 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/AbstractLSMWithBuddyDiskComponent.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/AbstractLSMWithBuddyDiskComponent.java
@@ -40,8 +40,19 @@
 
     @Override
     public void markAsValid(boolean persist, IPageWriteFailureCallback callback) throws HyracksDataException {
-        super.markAsValid(persist, callback);
-        ComponentUtils.markAsValid(getBuddyIndex(), persist, callback);
+        try {
+            super.markAsValid(persist, callback);
+            ComponentUtils.markAsValid(getBuddyIndex(), persist, callback);
+        } catch (HyracksDataException ex) {
+            returnPages();
+            throw ex;
+        }
+    }
+
+    @Override
+    public void returnPages() {
+        getBuddyIndex().getPageManager().returnAllPages();
+        super.returnPages();
     }
 
     @Override
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMDiskComponent.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMDiskComponent.java
index 029a59d..e41a17e 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMDiskComponent.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMDiskComponent.java
@@ -137,4 +137,9 @@
     ILSMDiskComponentBulkLoader createBulkLoader(ILSMIOOperation operation, float fillFactor, boolean verifyInput,
             long numElementsHint, boolean checkIfEmptyIndex, boolean withFilter, boolean cleanupEmptyComponent,
             IPageWriteCallback callback) throws HyracksDataException;
+
+    /**
+     * Returns all pages of the component to the buffer cache
+     */
+    void returnPages();
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIOOperationCallback.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIOOperationCallback.java
index a778a4c..542cf47 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIOOperationCallback.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIOOperationCallback.java
@@ -59,6 +59,12 @@
     void afterFinalize(ILSMIOOperation operation) throws HyracksDataException;
 
     /**
+     * This method is called on an IO operation after the operation fails
+     */
+    default void afterFailure(ILSMIOOperation operation) {
+    }
+
+    /**
      * This method is called after the schduler is done with the IO operation
      * For operation that are not scheduled, this call is skipped
      *
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 baf0931..c7240f3 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
@@ -158,8 +158,18 @@
      */
     @Override
     public void markAsValid(boolean persist, IPageWriteFailureCallback callback) throws HyracksDataException {
-        ComponentUtils.markAsValid(getMetadataHolder(), persist, callback);
-        LOGGER.debug("marked {} as valid component with id {}", getIndex(), getId());
+        try {
+            ComponentUtils.markAsValid(getMetadataHolder(), persist, callback);
+            LOGGER.debug("marked {} as valid component with id {}", getIndex(), getId());
+        } catch (Exception e) {
+            returnPages();
+            throw e;
+        }
+    }
+
+    @Override
+    public void returnPages() {
+        ComponentUtils.returnPages(getMetadataHolder());
     }
 
     @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 4f7c624..3f60978 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
@@ -583,7 +583,12 @@
             throws HyracksDataException {
         ILSMDiskComponent component = factory.createComponent(this,
                 new LSMComponentFileReferences(insertFileReference, deleteIndexFileReference, bloomFilterFileRef));
-        component.activate(createComponent);
+        try {
+            component.activate(createComponent);
+        } catch (HyracksDataException e) {
+            component.returnPages();
+            throw e;
+        }
         return component;
     }
 
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/ChainedLSMDiskComponentBulkLoader.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/ChainedLSMDiskComponentBulkLoader.java
index 0214712..f90da79 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/ChainedLSMDiskComponentBulkLoader.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/ChainedLSMDiskComponentBulkLoader.java
@@ -127,9 +127,20 @@
     @Override
     public void abort() throws HyracksDataException {
         operation.setStatus(LSMIOOperationStatus.FAILURE);
-        final int bulkloadersCount = bulkloaderChain.size();
-        for (int i = 0; i < bulkloadersCount; i++) {
-            bulkloaderChain.get(i).abort();
+        HyracksDataException failure = null;
+        for (IChainedComponentBulkLoader componentBulkLoader : bulkloaderChain) {
+            try {
+                componentBulkLoader.abort();
+            } catch (HyracksDataException e) {
+                if (failure == null) {
+                    failure = e;
+                } else {
+                    failure.addSuppressed(e);
+                }
+            }
+        }
+        if (failure != null) {
+            throw failure;
         }
     }
 
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/EmptyComponent.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/EmptyComponent.java
index 9a112ee..0dfb12e 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/EmptyComponent.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/EmptyComponent.java
@@ -157,4 +157,9 @@
     public String toString() {
         return "EmptyComponent";
     }
+
+    @Override
+    public void returnPages() {
+        // Do nothing
+    }
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMComponentFileReferences.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMComponentFileReferences.java
index 9346b56..3928724 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMComponentFileReferences.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMComponentFileReferences.java
@@ -21,7 +21,7 @@
 
 import org.apache.hyracks.api.io.FileReference;
 
-public final class LSMComponentFileReferences {
+public class LSMComponentFileReferences {
 
     // The FileReference for the index that is used for inserting records of the component. For instance, this will be the FileReference of the RTree in one component of the LSM-RTree.
     private final FileReference insertIndexFileReference;
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 d019a08..69c6ea6 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
@@ -266,7 +266,7 @@
             if (inactiveMemoryComponentsToBeCleanedUp != null) {
                 cleanupInactiveMemoryComponents(inactiveMemoryComponentsToBeCleanedUp);
             }
-            if (opType == LSMOperationType.FLUSH) {
+            if (opType == LSMOperationType.FLUSH && !failedOperation) {
                 ILSMMemoryComponent flushingComponent = (ILSMMemoryComponent) ctx.getComponentHolder().get(0);
                 // We must call flushed without synchronizing on opTracker to avoid deadlocks
                 flushingComponent.flushed();
@@ -575,6 +575,7 @@
         // if the operation failed, we need to cleanup files
         if (operation.getStatus() == LSMIOOperationStatus.FAILURE) {
             operation.cleanup(lsmIndex.getBufferCache());
+            operation.getCallback().afterFailure(operation);
         }
     }
 
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 3e17cb0..88fa105 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
@@ -53,7 +53,7 @@
         try {
             componentBulkLoader.add(tuple);
         } catch (Throwable th) {
-            opCtx.getIoOperation().setFailure(th);
+            fail(th);
             throw th;
         }
     }
@@ -63,7 +63,7 @@
         try {
             componentBulkLoader.delete(tuple);
         } catch (Throwable th) {
-            opCtx.getIoOperation().setFailure(th);
+            fail(th);
             throw th;
         }
     }
@@ -83,6 +83,7 @@
     @Override
     public void abort() throws HyracksDataException {
         opCtx.getIoOperation().setStatus(LSMIOOperationStatus.FAILURE);
+        fail(null);
         try {
             try {
                 componentBulkLoader.abort();
@@ -115,6 +116,7 @@
             componentBulkLoader.end();
         } catch (Throwable th) { // NOSONAR Must not call afterFinalize without setting failure
             fail(th);
+            componentBulkLoader.abort();
             throw th;
         } finally {
             lsmIndex.getIOOperationCallback().afterFinalize(opCtx.getIoOperation());
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMInvertedComponentFileReferences.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMInvertedComponentFileReferences.java
new file mode 100644
index 0000000..7d10c13
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMInvertedComponentFileReferences.java
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.hyracks.storage.am.lsm.common.impls;
+
+import org.apache.hyracks.api.io.FileReference;
+
+public class LSMInvertedComponentFileReferences extends LSMComponentFileReferences {
+
+    private final FileReference invListsFileReference;
+
+    public LSMInvertedComponentFileReferences(FileReference insertIndexFileReference,
+            FileReference deleteIndexFileReference, FileReference bloomFilterFileReference,
+            FileReference invListsFileReference) {
+        super(insertIndexFileReference, deleteIndexFileReference, bloomFilterFileReference);
+        this.invListsFileReference = invListsFileReference;
+    }
+
+    public FileReference getInvListsFileReference() {
+        return invListsFileReference;
+    }
+
+    @Override
+    public FileReference[] getFileReferences() {
+        return new FileReference[] { getInsertIndexFileReference(), getDeleteIndexFileReference(),
+                getBloomFilterFileReference(), invListsFileReference };
+    }
+}
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMTreeIndexAccessor.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMTreeIndexAccessor.java
index c768768..52bd07e 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMTreeIndexAccessor.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMTreeIndexAccessor.java
@@ -41,6 +41,8 @@
 import org.apache.hyracks.storage.am.lsm.common.api.LSMOperationType;
 import org.apache.hyracks.storage.common.IIndexCursor;
 import org.apache.hyracks.storage.common.ISearchPredicate;
+import org.apache.hyracks.util.ExponentialRetryPolicy;
+import org.apache.hyracks.util.IRetryPolicy;
 
 public class LSMTreeIndexAccessor implements ILSMIndexAccessor {
     @FunctionalInterface
@@ -122,6 +124,18 @@
     @Override
     public void flush(ILSMIOOperation operation) throws HyracksDataException {
         lsmHarness.flush(operation);
+        if (operation.getStatus() == ILSMIOOperation.LSMIOOperationStatus.FAILURE) {
+            IRetryPolicy policy = new ExponentialRetryPolicy();
+            while (operation.getStatus() == ILSMIOOperation.LSMIOOperationStatus.FAILURE) {
+                if (policy.retry(operation.getFailure())) {
+                    operation.setFailure(null);
+                    operation.setStatus(ILSMIOOperation.LSMIOOperationStatus.SUCCESS);
+                    lsmHarness.flush(operation);
+                } else {
+                    break;
+                }
+            }
+        }
     }
 
     @Override
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/util/ComponentUtils.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/util/ComponentUtils.java
index 842ec61..a8cbfdd 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/util/ComponentUtils.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/util/ComponentUtils.java
@@ -169,6 +169,22 @@
         }
     }
 
+    public static void returnPages(ITreeIndex treeIndex) {
+        treeIndex.getPageManager().returnAllPages();
+        IBufferCache bufferCache = treeIndex.getBufferCache();
+        // We need to return all pages to the buffer cache in case of a failure
+        try {
+            bufferCache.getCompressedPageWriter(treeIndex.getFileId()).abort();
+        } catch (IllegalStateException | NullPointerException ignored) {
+            // Since we call this method in multiple places, it is possible that the writer
+            // is not in the State.WRITABLE, which would throw an IllegalStateException.
+            // This means the writer has already written all the pages.
+            //
+            // We also catch NullPointerException in case the writer is not initialized.
+            // Or if the compressed page writer is not applicable to this case
+        }
+    }
+
     public static void markAsValid(IBufferCache bufferCache, BloomFilter filter, boolean forceToDisk)
             throws HyracksDataException {
         if (forceToDisk) {
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 a828d22..f98a056 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
@@ -18,6 +18,9 @@
  */
 package org.apache.hyracks.storage.am.lsm.invertedindex.impls;
 
+import static org.apache.hyracks.storage.am.lsm.common.impls.AbstractLSMIndexFileManager.DELIMITER;
+import static org.apache.hyracks.storage.am.lsm.invertedindex.impls.LSMInvertedIndexFileManager.INVLISTS_SUFFIX;
+
 import java.util.ArrayList;
 import java.util.List;
 
@@ -286,46 +289,56 @@
         // Create a scan cursor on the deleted keys BTree underlying the in-memory inverted index.
         IIndexCursor deletedKeysScanCursor = deletedKeysBTreeAccessor.createSearchCursor(false);
         try {
-            deletedKeysBTreeAccessor.search(deletedKeysScanCursor, nullPred);
             try {
-                while (deletedKeysScanCursor.hasNext()) {
-                    deletedKeysScanCursor.next();
-                    componentBulkLoader.delete(deletedKeysScanCursor.getTuple());
+                deletedKeysBTreeAccessor.search(deletedKeysScanCursor, nullPred);
+                try {
+                    while (deletedKeysScanCursor.hasNext()) {
+                        deletedKeysScanCursor.next();
+                        componentBulkLoader.delete(deletedKeysScanCursor.getTuple());
+                    }
+                } finally {
+                    deletedKeysScanCursor.close();
                 }
             } finally {
-                deletedKeysScanCursor.close();
+                deletedKeysScanCursor.destroy();
             }
-        } finally {
-            deletedKeysScanCursor.destroy();
-        }
-        // Scan the in-memory inverted index
-        InMemoryInvertedIndexAccessor memInvIndexAccessor =
-                flushingComponent.getIndex().createAccessor(NoOpIndexAccessParameters.INSTANCE);
-        BTreeAccessor memBTreeAccessor = memInvIndexAccessor.getBTreeAccessor();
-        IIndexCursor scanCursor = memBTreeAccessor.createSearchCursor(false);
-        try {
-            memBTreeAccessor.search(scanCursor, nullPred);
-            // Bulk load the disk inverted index from the in-memory inverted index.
+            // Scan the in-memory inverted index
+            InMemoryInvertedIndexAccessor memInvIndexAccessor =
+                    flushingComponent.getIndex().createAccessor(NoOpIndexAccessParameters.INSTANCE);
+            BTreeAccessor memBTreeAccessor = memInvIndexAccessor.getBTreeAccessor();
+            IIndexCursor scanCursor = memBTreeAccessor.createSearchCursor(false);
             try {
-                while (scanCursor.hasNext()) {
-                    scanCursor.next();
-                    componentBulkLoader.add(scanCursor.getTuple());
+                memBTreeAccessor.search(scanCursor, nullPred);
+                // Bulk load the disk inverted index from the in-memory inverted index.
+                try {
+                    while (scanCursor.hasNext()) {
+                        scanCursor.next();
+                        componentBulkLoader.add(scanCursor.getTuple());
+                    }
+                } finally {
+                    scanCursor.close();
                 }
             } finally {
-                scanCursor.close();
+                scanCursor.destroy();
             }
-        } finally {
-            scanCursor.destroy();
+            if (component.getLSMComponentFilter() != null) {
+                List<ITupleReference> filterTuples = new ArrayList<>();
+                filterTuples.add(flushingComponent.getLSMComponentFilter().getMinTuple());
+                filterTuples.add(flushingComponent.getLSMComponentFilter().getMaxTuple());
+                filterManager.updateFilter(component.getLSMComponentFilter(), filterTuples,
+                        NoOpOperationCallback.INSTANCE);
+                filterManager.writeFilter(component.getLSMComponentFilter(), component.getMetadataHolder());
+            }
+            flushingComponent.getMetadata().copy(component.getMetadata());
+            componentBulkLoader.end();
+        } catch (Throwable e) {
+            try {
+                componentBulkLoader.abort();
+            } catch (Throwable th) {
+                e.addSuppressed(th);
+            }
+            throw e;
         }
-        if (component.getLSMComponentFilter() != null) {
-            List<ITupleReference> filterTuples = new ArrayList<>();
-            filterTuples.add(flushingComponent.getLSMComponentFilter().getMinTuple());
-            filterTuples.add(flushingComponent.getLSMComponentFilter().getMaxTuple());
-            filterManager.updateFilter(component.getLSMComponentFilter(), filterTuples, NoOpOperationCallback.INSTANCE);
-            filterManager.writeFilter(component.getLSMComponentFilter(), component.getMetadataHolder());
-        }
-        flushingComponent.getMetadata().copy(component.getMetadata());
-        componentBulkLoader.end();
         return component;
     }
 
@@ -339,60 +352,71 @@
         // Create an inverted index instance.
         ILSMDiskComponent component = createDiskComponent(componentFactory, mergeOp.getTarget(),
                 mergeOp.getDeletedKeysBTreeTarget(), mergeOp.getBloomFilterTarget(), true);
-        ILSMDiskComponentBulkLoader componentBulkLoader;
+        ILSMDiskComponentBulkLoader componentBulkLoader = null;
         // In case we must keep the deleted-keys BTrees, then they must be merged *before* merging the inverted
         // indexes so that lsmHarness.endSearch() is called once when the inverted indexes have been merged.
-        if (mergeOp.getMergingComponents().get(mergeOp.getMergingComponents().size() - 1) != diskComponents
-                .get(diskComponents.size() - 1)) {
-            // Keep the deleted tuples since the oldest disk component is not included in the merge operation
-            LSMInvertedIndexDeletedKeysBTreeMergeCursor btreeCursor =
-                    new LSMInvertedIndexDeletedKeysBTreeMergeCursor(opCtx, mergeOp.getCursorStats());
-            try {
-                long numElements = 0L;
-                for (int i = 0; i < mergeOp.getMergingComponents().size(); ++i) {
-                    numElements += ((LSMInvertedIndexDiskComponent) mergeOp.getMergingComponents().get(i))
-                            .getBloomFilter().getNumElements();
-                }
-                componentBulkLoader = component.createBulkLoader(operation, 1.0f, false, numElements, false, false,
-                        false, pageWriteCallbackFactory.createPageWriteCallback());
-                loadDeleteTuples(opCtx, btreeCursor, mergePred, componentBulkLoader);
-            } finally {
-                btreeCursor.destroy();
-            }
-        } else {
-            componentBulkLoader = component.createBulkLoader(operation, 1.0f, false, 0L, false, false, false,
-                    pageWriteCallbackFactory.createPageWriteCallback());
-        }
-        search(opCtx, cursor, mergePred);
         try {
-            while (cursor.hasNext()) {
-                cursor.next();
-                componentBulkLoader.add(cursor.getTuple());
+            if (mergeOp.getMergingComponents().get(mergeOp.getMergingComponents().size() - 1) != diskComponents
+                    .get(diskComponents.size() - 1)) {
+                // Keep the deleted tuples since the oldest disk component is not included in the merge operation
+                LSMInvertedIndexDeletedKeysBTreeMergeCursor btreeCursor =
+                        new LSMInvertedIndexDeletedKeysBTreeMergeCursor(opCtx, mergeOp.getCursorStats());
+                try {
+                    long numElements = 0L;
+                    for (int i = 0; i < mergeOp.getMergingComponents().size(); ++i) {
+                        numElements += ((LSMInvertedIndexDiskComponent) mergeOp.getMergingComponents().get(i))
+                                .getBloomFilter().getNumElements();
+                    }
+                    componentBulkLoader = component.createBulkLoader(operation, 1.0f, false, numElements, false, false,
+                            false, pageWriteCallbackFactory.createPageWriteCallback());
+                    loadDeleteTuples(opCtx, btreeCursor, mergePred, componentBulkLoader);
+                } finally {
+                    btreeCursor.destroy();
+                }
+            } else {
+                componentBulkLoader = component.createBulkLoader(operation, 1.0f, false, 0L, false, false, false,
+                        pageWriteCallbackFactory.createPageWriteCallback());
             }
-        } finally {
+            search(opCtx, cursor, mergePred);
             try {
-                cursor.close();
+                while (cursor.hasNext()) {
+                    cursor.next();
+                    componentBulkLoader.add(cursor.getTuple());
+                }
             } finally {
-                cursor.destroy();
-            }
-        }
-        if (component.getLSMComponentFilter() != null) {
-            List<ITupleReference> filterTuples = new ArrayList<>();
-            for (int i = 0; i < mergeOp.getMergingComponents().size(); ++i) {
-                ITupleReference min = mergeOp.getMergingComponents().get(i).getLSMComponentFilter().getMinTuple();
-                ITupleReference max = mergeOp.getMergingComponents().get(i).getLSMComponentFilter().getMaxTuple();
-                if (min != null) {
-                    filterTuples.add(min);
-                }
-                if (max != null) {
-                    filterTuples.add(max);
+                try {
+                    cursor.close();
+                } finally {
+                    cursor.destroy();
                 }
             }
-            getFilterManager().updateFilter(component.getLSMComponentFilter(), filterTuples,
-                    NoOpOperationCallback.INSTANCE);
-            getFilterManager().writeFilter(component.getLSMComponentFilter(), component.getMetadataHolder());
+            if (component.getLSMComponentFilter() != null) {
+                List<ITupleReference> filterTuples = new ArrayList<>();
+                for (int i = 0; i < mergeOp.getMergingComponents().size(); ++i) {
+                    ITupleReference min = mergeOp.getMergingComponents().get(i).getLSMComponentFilter().getMinTuple();
+                    ITupleReference max = mergeOp.getMergingComponents().get(i).getLSMComponentFilter().getMaxTuple();
+                    if (min != null) {
+                        filterTuples.add(min);
+                    }
+                    if (max != null) {
+                        filterTuples.add(max);
+                    }
+                }
+                getFilterManager().updateFilter(component.getLSMComponentFilter(), filterTuples,
+                        NoOpOperationCallback.INSTANCE);
+                getFilterManager().writeFilter(component.getLSMComponentFilter(), component.getMetadataHolder());
+            }
+            componentBulkLoader.end();
+        } catch (Throwable e) {
+            try {
+                if (componentBulkLoader != null) {
+                    componentBulkLoader.abort();
+                }
+            } catch (Throwable th) {
+                e.addSuppressed(th);
+            }
+            throw e;
         }
-        componentBulkLoader.end();
 
         return component;
     }
@@ -487,7 +511,10 @@
             throws HyracksDataException {
         return new LSMInvertedIndexFlushOperation(new LSMInvertedIndexAccessor(getHarness(), opCtx),
                 componentFileRefs.getInsertIndexFileReference(), componentFileRefs.getDeleteIndexFileReference(),
-                componentFileRefs.getBloomFilterFileReference(), callback, getIndexIdentifier());
+                componentFileRefs.getBloomFilterFileReference(),
+                ioManager.resolve(
+                        getInvListsFilePath(componentFileRefs.getInsertIndexFileReference().getAbsolutePath())),
+                callback, getIndexIdentifier());
     }
 
     @Override
@@ -497,7 +524,14 @@
         IIndexCursorStats stats = new IndexCursorStats();
         IIndexCursor cursor = new LSMInvertedIndexMergeCursor(opCtx, stats);
         return new LSMInvertedIndexMergeOperation(accessor, cursor, stats, mergeFileRefs.getInsertIndexFileReference(),
-                mergeFileRefs.getDeleteIndexFileReference(), mergeFileRefs.getBloomFilterFileReference(), callback,
-                getIndexIdentifier());
+                mergeFileRefs.getDeleteIndexFileReference(), mergeFileRefs.getBloomFilterFileReference(),
+                ioManager.resolve(getInvListsFilePath(mergeFileRefs.getInsertIndexFileReference().getAbsolutePath())),
+                callback, getIndexIdentifier());
+    }
+
+    public String getInvListsFilePath(String dictBTreeFilePath) {
+        int index = dictBTreeFilePath.lastIndexOf(DELIMITER);
+        String file = dictBTreeFilePath.substring(0, index);
+        return file + DELIMITER + INVLISTS_SUFFIX;
     }
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexAccessor.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexAccessor.java
index 67312ce..d8900ad 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexAccessor.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexAccessor.java
@@ -37,6 +37,8 @@
 import org.apache.hyracks.storage.am.lsm.invertedindex.api.IInvertedListCursor;
 import org.apache.hyracks.storage.common.IIndexCursor;
 import org.apache.hyracks.storage.common.ISearchPredicate;
+import org.apache.hyracks.util.ExponentialRetryPolicy;
+import org.apache.hyracks.util.IRetryPolicy;
 
 public class LSMInvertedIndexAccessor implements ILSMIndexAccessor, IInvertedIndexAccessor {
 
@@ -93,6 +95,18 @@
     @Override
     public void flush(ILSMIOOperation operation) throws HyracksDataException {
         lsmHarness.flush(operation);
+        if (operation.getStatus() == ILSMIOOperation.LSMIOOperationStatus.FAILURE) {
+            IRetryPolicy policy = new ExponentialRetryPolicy();
+            while (operation.getStatus() == ILSMIOOperation.LSMIOOperationStatus.FAILURE) {
+                if (policy.retry(operation.getFailure())) {
+                    operation.setFailure(null);
+                    operation.setStatus(ILSMIOOperation.LSMIOOperationStatus.SUCCESS);
+                    lsmHarness.flush(operation);
+                } else {
+                    break;
+                }
+            }
+        }
     }
 
     @Override
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexDiskComponent.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexDiskComponent.java
index 6de6ab5..3edb5fe 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexDiskComponent.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexDiskComponent.java
@@ -105,17 +105,22 @@
 
     @Override
     public void markAsValid(boolean persist, IPageWriteFailureCallback callback) throws HyracksDataException {
-        ComponentUtils.markAsValid(getBloomFilterBufferCache(), getBloomFilter(), persist);
+        try {
+            ComponentUtils.markAsValid(getBloomFilterBufferCache(), getBloomFilter(), persist);
 
-        // Flush inverted index second.
-        invIndex.getBufferCache().force((invIndex).getInvListsFileId(), true);
-        ComponentUtils.markAsValid(getMetadataHolder(), persist, callback);
-        if (!callback.hasFailed()) {
-            // Flush deleted keys BTree.
-            ComponentUtils.markAsValid(getBuddyIndex(), persist, callback);
-        }
-        if (callback.hasFailed()) {
-            throw HyracksDataException.create(callback.getFailure());
+            // Flush inverted index second.
+            invIndex.getBufferCache().force((invIndex).getInvListsFileId(), true);
+            ComponentUtils.markAsValid(getMetadataHolder(), persist, callback);
+            if (!callback.hasFailed()) {
+                // Flush deleted keys BTree.
+                ComponentUtils.markAsValid(getBuddyIndex(), persist, callback);
+            }
+            if (callback.hasFailed()) {
+                throw HyracksDataException.create(callback.getFailure());
+            }
+        } catch (HyracksDataException ex) {
+            returnPages();
+            throw ex;
         }
     }
 
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexFileManager.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexFileManager.java
index 3b8f82a..442df0b 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexFileManager.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexFileManager.java
@@ -34,6 +34,7 @@
 import org.apache.hyracks.storage.am.lsm.common.impls.BTreeFactory;
 import org.apache.hyracks.storage.am.lsm.common.impls.IndexComponentFileReference;
 import org.apache.hyracks.storage.am.lsm.common.impls.LSMComponentFileReferences;
+import org.apache.hyracks.storage.am.lsm.common.impls.LSMInvertedComponentFileReferences;
 import org.apache.hyracks.storage.am.lsm.invertedindex.api.IInvertedIndexFileNameMapper;
 
 // TODO: Refactor for better code sharing with other file managers.
@@ -59,17 +60,19 @@
     @Override
     public LSMComponentFileReferences getRelFlushFileReference() throws HyracksDataException {
         String baseName = getNextComponentSequence(deletedKeysBTreeFilter);
-        return new LSMComponentFileReferences(baseDir.getChild(baseName + DELIMITER + DICT_BTREE_SUFFIX),
+        return new LSMInvertedComponentFileReferences(baseDir.getChild(baseName + DELIMITER + DICT_BTREE_SUFFIX),
                 baseDir.getChild(baseName + DELIMITER + DELETED_KEYS_BTREE_SUFFIX),
-                baseDir.getChild(baseName + DELIMITER + BLOOM_FILTER_SUFFIX));
+                baseDir.getChild(baseName + DELIMITER + BLOOM_FILTER_SUFFIX),
+                baseDir.getChild(baseName + DELIMITER + INVLISTS_SUFFIX));
     }
 
     @Override
     public LSMComponentFileReferences getRelMergeFileReference(String firstFileName, String lastFileName) {
         final String baseName = IndexComponentFileReference.getMergeSequence(firstFileName, lastFileName);
-        return new LSMComponentFileReferences(baseDir.getChild(baseName + DELIMITER + DICT_BTREE_SUFFIX),
+        return new LSMInvertedComponentFileReferences(baseDir.getChild(baseName + DELIMITER + DICT_BTREE_SUFFIX),
                 baseDir.getChild(baseName + DELIMITER + DELETED_KEYS_BTREE_SUFFIX),
-                baseDir.getChild(baseName + DELIMITER + BLOOM_FILTER_SUFFIX));
+                baseDir.getChild(baseName + DELIMITER + BLOOM_FILTER_SUFFIX),
+                baseDir.getChild(baseName + DELIMITER + INVLISTS_SUFFIX));
     }
 
     @Override
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexFlushOperation.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexFlushOperation.java
index 5b272ee..efd5060 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexFlushOperation.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexFlushOperation.java
@@ -24,17 +24,20 @@
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexAccessor;
 import org.apache.hyracks.storage.am.lsm.common.impls.FlushOperation;
 import org.apache.hyracks.storage.am.lsm.common.impls.LSMComponentFileReferences;
+import org.apache.hyracks.storage.am.lsm.common.impls.LSMInvertedComponentFileReferences;
 
 public class LSMInvertedIndexFlushOperation extends FlushOperation {
     private final FileReference deletedKeysBTreeFlushTarget;
     private final FileReference bloomFilterFlushTarget;
+    private final FileReference invListsFlushTarget;
 
     public LSMInvertedIndexFlushOperation(ILSMIndexAccessor accessor, FileReference flushTarget,
             FileReference deletedKeysBTreeFlushTarget, FileReference bloomFilterFlushTarget,
-            ILSMIOOperationCallback callback, String indexIdentifier) {
+            FileReference invListsFlushTarget, ILSMIOOperationCallback callback, String indexIdentifier) {
         super(accessor, flushTarget, callback, indexIdentifier);
         this.deletedKeysBTreeFlushTarget = deletedKeysBTreeFlushTarget;
         this.bloomFilterFlushTarget = bloomFilterFlushTarget;
+        this.invListsFlushTarget = invListsFlushTarget;
     }
 
     public FileReference getDeletedKeysBTreeTarget() {
@@ -47,6 +50,7 @@
 
     @Override
     public LSMComponentFileReferences getComponentFiles() {
-        return new LSMComponentFileReferences(target, deletedKeysBTreeFlushTarget, bloomFilterFlushTarget);
+        return new LSMInvertedComponentFileReferences(target, deletedKeysBTreeFlushTarget, bloomFilterFlushTarget,
+                invListsFlushTarget);
     }
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexMergeOperation.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexMergeOperation.java
index 31c79f0..4ebb9d3 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexMergeOperation.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexMergeOperation.java
@@ -23,6 +23,7 @@
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallback;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexAccessor;
 import org.apache.hyracks.storage.am.lsm.common.impls.LSMComponentFileReferences;
+import org.apache.hyracks.storage.am.lsm.common.impls.LSMInvertedComponentFileReferences;
 import org.apache.hyracks.storage.am.lsm.common.impls.MergeOperation;
 import org.apache.hyracks.storage.common.IIndexCursor;
 import org.apache.hyracks.storage.common.IIndexCursorStats;
@@ -30,13 +31,15 @@
 public class LSMInvertedIndexMergeOperation extends MergeOperation {
     private final FileReference deletedKeysBTreeMergeTarget;
     private final FileReference bloomFilterMergeTarget;
+    private final FileReference invListsMergeTarget;
 
     public LSMInvertedIndexMergeOperation(ILSMIndexAccessor accessor, IIndexCursor cursor, IIndexCursorStats stats,
             FileReference target, FileReference deletedKeysBTreeMergeTarget, FileReference bloomFilterMergeTarget,
-            ILSMIOOperationCallback callback, String indexIdentifier) {
+            FileReference invListsMergeTarget, ILSMIOOperationCallback callback, String indexIdentifier) {
         super(accessor, target, callback, indexIdentifier, cursor, stats);
         this.deletedKeysBTreeMergeTarget = deletedKeysBTreeMergeTarget;
         this.bloomFilterMergeTarget = bloomFilterMergeTarget;
+        this.invListsMergeTarget = invListsMergeTarget;
     }
 
     public FileReference getDeletedKeysBTreeTarget() {
@@ -49,7 +52,8 @@
 
     @Override
     public LSMComponentFileReferences getComponentFiles() {
-        return new LSMComponentFileReferences(target, deletedKeysBTreeMergeTarget, bloomFilterMergeTarget);
+        return new LSMInvertedComponentFileReferences(target, deletedKeysBTreeMergeTarget, bloomFilterMergeTarget,
+                invListsMergeTarget);
     }
 
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/ondisk/OnDiskInvertedIndex.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/ondisk/OnDiskInvertedIndex.java
index 2583d0a..47068a3 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/ondisk/OnDiskInvertedIndex.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/ondisk/OnDiskInvertedIndex.java
@@ -397,6 +397,9 @@
             if (btreeBulkloader != null) {
                 btreeBulkloader.abort();
             }
+            if (currentPage != null) {
+                bufferCache.returnPage(currentPage, false);
+            }
         }
 
         @Override
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 a8a1909..8b88e7e 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
@@ -110,7 +110,6 @@
         RangePredicate btreeNullPredicate = new RangePredicate(null, null, true, true, null, null);
         BTreeAccessor memBTreeAccessor =
                 flushingComponent.getBuddyIndex().createAccessor(NoOpIndexAccessParameters.INSTANCE);
-        boolean abort = true;
         try {
             try {
                 rTreeTupleSorter = getRTreeTupleSorter(flushingComponent, memBTreeAccessor, btreeNullPredicate,
@@ -142,12 +141,16 @@
             }
             // Note. If we change the filter to write to metadata object, we don't need the if block above
             flushingComponent.getMetadata().copy(component.getMetadata());
-            abort = false;
             componentBulkLoader.end();
-        } finally {
-            if (abort && componentBulkLoader != null) {
-                componentBulkLoader.abort();
+        } catch (Throwable e) {
+            try {
+                if (componentBulkLoader != null) {
+                    componentBulkLoader.abort();
+                }
+            } catch (Throwable th) {
+                e.addSuppressed(th);
             }
+            throw e;
         }
         return component;
     }
@@ -264,32 +267,35 @@
         IIndexCursor cursor = mergeOp.getCursor();
         ILSMDiskComponentBulkLoader componentBulkLoader = null;
         ILSMDiskComponent mergedComponent = null;
-        boolean abort = true;
         try {
-            mergedComponent = createDiskComponent(componentFactory, mergeOp.getTarget(), mergeOp.getBTreeTarget(),
-                    mergeOp.getBloomFilterTarget(), true);
-            componentBulkLoader = loadMergeBulkLoader(mergeOp, cursor, mergedComponent);
-            if (mergedComponent.getLSMComponentFilter() != null) {
-                List<ITupleReference> filterTuples = new ArrayList<>();
-                for (int i = 0; i < mergeOp.getMergingComponents().size(); ++i) {
-                    filterTuples.add(mergeOp.getMergingComponents().get(i).getLSMComponentFilter().getMinTuple());
-                    filterTuples.add(mergeOp.getMergingComponents().get(i).getLSMComponentFilter().getMaxTuple());
-                }
-                getFilterManager().updateFilter(mergedComponent.getLSMComponentFilter(), filterTuples,
-                        NoOpOperationCallback.INSTANCE);
-                getFilterManager().writeFilter(mergedComponent.getLSMComponentFilter(),
-                        mergedComponent.getMetadataHolder());
-            }
-            abort = false;
-            componentBulkLoader.end();
-        } finally {
             try {
-                cursor.destroy();
+                mergedComponent = createDiskComponent(componentFactory, mergeOp.getTarget(), mergeOp.getBTreeTarget(),
+                        mergeOp.getBloomFilterTarget(), true);
+                componentBulkLoader = loadMergeBulkLoader(mergeOp, cursor, mergedComponent);
+                if (mergedComponent.getLSMComponentFilter() != null) {
+                    List<ITupleReference> filterTuples = new ArrayList<>();
+                    for (int i = 0; i < mergeOp.getMergingComponents().size(); ++i) {
+                        filterTuples.add(mergeOp.getMergingComponents().get(i).getLSMComponentFilter().getMinTuple());
+                        filterTuples.add(mergeOp.getMergingComponents().get(i).getLSMComponentFilter().getMaxTuple());
+                    }
+                    getFilterManager().updateFilter(mergedComponent.getLSMComponentFilter(), filterTuples,
+                            NoOpOperationCallback.INSTANCE);
+                    getFilterManager().writeFilter(mergedComponent.getLSMComponentFilter(),
+                            mergedComponent.getMetadataHolder());
+                }
+                componentBulkLoader.end();
             } finally {
-                if (abort && componentBulkLoader != null) {
+                cursor.destroy();
+            }
+        } catch (Throwable e) {
+            try {
+                if (componentBulkLoader != null) {
                     componentBulkLoader.abort();
                 }
+            } catch (Throwable th) {
+                e.addSuppressed(th);
             }
+            throw e;
         }
         return mergedComponent;
     }
@@ -297,46 +303,38 @@
     private ILSMDiskComponentBulkLoader loadMergeBulkLoader(LSMRTreeMergeOperation mergeOp, IIndexCursor cursor,
             ILSMDiskComponent mergedComponent) throws HyracksDataException {
         ILSMDiskComponentBulkLoader componentBulkLoader = null;
-        boolean abort = true;
         ISearchPredicate rtreeSearchPred = new SearchPredicate(null, null);
         ILSMIndexOperationContext opCtx = ((LSMRTreeSortedCursor) cursor).getOpCtx();
         search(opCtx, cursor, rtreeSearchPred);
         try {
-            try {
-                // In case we must keep the deleted-keys BTrees, then they must be merged
-                // *before* merging the r-trees so that
-                // lsmHarness.endSearch() is called once when the r-trees have been merged.
-                if (mergeOp.getMergingComponents().get(mergeOp.getMergingComponents().size() - 1) != diskComponents
-                        .get(diskComponents.size() - 1)) {
-                    // Keep the deleted tuples since the oldest disk component
-                    // is not included in the merge operation
-                    long numElements = 0L;
-                    for (int i = 0; i < mergeOp.getMergingComponents().size(); ++i) {
-                        numElements += ((LSMRTreeDiskComponent) mergeOp.getMergingComponents().get(i)).getBloomFilter()
-                                .getNumElements();
-                    }
-                    componentBulkLoader = mergedComponent.createBulkLoader(mergeOp, 1.0f, false, numElements, false,
-                            false, false, pageWriteCallbackFactory.createPageWriteCallback());
-                    mergeLoadBTree(mergeOp, opCtx, rtreeSearchPred, componentBulkLoader);
-                } else {
-                    //no buddy-btree needed
-                    componentBulkLoader = mergedComponent.createBulkLoader(mergeOp, 1.0f, false, 0L, false, false,
-                            false, pageWriteCallbackFactory.createPageWriteCallback());
+            // In case we must keep the deleted-keys BTrees, then they must be merged
+            // *before* merging the r-trees so that
+            // lsmHarness.endSearch() is called once when the r-trees have been merged.
+            if (mergeOp.getMergingComponents().get(mergeOp.getMergingComponents().size() - 1) != diskComponents
+                    .get(diskComponents.size() - 1)) {
+                // Keep the deleted tuples since the oldest disk component
+                // is not included in the merge operation
+                long numElements = 0L;
+                for (int i = 0; i < mergeOp.getMergingComponents().size(); ++i) {
+                    numElements += ((LSMRTreeDiskComponent) mergeOp.getMergingComponents().get(i)).getBloomFilter()
+                            .getNumElements();
                 }
-                //search old rtree components
-                while (cursor.hasNext()) {
-                    cursor.next();
-                    ITupleReference frameTuple = cursor.getTuple();
-                    componentBulkLoader.add(frameTuple);
-                }
-            } finally {
-                cursor.close();
+                componentBulkLoader = mergedComponent.createBulkLoader(mergeOp, 1.0f, false, numElements, false, false,
+                        false, pageWriteCallbackFactory.createPageWriteCallback());
+                mergeLoadBTree(mergeOp, opCtx, rtreeSearchPred, componentBulkLoader);
+            } else {
+                //no buddy-btree needed
+                componentBulkLoader = mergedComponent.createBulkLoader(mergeOp, 1.0f, false, 0L, false, false, false,
+                        pageWriteCallbackFactory.createPageWriteCallback());
             }
-            abort = false;
+            //search old rtree components
+            while (cursor.hasNext()) {
+                cursor.next();
+                ITupleReference frameTuple = cursor.getTuple();
+                componentBulkLoader.add(frameTuple);
+            }
         } finally {
-            if (abort && componentBulkLoader != null) {
-                componentBulkLoader.abort();
-            }
+            cursor.close();
         }
         return componentBulkLoader;
     }
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 729ca74..6968b3c 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
@@ -103,107 +103,111 @@
         TreeTupleSorter rTreeTupleSorter = null;
         TreeTupleSorter bTreeTupleSorter = null;
         boolean isEmpty = true;
-        boolean abort = true;
         try {
-            RTreeAccessor memRTreeAccessor =
-                    flushingComponent.getIndex().createAccessor(NoOpIndexAccessParameters.INSTANCE);
+            try {
+                RTreeAccessor memRTreeAccessor =
+                        flushingComponent.getIndex().createAccessor(NoOpIndexAccessParameters.INSTANCE);
 
-            try {
-                RTreeSearchCursor rtreeScanCursor = memRTreeAccessor.createSearchCursor(false);
                 try {
-                    memRTreeAccessor.search(rtreeScanCursor, rtreeNullPredicate);
-                    component = createDiskComponent(componentFactory, flushOp.getTarget(), null, null, true);
-                    componentBulkLoader = component.createBulkLoader(operation, 1.0f, false, 0L, false, false, false,
-                            pageWriteCallbackFactory.createPageWriteCallback());
-                    // Since the LSM-RTree is used as a secondary assumption, the
-                    // primary key will be the last comparator in the BTree comparators
-                    rTreeTupleSorter = new TreeTupleSorter(flushingComponent.getIndex().getFileId(), linearizerArray,
-                            rtreeLeafFrameFactory.createFrame(), rtreeLeafFrameFactory.createFrame(),
-                            flushingComponent.getIndex().getBufferCache(), comparatorFields);
+                    RTreeSearchCursor rtreeScanCursor = memRTreeAccessor.createSearchCursor(false);
                     try {
-                        isEmpty = scanAndSort(rtreeScanCursor, rTreeTupleSorter);
+                        memRTreeAccessor.search(rtreeScanCursor, rtreeNullPredicate);
+                        component = createDiskComponent(componentFactory, flushOp.getTarget(), null, null, true);
+                        componentBulkLoader = component.createBulkLoader(operation, 1.0f, false, 0L, false, false,
+                                false, pageWriteCallbackFactory.createPageWriteCallback());
+                        // Since the LSM-RTree is used as a secondary assumption, the
+                        // primary key will be the last comparator in the BTree comparators
+                        rTreeTupleSorter =
+                                new TreeTupleSorter(flushingComponent.getIndex().getFileId(), linearizerArray,
+                                        rtreeLeafFrameFactory.createFrame(), rtreeLeafFrameFactory.createFrame(),
+                                        flushingComponent.getIndex().getBufferCache(), comparatorFields);
+                        try {
+                            isEmpty = scanAndSort(rtreeScanCursor, rTreeTupleSorter);
+                        } finally {
+                            rtreeScanCursor.close();
+                        }
                     } finally {
-                        rtreeScanCursor.close();
+                        rtreeScanCursor.destroy();
                     }
                 } finally {
-                    rtreeScanCursor.destroy();
+                    memRTreeAccessor.destroy();
                 }
-            } finally {
-                memRTreeAccessor.destroy();
-            }
-            if (!isEmpty) {
-                rTreeTupleSorter.sort();
-            }
-            // scan the memory BTree
-            RangePredicate btreeNullPredicate = new RangePredicate(null, null, true, true, null, null);
-            BTreeAccessor memBTreeAccessor =
-                    flushingComponent.getBuddyIndex().createAccessor(NoOpIndexAccessParameters.INSTANCE);
-            try {
-                bTreeTupleSorter = new TreeTupleSorter(flushingComponent.getBuddyIndex().getFileId(), linearizerArray,
-                        btreeLeafFrameFactory.createFrame(), btreeLeafFrameFactory.createFrame(),
-                        flushingComponent.getBuddyIndex().getBufferCache(), comparatorFields);
-                BTreeRangeSearchCursor btreeScanCursor =
-                        (BTreeRangeSearchCursor) memBTreeAccessor.createSearchCursor(false);
+                if (!isEmpty) {
+                    rTreeTupleSorter.sort();
+                }
+                // scan the memory BTree
+                RangePredicate btreeNullPredicate = new RangePredicate(null, null, true, true, null, null);
+                BTreeAccessor memBTreeAccessor =
+                        flushingComponent.getBuddyIndex().createAccessor(NoOpIndexAccessParameters.INSTANCE);
                 try {
-                    isEmpty = true;
-                    memBTreeAccessor.search(btreeScanCursor, btreeNullPredicate);
+                    bTreeTupleSorter = new TreeTupleSorter(flushingComponent.getBuddyIndex().getFileId(),
+                            linearizerArray, btreeLeafFrameFactory.createFrame(), btreeLeafFrameFactory.createFrame(),
+                            flushingComponent.getBuddyIndex().getBufferCache(), comparatorFields);
+                    BTreeRangeSearchCursor btreeScanCursor =
+                            (BTreeRangeSearchCursor) memBTreeAccessor.createSearchCursor(false);
                     try {
-                        isEmpty = scanAndSort(btreeScanCursor, bTreeTupleSorter);
+                        isEmpty = true;
+                        memBTreeAccessor.search(btreeScanCursor, btreeNullPredicate);
+                        try {
+                            isEmpty = scanAndSort(btreeScanCursor, bTreeTupleSorter);
+                        } finally {
+                            btreeScanCursor.close();
+                        }
                     } finally {
-                        btreeScanCursor.close();
+                        btreeScanCursor.destroy();
                     }
                 } finally {
-                    btreeScanCursor.destroy();
+                    memBTreeAccessor.destroy();
                 }
-            } finally {
-                memBTreeAccessor.destroy();
-            }
-            if (!isEmpty) {
-                bTreeTupleSorter.sort();
-            }
-            LSMRTreeWithAntiMatterTuplesFlushCursor cursor = new LSMRTreeWithAntiMatterTuplesFlushCursor(
-                    rTreeTupleSorter, bTreeTupleSorter, comparatorFields, linearizerArray);
-            try {
-                cursor.open(null, null);
+                if (!isEmpty) {
+                    bTreeTupleSorter.sort();
+                }
+                LSMRTreeWithAntiMatterTuplesFlushCursor cursor = new LSMRTreeWithAntiMatterTuplesFlushCursor(
+                        rTreeTupleSorter, bTreeTupleSorter, comparatorFields, linearizerArray);
                 try {
-                    while (cursor.hasNext()) {
-                        cursor.next();
-                        ITupleReference frameTuple = cursor.getTuple();
-                        componentBulkLoader.add(frameTuple);
+                    cursor.open(null, null);
+                    try {
+                        while (cursor.hasNext()) {
+                            cursor.next();
+                            ITupleReference frameTuple = cursor.getTuple();
+                            componentBulkLoader.add(frameTuple);
+                        }
+                    } finally {
+                        cursor.close();
                     }
                 } finally {
-                    cursor.close();
+                    cursor.destroy();
                 }
-            } finally {
-                cursor.destroy();
-            }
-            if (component.getLSMComponentFilter() != null) {
-                List<ITupleReference> filterTuples = new ArrayList<>();
-                filterTuples.add(flushingComponent.getLSMComponentFilter().getMinTuple());
-                filterTuples.add(flushingComponent.getLSMComponentFilter().getMaxTuple());
-                getFilterManager().updateFilter(component.getLSMComponentFilter(), filterTuples,
-                        NoOpOperationCallback.INSTANCE);
-                getFilterManager().writeFilter(component.getLSMComponentFilter(), component.getMetadataHolder());
-            }
-            flushingComponent.getMetadata().copy(component.getMetadata());
-            abort = false;
-            componentBulkLoader.end();
-        } finally {
-            try {
-                if (rTreeTupleSorter != null) {
-                    rTreeTupleSorter.destroy();
+                if (component.getLSMComponentFilter() != null) {
+                    List<ITupleReference> filterTuples = new ArrayList<>();
+                    filterTuples.add(flushingComponent.getLSMComponentFilter().getMinTuple());
+                    filterTuples.add(flushingComponent.getLSMComponentFilter().getMaxTuple());
+                    getFilterManager().updateFilter(component.getLSMComponentFilter(), filterTuples,
+                            NoOpOperationCallback.INSTANCE);
+                    getFilterManager().writeFilter(component.getLSMComponentFilter(), component.getMetadataHolder());
                 }
+                flushingComponent.getMetadata().copy(component.getMetadata());
+                componentBulkLoader.end();
             } finally {
                 try {
+                    if (rTreeTupleSorter != null) {
+                        rTreeTupleSorter.destroy();
+                    }
+                } finally {
                     if (bTreeTupleSorter != null) {
                         bTreeTupleSorter.destroy();
                     }
-                } finally {
-                    if (abort && componentBulkLoader != null) {
-                        componentBulkLoader.abort();
-                    }
                 }
             }
+        } catch (Throwable e) {
+            try {
+                if (componentBulkLoader != null) {
+                    componentBulkLoader.abort();
+                }
+            } catch (Throwable th) {
+                e.addSuppressed(th);
+            }
+            throw e;
         }
         return component;
     }
@@ -243,26 +247,35 @@
         ILSMDiskComponentBulkLoader componentBulkLoader = component.createBulkLoader(operation, 1.0f, false, 0L, false,
                 false, false, pageWriteCallbackFactory.createPageWriteCallback());
         try {
-            while (cursor.hasNext()) {
-                cursor.next();
-                ITupleReference frameTuple = cursor.getTuple();
-                componentBulkLoader.add(frameTuple);
+            try {
+                while (cursor.hasNext()) {
+                    cursor.next();
+                    ITupleReference frameTuple = cursor.getTuple();
+                    componentBulkLoader.add(frameTuple);
+                }
+            } finally {
+                cursor.close();
             }
-        } finally {
-            cursor.close();
-        }
-        if (component.getLSMComponentFilter() != null) {
-            List<ITupleReference> filterTuples = new ArrayList<>();
-            for (int i = 0; i < mergeOp.getMergingComponents().size(); ++i) {
-                filterTuples.add(mergeOp.getMergingComponents().get(i).getLSMComponentFilter().getMinTuple());
-                filterTuples.add(mergeOp.getMergingComponents().get(i).getLSMComponentFilter().getMaxTuple());
+            if (component.getLSMComponentFilter() != null) {
+                List<ITupleReference> filterTuples = new ArrayList<>();
+                for (int i = 0; i < mergeOp.getMergingComponents().size(); ++i) {
+                    filterTuples.add(mergeOp.getMergingComponents().get(i).getLSMComponentFilter().getMinTuple());
+                    filterTuples.add(mergeOp.getMergingComponents().get(i).getLSMComponentFilter().getMaxTuple());
+                }
+                getFilterManager().updateFilter(component.getLSMComponentFilter(), filterTuples,
+                        NoOpOperationCallback.INSTANCE);
+                getFilterManager().writeFilter(component.getLSMComponentFilter(), component.getMetadataHolder());
             }
-            getFilterManager().updateFilter(component.getLSMComponentFilter(), filterTuples,
-                    NoOpOperationCallback.INSTANCE);
-            getFilterManager().writeFilter(component.getLSMComponentFilter(), component.getMetadataHolder());
-        }
 
-        componentBulkLoader.end();
+            componentBulkLoader.end();
+        } catch (Throwable e) {
+            try {
+                componentBulkLoader.abort();
+            } catch (Throwable th) {
+                e.addSuppressed(th);
+            }
+            throw e;
+        }
 
         return component;
     }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/impls/RTree.java b/hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/impls/RTree.java
index 35785d3..33ba01e 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/impls/RTree.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/impls/RTree.java
@@ -976,17 +976,22 @@
 
         @Override
         public void end() throws HyracksDataException {
-            pagesToWrite.clear();
-            //if writing a trivial 1-page tree, don't try and propagate up
-            if (nodeFrontiers.size() > 1) {
-                propagateBulk(1, true, pagesToWrite);
-            }
+            try {
+                pagesToWrite.clear();
+                //if writing a trivial 1-page tree, don't try and propagate up
+                if (nodeFrontiers.size() > 1) {
+                    propagateBulk(1, true, pagesToWrite);
+                }
 
-            for (ICachedPage c : pagesToWrite) {
-                write(c);
+                for (ICachedPage c : pagesToWrite) {
+                    write(c);
+                }
+                finish();
+                super.end();
+            } catch (HyracksDataException e) {
+                handleException();
+                throw e;
             }
-            finish();
-            super.end();
         }
 
         @Override
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/BufferCache.java b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/BufferCache.java
index 4c2395f..a4ecb6f 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/BufferCache.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/BufferCache.java
@@ -499,11 +499,13 @@
         synchronized (cachedPages) {
             for (ICachedPageInternal internalPage : cachedPages) {
                 CachedPage c = (CachedPage) internalPage;
-                if (c.confiscated() || c.latch.getReadLockCount() != 0 || c.latch.getWriteHoldCount() != 0) {
-                    return false;
-                }
-                if (c.valid) {
-                    reachableDpids.add(c.dpid);
+                if (c != null) {
+                    if (c.confiscated() || c.latch.getReadLockCount() != 0 || c.latch.getWriteHoldCount() != 0) {
+                        return false;
+                    }
+                    if (c.valid) {
+                        reachableDpids.add(c.dpid);
+                    }
                 }
             }
         }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/FIFOLocalWriter.java b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/FIFOLocalWriter.java
index 7d4e119..219275e 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/FIFOLocalWriter.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/FIFOLocalWriter.java
@@ -15,6 +15,8 @@
 
 package org.apache.hyracks.storage.common.buffercache;
 
+import org.apache.hyracks.api.exceptions.ErrorCode;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
 import org.apache.hyracks.storage.common.buffercache.context.IBufferCacheWriteContext;
 import org.apache.hyracks.util.ExitUtil;
 import org.apache.logging.log4j.LogManager;
@@ -38,18 +40,21 @@
         this.context = context;
     }
 
-    @SuppressWarnings("squid:S1181") // System must halt on all IO errors
+    @SuppressWarnings("squid:S1181")
     @Override
-    public void write(ICachedPage page) {
+    public void write(ICachedPage page) throws HyracksDataException {
         CachedPage cPage = (CachedPage) page;
         try {
             callback.beforeWrite(cPage);
             bufferCache.write(cPage, context);
             callback.afterWrite(cPage);
+        } catch (HyracksDataException e) {
+            LOGGER.warn("Failed to write page {}", cPage, e);
+            throw HyracksDataException.create(ErrorCode.FAILED_IO_OPERATION, e);
         } catch (Throwable th) {
             // Halt
-            LOGGER.error("Failed to write page {}", cPage, th);
-            ExitUtil.halt(ExitUtil.EC_IO_OPERATION_FAILED);
+            LOGGER.error("FIFOLocalWriter has encountered a fatal error", th);
+            ExitUtil.halt(ExitUtil.EC_ABNORMAL_TERMINATION);
         } finally {
             bufferCache.returnPage(cPage);
             if (DEBUG) {
@@ -57,5 +62,4 @@
             }
         }
     }
-
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/IFIFOPageWriter.java b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/IFIFOPageWriter.java
index 31e0e85..2dc4a58 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/IFIFOPageWriter.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/IFIFOPageWriter.java
@@ -15,7 +15,9 @@
 
 package org.apache.hyracks.storage.common.buffercache;
 
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+
 @FunctionalInterface
 public interface IFIFOPageWriter {
-    void write(ICachedPage page);
+    void write(ICachedPage page) throws HyracksDataException;
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-util/src/main/java/org/apache/hyracks/util/ExponentialRetryPolicy.java b/hyracks-fullstack/hyracks/hyracks-util/src/main/java/org/apache/hyracks/util/ExponentialRetryPolicy.java
index b332a2d..080b9ea 100644
--- a/hyracks-fullstack/hyracks/hyracks-util/src/main/java/org/apache/hyracks/util/ExponentialRetryPolicy.java
+++ b/hyracks-fullstack/hyracks/hyracks-util/src/main/java/org/apache/hyracks/util/ExponentialRetryPolicy.java
@@ -21,22 +21,31 @@
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.TimeUnit;
 
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
 public class ExponentialRetryPolicy implements IRetryPolicy {
 
+    private static final Logger LOGGER = LogManager.getLogger();
+
+    public static final String CLOUD_UNSTABLE_MODE = "cloud.unstable.mode";
     private static final int DEFAULT_MAX_RETRIES = 10;
     private static final long DEFAULT_INITIAL_DELAY_IN_MILLIS = 100;
     private static final long DEFAULT_MAX_DELAY_IN_MILLIS = Long.MAX_VALUE - 1;
+    private static final int UNSTABLE_NUMBER_OF_RETRIES = 100;
     private final int maxRetries;
     private final long maxDelay;
     private int attempt = 0;
     private long delay;
+    private boolean printDebugLines = true;
 
     /**
      * Default constructor for ExponentialRetryPolicy.
      * Initializes with default max retries, initial delay, and max delay.
      */
     public ExponentialRetryPolicy() {
-        this(DEFAULT_MAX_RETRIES, DEFAULT_INITIAL_DELAY_IN_MILLIS, DEFAULT_MAX_DELAY_IN_MILLIS);
+        this(isUnstable() ? UNSTABLE_NUMBER_OF_RETRIES : DEFAULT_MAX_RETRIES, DEFAULT_INITIAL_DELAY_IN_MILLIS,
+                isUnstable() ? 0 : DEFAULT_MAX_DELAY_IN_MILLIS);
     }
 
     /**
@@ -71,13 +80,18 @@
      */
     public ExponentialRetryPolicy(int maxRetries, long maxDelay) {
         this(maxRetries, DEFAULT_INITIAL_DELAY_IN_MILLIS, maxDelay);
+        printDebugLines = false;
     }
 
     @Override
     public boolean retry(Throwable failure) {
         if (attempt < maxRetries) {
             try {
-                TimeUnit.MILLISECONDS.sleep(ThreadLocalRandom.current().nextLong(1 + delay));
+                long sleepTime = ThreadLocalRandom.current().nextLong(1 + delay);
+                if (printDebugLines) {
+                    LOGGER.info("Retrying after {}ms, attempt: {}/{}", sleepTime, attempt + 1, maxRetries);
+                }
+                TimeUnit.MILLISECONDS.sleep(sleepTime);
             } catch (InterruptedException e) {
                 Thread.currentThread().interrupt();
             }
@@ -87,4 +101,8 @@
         }
         return false;
     }
+
+    private static boolean isUnstable() {
+        return Boolean.getBoolean(CLOUD_UNSTABLE_MODE);
+    }
 }