HDDS-13026. KeyDeletingService should also delete RenameEntries

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/snapshot/TestSnapshotDeletingServiceIntegrationTest.java

@@ -620,9 +620,9 @@ public void testParallelExcecutionOfKeyDeletionAndSnapshotDeletion() throws Exce
              om.getOmSnapshotManager().getSnapshot(testBucket.getVolumeName(), testBucket.getName(),
                  testBucket.getName() + "snap2")) {
       renamesKeyEntries = snapshot.get().getKeyManager().getRenamesKeyEntries(testBucket.getVolumeName(),
-          testBucket.getName(), "", 1000);
+          testBucket.getName(), "", (kv) -> true, 1000);
       deletedKeyEntries = snapshot.get().getKeyManager().getDeletedKeyEntries(testBucket.getVolumeName(),
-          testBucket.getName(), "", 1000);
+          testBucket.getName(), "", (kv) -> true, 1000);
       deletedDirEntries = snapshot.get().getKeyManager().getDeletedDirEntries(testBucket.getVolumeName(),
           testBucket.getName(), 1000);
     }
@@ -658,20 +658,20 @@ public void testParallelExcecutionOfKeyDeletionAndSnapshotDeletion() throws Exce
                  testBucket.getName() + "snap2")) {
       Assertions.assertEquals(Collections.emptyList(),
           snapshot.get().getKeyManager().getRenamesKeyEntries(testBucket.getVolumeName(),
-          testBucket.getName(), "", 1000));
+          testBucket.getName(), "", (kv) -> true, 1000));
       Assertions.assertEquals(Collections.emptyList(),
           snapshot.get().getKeyManager().getDeletedKeyEntries(testBucket.getVolumeName(),
-          testBucket.getName(), "", 1000));
+          testBucket.getName(), "", (kv) -> true, 1000));
       Assertions.assertEquals(Collections.emptyList(),
           snapshot.get().getKeyManager().getDeletedDirEntries(testBucket.getVolumeName(),
           testBucket.getName(), 1000));
     }
     List<Table.KeyValue<String, String>> aosRenamesKeyEntries =
         om.getKeyManager().getRenamesKeyEntries(testBucket.getVolumeName(),
-            testBucket.getName(), "", 1000);
+            testBucket.getName(), "", (kv) -> true, 1000);
     List<Table.KeyValue<String, List<OmKeyInfo>>> aosDeletedKeyEntries =
         om.getKeyManager().getDeletedKeyEntries(testBucket.getVolumeName(),
-            testBucket.getName(), "", 1000);
+            testBucket.getName(), "", (kv) -> true, 1000);
     List<Table.KeyValue<String, OmKeyInfo>> aosDeletedDirEntries =
         om.getKeyManager().getDeletedDirEntries(testBucket.getVolumeName(),
             testBucket.getName(), 1000);

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

@@ -1392,6 +1392,7 @@ message PurgeKeysRequest {
     repeated SnapshotMoveKeyInfos keysToUpdate = 3;
     // previous snapshotID can also be null & this field would be absent in older requests.
     optional NullableUUID expectedPreviousSnapshotID = 4;
+    repeated string renamedKeys = 5;
 }
 
 message PurgeKeysResponse {

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

@@ -65,6 +65,8 @@ public final class DeletingServiceMetrics {
    */
   @Metric("Total no. of keys purged")
   private MutableGaugeLong numKeysPurged;
+  @Metric("Total no. of rename entries purged")
+  private MutableGaugeLong numRenameEntriesPurged;
 
   private DeletingServiceMetrics() {
     this.registry = new MetricsRegistry(METRICS_SOURCE_NAME);
@@ -154,6 +156,10 @@ public void incrNumKeysPurged(long keysPurged) {
     this.numKeysPurged.incr(keysPurged);
   }
 
+  public void incrNumRenameEntriesPurged(long renameEntriesPurged) {
+    this.numRenameEntriesPurged.incr(renameEntriesPurged);
+  }
+
   @VisibleForTesting
   public void resetDirectoryMetrics() {
     numDirsPurged.set(0);

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

@@ -36,6 +36,7 @@
 import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
 import org.apache.hadoop.ozone.om.helpers.OmMultipartUploadList;
 import org.apache.hadoop.ozone.om.helpers.OmMultipartUploadListParts;
+import org.apache.hadoop.ozone.om.helpers.RepeatedOmKeyInfo;
 import org.apache.hadoop.ozone.om.service.CompactionService;
 import org.apache.hadoop.ozone.om.service.DirectoryDeletingService;
 import org.apache.hadoop.ozone.om.service.KeyDeletingService;
@@ -148,13 +149,16 @@ PendingKeysDeletion getPendingDeletionKeys(
   /**
    * Returns a list rename entries from the snapshotRenamedTable.
    *
-   * @param size max number of keys to return.
+   * @param count max number of keys to return.
+   * @param filter filter to apply on the entries.
    * @return a Pair of list of {@link org.apache.hadoop.hdds.utils.db.Table.KeyValue} representing the keys in the
    * underlying metadataManager.
    * @throws IOException
    */
   List<Table.KeyValue<String, String>> getRenamesKeyEntries(
-      String volume, String bucket, String startKey, int size) throws IOException;
+      String volume, String bucket, String startKey,
+      CheckedFunction<Table.KeyValue<String, String>, Boolean, IOException> filter, int count)
+      throws IOException;
 
 
   /**
@@ -178,13 +182,16 @@ CheckedFunction<KeyManager, OmKeyInfo, IOException> getPreviousSnapshotOzoneKeyI
   /**
    * Returns a list deleted entries from the deletedTable.
    *
-   * @param size max number of keys to return.
+   * @param count max number of keys to return.
+   * @param filter filter to apply on the entries.
    * @return a Pair of list of {@link org.apache.hadoop.hdds.utils.db.Table.KeyValue} representing the keys in the
    * underlying metadataManager.
    * @throws IOException
    */
   List<Table.KeyValue<String, List<OmKeyInfo>>> getDeletedKeyEntries(
-      String volume, String bucket, String startKey, int size) throws IOException;
+      String volume, String bucket, String startKey,
+      CheckedFunction<Table.KeyValue<String, RepeatedOmKeyInfo>, Boolean, IOException> filter,
+      int count) throws IOException;
 
   /**
    * Returns the names of up to {@code count} open keys whose age is

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

@@ -797,7 +797,9 @@ public PendingKeysDeletion getPendingDeletionKeys(
 
   private <V, R> List<Table.KeyValue<String, R>> getTableEntries(String startKey,
           TableIterator<String, ? extends Table.KeyValue<String, V>> tableIterator,
-          Function<V, R> valueFunction, int size) throws IOException {
+          Function<V, R> valueFunction,
+          CheckedFunction<Table.KeyValue<String, V>, Boolean, IOException> filter,
+          int size) throws IOException {
     List<Table.KeyValue<String, R>> entries = new ArrayList<>();
     /* Seek to the start key if it's not null. The next key in queue is ensured to start with the bucket
          prefix, {@link org.apache.hadoop.hdds.utils.db.Table#iterator(bucketPrefix)} would ensure this.
@@ -810,7 +812,7 @@ private <V, R> List<Table.KeyValue<String, R>> getTableEntries(String startKey,
     int currentCount = 0;
     while (tableIterator.hasNext() && currentCount < size) {
       Table.KeyValue<String, V> kv = tableIterator.next();
-      if (kv != null) {
+      if (kv != null && filter.apply(kv)) {
         entries.add(Table.newKeyValue(kv.getKey(), valueFunction.apply(kv.getValue())));
         currentCount++;
       }
@@ -832,11 +834,12 @@ private Optional<String> getBucketPrefix(String volumeName, String bucketName, b
 
   @Override
   public List<Table.KeyValue<String, String>> getRenamesKeyEntries(
-      String volume, String bucket, String startKey, int size) throws IOException {
+      String volume, String bucket, String startKey,
+      CheckedFunction<Table.KeyValue<String, String>, Boolean, IOException> filter, int size) throws IOException {
     Optional<String> bucketPrefix = getBucketPrefix(volume, bucket, false);
     try (TableIterator<String, ? extends Table.KeyValue<String, String>>
              renamedKeyIter = metadataManager.getSnapshotRenamedTable().iterator(bucketPrefix.orElse(""))) {
-      return getTableEntries(startKey, renamedKeyIter, Function.identity(), size);
+      return getTableEntries(startKey, renamedKeyIter, Function.identity(), filter, size);
     }
   }
 
@@ -880,11 +883,13 @@ private <T> CheckedFunction<KeyManager, T, IOException> getPreviousSnapshotOzone
 
   @Override
   public List<Table.KeyValue<String, List<OmKeyInfo>>> getDeletedKeyEntries(
-      String volume, String bucket, String startKey, int size) throws IOException {
+      String volume, String bucket, String startKey,
+      CheckedFunction<Table.KeyValue<String, RepeatedOmKeyInfo>, Boolean, IOException> filter,
+      int size) throws IOException {
     Optional<String> bucketPrefix = getBucketPrefix(volume, bucket, false);
     try (TableIterator<String, ? extends Table.KeyValue<String, RepeatedOmKeyInfo>>
              delKeyIter = metadataManager.getDeletedTable().iterator(bucketPrefix.orElse(""))) {
-      return getTableEntries(startKey, delKeyIter, RepeatedOmKeyInfo::cloneOmKeyInfoList, size);
+      return getTableEntries(startKey, delKeyIter, RepeatedOmKeyInfo::cloneOmKeyInfoList, filter, size);
     }
   }
 

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

@@ -91,15 +91,17 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
     List<String> keysToBePurgedList = new ArrayList<>();
 
     int numKeysDeleted = 0;
+    List<String> renamedKeysToBePurged = new ArrayList<>(purgeKeysRequest.getRenamedKeysList());
     for (DeletedKeys bucketWithDeleteKeys : bucketDeletedKeysList) {
       List<String> keysList = bucketWithDeleteKeys.getKeysList();
       keysToBePurgedList.addAll(keysList);
       numKeysDeleted = numKeysDeleted + keysList.size();
     }
     DeletingServiceMetrics deletingServiceMetrics = ozoneManager.getDeletionMetrics();
     deletingServiceMetrics.incrNumKeysPurged(numKeysDeleted);
+    deletingServiceMetrics.incrNumRenameEntriesPurged(renamedKeysToBePurged.size());
 
-    if (keysToBePurgedList.isEmpty()) {
+    if (keysToBePurgedList.isEmpty() && renamedKeysToBePurged.isEmpty()) {
       return new OMKeyPurgeResponse(createErrorOMResponse(omResponse,
           new OMException("None of the keys can be purged be purged since a new snapshot was created for all the " +
               "buckets, making this request invalid", OMException.ResultCodes.KEY_DELETION_ERROR)));
@@ -118,7 +120,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
     }
 
     return new OMKeyPurgeResponse(omResponse.build(),
-        keysToBePurgedList, fromSnapshotInfo, keysToUpdateList);
+        keysToBePurgedList, renamedKeysToBePurged, fromSnapshotInfo, keysToUpdateList);
   }
 
 }

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/key/OMKeyPurgeResponse.java

@@ -45,15 +45,18 @@
 @CleanupTableInfo(cleanupTables = {DELETED_TABLE, SNAPSHOT_INFO_TABLE})
 public class OMKeyPurgeResponse extends OmKeyResponse {
   private List<String> purgeKeyList;
+  private List<String> renamedList;
   private SnapshotInfo fromSnapshot;
   private List<SnapshotMoveKeyInfos> keysToUpdateList;
 
   public OMKeyPurgeResponse(@Nonnull OMResponse omResponse,
       @Nonnull List<String> keyList,
+      @Nonnull List<String> renamedList,
       SnapshotInfo fromSnapshot,
       List<SnapshotMoveKeyInfos> keysToUpdate) {
     super(omResponse);
     this.purgeKeyList = keyList;
+    this.renamedList = renamedList;
     this.fromSnapshot = fromSnapshot;
     this.keysToUpdateList = keysToUpdate;
   }
@@ -103,19 +106,21 @@ private void processKeysToUpdate(BatchOperation batchOp,
 
     for (SnapshotMoveKeyInfos keyToUpdate : keysToUpdateList) {
       List<KeyInfo> keyInfosList = keyToUpdate.getKeyInfosList();
-      RepeatedOmKeyInfo repeatedOmKeyInfo =
-          createRepeatedOmKeyInfo(keyInfosList);
+      RepeatedOmKeyInfo repeatedOmKeyInfo = createRepeatedOmKeyInfo(keyInfosList);
       metadataManager.getDeletedTable().putWithBatch(batchOp,
           keyToUpdate.getKey(), repeatedOmKeyInfo);
     }
   }
 
-  private void processKeys(BatchOperation batchOp,
-      OMMetadataManager metadataManager) throws IOException {
+  private void processKeys(BatchOperation batchOp, OMMetadataManager metadataManager) throws IOException {
     for (String key : purgeKeyList) {
       metadataManager.getDeletedTable().deleteWithBatch(batchOp,
           key);
     }
+    // Delete rename entries.
+    for (String key : renamedList) {
+      metadataManager.getSnapshotRenamedTable().deleteWithBatch(batchOp, key);
+    }
   }
 
 }

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

@@ -102,7 +102,7 @@ public AbstractKeyDeletingService(String serviceName, long interval,
   }
 
   protected Pair<Integer, Boolean> processKeyDeletes(List<BlockGroup> keyBlocksList,
-      Map<String, RepeatedOmKeyInfo> keysToModify,
+      Map<String, RepeatedOmKeyInfo> keysToModify, List<String> renameEntries,
       String snapTableKey, UUID expectedPreviousSnapshotId) throws IOException, InterruptedException {
 
     long startTime = Time.monotonicNow();
@@ -125,7 +125,7 @@ protected Pair<Integer, Boolean> processKeyDeletes(List<BlockGroup> keyBlocksLis
     if (blockDeletionResults != null) {
       long purgeStartTime = Time.monotonicNow();
       purgeResult = submitPurgeKeysRequest(blockDeletionResults,
-          keysToModify, snapTableKey, expectedPreviousSnapshotId);
+          keysToModify, renameEntries, snapTableKey, expectedPreviousSnapshotId);
       int limit = ozoneManager.getConfiguration().getInt(OMConfigKeys.OZONE_KEY_DELETING_LIMIT_PER_TASK,
           OMConfigKeys.OZONE_KEY_DELETING_LIMIT_PER_TASK_DEFAULT);
       LOG.info("Blocks for {} (out of {}) keys are deleted from DB in {} ms. Limit per task is {}.",
@@ -142,8 +142,8 @@ protected Pair<Integer, Boolean> processKeyDeletes(List<BlockGroup> keyBlocksLis
    * @param keysToModify Updated list of RepeatedOmKeyInfo
    */
   private Pair<Integer, Boolean> submitPurgeKeysRequest(List<DeleteBlockGroupResult> results,
-      Map<String, RepeatedOmKeyInfo> keysToModify, String snapTableKey, UUID expectedPreviousSnapshotId)
-      throws InterruptedException {
+      Map<String, RepeatedOmKeyInfo> keysToModify,  List<String> renameEntriesToBeDeleted,
+      String snapTableKey, UUID expectedPreviousSnapshotId) throws InterruptedException {
     List<String> purgeKeys = new ArrayList<>();
 
     // Put all keys to be purged in a list
@@ -191,7 +191,10 @@ private Pair<Integer, Boolean> submitPurgeKeysRequest(List<DeleteBlockGroupResul
         .addAllKeys(purgeKeys)
         .build();
     purgeKeysRequest.addDeletedKeys(deletedKeys);
-
+    // Adding rename entries to be purged.
+    if (renameEntriesToBeDeleted != null) {
+      purgeKeysRequest.addAllRenamedKeys(renameEntriesToBeDeleted);
+    }
     List<SnapshotMoveKeyInfos> keysToUpdateList = new ArrayList<>();
     if (keysToModify != null) {
       for (Map.Entry<String, RepeatedOmKeyInfo> keyToModify :

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

@@ -55,6 +55,7 @@
 import org.apache.hadoop.ozone.om.lock.IOzoneManagerLock;
 import org.apache.hadoop.ozone.om.snapshot.SnapshotUtils;
 import org.apache.hadoop.ozone.om.snapshot.filter.ReclaimableKeyFilter;
+import org.apache.hadoop.ozone.om.snapshot.filter.ReclaimableRenameEntryFilter;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.SetSnapshotPropertyRequest;
 import org.apache.ratis.util.function.UncheckedAutoCloseableSupplier;
@@ -230,20 +231,33 @@ private void processDeletedKeysForStore(SnapshotInfo currentSnapshotInfo, KeyMan
         // Purge deleted Keys in the deletedTable && rename entries in the snapshotRenamedTable which doesn't have a
         // reference in the previous snapshot.
         try (ReclaimableKeyFilter reclaimableKeyFilter = new ReclaimableKeyFilter(getOzoneManager(),
-            omSnapshotManager, snapshotChainManager, currentSnapshotInfo, keyManager, lock)) {
+            omSnapshotManager, snapshotChainManager, currentSnapshotInfo, keyManager, lock);
+             ReclaimableRenameEntryFilter renameEntryFilter = new ReclaimableRenameEntryFilter(
+                 getOzoneManager(), omSnapshotManager, snapshotChainManager, currentSnapshotInfo,
+                 keyManager, lock)) {
+          List<String> renamedTableEntries =
+              keyManager.getRenamesKeyEntries(volume, bucket, null, renameEntryFilter, remainNum).stream()
+                  .map(entry -> {
+                    try {
+                      return entry.getKey();
+                    } catch (IOException e) {
+                      throw new UncheckedIOException(e);
+                    }
+                  }).collect(Collectors.toList());
+          remainNum -= renamedTableEntries.size();
+
           // Get pending keys that can be deleted
           PendingKeysDeletion pendingKeysDeletion = currentSnapshotInfo == null
               ? keyManager.getPendingDeletionKeys(reclaimableKeyFilter, remainNum)
               : keyManager.getPendingDeletionKeys(volume, bucket, null, reclaimableKeyFilter, remainNum);
           List<BlockGroup> keyBlocksList = pendingKeysDeletion.getKeyBlocksList();
           //submit purge requests if there are renamed entries to be purged or keys to be purged.
-          if (keyBlocksList != null && !keyBlocksList.isEmpty()) {
+          if (!renamedTableEntries.isEmpty() || keyBlocksList != null && !keyBlocksList.isEmpty()) {
             // Validating if the previous snapshot is still the same before purging the blocks.
             SnapshotUtils.validatePreviousSnapshotId(currentSnapshotInfo, snapshotChainManager,
                 expectedPreviousSnapshotId);
-            Pair<Integer, Boolean> purgeResult = processKeyDeletes(keyBlocksList,
-                 pendingKeysDeletion.getKeysToModify(), snapshotTableKey,
-                 expectedPreviousSnapshotId);
+            Pair<Integer, Boolean> purgeResult = processKeyDeletes(keyBlocksList, pendingKeysDeletion.getKeysToModify(),
+                renamedTableEntries, snapshotTableKey, expectedPreviousSnapshotId);
             remainNum -= purgeResult.getKey();
             successStatus = purgeResult.getValue();
             metrics.incrNumKeysProcessed(keyBlocksList.size());

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

@@ -193,15 +193,15 @@ public BackgroundTaskResult call() throws InterruptedException {
             // Get all entries from deletedKeyTable.
             List<Table.KeyValue<String, List<OmKeyInfo>>> deletedKeyEntries =
                 snapshotKeyManager.getDeletedKeyEntries(snapInfo.getVolumeName(), snapInfo.getBucketName(),
-                    null, remaining);
+                    null, (kv) -> true, remaining);
             moveCount += deletedKeyEntries.size();
             // Get all entries from deletedDirTable.
             List<Table.KeyValue<String, OmKeyInfo>> deletedDirEntries = snapshotKeyManager.getDeletedDirEntries(
                 snapInfo.getVolumeName(), snapInfo.getBucketName(), remaining - moveCount);
             moveCount += deletedDirEntries.size();
             // Get all entries from snapshotRenamedTable.
             List<Table.KeyValue<String, String>> renameEntries = snapshotKeyManager.getRenamesKeyEntries(
-                snapInfo.getVolumeName(), snapInfo.getBucketName(), null, remaining - moveCount);
+                snapInfo.getVolumeName(), snapInfo.getBucketName(), null, (kv) -> true, remaining - moveCount);
             moveCount += renameEntries.size();
             if (moveCount > 0) {
               List<SnapshotMoveKeyInfos> deletedKeys = new ArrayList<>(deletedKeyEntries.size());