diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/StoragePathUtil.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/StoragePathUtil.java
index 32a226e..f84472e 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/StoragePathUtil.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/StoragePathUtil.java
@@ -90,9 +90,11 @@
     }
 
     public static int getPartitionNumFromRelativePath(String relativePath) {
-        int startIdx = relativePath.indexOf(StorageConstants.PARTITION_DIR_PREFIX)
+        int startIdx = relativePath.lastIndexOf(StorageConstants.PARTITION_DIR_PREFIX)
                 + StorageConstants.PARTITION_DIR_PREFIX.length();
-        String partition = relativePath.substring(startIdx, relativePath.indexOf(File.separatorChar, startIdx));
+        int partitionEndIdx = relativePath.indexOf(File.separatorChar, startIdx);
+        int idxEnd = partitionEndIdx != -1 ? partitionEndIdx : relativePath.length();
+        String partition = relativePath.substring(startIdx, idxEnd);
         return Integer.parseInt(partition);
     }
 
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/PersistentLocalResourceRepository.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/PersistentLocalResourceRepository.java
index 3b192aa..4d15385 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/PersistentLocalResourceRepository.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/PersistentLocalResourceRepository.java
@@ -33,6 +33,7 @@
 import java.nio.file.Paths;
 import java.text.ParseException;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -646,4 +647,28 @@
     public Path[] getStorageRoots() {
         return storageRoots;
     }
+
+    public void keepPartitions(Set<Integer> keepPartitions) {
+        List<File> onDiskPartitions = getOnDiskPartitions();
+        for (File onDiskPartition : onDiskPartitions) {
+            int partitionNum = StoragePathUtil.getPartitionNumFromRelativePath(onDiskPartition.getAbsolutePath());
+            if (!keepPartitions.contains(partitionNum)) {
+                LOGGER.warn("deleting partition {} since it is not on partitions to keep {}", partitionNum,
+                        keepPartitions);
+                FileUtils.deleteQuietly(onDiskPartition);
+            }
+        }
+    }
+
+    public List<File> getOnDiskPartitions() {
+        List<File> onDiskPartitions = new ArrayList<>();
+        for (Path root : storageRoots) {
+            File[] partitions = root.toFile().listFiles(
+                    (dir, name) -> dir.isDirectory() && name.startsWith(StorageConstants.PARTITION_DIR_PREFIX));
+            if (partitions != null) {
+                onDiskPartitions.addAll(Arrays.asList(partitions));
+            }
+        }
+        return onDiskPartitions;
+    }
 }