HDDS-13859. OmSnapshotLocalDataManager should handle needsDefrag flag in the yaml file

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/SnapshotDefragService.java

@@ -134,12 +134,9 @@ private boolean needsDefragmentation(SnapshotInfo snapshotInfo) {
     try (OmSnapshotLocalDataManager.ReadableOmSnapshotLocalDataProvider readableOmSnapshotLocalDataProvider =
              ozoneManager.getOmSnapshotManager().getSnapshotLocalDataManager().getOmSnapshotLocalData(snapshotInfo)) {
       // Read snapshot local metadata from YAML
-      OmSnapshotLocalData snapshotLocalData = readableOmSnapshotLocalDataProvider.getSnapshotLocalData();
-
       // Check if snapshot needs compaction (defragmentation)
-      boolean needsDefrag = snapshotLocalData.getNeedsDefrag();
-      LOG.debug("Snapshot {} needsDefragmentation field value: {}",
-          snapshotInfo.getName(), needsDefrag);
+      boolean needsDefrag = readableOmSnapshotLocalDataProvider.needsDefrag();
+      LOG.debug("Snapshot {} needsDefragmentation field value: {}", snapshotInfo.getName(), needsDefrag);
 
       return needsDefrag;
     } catch (IOException e) {

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/OmSnapshotLocalDataManager.java

@@ -575,6 +575,19 @@ private LockDataProviderInitResult initialize(
       }
     }
 
+    public boolean needsDefrag() {
+      if (snapshotLocalData.getNeedsDefrag()) {
+        return true;
+      }
+      if (resolvedPreviousSnapshotId != null) {
+        int snapshotVersion = snapshotLocalData.getVersion();
+        int previousResolvedSnapshotVersion = snapshotLocalData.getVersionSstFileInfos().get(snapshotVersion)
+            .getPreviousSnapshotVersion();
+        return previousResolvedSnapshotVersion < getVersionNodeMap().get(resolvedPreviousSnapshotId).getVersion();
+      }
+      return false;
+    }
+
     @Override
     public void close() throws IOException {
       if (previousLock != null) {
@@ -642,6 +655,9 @@ private SnapshotVersionsMeta validateModification(OmSnapshotLocalData snapshotLo
         if (existingVersionsMeta == null || !Objects.equals(versionsToBeAdded.getPreviousSnapshotId(),
             existingVersionsMeta.getPreviousSnapshotId())) {
           setDirty();
+          // Set the needsDefrag if the new previous snapshotId is different from the existing one or if this is a new
+          // snapshot yaml file.
+          snapshotLocalData.setNeedsDefrag(true);
         }
         return versionsToBeAdded;
       } finally {
@@ -654,6 +670,8 @@ public void addSnapshotVersion(RDBStore snapshotStore) throws IOException {
       OmSnapshotLocalData previousSnapshotLocalData = getPreviousSnapshotLocalData();
       this.getSnapshotLocalData().addVersionSSTFileInfos(sstFiles, previousSnapshotLocalData == null ? 0 :
           previousSnapshotLocalData.getVersion());
+      // Adding a new snapshot version means it has been defragged thus the flag needs to be reset.
+      this.getSnapshotLocalData().setNeedsDefrag(false);
       // Set Dirty if a version is added.
       setDirty();
     }

hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/snapshot/TestOmSnapshotLocalDataManager.java

@@ -435,6 +435,32 @@ private void validateVersions(OmSnapshotLocalDataManager snapshotLocalDataManage
     }
   }
 
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
+  public void testWriteWithChainUpdate(boolean previousSnapshotExisting) throws IOException {
+    localDataManager = new OmSnapshotLocalDataManager(omMetadataManager);
+    List<UUID> snapshotIds = createSnapshotLocalData(localDataManager, 3 + (previousSnapshotExisting ? 1 : 0));
+    int snapshotIdx = 1 + (previousSnapshotExisting ? 1 : 0);
+    for (UUID snapshotId : snapshotIds) {
+      addVersionsToLocalData(localDataManager, snapshotId, ImmutableMap.of(1, 1));
+    }
+
+    UUID snapshotId = snapshotIds.get(snapshotIdx);
+    UUID toUpdatePreviousSnapshotId = snapshotIdx - 2 >= 0 ? snapshotIds.get(snapshotIdx - 2) : null;
+
+    try (WritableOmSnapshotLocalDataProvider snap =
+             localDataManager.getWritableOmSnapshotLocalData(snapshotId, toUpdatePreviousSnapshotId)) {
+      assertFalse(snap.needsDefrag());
+      snap.commit();
+      assertTrue(snap.needsDefrag());
+    }
+    try (ReadableOmSnapshotLocalDataProvider snap =
+             localDataManager.getOmSnapshotLocalData(snapshotId)) {
+      assertEquals(toUpdatePreviousSnapshotId, snap.getSnapshotLocalData().getPreviousSnapshotId());
+      assertTrue(snap.needsDefrag());
+    }
+  }
+
   /**
    * Validates write-time version propagation and removal rules when the previous
    * snapshot already has a concrete version recorded.
@@ -535,6 +561,26 @@ private void addVersionsToLocalData(OmSnapshotLocalDataManager snapshotLocalData
     }
   }
 
+  @ParameterizedTest
+  @ValueSource(ints = {1, 2, 3})
+  public void testNeedsDefrag(int previousVersion) throws IOException {
+    localDataManager = new OmSnapshotLocalDataManager(omMetadataManager);
+    List<UUID> snapshotIds = createSnapshotLocalData(localDataManager, 2);
+    for (UUID snapshotId : snapshotIds) {
+      try (ReadableOmSnapshotLocalDataProvider snap = localDataManager.getOmSnapshotLocalData(snapshotId)) {
+        assertTrue(snap.needsDefrag());
+      }
+    }
+    addVersionsToLocalData(localDataManager, snapshotIds.get(0), ImmutableMap.of(1, 1, 2, 2, 3, 3));
+    try (ReadableOmSnapshotLocalDataProvider snap = localDataManager.getOmSnapshotLocalData(snapshotIds.get(0))) {
+      assertFalse(snap.needsDefrag());
+    }
+    addVersionsToLocalData(localDataManager, snapshotIds.get(1), ImmutableMap.of(1, 3, 2, previousVersion));
+    try (ReadableOmSnapshotLocalDataProvider snap = localDataManager.getOmSnapshotLocalData(snapshotIds.get(1))) {
+      assertEquals(previousVersion < snap.getPreviousSnapshotLocalData().getVersion(), snap.needsDefrag());
+    }
+  }
+
   @ParameterizedTest
   @ValueSource(booleans = {true, false})
   public void testVersionResolution(boolean read) throws IOException {
@@ -645,7 +691,7 @@ public void testCreateNewSnapshotLocalYaml() throws IOException {
     assertEquals(notDefraggedVersionMeta, localData.getVersionSstFileInfos().get(0));
     assertFalse(localData.getSstFiltered());
     assertEquals(0L, localData.getLastDefragTime());
-    assertFalse(localData.getNeedsDefrag());
+    assertTrue(localData.getNeedsDefrag());
     assertEquals(1, localData.getVersionSstFileInfos().size());
   }
 
@@ -681,6 +727,8 @@ public void testCreateNewOmSnapshotLocalDataFile() throws IOException {
       OmSnapshotLocalData.VersionMeta expectedVersionMeta =
           new OmSnapshotLocalData.VersionMeta(0, sstFileInfos);
       assertEquals(expectedVersionMeta, versionMeta);
+      // New Snapshot create needs to be defragged always.
+      assertTrue(snapshotLocalData.needsDefrag());
     }
   }