HDDS-12501. Remove OMKeyInfo from SST files in backup directory.

hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConfigKeys.java

@@ -639,6 +639,17 @@ public final class OzoneConfigKeys {
       OZONE_OM_SNAPSHOT_PRUNE_COMPACTION_DAG_DAEMON_RUN_INTERVAL_DEFAULT =
       TimeUnit.HOURS.toMillis(1);
 
+  public static final String
+      OZONE_OM_SNAPSHOT_PRUNE_COMPACTION_BACKUP_BATCH_SIZE =
+      "ozone.om.snapshot.prune.compaction.backup.batch.size";
+
+  public static final int
+      OZONE_OM_SNAPSHOT_PRUNE_COMPACTION_BACKUP_BATCH_SIZE_DEFAULT = 2000;
+
+  public static final String OZONE_OM_SNAPSHOT_LOAD_NATIVE_LIB =
+      "ozone.om.snapshot.load.native.lib";
+  public static final boolean OZONE_OM_SNAPSHOT_LOAD_NATIVE_LIB_DEFAULT = true;
+
   public static final String OZONE_OM_DELTA_UPDATE_DATA_SIZE_MAX_LIMIT =
       "ozone.om.delta.update.data.size.max.limit";
   public static final String

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

@@ -4335,6 +4335,16 @@
     </description>
   </property>
 
+  <property>
+    <name>ozone.om.snapshot.prune.compaction.backup.batch.size</name>
+    <value>2000</value>
+    <tag>OZONE, OM</tag>
+    <description>
+      Prune SST files in Compaction backup directory in batches every
+      ozone.om.snapshot.compaction.dag.prune.daemon.run.interval.
+    </description>
+  </property>
+
   <property>
     <name>ozone.om.snapshot.compaction.dag.prune.daemon.run.interval</name>
     <value>3600s</value>

hadoop-hdds/interface-client/src/main/proto/hdds.proto

@@ -515,6 +515,7 @@ message CompactionFileInfoProto {
     optional string startKey = 2;
     optional string endKey = 3;
     optional string columnFamily = 4;
+    optional bool pruned = 5;
 }
 
 message CompactionLogEntryProto {

hadoop-hdds/rocksdb-checkpoint-differ/src/main/java/org/apache/ozone/compaction/log/CompactionFileInfo.java

@@ -32,16 +32,26 @@ public final class CompactionFileInfo {
   private final String startKey;
   private final String endKey;
   private final String columnFamily;
+  private boolean pruned;
 
   @VisibleForTesting
   public CompactionFileInfo(String fileName,
                             String startRange,
                             String endRange,
                             String columnFamily) {
+    this(fileName, startRange, endRange, columnFamily, false);
+  }
+
+  public CompactionFileInfo(String fileName,
+                            String startRange,
+                            String endRange,
+                            String columnFamily,
+                            boolean pruned) {
     this.fileName = fileName;
     this.startKey = startRange;
     this.endKey = endRange;
     this.columnFamily = columnFamily;
+    this.pruned = pruned;
   }
 
   public String getFileName() {
@@ -60,10 +70,19 @@ public String getColumnFamily() {
     return columnFamily;
   }
 
+  public boolean isPruned() {
+    return pruned;
+  }
+
+  public void setPruned() {
+    this.pruned = true;
+  }
+
   public HddsProtos.CompactionFileInfoProto getProtobuf() {
     HddsProtos.CompactionFileInfoProto.Builder builder =
         HddsProtos.CompactionFileInfoProto.newBuilder()
-            .setFileName(fileName);
+            .setFileName(fileName)
+            .setPruned(pruned);
     if (startKey != null) {
       builder = builder.setStartKey(startKey);
     }
@@ -89,14 +108,17 @@ public static CompactionFileInfo getFromProtobuf(
     if (proto.hasColumnFamily()) {
       builder.setColumnFamily(proto.getColumnFamily());
     }
+    if (proto.hasPruned() && proto.getPruned()) {
+      builder.setPruned();
+    }
 
     return builder.build();
   }
 
   @Override
   public String toString() {
     return String.format("fileName: '%s', startKey: '%s', endKey: '%s'," +
-        " columnFamily: '%s'", fileName, startKey, endKey, columnFamily);
+        " columnFamily: '%s', isPruned: '%b'", fileName, startKey, endKey, columnFamily, pruned);
   }
 
   /**
@@ -107,6 +129,7 @@ public static class Builder {
     private String startRange;
     private String endRange;
     private String columnFamily;
+    private boolean pruned = false;
 
     public Builder(String fileName) {
       Preconditions.checkNotNull(fileName, "FileName is required parameter.");
@@ -138,6 +161,11 @@ public Builder setValues(LiveFileMetaData fileMetaData) {
       return this;
     }
 
+    public Builder setPruned() {
+      this.pruned = true;
+      return this;
+    }
+
     public CompactionFileInfo build() {
       if ((startRange != null || endRange != null || columnFamily != null) &&
           (startRange == null || endRange == null || columnFamily == null)) {
@@ -149,7 +177,7 @@ public CompactionFileInfo build() {
       }
 
       return new CompactionFileInfo(fileName, startRange, endRange,
-          columnFamily);
+          columnFamily, pruned);
     }
   }
 

hadoop-hdds/rocksdb-checkpoint-differ/src/main/java/org/apache/ozone/compaction/log/CompactionLogEntry.java

@@ -129,13 +129,23 @@ public String toString() {
         inputFileInfoList, outputFileInfoList, compactionReason);
   }
 
+  public Builder toBuilder() {
+    Builder builder = new Builder(this.getDbSequenceNumber(), this.getCompactionTime(),
+        this.getInputFileInfoList(), this.getOutputFileInfoList());
+    String reason = this.getCompactionReason();
+    if (this.getCompactionReason() != null) {
+      builder.setCompactionReason(reason);
+    }
+    return builder;
+  }
+
   /**
    * Builder of CompactionLogEntry.
    */
   public static class Builder {
     private final long dbSequenceNumber;
     private final long compactionTime;
-    private final List<CompactionFileInfo> inputFileInfoList;
+    private List<CompactionFileInfo> inputFileInfoList;
     private final List<CompactionFileInfo> outputFileInfoList;
     private String compactionReason;
 
@@ -157,6 +167,11 @@ public Builder setCompactionReason(String compactionReason) {
       return this;
     }
 
+    public Builder updateInputFileInfoList(List<CompactionFileInfo> fileInfoList) {
+      this.inputFileInfoList = fileInfoList;
+      return this;
+    }
+
     public CompactionLogEntry build() {
       return new CompactionLogEntry(dbSequenceNumber, compactionTime,
           inputFileInfoList, outputFileInfoList, compactionReason);