diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/SnapshotInfo.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/SnapshotInfo.java
index ec429bc6a65f..e72e4978d446 100644
--- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/SnapshotInfo.java
+++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/SnapshotInfo.java
@@ -128,6 +128,7 @@ public static SnapshotStatus valueOf(SnapshotStatusProto status) {
   private long referencedReplicatedSize;
   private long exclusiveSize;
   private long exclusiveReplicatedSize;
+  private boolean expandedDeletedDir;
 
   /**
    * Private constructor, constructed via builder.
@@ -149,6 +150,7 @@ public static SnapshotStatus valueOf(SnapshotStatusProto status) {
    * @param referencedReplicatedSize - Snapshot referenced size w/ replication.
    * @param exclusiveSize - Snapshot exclusive size.
    * @param exclusiveReplicatedSize - Snapshot exclusive size w/ replication.
+   * @param expandedDeletedDir - Indicates if the deleted dir is expanded.
    */
   @SuppressWarnings("checkstyle:ParameterNumber")
   private SnapshotInfo(UUID snapshotId,
@@ -168,7 +170,8 @@ private SnapshotInfo(UUID snapshotId,
                        long referencedSize,
                        long referencedReplicatedSize,
                        long exclusiveSize,
-                       long exclusiveReplicatedSize) {
+                       long exclusiveReplicatedSize,
+                       boolean expandedDeletedDir) {
     this.snapshotId = snapshotId;
     this.name = name;
     this.volumeName = volumeName;
@@ -187,6 +190,7 @@ private SnapshotInfo(UUID snapshotId,
     this.referencedReplicatedSize = referencedReplicatedSize;
     this.exclusiveSize = exclusiveSize;
     this.exclusiveReplicatedSize = exclusiveReplicatedSize;
+    this.expandedDeletedDir = expandedDeletedDir;
   }
 
   public void setName(String name) {
@@ -285,6 +289,14 @@ public void setSstFiltered(boolean sstFiltered) {
     this.sstFiltered = sstFiltered;
   }
 
+  public boolean getExpandedDeletedDir() {
+    return expandedDeletedDir;
+  }
+
+  public void setExpandedDeletedDir(boolean expandedDeletedDir) {
+    this.expandedDeletedDir = expandedDeletedDir;
+  }
+
   public static org.apache.hadoop.ozone.om.helpers.SnapshotInfo.Builder
       newBuilder() {
     return new org.apache.hadoop.ozone.om.helpers.SnapshotInfo.Builder();
@@ -308,7 +320,8 @@ public SnapshotInfo.Builder toBuilder() {
         .setReferencedSize(referencedSize)
         .setReferencedReplicatedSize(referencedReplicatedSize)
         .setExclusiveSize(exclusiveSize)
-        .setExclusiveReplicatedSize(exclusiveReplicatedSize);
+        .setExclusiveReplicatedSize(exclusiveReplicatedSize)
+        .setExpandedDeletedDir(expandedDeletedDir);
   }
 
   /**
@@ -333,6 +346,7 @@ public static class Builder {
     private long referencedReplicatedSize;
     private long exclusiveSize;
     private long exclusiveReplicatedSize;
+    private boolean expandedDeletedDir;
 
     public Builder() {
       // default values
@@ -428,6 +442,11 @@ public Builder setExclusiveReplicatedSize(long exclusiveReplicatedSize) {
       this.exclusiveReplicatedSize = exclusiveReplicatedSize;
       return this;
     }
+    public Builder setExpandedDeletedDir(boolean expandedDeletedDir) {
+      this.expandedDeletedDir = expandedDeletedDir;
+      return this;
+    }
+
 
     public SnapshotInfo build() {
       Preconditions.checkNotNull(name);
@@ -449,7 +468,8 @@ public SnapshotInfo build() {
           referencedSize,
           referencedReplicatedSize,
           exclusiveSize,
-          exclusiveReplicatedSize
+          exclusiveReplicatedSize,
+          expandedDeletedDir
       );
     }
   }
@@ -471,7 +491,8 @@ public OzoneManagerProtocolProtos.SnapshotInfo getProtobuf() {
             .setReferencedSize(referencedSize)
             .setReferencedReplicatedSize(referencedReplicatedSize)
             .setExclusiveSize(exclusiveSize)
-            .setExclusiveReplicatedSize(exclusiveReplicatedSize);
+            .setExclusiveReplicatedSize(exclusiveReplicatedSize)
+            .setExpandedDeletedDir(expandedDeletedDir);
 
     if (pathPreviousSnapshotId != null) {
       sib.setPathPreviousSnapshotID(toProtobuf(pathPreviousSnapshotId));
@@ -544,6 +565,11 @@ public static SnapshotInfo getFromProtobuf(
           snapshotInfoProto.getExclusiveReplicatedSize());
     }
 
+    if (snapshotInfoProto.hasExpandedDeletedDir()) {
+      osib.setExpandedDeletedDir(
+          snapshotInfoProto.getExpandedDeletedDir());
+    }
+
     osib.setSnapshotPath(snapshotInfoProto.getSnapshotPath())
         .setCheckpointDir(snapshotInfoProto.getCheckpointDir())
         .setDbTxSequenceNumber(snapshotInfoProto.getDbTxSequenceNumber());
@@ -661,7 +687,8 @@ public static SnapshotInfo newInstance(String volumeName,
         .setSnapshotPath(volumeName + OM_KEY_PREFIX + bucketName)
         .setVolumeName(volumeName)
         .setBucketName(bucketName)
-        .setDeepClean(true);
+        .setDeepClean(true)
+        .setExpandedDeletedDir(false);
 
     if (snapshotId != null) {
       builder.setCheckpointDir(getCheckpointDirName(snapshotId));
@@ -694,7 +721,8 @@ public boolean equals(Object o) {
         referencedSize == that.referencedSize &&
         referencedReplicatedSize == that.referencedReplicatedSize &&
         exclusiveSize == that.exclusiveSize &&
-        exclusiveReplicatedSize == that.exclusiveReplicatedSize;
+        exclusiveReplicatedSize == that.exclusiveReplicatedSize &&
+        expandedDeletedDir == that.expandedDeletedDir;
   }
 
   @Override
@@ -705,7 +733,7 @@ public int hashCode() {
         globalPreviousSnapshotId, snapshotPath, checkpointDir,
         deepClean, sstFiltered,
         referencedSize, referencedReplicatedSize,
-        exclusiveSize, exclusiveReplicatedSize);
+        exclusiveSize, exclusiveReplicatedSize, expandedDeletedDir);
   }
 
   /**
@@ -732,6 +760,7 @@ public SnapshotInfo copyObject() {
         .setReferencedReplicatedSize(referencedReplicatedSize)
         .setExclusiveSize(exclusiveSize)
         .setExclusiveReplicatedSize(exclusiveReplicatedSize)
+        .setExpandedDeletedDir(expandedDeletedDir)
         .build();
   }
 }
diff --git a/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/helpers/TestOmSnapshotInfo.java b/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/helpers/TestOmSnapshotInfo.java
index bbc416150ca0..0cd0eb8d1dcd 100644
--- a/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/helpers/TestOmSnapshotInfo.java
+++ b/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/helpers/TestOmSnapshotInfo.java
@@ -72,6 +72,7 @@ private SnapshotInfo createSnapshotInfo() {
         .setReferencedReplicatedSize(6000L)
         .setExclusiveSize(1000L)
         .setExclusiveReplicatedSize(3000L)
+        .setExpandedDeletedDir(false)
         .build();
   }
 
@@ -95,6 +96,7 @@ private OzoneManagerProtocolProtos.SnapshotInfo createSnapshotInfoProto() {
         .setReferencedReplicatedSize(6000L)
         .setExclusiveSize(1000L)
         .setExclusiveReplicatedSize(3000L)
+        .setExpandedDeletedDir(false)
         .build();
   }
 
@@ -138,7 +140,8 @@ public void testSnapshotInfoToProto() {
         snapshotInfoEntryActual.getExclusiveSize());
     Assert.assertEquals(snapshotInfoEntryExpected.getExclusiveReplicatedSize(),
         snapshotInfoEntryActual.getExclusiveReplicatedSize());
-
+    Assert.assertEquals(snapshotInfoEntryExpected.getExpandedDeletedDir(),
+        snapshotInfoEntryActual.getExpandedDeletedDir());
     Assert.assertEquals(snapshotInfoEntryExpected, snapshotInfoEntryActual);
   }
 
@@ -174,6 +177,8 @@ public void testSnapshotInfoProtoToSnapshotInfo() {
         snapshotInfoActual.getExclusiveSize());
     Assert.assertEquals(snapshotInfoExpected.getExclusiveReplicatedSize(),
         snapshotInfoActual.getExclusiveReplicatedSize());
+    Assert.assertEquals(snapshotInfoExpected.getExpandedDeletedDir(),
+        snapshotInfoActual.getExpandedDeletedDir());
 
     Assert.assertEquals(snapshotInfoExpected, snapshotInfoActual);
   }
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestDirectoryDeletingServiceWithFSO.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestDirectoryDeletingServiceWithFSO.java
index d5c042bb036a..5f120b429b0c 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestDirectoryDeletingServiceWithFSO.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestDirectoryDeletingServiceWithFSO.java
@@ -33,7 +33,9 @@
 import org.apache.hadoop.ozone.TestDataUtil;
 import org.apache.hadoop.ozone.client.OzoneBucket;
 import org.apache.hadoop.ozone.client.OzoneClient;
+import org.apache.hadoop.ozone.om.IOmMetadataReader;
 import org.apache.hadoop.ozone.om.OMMetadataManager;
+import org.apache.hadoop.ozone.om.OmSnapshot;
 import org.apache.hadoop.ozone.om.helpers.SnapshotInfo;
 import org.apache.hadoop.ozone.om.service.DirectoryDeletingService;
 import org.apache.hadoop.ozone.om.service.KeyDeletingService;
@@ -42,6 +44,8 @@
 import org.apache.hadoop.ozone.om.helpers.OmDirectoryInfo;
 import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
 import org.apache.hadoop.ozone.om.helpers.RepeatedOmKeyInfo;
+import org.apache.hadoop.ozone.om.snapshot.ReferenceCounted;
+import org.apache.hadoop.ozone.om.snapshot.SnapshotCache;
 import org.apache.ozone.test.GenericTestUtils;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.AfterAll;
@@ -57,6 +61,8 @@
 import java.util.concurrent.TimeoutException;
 import java.util.function.LongSupplier;
 
+import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_SNAPSHOT_DELETING_SERVICE_INTERVAL;
+import static org.apache.hadoop.ozone.om.OmSnapshotManager.getSnapshotPrefix;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
@@ -84,10 +90,12 @@ public class TestDirectoryDeletingServiceWithFSO {
   @BeforeAll
   public static void init() throws Exception {
     OzoneConfiguration conf = new OzoneConfiguration();
-    conf.setInt(OMConfigKeys.OZONE_DIR_DELETING_SERVICE_INTERVAL, 2000);
-    conf.setInt(OMConfigKeys.OZONE_PATH_DELETING_LIMIT_PER_TASK, 5);
+    conf.setInt(OMConfigKeys.OZONE_DIR_DELETING_SERVICE_INTERVAL, 100);
+    conf.setInt(OMConfigKeys.OZONE_PATH_DELETING_LIMIT_PER_TASK, 10);
     conf.setTimeDuration(OZONE_BLOCK_DELETING_SERVICE_INTERVAL, 100,
         TimeUnit.MILLISECONDS);
+    conf.setTimeDuration(OZONE_SNAPSHOT_DELETING_SERVICE_INTERVAL, 100,
+        TimeUnit.MILLISECONDS);
     conf.setBoolean(OMConfigKeys.OZONE_OM_RATIS_ENABLE_KEY, omRatisEnabled);
     conf.setBoolean(OZONE_ACL_ENABLED, true);
     cluster = MiniOzoneCluster.newBuilder(conf)
@@ -323,9 +331,8 @@ public void testDeleteFilesAndSubFiles() throws Exception {
       }
     }
 
-    KeyDeletingService keyDeletingService =
-        (KeyDeletingService) cluster.getOzoneManager().getKeyManager()
-            .getDeletingService();
+    KeyDeletingService keyDeletingService =  cluster.getOzoneManager()
+        .getKeyManager().getDeletingService();
 
     // Before delete
     assertTableRowCount(deletedDirTable, 0);
@@ -493,7 +500,103 @@ public void testDirDeletedTableCleanUpForSnapshot() throws Exception {
     cleanupTables();
   }
 
-  private void cleanupTables() throws IOException {
+  @Test
+  public void testExpandDirDeletedTableForAllSnapshot() throws Exception {
+    Table<String, SnapshotInfo> snapshotInfoTable =
+        cluster.getOzoneManager().getMetadataManager().getSnapshotInfoTable();
+    DirectoryDeletingService dirDeletingService =
+        (DirectoryDeletingService) cluster.getOzoneManager().getKeyManager()
+            .getDirDeletingService();
+    dirDeletingService.suspend();
+    /* For each snapshot
+        DirTable                               KeyTable
+    /v/b/snapDir                       /v/b/snapDir/testKey0 - testKey5
+    /v/b/snapDir/root/              /v/b/snapDir/root/parentDir0/file1 - file10
+    /v/b/snapDir/root/parentDir0/   /v/b/snapDir/root/parentDir1/file1 - file10
+    /v/b/snapDir/root/parentDir1/   /v/b/snapDir/root/parentDir2/file1 - file10
+    /v/b/snapDir/root/parentDir2/
+     */
+
+    for (int i = 1; i <= 5; i++) {
+      createSnapshotForDirectoryDeletingServiceTest("snap" + i, i, 11 - i);
+    }
+
+    dirDeletingService.resume();
+    long prevDDSRunCount = dirDeletingService.getRunCount().get();
+
+    GenericTestUtils.waitFor(() -> dirDeletingService.getRunCount().get() >
+        prevDDSRunCount + 25, 100, 10000);
+
+    try (TableIterator<String, ? extends Table.KeyValue
+        <String, SnapshotInfo>> iterator = snapshotInfoTable.iterator()) {
+      int fileCountMultiplier = 10;
+      while (iterator.hasNext()) {
+        SnapshotInfo currSnapInfo = iterator.next().getValue();
+        assertTrue(currSnapInfo.getExpandedDeletedDir());
+        try (ReferenceCounted<IOmMetadataReader, SnapshotCache>
+                 rcCurrOmSnapshot = cluster.getOzoneManager()
+            .getOmSnapshotManager().checkForSnapshot(
+                currSnapInfo.getVolumeName(),
+                currSnapInfo.getBucketName(),
+                getSnapshotPrefix(currSnapInfo.getName()),
+                true)) {
+
+          OmSnapshot currOmSnapshot = (OmSnapshot) rcCurrOmSnapshot.get();
+          Table<String, OmKeyInfo> snapDeletedDirTable =
+              currOmSnapshot.getMetadataManager().getDeletedDirTable();
+          Table<String, RepeatedOmKeyInfo> deletedTable =
+              currOmSnapshot.getMetadataManager().getDeletedTable();
+          assertTrue(snapDeletedDirTable.isEmpty());
+          // Created 5 files. And two subdirectories with fileCountMultiplier.
+          assertTableRowCount(deletedTable, fileCountMultiplier * 2 + 5);
+          fileCountMultiplier--;
+        }
+      }
+    }
+    cleanupTables();
+  }
+
+  private void createSnapshotForDirectoryDeletingServiceTest(
+      String rootPath, int count, int fileCount) throws Exception {
+
+    Table<String, OmKeyInfo> deletedDirTable =
+        cluster.getOzoneManager().getMetadataManager().getDeletedDirTable();
+    Table<String, SnapshotInfo> snapshotInfoTable =
+        cluster.getOzoneManager().getMetadataManager().getSnapshotInfoTable();
+    Path root = new Path("/" + rootPath);
+    Path appRoot = new Path(root, "root");
+    // Create  parent dir from root.
+    fs.mkdirs(root);
+
+    // Added 5 sub files inside root dir
+    for (int i = 0; i < 5; i++) {
+      Path path = new Path(root, "testKey" + i);
+      try (FSDataOutputStream stream = fs.create(path)) {
+        stream.write(1);
+      }
+    }
+
+    // Add 2*5 more sub files in different level
+    for (int i = 0; i < 2; i++) {
+      Path parent = new Path(appRoot, "parentDir" + i);
+      for (int j = 1; j <= fileCount; j++) {
+        Path child = new Path(parent, "file" + j);
+        ContractTestUtils.touch(fs, child);
+      }
+    }
+
+    // Delete dir
+    fs.delete(root, true);
+    // For 1st snapshot deletedDirTable will be expanded, as DeletedDirService
+    // runs every 100ms for this test
+    if (count != 1) {
+      assertTableRowCount(deletedDirTable, 1);
+    }
+    client.getObjectStore().createSnapshot(volumeName, bucketName, rootPath);
+    assertTableRowCount(snapshotInfoTable, count);
+  }
+
+  private void cleanupTables() throws Exception {
     OMMetadataManager metadataManager =
         cluster.getOzoneManager().getMetadataManager();
 
@@ -508,6 +611,13 @@ private void cleanupTables() throws IOException {
         .iterator()) {
       removeAllFromDB(it);
     }
+
+    try (TableIterator<?, ?> it = metadataManager.getSnapshotInfoTable()
+        .iterator()) {
+      removeAllFromDB(it);
+      cluster.getOzoneManager().getOmSnapshotManager()
+          .getSnapshotCache().invalidateAll();
+    }
   }
 
   private static void removeAllFromDB(TableIterator<?, ?> iterator)
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestSnapshotDeletingService.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestSnapshotDeletingService.java
index 98c23b8076f8..acb965b65330 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestSnapshotDeletingService.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestSnapshotDeletingService.java
@@ -198,7 +198,6 @@ public void testMultipleSnapshotKeyReclaim() throws Exception {
   }
 
   @SuppressWarnings("checkstyle:MethodLength")
-  @Flaky("HDDS-9023")
   @Test
   public void testSnapshotWithFSO() throws Exception {
     Table<String, OmDirectoryInfo> dirTable =
@@ -248,7 +247,7 @@ public void testSnapshotWithFSO() throws Exception {
       }
     }
 
-    // Total 12 dirs, 19 keys.
+    // Total 12 dirs, 24 keys.
     assertTableRowCount(dirTable, 12);
     assertTableRowCount(keyTable, 24);
     assertTableRowCount(deletedDirTable, 0);
@@ -367,17 +366,15 @@ public void testSnapshotWithFSO() throws Exception {
         .checkForSnapshot(VOLUME_NAME, BUCKET_NAME_TWO,
             getSnapshotPrefix("snap3"), true).get();
 
-    Table<String, OmKeyInfo> snapDeletedDirTable =
-        snap3.getMetadataManager().getDeletedDirTable();
-    Table<String, String> snapRenamedTable =
+    Table<String, String> snap3RenamedTable =
         snap3.getMetadataManager().getSnapshotRenamedTable();
-    Table<String, RepeatedOmKeyInfo> snapDeletedTable =
+    Table<String, RepeatedOmKeyInfo> snap3DeletedTable =
         snap3.getMetadataManager().getDeletedTable();
 
-    assertTableRowCount(snapRenamedTable, 4);
-    assertTableRowCount(snapDeletedDirTable, 3);
+    // Renamed Dir should not be moved.
+    assertTableRowCount(snap3RenamedTable, 3);
     // All the keys deleted before snapshot2 is moved to snap3
-    assertTableRowCount(snapDeletedTable, 15);
+    assertTableRowCount(snap3DeletedTable, 24);
 
     // Before deleting the last snapshot
     assertTableRowCount(renamedTable, 0);
@@ -389,8 +386,12 @@ public void testSnapshotWithFSO() throws Exception {
 
     // Check entries moved to active DB
     assertTableRowCount(snapshotInfoTable, 1);
-    assertTableRowCount(renamedTable, 4);
-    assertTableRowCount(deletedDirTable, 3);
+    assertTableRowCount(renamedTable, 3);
+    // No directories are moved between snapshots.
+    // They are either expanded by KeyDeletingService for
+    // Active Snapshots or SnapshotDeletingService for
+    // Deleted Snapshots.
+    assertTableRowCount(deletedDirTable, 0);
 
     ReferenceCounted<IOmMetadataReader, SnapshotCache> rcSnap1 =
         om.getOmSnapshotManager().checkForSnapshot(
@@ -426,7 +427,7 @@ public void testSnapshotWithFSO() throws Exception {
         }
       }
     }
-    assertTableRowCount(deletedTable, 15);
+    assertTableRowCount(deletedTable, 24);
 
     snap1 = null;
     rcSnap1.close();
diff --git a/hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto b/hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto
index fb3c6925fc02..080ade872d1e 100644
--- a/hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto
+++ b/hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto
@@ -841,6 +841,7 @@ message SnapshotInfo {
   // snapshot exclusive size after replication
   optional uint64 exclusiveReplicatedSize = 18;
   // note: shared sizes can be calculated from: referenced - exclusive
+  optional bool expandedDeletedDir = 19;
 }
 
 message SnapshotDiffJobProto {
@@ -1863,11 +1864,12 @@ message SnapshotPurgeRequest {
 }
 
 message SetSnapshotPropertyRequest {
-  optional SnapshotProperty snapshotProperty = 1;
+  optional string snapshotKey = 1;
+  optional SnapshotSize snapshotSize = 2;
+  optional bool expandedDeletedDir = 3;
 }
 
-message SnapshotProperty {
-  optional string snapshotKey = 1;
+message SnapshotSize {
   optional uint64 exclusiveSize = 2;
   optional uint64 exclusiveReplicatedSize = 3;
 }
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/snapshot/OMSnapshotMoveDeletedKeysRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/snapshot/OMSnapshotMoveDeletedKeysRequest.java
index 7dacebea9ecc..03dd7f895dea 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/snapshot/OMSnapshotMoveDeletedKeysRequest.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/snapshot/OMSnapshotMoveDeletedKeysRequest.java
@@ -89,12 +89,10 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
           moveDeletedKeysRequest.getReclaimKeysList();
       List<HddsProtos.KeyValue> renamedKeysList =
           moveDeletedKeysRequest.getRenamedKeysList();
-      List<String> movedDirs =
-          moveDeletedKeysRequest.getDeletedDirsToMoveList();
 
       omClientResponse = new OMSnapshotMoveDeletedKeysResponse(
           omResponse.build(), fromSnapshot, nextSnapshot,
-          nextDBKeysList, reclaimKeysList, renamedKeysList, movedDirs);
+          nextDBKeysList, reclaimKeysList, renamedKeysList);
 
     } catch (IOException ex) {
       omClientResponse = new OMSnapshotMoveDeletedKeysResponse(
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/snapshot/OMSnapshotPurgeRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/snapshot/OMSnapshotPurgeRequest.java
index 9d79063864d2..d725eabb5b25 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/snapshot/OMSnapshotPurgeRequest.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/snapshot/OMSnapshotPurgeRequest.java
@@ -99,8 +99,11 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
             .getNextActiveSnapshot(fromSnapshot,
                 snapshotChainManager, omSnapshotManager);
 
-        updateSnapshotInfoAndCache(nextSnapshot, omMetadataManager,
-            trxnLogIndex, updatedSnapInfos, true);
+        if (nextSnapshot != null) {
+          updateSnapshotInfoAndCache(nextSnapshot, omMetadataManager,
+              trxnLogIndex, updatedSnapInfos, true);
+        }
+
         updateSnapshotChainAndCache(omMetadataManager, fromSnapshot,
             trxnLogIndex, updatedPathPreviousAndGlobalSnapshots);
         ozoneManager.getOmSnapshotManager().getSnapshotCache()
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/snapshot/OMSnapshotSetPropertyRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/snapshot/OMSnapshotSetPropertyRequest.java
index c0b1b4f3ae81..6a81f5269700 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/snapshot/OMSnapshotSetPropertyRequest.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/snapshot/OMSnapshotSetPropertyRequest.java
@@ -30,7 +30,7 @@
 import org.apache.hadoop.ozone.om.response.snapshot.OMSnapshotSetPropertyResponse;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest;
-import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.SnapshotProperty;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.SnapshotSize;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -62,15 +62,10 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
         setSnapshotPropertyRequest = getOmRequest()
         .getSetSnapshotPropertyRequest();
 
-    SnapshotProperty snapshotProperty = setSnapshotPropertyRequest
-        .getSnapshotProperty();
+    String snapshotKey = setSnapshotPropertyRequest.getSnapshotKey();
     SnapshotInfo updatedSnapInfo = null;
 
     try {
-      String snapshotKey = snapshotProperty.getSnapshotKey();
-      long exclusiveSize = snapshotProperty.getExclusiveSize();
-      long exclusiveReplicatedSize = snapshotProperty
-          .getExclusiveReplicatedSize();
       updatedSnapInfo = metadataManager.getSnapshotInfoTable()
           .get(snapshotKey);
 
@@ -80,9 +75,22 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
             " is not found", INVALID_SNAPSHOT_ERROR);
       }
 
-      // Set Exclusive size.
-      updatedSnapInfo.setExclusiveSize(exclusiveSize);
-      updatedSnapInfo.setExclusiveReplicatedSize(exclusiveReplicatedSize);
+      if (setSnapshotPropertyRequest.hasExpandedDeletedDir()) {
+        updatedSnapInfo.setExpandedDeletedDir(setSnapshotPropertyRequest
+            .getExpandedDeletedDir());
+      }
+
+      if (setSnapshotPropertyRequest.hasSnapshotSize()) {
+        SnapshotSize snapshotSize = setSnapshotPropertyRequest
+            .getSnapshotSize();
+        long exclusiveSize = snapshotSize.getExclusiveSize();
+        long exclusiveReplicatedSize = snapshotSize
+            .getExclusiveReplicatedSize();
+        // Set Exclusive size.
+        updatedSnapInfo.setExclusiveSize(exclusiveSize);
+        updatedSnapInfo.setExclusiveReplicatedSize(exclusiveReplicatedSize);
+      }
+
       // Update Table Cache
       metadataManager.getSnapshotInfoTable().addCacheEntry(
           new CacheKey<>(snapshotKey),
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/snapshot/OMSnapshotMoveDeletedKeysResponse.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/snapshot/OMSnapshotMoveDeletedKeysResponse.java
index f4142400d7cf..87c40c0bceba 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/snapshot/OMSnapshotMoveDeletedKeysResponse.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/snapshot/OMSnapshotMoveDeletedKeysResponse.java
@@ -55,22 +55,19 @@ public class OMSnapshotMoveDeletedKeysResponse extends OMClientResponse {
   private List<SnapshotMoveKeyInfos> nextDBKeysList;
   private List<SnapshotMoveKeyInfos> reclaimKeysList;
   private List<HddsProtos.KeyValue> renamedKeysList;
-  private List<String> movedDirs;
 
   public OMSnapshotMoveDeletedKeysResponse(OMResponse omResponse,
       @Nonnull SnapshotInfo fromSnapshot,
       SnapshotInfo nextSnapshot,
       List<SnapshotMoveKeyInfos> nextDBKeysList,
       List<SnapshotMoveKeyInfos> reclaimKeysList,
-      List<HddsProtos.KeyValue> renamedKeysList,
-      List<String> movedDirs) {
+      List<HddsProtos.KeyValue> renamedKeysList) {
     super(omResponse);
     this.fromSnapshot = fromSnapshot;
     this.nextSnapshot = nextSnapshot;
     this.nextDBKeysList = nextDBKeysList;
     this.reclaimKeysList = reclaimKeysList;
     this.renamedKeysList = renamedKeysList;
-    this.movedDirs = movedDirs;
   }
 
   /**
@@ -117,8 +114,6 @@ protected void addToDBBatch(OMMetadataManager omMetadataManager,
           try (BatchOperation writeBatch =
               nextSnapshotStore.initBatchOperation()) {
             processKeys(writeBatch, nextOmSnapshot.getMetadataManager());
-            processDirs(writeBatch, nextOmSnapshot.getMetadataManager(),
-                fromOmSnapshot);
             nextSnapshotStore.commitBatchOperation(writeBatch);
             nextSnapshotStore.getDb().flushWal(true);
             nextSnapshotStore.getDb().flush();
@@ -127,7 +122,6 @@ protected void addToDBBatch(OMMetadataManager omMetadataManager,
       } else {
         // Handle the case where there is no next Snapshot.
         processKeys(batchOperation, omMetadataManager);
-        processDirs(batchOperation, omMetadataManager, fromOmSnapshot);
       }
 
       // Update From Snapshot Deleted Table.
@@ -137,7 +131,6 @@ protected void addToDBBatch(OMMetadataManager omMetadataManager,
           fromSnapshotStore.initBatchOperation()) {
         processReclaimKeys(fromSnapshotBatchOp,
             fromOmSnapshot.getMetadataManager());
-        deleteDirsFromSnapshot(fromSnapshotBatchOp, fromOmSnapshot);
         fromSnapshotStore.commitBatchOperation(fromSnapshotBatchOp);
         fromSnapshotStore.getDb().flushWal(true);
         fromSnapshotStore.getDb().flush();
@@ -146,18 +139,6 @@ protected void addToDBBatch(OMMetadataManager omMetadataManager,
 
   }
 
-  private void deleteDirsFromSnapshot(BatchOperation batchOp,
-      OmSnapshot fromOmSnapshot)
-      throws IOException {
-    for (String movedDirsKey : movedDirs) {
-      // Delete dirs from current snapshot that are moved to next snapshot.
-      fromOmSnapshot
-          .getMetadataManager()
-          .getDeletedDirTable()
-          .deleteWithBatch(batchOp, movedDirsKey);
-    }
-  }
-
   private void processReclaimKeys(BatchOperation batchOp,
                                   OMMetadataManager metadataManager)
       throws IOException {
@@ -178,23 +159,6 @@ private void processReclaimKeys(BatchOperation batchOp,
     }
   }
 
-  private void processDirs(BatchOperation batchOp,
-                           OMMetadataManager omMetadataManager,
-                           OmSnapshot fromOmSnapshot)
-      throws IOException {
-    for (String movedDirsKey : movedDirs) {
-      OmKeyInfo keyInfo = fromOmSnapshot.getMetadataManager()
-          .getDeletedDirTable()
-          .get(movedDirsKey);
-      if (keyInfo == null) {
-        continue;
-      }
-      // Move deleted dirs to next snapshot or active DB
-      omMetadataManager.getDeletedDirTable().putWithBatch(
-          batchOp, movedDirsKey, keyInfo);
-    }
-  }
-
   private void processKeys(BatchOperation batchOp,
       OMMetadataManager metadataManager) throws IOException {
 
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/DirectoryDeletingService.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/DirectoryDeletingService.java
index 71f352588d1b..26c1c0141132 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/DirectoryDeletingService.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/DirectoryDeletingService.java
@@ -17,6 +17,7 @@
 package org.apache.hadoop.ozone.om.service;
 
 import com.google.common.annotations.VisibleForTesting;
+import com.google.protobuf.ServiceException;
 import org.apache.commons.lang3.tuple.Pair;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
 import org.apache.hadoop.hdds.conf.StorageUnit;
@@ -32,13 +33,22 @@
 import org.apache.hadoop.ozone.om.OmSnapshot;
 import org.apache.hadoop.ozone.om.OmSnapshotManager;
 import org.apache.hadoop.ozone.om.OzoneManager;
+import org.apache.hadoop.ozone.om.helpers.OMRatisHelper;
+import org.apache.hadoop.ozone.om.helpers.OmBucketInfo;
 import org.apache.hadoop.ozone.om.helpers.OmDirectoryInfo;
 import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
+import org.apache.hadoop.ozone.om.helpers.SnapshotInfo;
+import org.apache.hadoop.ozone.om.ratis.OzoneManagerRatisServer;
 import org.apache.hadoop.ozone.om.snapshot.ReferenceCounted;
 import org.apache.hadoop.ozone.om.snapshot.SnapshotCache;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.PurgePathRequest;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.SetSnapshotPropertyRequest;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Type;
 import org.apache.hadoop.util.Time;
 import org.apache.ratis.protocol.ClientId;
+import org.apache.ratis.protocol.Message;
+import org.apache.ratis.protocol.RaftClientRequest;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -48,8 +58,11 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import static org.apache.hadoop.ozone.OzoneConsts.OM_KEY_PREFIX;
 import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_PATH_DELETING_LIMIT_PER_TASK;
 import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_PATH_DELETING_LIMIT_PER_TASK_DEFAULT;
+import static org.apache.hadoop.ozone.om.OmSnapshotManager.getSnapshotPrefix;
+import static org.apache.hadoop.ozone.om.helpers.SnapshotInfo.SnapshotStatus.SNAPSHOT_ACTIVE;
 
 /**
  * This is a background service to delete orphan directories and its
@@ -72,8 +85,6 @@ public class DirectoryDeletingService extends AbstractKeyDeletingService {
   public static final Logger LOG =
       LoggerFactory.getLogger(DirectoryDeletingService.class);
 
-  private static ClientId clientId = ClientId.randomId();
-
   // Use only a single thread for DirDeletion. Multiple threads would read
   // or write to same tables and can send deletion requests for same key
   // multiple times.
@@ -212,8 +223,7 @@ public BackgroundTaskResult call() {
           optimizeDirDeletesAndSubmitRequest(
               remainNum, dirNum, subDirNum, subFileNum,
               allSubDirList, purgePathRequestList, null, startTime,
-              ratisByteLimit - consumedSize,
-              getOzoneManager().getKeyManager());
+              ratisByteLimit - consumedSize, getOzoneManager().getKeyManager());
 
         } catch (IOException e) {
           LOG.error("Error while running delete directories and files " +
@@ -223,12 +233,194 @@ public BackgroundTaskResult call() {
           getOzoneManager().getMetadataManager().getTableLock(
               OmMetadataManagerImpl.DELETED_DIR_TABLE).writeLock().unlock();
         }
+
+        try {
+          if (remainNum > 0) {
+            expandSnapshotDirectories(remainNum);
+          }
+        } catch (Exception e) {
+          LOG.error("Error while running deep clean on snapshots. Will " +
+              "retry at next run.", e);
+        }
       }
 
       // place holder by returning empty results of this call back.
       return BackgroundTaskResult.EmptyTaskResult.newResult();
     }
 
+    private void expandSnapshotDirectories(long remainNum) throws IOException {
+      OmSnapshotManager omSnapshotManager =
+          getOzoneManager().getOmSnapshotManager();
+      Table<String, SnapshotInfo> snapshotInfoTable =
+          getOzoneManager().getMetadataManager().getSnapshotInfoTable();
+
+      long dirNum = 0L;
+      long subDirNum = 0L;
+      long subFileNum = 0L;
+      int consumedSize = 0;
+      List<PurgePathRequest> purgePathRequestList = new ArrayList<>();
+      try (TableIterator<String, ? extends Table.KeyValue
+          <String, SnapshotInfo>> iterator = snapshotInfoTable.iterator()) {
+
+        while (remainNum > 0 && iterator.hasNext()) {
+          SnapshotInfo currSnapInfo = iterator.next().getValue();
+
+          // Expand deleted dirs only on active snapshot. Deleted Snapshots
+          // will be cleaned up by SnapshotDeletingService.
+          if (currSnapInfo.getSnapshotStatus() != SNAPSHOT_ACTIVE ||
+              currSnapInfo.getExpandedDeletedDir()) {
+            continue;
+          }
+
+          long volumeId = getOzoneManager().getMetadataManager()
+              .getVolumeId(currSnapInfo.getVolumeName());
+          // Get bucketInfo for the snapshot bucket to get bucket layout.
+          String dbBucketKey = getOzoneManager().getMetadataManager()
+              .getBucketKey(currSnapInfo.getVolumeName(),
+                  currSnapInfo.getBucketName());
+          OmBucketInfo bucketInfo = getOzoneManager().getMetadataManager()
+              .getBucketTable().get(dbBucketKey);
+
+          if (bucketInfo == null) {
+            throw new IllegalStateException("Bucket " + "/" +
+                currSnapInfo.getVolumeName() + "/" + currSnapInfo
+                .getBucketName() + " is not found. BucketInfo should not be " +
+                "null for snapshotted bucket. The OM is in unexpected state.");
+          }
+
+          String dbBucketKeyForDir = getOzoneManager().getMetadataManager()
+              .getBucketKey(Long.toString(volumeId),
+                  Long.toString(bucketInfo.getObjectID())) + OM_KEY_PREFIX;
+
+          try (ReferenceCounted<IOmMetadataReader, SnapshotCache>
+                   rcCurrOmSnapshot = omSnapshotManager.checkForSnapshot(
+              currSnapInfo.getVolumeName(),
+              currSnapInfo.getBucketName(),
+              getSnapshotPrefix(currSnapInfo.getName()),
+              true)) {
+
+            OmSnapshot currOmSnapshot = (OmSnapshot) rcCurrOmSnapshot.get();
+            Table<String, OmKeyInfo> snapDeletedDirTable =
+                currOmSnapshot.getMetadataManager().getDeletedDirTable();
+
+            if (snapDeletedDirTable.isEmpty()) {
+              updateExpandedSnapshotDir(currSnapInfo.getTableKey());
+              continue;
+            }
+
+            List<Pair<String, OmKeyInfo>> allSubDirList
+                = new ArrayList<>((int) remainNum);
+
+            try (TableIterator<String, ? extends Table.KeyValue<String,
+                OmKeyInfo>> deletedIterator = snapDeletedDirTable.iterator()) {
+
+              long startTime = Time.monotonicNow();
+              deletedIterator.seek(dbBucketKeyForDir);
+
+              while (remainNum > 0 && deletedIterator.hasNext()) {
+                Table.KeyValue<String, OmKeyInfo> deletedDirInfo =
+                    deletedIterator.next();
+                String deletedDirKey = deletedDirInfo.getKey();
+
+                // Exit if it is out of the bucket scope.
+                if (!deletedDirKey.startsWith(dbBucketKeyForDir)) {
+                  break;
+                }
+
+                PurgePathRequest request = prepareDeleteDirRequest(
+                    remainNum, deletedDirInfo.getValue(),
+                    deletedDirInfo.getKey(), allSubDirList,
+                    currOmSnapshot.getKeyManager());
+                if (isBufferLimitCrossed(ratisByteLimit, consumedSize,
+                    request.getSerializedSize())) {
+                  if (purgePathRequestList.size() != 0) {
+                    // if message buffer reaches max limit,
+                    // avoid sending further
+                    remainNum = 0;
+                    break;
+                  }
+                  // if directory itself is having a lot of keys / files,
+                  // reduce capacity to minimum level
+                  remainNum = MIN_ERR_LIMIT_PER_TASK;
+                  request = prepareDeleteDirRequest(
+                      remainNum, deletedDirInfo.getValue(),
+                      deletedDirInfo.getKey(), allSubDirList,
+                      currOmSnapshot.getKeyManager());
+                }
+
+                consumedSize += request.getSerializedSize();
+                purgePathRequestList.add(request);
+                remainNum = remainNum - request.getDeletedSubFilesCount();
+                remainNum = remainNum - request.getMarkDeletedSubDirsCount();
+                // Count up the purgeDeletedDir, subDirs and subFiles
+                if (request.hasDeletedDir() &&
+                    !request.getDeletedDir().isEmpty()) {
+                  dirNum++;
+                }
+                subDirNum += request.getMarkDeletedSubDirsCount();
+                subFileNum += request.getDeletedSubFilesCount();
+              }
+
+              optimizeDirDeletesAndSubmitRequest(
+                  remainNum, dirNum, subDirNum, subFileNum,
+                  allSubDirList, purgePathRequestList,
+                  currSnapInfo.getTableKey(), startTime,
+                  ratisByteLimit - consumedSize,
+                  currOmSnapshot.getKeyManager());
+
+            } catch (IOException e) {
+              LOG.error("Error while expanding snapshot " +
+                  currSnapInfo.getTableKey() + " deleted directories and " +
+                  "files. Will retry at next run.", e);
+            }
+          }
+        }
+      }
+    }
+
+    private void updateExpandedSnapshotDir(String snapshotKeyTable) {
+      ClientId clientId = ClientId.randomId();
+      SetSnapshotPropertyRequest setSnapshotPropertyRequest =
+          SetSnapshotPropertyRequest.newBuilder()
+              .setSnapshotKey(snapshotKeyTable)
+              .setExpandedDeletedDir(true)
+              .build();
+
+      OMRequest omRequest = OMRequest.newBuilder()
+          .setCmdType(Type.SetSnapshotProperty)
+          .setSetSnapshotPropertyRequest(setSnapshotPropertyRequest)
+          .setClientId(clientId.toString())
+          .build();
+
+      submitRequest(omRequest, clientId);
+    }
+
+    public void submitRequest(OMRequest omRequest, ClientId clientId) {
+      try {
+        if (isRatisEnabled()) {
+          OzoneManagerRatisServer server = getOzoneManager().getOmRatisServer();
+
+          RaftClientRequest raftClientRequest = RaftClientRequest.newBuilder()
+              .setClientId(clientId)
+              .setServerId(server.getRaftPeerId())
+              .setGroupId(server.getRaftGroupId())
+              .setCallId(getRunCount().get())
+              .setMessage(Message.valueOf(
+                  OMRatisHelper.convertRequestToByteString(omRequest)))
+              .setType(RaftClientRequest.writeRequestType())
+              .build();
+
+          server.submitRequest(omRequest, raftClientRequest);
+        } else {
+          getOzoneManager().getOmServerProtocol()
+              .submitRequest(null, omRequest);
+        }
+      } catch (ServiceException e) {
+        LOG.error("Snapshot deep cleaning request failed. " +
+            "Will retry at next run.", e);
+      }
+    }
+
     private boolean previousSnapshotHasDir(
         KeyValue<String, OmKeyInfo> pendingDeletedDirInfo) throws IOException {
       String key = pendingDeletedDirInfo.getKey();
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/KeyDeletingService.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/KeyDeletingService.java
index a0ca1ae547ac..aa549219a625 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/KeyDeletingService.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/KeyDeletingService.java
@@ -49,7 +49,7 @@
 import org.apache.hadoop.ozone.om.snapshot.SnapshotCache;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.SnapshotPurgeRequest;
-import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.SnapshotProperty;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.SnapshotSize;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.SetSnapshotPropertyRequest;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Type;
 import org.apache.hadoop.hdds.utils.BackgroundTask;
@@ -98,6 +98,7 @@ public class KeyDeletingService extends AbstractKeyDeletingService {
   private final Map<String, Long> exclusiveSizeMap;
   private final Map<String, Long> exclusiveReplicatedSizeMap;
   private final Set<String> completedExclusiveSizeSet;
+  private final Map<String, String> snapshotSeekMap;
 
   public KeyDeletingService(OzoneManager ozoneManager,
       ScmBlockLocationProtocol scmClient,
@@ -116,6 +117,7 @@ public KeyDeletingService(OzoneManager ozoneManager,
     this.exclusiveSizeMap = new HashMap<>();
     this.exclusiveReplicatedSizeMap = new HashMap<>();
     this.completedExclusiveSizeSet = new HashSet<>();
+    this.snapshotSeekMap = new HashMap<>();
   }
 
   /**
@@ -341,11 +343,22 @@ private void processSnapshotDeepClean(int delCount)
                 RepeatedOmKeyInfo>> deletedIterator = snapDeletedTable
                 .iterator()) {
 
-              deletedIterator.seek(snapshotBucketKey);
+              String lastKeyInCurrentRun = null;
+              String deletedTableSeek = snapshotSeekMap.getOrDefault(
+                  currSnapInfo.getTableKey(), snapshotBucketKey);
+              deletedIterator.seek(deletedTableSeek);
+              // To avoid processing the last key from the previous
+              // run again.
+              if (!deletedTableSeek.equals(snapshotBucketKey) &&
+                  deletedIterator.hasNext()) {
+                deletedIterator.next();
+              }
+
               while (deletedIterator.hasNext() && delCount < keyLimitPerTask) {
                 Table.KeyValue<String, RepeatedOmKeyInfo>
                     deletedKeyValue = deletedIterator.next();
                 String deletedKey = deletedKeyValue.getKey();
+                lastKeyInCurrentRun = deletedKey;
 
                 // Exit if it is out of the bucket scope.
                 if (!deletedKey.startsWith(snapshotBucketKey)) {
@@ -405,6 +418,15 @@ private void processSnapshotDeepClean(int delCount)
                   completedExclusiveSizeSet.add(
                       previousSnapshot.getTableKey());
                 }
+
+                snapshotSeekMap.remove(currSnapInfo.getTableKey());
+              } else {
+                // There are keys that still needs processing
+                // we can continue from it in the next iteration
+                if (lastKeyInCurrentRun != null) {
+                  snapshotSeekMap.put(currSnapInfo.getTableKey(),
+                      lastKeyInCurrentRun);
+                }
               }
 
               if (!keysToPurge.isEmpty()) {
@@ -524,15 +546,15 @@ private void updateSnapshotExclusiveSize() {
       while (completedSnapshotIterator.hasNext()) {
         ClientId clientId = ClientId.randomId();
         String dbKey = completedSnapshotIterator.next();
-        SnapshotProperty snapshotProperty = SnapshotProperty.newBuilder()
-                .setSnapshotKey(dbKey)
+        SnapshotSize snapshotSize = SnapshotSize.newBuilder()
                 .setExclusiveSize(exclusiveSizeMap.get(dbKey))
                 .setExclusiveReplicatedSize(
                     exclusiveReplicatedSizeMap.get(dbKey))
                 .build();
         SetSnapshotPropertyRequest setSnapshotPropertyRequest =
             SetSnapshotPropertyRequest.newBuilder()
-                .setSnapshotProperty(snapshotProperty)
+                .setSnapshotKey(dbKey)
+                .setSnapshotSize(snapshotSize)
                 .build();
 
         OMRequest omRequest = OMRequest.newBuilder()
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/SnapshotDeletingService.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/SnapshotDeletingService.java
index 5aa2fc78a0af..fde412e4541c 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/SnapshotDeletingService.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/SnapshotDeletingService.java
@@ -44,7 +44,6 @@
 import org.apache.hadoop.ozone.om.SnapshotChainManager;
 import org.apache.hadoop.ozone.om.helpers.OMRatisHelper;
 import org.apache.hadoop.ozone.om.helpers.OmBucketInfo;
-import org.apache.hadoop.ozone.om.helpers.OmDirectoryInfo;
 import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
 import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo;
 import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfoGroup;
@@ -92,7 +91,6 @@ public class SnapshotDeletingService extends AbstractKeyDeletingService {
   // multiple times.
   private static final int SNAPSHOT_DELETING_CORE_POOL_SIZE = 1;
   private static final int MIN_ERR_LIMIT_PER_TASK = 1000;
-  private final ClientId clientId = ClientId.randomId();
 
   private final OzoneManager ozoneManager;
   private final OmSnapshotManager omSnapshotManager;
@@ -220,7 +218,6 @@ public BackgroundTaskResult call() throws InterruptedException {
           SnapshotInfo previousSnapshot = getPreviousActiveSnapshot(
               snapInfo, chainManager, omSnapshotManager);
           Table<String, OmKeyInfo> previousKeyTable = null;
-          Table<String, OmDirectoryInfo> previousDirTable = null;
           OmSnapshot omPreviousSnapshot = null;
 
           // Split RepeatedOmKeyInfo and update current snapshot deletedKeyTable
@@ -235,8 +232,6 @@ public BackgroundTaskResult call() throws InterruptedException {
 
             previousKeyTable = omPreviousSnapshot
                 .getMetadataManager().getKeyTable(bucketInfo.getBucketLayout());
-            previousDirTable = omPreviousSnapshot
-                .getMetadataManager().getDirectoryTable();
           }
 
           // Move key to either next non deleted snapshot's deletedTable
@@ -245,11 +240,9 @@ public BackgroundTaskResult call() throws InterruptedException {
           List<SnapshotMoveKeyInfos> toNextDBList = new ArrayList<>();
           // A list of renamed keys/files/dirs
           List<HddsProtos.KeyValue> renamedList = new ArrayList<>();
-          List<String> dirsToMove = new ArrayList<>();
 
           long remainNum = handleDirectoryCleanUp(snapshotDeletedDirTable,
-              previousDirTable, renamedTable, dbBucketKeyForDir, snapInfo,
-              omSnapshot, dirsToMove, renamedList);
+              dbBucketKeyForDir, snapInfo, omSnapshot);
           int deletionCount = 0;
 
           try (TableIterator<String, ? extends Table.KeyValue<String,
@@ -325,7 +318,7 @@ public BackgroundTaskResult call() throws InterruptedException {
           snapshotLimit--;
           // Submit Move request to OM.
           submitSnapshotMoveDeletedKeys(snapInfo, toReclaimList,
-              toNextDBList, renamedList, dirsToMove);
+              toNextDBList, renamedList);
 
           // Properly decrement ref count for rcOmPreviousSnapshot
           if (rcOmPreviousSnapshot != null) {
@@ -378,14 +371,10 @@ private boolean isSnapshotReclaimable(
           (isKeyTableCleanedUp || snapshotDeletedTable.isEmpty());
     }
 
-    @SuppressWarnings("checkstyle:ParameterNumber")
     private long handleDirectoryCleanUp(
         Table<String, OmKeyInfo> snapshotDeletedDirTable,
-        Table<String, OmDirectoryInfo> previousDirTable,
-        Table<String, String> renamedTable,
         String dbBucketKeyForDir, SnapshotInfo snapInfo,
-        OmSnapshot omSnapshot, List<String> dirsToMove,
-        List<HddsProtos.KeyValue> renamedList) {
+        OmSnapshot omSnapshot) {
 
       long dirNum = 0L;
       long subDirNum = 0L;
@@ -402,44 +391,44 @@ private long handleDirectoryCleanUp(
         long startTime = Time.monotonicNow();
         deletedDirIterator.seek(dbBucketKeyForDir);
 
-        while (deletedDirIterator.hasNext()) {
+        while (remainNum > 0 && deletedDirIterator.hasNext()) {
           Table.KeyValue<String, OmKeyInfo> deletedDir =
               deletedDirIterator.next();
+          String deletedDirKey = deletedDir.getKey();
+
+          // Exit for dirs out of snapshot scope.
+          if (!deletedDirKey.startsWith(dbBucketKeyForDir)) {
+            break;
+          }
 
-          if (isDirReclaimable(deletedDir, previousDirTable,
-              renamedTable, renamedList)) {
-            // Reclaim here
-            PurgePathRequest request = prepareDeleteDirRequest(
+          PurgePathRequest request = prepareDeleteDirRequest(
+              remainNum, deletedDir.getValue(), deletedDir.getKey(),
+              allSubDirList, omSnapshot.getKeyManager());
+          if (isBufferLimitCrossed(ratisByteLimit, consumedSize,
+              request.getSerializedSize())) {
+            if (purgePathRequestList.size() != 0) {
+              // if message buffer reaches max limit, avoid sending further
+              remainNum = 0;
+              break;
+            }
+            // if directory itself is having a lot of keys / files,
+            // reduce capacity to minimum level
+            remainNum = MIN_ERR_LIMIT_PER_TASK;
+            request = prepareDeleteDirRequest(
                 remainNum, deletedDir.getValue(), deletedDir.getKey(),
                 allSubDirList, omSnapshot.getKeyManager());
-            if (isBufferLimitCrossed(ratisByteLimit, consumedSize,
-                request.getSerializedSize())) {
-              if (purgePathRequestList.size() != 0) {
-                // if message buffer reaches max limit, avoid sending further
-                remainNum = 0;
-                break;
-              }
-              // if directory itself is having a lot of keys / files,
-              // reduce capacity to minimum level
-              remainNum = MIN_ERR_LIMIT_PER_TASK;
-              request = prepareDeleteDirRequest(
-                  remainNum, deletedDir.getValue(), deletedDir.getKey(),
-                  allSubDirList, omSnapshot.getKeyManager());
-            }
-            consumedSize += request.getSerializedSize();
-            purgePathRequestList.add(request);
-            remainNum = remainNum - request.getDeletedSubFilesCount();
-            remainNum = remainNum - request.getMarkDeletedSubDirsCount();
-            // Count up the purgeDeletedDir, subDirs and subFiles
-            if (request.getDeletedDir() != null
-                && !request.getDeletedDir().isEmpty()) {
-              dirNum++;
-            }
-            subDirNum += request.getMarkDeletedSubDirsCount();
-            subFileNum += request.getDeletedSubFilesCount();
-          } else {
-            dirsToMove.add(deletedDir.getKey());
           }
+          consumedSize += request.getSerializedSize();
+          purgePathRequestList.add(request);
+          remainNum = remainNum - request.getDeletedSubFilesCount();
+          remainNum = remainNum - request.getMarkDeletedSubDirsCount();
+          // Count up the purgeDeletedDir, subDirs and subFiles
+          if (request.getDeletedDir() != null
+              && !request.getDeletedDir().isEmpty()) {
+            dirNum++;
+          }
+          subDirNum += request.getMarkDeletedSubDirsCount();
+          subFileNum += request.getDeletedSubFilesCount();
         }
 
         remainNum = optimizeDirDeletesAndSubmitRequest(remainNum, dirNum,
@@ -457,6 +446,7 @@ private long handleDirectoryCleanUp(
 
     private void submitSnapshotPurgeRequest(List<String> purgeSnapshotKeys) {
       if (!purgeSnapshotKeys.isEmpty()) {
+        ClientId clientId = ClientId.randomId();
         SnapshotPurgeRequest snapshotPurgeRequest = SnapshotPurgeRequest
             .newBuilder()
             .addAllSnapshotDBKeys(purgeSnapshotKeys)
@@ -468,7 +458,7 @@ private void submitSnapshotPurgeRequest(List<String> purgeSnapshotKeys) {
             .setClientId(clientId.toString())
             .build();
 
-        submitRequest(omRequest);
+        submitRequest(omRequest, clientId);
       }
     }
 
@@ -492,57 +482,11 @@ private void splitRepeatedOmKeyInfo(SnapshotMoveKeyInfos.Builder toReclaim,
       }
     }
 
-    private boolean isDirReclaimable(
-        Table.KeyValue<String, OmKeyInfo> deletedDir,
-        Table<String, OmDirectoryInfo> previousDirTable,
-        Table<String, String> renamedTable,
-        List<HddsProtos.KeyValue> renamedList) throws IOException {
-
-      if (previousDirTable == null) {
-        return true;
-      }
-
-      String deletedDirDbKey = deletedDir.getKey();
-      OmKeyInfo deletedDirInfo = deletedDir.getValue();
-      String dbRenameKey = ozoneManager.getMetadataManager().getRenameKey(
-          deletedDirInfo.getVolumeName(), deletedDirInfo.getBucketName(),
-          deletedDirInfo.getObjectID());
-
-      /*
-      snapshotRenamedTable: /volumeName/bucketName/objectID ->
-          /volumeId/bucketId/parentId/dirName
-       */
-      String dbKeyBeforeRename = renamedTable.getIfExist(dbRenameKey);
-      String prevDbKey = null;
-
-      if (dbKeyBeforeRename != null) {
-        prevDbKey = dbKeyBeforeRename;
-        HddsProtos.KeyValue renamedDir = HddsProtos.KeyValue
-            .newBuilder()
-            .setKey(dbRenameKey)
-            .setValue(dbKeyBeforeRename)
-            .build();
-        renamedList.add(renamedDir);
-      } else {
-        // In OMKeyDeleteResponseWithFSO OzonePathKey is converted to
-        // OzoneDeletePathKey. Changing it back to check the previous DirTable.
-        prevDbKey = ozoneManager.getMetadataManager()
-            .getOzoneDeletePathDirKey(deletedDirDbKey);
-      }
-
-      OmDirectoryInfo prevDirectoryInfo = previousDirTable.get(prevDbKey);
-      if (prevDirectoryInfo == null) {
-        return true;
-      }
-
-      return prevDirectoryInfo.getObjectID() != deletedDirInfo.getObjectID();
-    }
-
     public void submitSnapshotMoveDeletedKeys(SnapshotInfo snapInfo,
         List<SnapshotMoveKeyInfos> toReclaimList,
         List<SnapshotMoveKeyInfos> toNextDBList,
-        List<HddsProtos.KeyValue> renamedList,
-        List<String> dirsToMove) throws InterruptedException {
+        List<HddsProtos.KeyValue> renamedList) throws InterruptedException {
+      ClientId clientId = ClientId.randomId();
 
       SnapshotMoveDeletedKeysRequest.Builder moveDeletedKeysBuilder =
           SnapshotMoveDeletedKeysRequest.newBuilder()
@@ -552,7 +496,6 @@ public void submitSnapshotMoveDeletedKeys(SnapshotInfo snapInfo,
           .addAllReclaimKeys(toReclaimList)
           .addAllNextDBKeys(toNextDBList)
           .addAllRenamedKeys(renamedList)
-          .addAllDeletedDirsToMove(dirsToMove)
           .build();
 
       OMRequest omRequest = OMRequest.newBuilder()
@@ -562,11 +505,11 @@ public void submitSnapshotMoveDeletedKeys(SnapshotInfo snapInfo,
           .build();
 
       try (BootstrapStateHandler.Lock lock = new BootstrapStateHandler.Lock()) {
-        submitRequest(omRequest);
+        submitRequest(omRequest, clientId);
       }
     }
 
-    public void submitRequest(OMRequest omRequest) {
+    public void submitRequest(OMRequest omRequest, ClientId clientId) {
       try {
         if (isRatisEnabled()) {
           OzoneManagerRatisServer server = ozoneManager.getOmRatisServer();
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/SnapshotUtils.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/SnapshotUtils.java
index ef7e4e895e7a..b4f8a7e1b8b6 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/SnapshotUtils.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/SnapshotUtils.java
@@ -32,6 +32,7 @@
 
 import java.io.File;
 import java.io.IOException;
+import java.util.NoSuchElementException;
 import java.util.UUID;
 
 import static org.apache.hadoop.ozone.om.exceptions.OMException.ResultCodes.FILE_NOT_FOUND;
@@ -135,23 +136,33 @@ public static void checkSnapshotActive(SnapshotInfo snapInfo,
   public static SnapshotInfo getNextActiveSnapshot(SnapshotInfo snapInfo,
       SnapshotChainManager chainManager, OmSnapshotManager omSnapshotManager)
       throws IOException {
-    while (chainManager.hasNextPathSnapshot(snapInfo.getSnapshotPath(),
-        snapInfo.getSnapshotId())) {
 
-      UUID nextPathSnapshot =
-          chainManager.nextPathSnapshot(
-              snapInfo.getSnapshotPath(), snapInfo.getSnapshotId());
+    // If the snapshot is deleted in the previous run, then the in-memory
+    // SnapshotChainManager might throw NoSuchElementException as the snapshot
+    // is removed in-memory but OMDoubleBuffer has not flushed yet.
+    try {
+      while (chainManager.hasNextPathSnapshot(snapInfo.getSnapshotPath(),
+          snapInfo.getSnapshotId())) {
 
-      String tableKey = chainManager.getTableKey(nextPathSnapshot);
-      SnapshotInfo nextSnapshotInfo =
-          omSnapshotManager.getSnapshotInfo(tableKey);
+        UUID nextPathSnapshot =
+            chainManager.nextPathSnapshot(
+                snapInfo.getSnapshotPath(), snapInfo.getSnapshotId());
 
-      if (nextSnapshotInfo.getSnapshotStatus().equals(
-          SnapshotInfo.SnapshotStatus.SNAPSHOT_ACTIVE)) {
-        return nextSnapshotInfo;
-      }
+        String tableKey = chainManager.getTableKey(nextPathSnapshot);
+        SnapshotInfo nextSnapshotInfo =
+            omSnapshotManager.getSnapshotInfo(tableKey);
 
-      snapInfo = nextSnapshotInfo;
+        if (nextSnapshotInfo.getSnapshotStatus().equals(
+            SnapshotInfo.SnapshotStatus.SNAPSHOT_ACTIVE)) {
+          return nextSnapshotInfo;
+        }
+
+        snapInfo = nextSnapshotInfo;
+      }
+    } catch (NoSuchElementException ex) {
+      LOG.error("The snapshot {} is not longer in snapshot chain, It " +
+          "maybe removed in the previous Snapshot purge request.",
+          snapInfo.getTableKey());
     }
     return null;
   }
diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/snapshot/TestOMSnapshotSetPropertyRequestAndResponse.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/snapshot/TestOMSnapshotSetPropertyRequestAndResponse.java
index 6ab86609dafe..87fdad8b24bb 100644
--- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/snapshot/TestOMSnapshotSetPropertyRequestAndResponse.java
+++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/snapshot/TestOMSnapshotSetPropertyRequestAndResponse.java
@@ -37,7 +37,7 @@
 import org.apache.hadoop.ozone.om.upgrade.OMLayoutVersionManager;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest;
-import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.SnapshotProperty;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.SnapshotSize;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.SetSnapshotPropertyRequest;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.BeforeEach;
@@ -52,6 +52,7 @@
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
@@ -137,6 +138,7 @@ public void testValidateAndUpdateCache() throws IOException {
             getExclusiveSize());
         assertEquals(exclusiveSizeAfterRepl, snapshotEntry.getValue()
             .getExclusiveReplicatedSize());
+        assertTrue(snapshotEntry.getValue().getExpandedDeletedDir());
       }
     }
   }
@@ -148,6 +150,7 @@ private void assertCacheValues(String dbKey) {
     assertEquals(exclusiveSize, cacheValue.getCacheValue().getExclusiveSize());
     assertEquals(exclusiveSizeAfterRepl, cacheValue.getCacheValue()
         .getExclusiveReplicatedSize());
+    assertTrue(cacheValue.getCacheValue().getExpandedDeletedDir());
   }
 
   private List<OMRequest> createSnapshotUpdateSizeRequest()
@@ -157,14 +160,15 @@ private List<OMRequest> createSnapshotUpdateSizeRequest()
              iterator = omMetadataManager.getSnapshotInfoTable().iterator()) {
       while (iterator.hasNext()) {
         String snapDbKey = iterator.next().getKey();
-        SnapshotProperty snapshotSize = SnapshotProperty.newBuilder()
-            .setSnapshotKey(snapDbKey)
+        SnapshotSize snapshotSize = SnapshotSize.newBuilder()
             .setExclusiveSize(exclusiveSize)
             .setExclusiveReplicatedSize(exclusiveSizeAfterRepl)
             .build();
         SetSnapshotPropertyRequest snapshotUpdateSizeRequest =
             SetSnapshotPropertyRequest.newBuilder()
-                .setSnapshotProperty(snapshotSize)
+                .setSnapshotKey(snapDbKey)
+                .setSnapshotSize(snapshotSize)
+                .setExpandedDeletedDir(true)
                 .build();
 
         OMRequest omRequest = OMRequest.newBuilder()
diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/service/TestKeyDeletingService.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/service/TestKeyDeletingService.java
index 22899947fd74..48c949f737a1 100644
--- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/service/TestKeyDeletingService.java
+++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/service/TestKeyDeletingService.java
@@ -592,7 +592,7 @@ public void testSnapshotExclusiveSize() throws Exception {
 
     KeyDeletingService keyDeletingService = keyManager.getDeletingService();
     // Supspend KDS
-    keyDeletingService.suspend();
+    // keyDeletingService.suspend();
 
     String volumeName = "volume1";
     String bucketName = "bucket1";