Introduce MVCC key reclaimability checks with seqNumMin & seqNumMax

Signed-off-by: peterxcli <peterxcli@gmail.com>

Author: peterxcli

hadoop-hdds/common/src/main/resources/ozone-default.xml

@@ -2429,6 +2429,19 @@
     </description>
   </property>
 
+  <property>
+    <name>ozone.om.snapshot.key.reclaim.interval.enabled</name>
+    <value>true</value>
+    <tag>OZONE, OM, MANAGEMENT, PERFORMANCE</tag>
+    <description>
+      Enables MVCC-style deleted key reclaimability checks using optional
+      seqNumMin and seqNumMax fields on key versions. Entries without complete
+      sequence interval metadata, or buckets whose active snapshot create
+      sequence numbers cannot be built exactly, fall back to the previous
+      snapshot lookup based reclaimability logic.
+    </description>
+  </property>
+
   <property>
     <name>ozone.acl.authorizer.class</name>
     <value>org.apache.hadoop.ozone.security.acl.OzoneAccessAuthorizer</value>

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/OmUtils.java

@@ -640,6 +640,9 @@ public static File createOMDir(String dirPath) {
    * create a new instance to include this key, else we update the existing
    * repeatedOmKeyInfo instance.
    * 3. Set the updateID to the transactionLogIndex.
+   * 4. Set seqNumMax to the transactionLogIndex when seqNumMin is present,
+   *    making the interval exclusive at the transaction that deleted or
+   *    overwrote this version.
    * @param keyInfo args supplied by client
    * @param bucketId bucket id
    * @param trxnLogIndex For Multipart keys, this is the transactionLogIndex
@@ -667,6 +670,9 @@ public static RepeatedOmKeyInfo prepareKeyForDelete(long bucketId, OmKeyInfo key
 
     // Set the updateID
     builder.setUpdateID(trxnLogIndex);
+    if (keyInfo.hasSeqNumMin()) {
+      builder.setSeqNumMax(trxnLogIndex);
+    }
 
     //The key doesn't exist in deletedTable, so create a new instance.
     return new RepeatedOmKeyInfo(builder.build(), bucketId);

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/OMConfigKeys.java

@@ -292,6 +292,11 @@ public final class OMConfigKeys {
       "ozone.om.fs.snapshot.max.limit";
   public static final int OZONE_OM_FS_SNAPSHOT_MAX_LIMIT_DEFAULT = 10000;
 
+  public static final String OZONE_OM_SNAPSHOT_KEY_RECLAIM_INTERVAL_ENABLED =
+      "ozone.om.snapshot.key.reclaim.interval.enabled";
+  public static final boolean
+      OZONE_OM_SNAPSHOT_KEY_RECLAIM_INTERVAL_ENABLED_DEFAULT = true;
+
   public static final String OZONE_OM_KERBEROS_KEYTAB_FILE_KEY = "ozone.om."
       + "kerberos.keytab.file";
   public static final String OZONE_OM_KERBEROS_PRINCIPAL_KEY = "ozone.om"

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmKeyInfo.java

@@ -111,6 +111,8 @@ public final class OmKeyInfo extends WithParentObjectId
   // been modified.
   private Long expectedDataGeneration = null;
   private String expectedETag;
+  private Long seqNumMin;
+  private Long seqNumMax;
 
   private OmKeyInfo(Builder b) {
     super(b);
@@ -131,6 +133,8 @@ private OmKeyInfo(Builder b) {
     this.tags = b.tags.build();
     this.expectedDataGeneration = b.expectedDataGeneration;
     this.expectedETag = b.expectedETag;
+    this.seqNumMin = b.seqNumMin;
+    this.seqNumMax = b.seqNumMax;
   }
 
   private static Codec<OmKeyInfo> newCodec(boolean ignorePipeline) {
@@ -199,6 +203,30 @@ public String getExpectedETag() {
     return expectedETag;
   }
 
+  public void setSeqNumMin(Long seqNumMin) {
+    this.seqNumMin = seqNumMin;
+  }
+
+  public Long getSeqNumMin() {
+    return seqNumMin;
+  }
+
+  public boolean hasSeqNumMin() {
+    return seqNumMin != null;
+  }
+
+  public void setSeqNumMax(Long seqNumMax) {
+    this.seqNumMax = seqNumMax;
+  }
+
+  public Long getSeqNumMax() {
+    return seqNumMax;
+  }
+
+  public boolean hasSeqNumMax() {
+    return seqNumMax != null;
+  }
+
   public String getOwnerName() {
     return ownerName;
   }
@@ -475,6 +503,8 @@ public String toString() {
         ", fileChecksum=" + fileChecksum +
         ", isFile=" + isFile +
         ", fileName='" + fileName + '\'' +
+        ", seqNumMin=" + seqNumMin +
+        ", seqNumMax=" + seqNumMax +
         ", acls=" + acls +
         '}';
   }
@@ -503,6 +533,8 @@ public static class Builder extends WithParentObjectId.Builder<OmKeyInfo> {
     private final MapBuilder<String, String> tags;
     private Long expectedDataGeneration = null;
     private String expectedETag;
+    private Long seqNumMin;
+    private Long seqNumMax;
 
     public Builder() {
       this.acls = AclListBuilder.empty();
@@ -526,6 +558,8 @@ public Builder(OmKeyInfo obj) {
       this.isFile = obj.isFile;
       this.expectedDataGeneration = obj.expectedDataGeneration;
       this.expectedETag = obj.expectedETag;
+      this.seqNumMin = obj.seqNumMin;
+      this.seqNumMax = obj.seqNumMax;
       this.tags = MapBuilder.of(obj.tags);
       obj.keyLocationVersions.forEach(keyLocationVersion ->
           this.omKeyLocationInfoGroups.add(
@@ -702,6 +736,16 @@ public Builder setExpectedETag(String eTag) {
       return this;
     }
 
+    public Builder setSeqNumMin(Long seqNum) {
+      this.seqNumMin = seqNum;
+      return this;
+    }
+
+    public Builder setSeqNumMax(Long seqNum) {
+      this.seqNumMax = seqNum;
+      return this;
+    }
+
     @Override
     protected void validate() {
       super.validate();
@@ -824,6 +868,12 @@ private KeyInfo getProtobuf(boolean ignorePipeline, String fullKeyName,
     if (expectedETag != null) {
       kb.setExpectedETag(expectedETag);
     }
+    if (seqNumMin != null) {
+      kb.setSeqNumMin(seqNumMin);
+    }
+    if (seqNumMax != null) {
+      kb.setSeqNumMax(seqNumMax);
+    }
     if (ownerName != null) {
       kb.setOwnerName(ownerName);
     }
@@ -880,6 +930,12 @@ public static Builder builderFromProtobuf(KeyInfo keyInfo) {
     if (keyInfo.hasExpectedETag()) {
       builder.setExpectedETag(keyInfo.getExpectedETag());
     }
+    if (keyInfo.hasSeqNumMin()) {
+      builder.setSeqNumMin(keyInfo.getSeqNumMin());
+    }
+    if (keyInfo.hasSeqNumMax()) {
+      builder.setSeqNumMax(keyInfo.getSeqNumMax());
+    }
 
     if (keyInfo.hasOwnerName()) {
       builder.setOwnerName(keyInfo.getOwnerName());
@@ -903,6 +959,8 @@ public String getObjectInfo() {
         ", creationTime='" + creationTime + '\'' +
         ", objectID='" + getObjectID() + '\'' +
         ", parentID='" + getParentObjectID() + '\'' +
+        ", seqNumMin='" + seqNumMin + '\'' +
+        ", seqNumMax='" + seqNumMax + '\'' +
         ", replication='" + replicationConfig + '\'' +
         ", fileChecksum='" + fileChecksum +
         '}';
@@ -921,6 +979,8 @@ public boolean isKeyInfoSame(OmKeyInfo omKeyInfo, boolean checkPath,
         Objects.equals(getMetadata(), omKeyInfo.getMetadata()) &&
         Objects.equals(acls, omKeyInfo.acls) &&
         Objects.equals(getTags(), omKeyInfo.getTags()) &&
+        Objects.equals(seqNumMin, omKeyInfo.seqNumMin) &&
+        Objects.equals(seqNumMax, omKeyInfo.seqNumMax) &&
         getObjectID() == omKeyInfo.getObjectID();
 
     if (isEqual && checkUpdateID) {

hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/helpers/TestOmKeyInfo.java

@@ -25,6 +25,7 @@
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import java.io.IOException;
@@ -41,6 +42,7 @@
 import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
 import org.apache.hadoop.hdds.scm.pipeline.PipelineID;
 import org.apache.hadoop.ozone.ClientVersion;
+import org.apache.hadoop.ozone.OmUtils;
 import org.apache.hadoop.ozone.OzoneAcl;
 import org.apache.hadoop.ozone.OzoneConsts;
 import org.apache.hadoop.ozone.om.helpers.OmKeyInfo.Builder;
@@ -73,6 +75,62 @@ public void protobufConversion() throws IOException {
     assertEquals(5678L, key.getExpectedDataGeneration());
   }
 
+  @Test
+  public void protobufConversionPreservesSeqNumIntervalPresence()
+      throws IOException {
+    OmKeyInfo key = createOmKeyInfo(
+        RatisReplicationConfig.getInstance(ReplicationFactor.THREE));
+    OzoneManagerProtocolProtos.KeyInfo proto =
+        key.getProtobuf(ClientVersion.CURRENT_VERSION);
+    assertFalse(proto.hasSeqNumMin());
+    assertFalse(proto.hasSeqNumMax());
+
+    OmKeyInfo recovered = OmKeyInfo.getFromProtobuf(proto);
+    assertNull(recovered.getSeqNumMin());
+    assertNull(recovered.getSeqNumMax());
+    assertFalse(recovered.hasSeqNumMin());
+    assertFalse(recovered.hasSeqNumMax());
+
+    key = key.toBuilder()
+        .setSeqNumMin(1234L)
+        .setSeqNumMax(5678L)
+        .build();
+    proto = key.getProtobuf(ClientVersion.CURRENT_VERSION);
+    assertTrue(proto.hasSeqNumMin());
+    assertTrue(proto.hasSeqNumMax());
+    assertEquals(1234L, proto.getSeqNumMin());
+    assertEquals(5678L, proto.getSeqNumMax());
+
+    recovered = OmKeyInfo.getFromProtobuf(proto);
+    assertEquals(1234L, recovered.getSeqNumMin());
+    assertEquals(5678L, recovered.getSeqNumMax());
+    assertTrue(recovered.hasSeqNumMin());
+    assertTrue(recovered.hasSeqNumMax());
+    assertEquals(key, recovered);
+  }
+
+  @Test
+  public void prepareKeyForDeletePreservesSeqNumMinAndSetsSeqNumMax() {
+    OmKeyInfo key = createOmKeyInfo(
+        RatisReplicationConfig.getInstance(ReplicationFactor.THREE))
+        .toBuilder()
+        .setSeqNumMin(11L)
+        .build();
+
+    RepeatedOmKeyInfo deletedKeyInfo =
+        OmUtils.prepareKeyForDelete(1L, key, 22L);
+    OmKeyInfo deletedKey = deletedKeyInfo.getOmKeyInfoList().get(0);
+    assertEquals(11L, deletedKey.getSeqNumMin());
+    assertEquals(22L, deletedKey.getSeqNumMax());
+
+    key = createOmKeyInfo(
+        RatisReplicationConfig.getInstance(ReplicationFactor.THREE));
+    deletedKeyInfo = OmUtils.prepareKeyForDelete(1L, key, 22L);
+    deletedKey = deletedKeyInfo.getOmKeyInfoList().get(0);
+    assertNull(deletedKey.getSeqNumMin());
+    assertNull(deletedKey.getSeqNumMax());
+  }
+
   @Test
   public void getProtobufMessageEC() throws IOException {
     OmKeyInfo key = createOmKeyInfo(

hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto

@@ -1197,6 +1197,12 @@ message KeyInfo {
     // the given ETag for the operation to succeed. This is used for
     // S3 conditional writes with the If-Match header.
     optional string expectedETag = 23;
+
+    // MVCC-style visibility interval for snapshot key reclamation.
+    // A key version is visible to a snapshot when:
+    // seqNumMin <= snapshotCreateSeqNum < seqNumMax.
+    optional uint64 seqNumMin = 24;
+    optional uint64 seqNumMax = 25;
 }
 
 // KeyInfoProtoLight is a lightweight subset of KeyInfo message containing

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

@@ -109,6 +109,10 @@ public final class DeletingServiceMetrics {
   private MutableGaugeLong snapKeysIteratedLast;
   @Metric("Snapshot: No. of not reclaimable keys the last run")
   private MutableGaugeLong snapKeysNotReclaimableLast;
+  @Metric("Snapshot: No. of key reclaimability decisions optimized by seqNum intervals")
+  private MutableGaugeLong snapKeyReclaimIntervalOptimized;
+  @Metric("Snapshot: No. of key reclaimability decisions falling back from seqNum intervals")
+  private MutableGaugeLong snapKeyReclaimIntervalFallback;
 
   /**
    * Metric to track the term ID of the last key that was purged from the
@@ -299,6 +303,22 @@ public long getSnapKeysNotReclaimableLast() {
     return snapKeysNotReclaimableLast.value();
   }
 
+  public void incrSnapKeyReclaimIntervalOptimized() {
+    this.snapKeyReclaimIntervalOptimized.incr(1L);
+  }
+
+  public long getSnapKeyReclaimIntervalOptimized() {
+    return snapKeyReclaimIntervalOptimized.value();
+  }
+
+  public void incrSnapKeyReclaimIntervalFallback() {
+    this.snapKeyReclaimIntervalFallback.incr(1L);
+  }
+
+  public long getSnapKeyReclaimIntervalFallback() {
+    return snapKeyReclaimIntervalFallback.value();
+  }
+
   public synchronized TransactionInfo getLastAOSTransactionInfo() {
     return TransactionInfo.valueOf(lastAOSPurgeTermId.value(), lastAOSPurgeTransactionId.value());
   }

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyCommitRequest.java

@@ -313,6 +313,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
           .addAllMetadata(KeyValueUtil.getFromProtobuf(
                 commitKeyArgs.getMetadataList()))
           .setUpdateID(trxnLogIndex)
+          .setSeqNumMin(trxnLogIndex)
           .setDataSize(commitKeyArgs.getDataSize())
           .build();
 

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyCommitRequestWithFSO.java

@@ -231,6 +231,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
               commitKeyArgs.getMetadataList()))
           .setDataSize(commitKeyArgs.getDataSize())
           .setUpdateID(trxnLogIndex)
+          .setSeqNumMin(trxnLogIndex)
           .build();
 
       List<OmKeyLocationInfo> uncommitted =
@@ -318,8 +319,9 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
         if (null == oldKeyVersionsToDeleteMap) {
           oldKeyVersionsToDeleteMap = new HashMap<>();
         }
+        OmKeyInfo keyInfoToDelete = prepareKeyInfoForDeleteMap(trxnLogIndex, pseudoKeyInfo);
         oldKeyVersionsToDeleteMap.computeIfAbsent(delKeyName,
-            key -> new RepeatedOmKeyInfo(omBucketInfo.getObjectID())).addOmKeyInfo(pseudoKeyInfo);
+            key -> new RepeatedOmKeyInfo(omBucketInfo.getObjectID())).addOmKeyInfo(keyInfoToDelete);
       }
 
       // Add to cache of open key table and key table.

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyRequest.java

@@ -1185,6 +1185,7 @@ protected static Map<String, RepeatedOmKeyInfo> addKeyInfoToDeleteMap(OzoneManag
     if (keyInfo == null) {
       return deleteMap;
     }
+    keyInfo = prepareKeyInfoForDeleteMap(trxnLogIndex, keyInfo);
     final long pseudoObjId = om.getObjectIdFromTxId(trxnLogIndex);
     final String delKeyName = om.getMetadataManager().getOzoneDeletePathKey(pseudoObjId, ozoneKey);
     if (deleteMap == null) {
@@ -1195,6 +1196,17 @@ protected static Map<String, RepeatedOmKeyInfo> addKeyInfoToDeleteMap(OzoneManag
     return deleteMap;
   }
 
+  protected static OmKeyInfo prepareKeyInfoForDeleteMap(long trxnLogIndex, OmKeyInfo keyInfo) {
+    if (keyInfo.getObjectID() == OBJECT_ID_RECLAIM_BLOCKS) {
+      return keyInfo.toBuilder()
+          .setUpdateID(trxnLogIndex)
+          .setSeqNumMin(trxnLogIndex)
+          .setSeqNumMax(trxnLogIndex)
+          .build();
+    }
+    return keyInfo;
+  }
+
   /**
    * Remove blocks in-place from keysToBeFiltered that exist in referenceKey.
    * <p>