HDDS-12960. SST Statistics-Based RocksDB Compaction Scheduling

.github/workflows/check.yml

@@ -136,6 +136,8 @@ env:
   OZONE_IMAGE: ghcr.io/apache/ozone
   OZONE_RUNNER_IMAGE: ghcr.io/apache/ozone-runner
   OZONE_VOLUME_OWNER: 1000
+  GITHUB_ACTOR: ${{ github.actor }}
+  GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
 
 jobs:
   check:

.github/workflows/ci.yml

@@ -34,6 +34,9 @@ env:
   MAVEN_ARGS: --batch-mode --settings ${{ github.workspace }}/dev-support/ci/maven-settings.xml
   MAVEN_OPTS: -Dhttp.keepAlive=false -Dmaven.wagon.http.pool=false -Dmaven.wagon.http.retryHandler.class=standard -Dmaven.wagon.http.retryHandler.count=3
   OZONE_WITH_COVERAGE: ${{ github.event_name == 'push' }}
+  GITHUB_ACTOR: ${{ github.actor }}
+  GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+
 
 jobs:
   build-info:

dev-support/ci/maven-settings.xml

@@ -32,4 +32,12 @@
       <blocked>true</blocked>
     </mirror>
   </mirrors>
+
+  <servers>
+    <server>
+      <id>github</id>
+      <username>${env.GITHUB_ACTOR}</username>
+      <password>${env.GITHUB_TOKEN}</password>
+    </server>
+  </servers>
 </settings>

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/StringUtils.java

@@ -97,4 +97,16 @@ public static String bytes2String(byte[] bytes) {
   public static byte[] string2Bytes(String str) {
     return str.getBytes(UTF8);
   }
+
+  public static String getKeyPrefixUpperBound(String key) {
+    return key.substring(0, key.length() - 1) + (char)(key.charAt(key.length() - 1) + 1);
+  }
+
+  public static String max(String str1, String str2) {
+    return str1.compareTo(str2) > 0 ? str1 : str2;
+  }
+
+  public static String min(String str1, String str2) {
+    return str1.compareTo(str2) < 0 ? str1 : str2;
+  }
 }

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

@@ -4819,4 +4819,54 @@
       warm up edek cache if none of key successful on OM start up.
     </description>
   </property>
+
+  <!-- Range Compaction Service Configuration -->
+  <property>
+    <name>ozone.om.range.compaction.service.enabled</name>
+    <value>false</value>
+    <tag>OZONE, OM, PERFORMANCE</tag>
+    <description>Enable or disable the range compaction service that compacts specific ranges of data based on tombstone ratio. This service helps improve performance and reduce storage overhead by compacting ranges with high tombstone density.</description>
+  </property>
+
+  <property>
+    <name>ozone.om.range.compaction.service.interval</name>
+    <value>30s</value>
+    <tag>OZONE, OM, PERFORMANCE</tag>
+    <description>The interval at which the range compaction service checks for ranges that need compaction. The service will scan for ranges with high tombstone density at this interval.</description>
+  </property>
+
+  <property>
+    <name>ozone.om.range.compaction.service.max.compaction.entries</name>
+    <value>1000000</value>
+    <tag>OZONE, OM, PERFORMANCE</tag>
+    <description>The maximum number of entries that can be compacted in a single range compaction operation. This limit helps prevent long-running compaction operations that could impact system performance.</description>
+  </property>
+
+  <property>
+    <name>ozone.om.range.compaction.service.min.tombstones</name>
+    <value>10000</value>
+    <tag>OZONE, OM, PERFORMANCE</tag>
+    <description>The minimum number of tombstones required in a range before it is considered for compaction. This threshold helps ensure that only ranges with significant tombstone density are compacted.</description>
+  </property>
+
+  <property>
+    <name>ozone.om.range.compaction.service.ranges.per.run</name>
+    <value>20</value>
+    <tag>OZONE, OM, PERFORMANCE</tag>
+    <description>The number of ranges to process in each run of the range compaction service. This controls the granularity of compaction operations and helps balance system load.</description>
+  </property>
+
+  <property>
+    <name>ozone.om.range.compaction.service.timeout</name>
+    <value>10m</value>
+    <tag>OZONE, OM, PERFORMANCE</tag>
+    <description>The timeout for range compaction operations. If a compaction operation takes longer than this, it will be cancelled to prevent long-running operations from impacting system performance.</description>
+  </property>
+
+  <property>
+    <name>ozone.om.range.compaction.service.tombstone.ratio</name>
+    <value>0.3</value>
+    <tag>OZONE, OM, PERFORMANCE</tag>
+    <description>The ratio of tombstones to total entries that triggers compaction of a range. When the ratio exceeds this threshold, the range will be considered for compaction to improve storage efficiency.</description>
+  </property>
 </configuration>

hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/DBStore.java

@@ -18,6 +18,7 @@
 package org.apache.hadoop.hdds.utils.db;
 
 import java.io.File;
+import java.io.IOException;
 import java.util.List;
 import java.util.Map;
 import org.apache.hadoop.hdds.annotation.InterfaceStability;
@@ -26,6 +27,7 @@
 import org.apache.hadoop.hdds.utils.db.managed.ManagedCompactRangeOptions;
 import org.apache.ozone.rocksdiff.RocksDBCheckpointDiffer;
 import org.apache.ratis.util.UncheckedAutoCloseable;
+import org.rocksdb.TableProperties;
 
 /**
  * The DBStore interface provides the ability to create Tables, which store
@@ -108,6 +110,33 @@ <KEY, VALUE> TypedTable<KEY, VALUE> getTable(
    */
   void compactTable(String tableName, ManagedCompactRangeOptions options) throws RocksDatabaseException;
 
+  /**
+   * Compact a range of keys in the database.
+   *
+   * @param tableName - The table name to compact.
+   * @param startKey - The starting key of the range to compact.
+   * @param endKey - The ending key of the range to compact.
+   * @throws IOException on Failure
+   */
+  void compactTable(String tableName, String startKey, String endKey) throws IOException;
+
+  void compactTable(String tableName, String startKey, String endKey,
+          ManagedCompactRangeOptions options) throws IOException;
+
+  /**
+   * Get the properties of a column family in a range.
+   * @param tableName - The table name to get the properties of.
+   * @param startKey - The starting key of the range.
+   * @param endKey - The ending key of the range.
+   * @return - The properties of the column family in the range.
+   * @throws IOException on Failure
+   */
+  Map<String, TableProperties> getPropertiesOfTableInRange(String tableName, String startKey,
+      String endKey) throws IOException;
+
+  Map<String, TableProperties> getPropertiesOfTableInRange(String tableName, List<KeyRange> ranges)
+      throws IOException;
+
   /**
    * Returns an estimated count of keys in this DB.
    *

hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/KeyRange.java

@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hdds.utils.db;
+
+import org.apache.hadoop.hdds.StringUtils;
+
+/**
+ * Represents a range of keys for compaction.
+ */
+public class KeyRange {
+  private final String startKey;
+  private final String endKey;
+
+  public KeyRange(String startKey, String endKey) {
+    this.startKey = startKey;
+    this.endKey = endKey;
+  }
+
+  public KeyRange(byte[] startKey, byte[] endKey) {
+    this.startKey = StringUtils.bytes2String(startKey);
+    this.endKey = StringUtils.bytes2String(endKey);
+  }
+
+  public String getStartKey() {
+    return startKey;
+  }
+
+  public String getEndKey() {
+    return endKey;
+  }
+
+  @Override
+  public String toString() {
+    return "KeyRange{" +
+        "startKey='" + startKey + '\'' +
+        ", endKey='" + endKey + '\'' +
+        '}';
+  }
+}

hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/RDBStore.java

@@ -34,22 +34,27 @@
 import java.nio.file.Paths;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import org.apache.hadoop.hdds.StringUtils;
 import org.apache.hadoop.hdds.conf.ConfigurationSource;
 import org.apache.hadoop.hdds.utils.IOUtils;
 import org.apache.hadoop.hdds.utils.RocksDBStoreMetrics;
 import org.apache.hadoop.hdds.utils.db.RocksDatabase.ColumnFamily;
 import org.apache.hadoop.hdds.utils.db.cache.TableCache;
 import org.apache.hadoop.hdds.utils.db.managed.ManagedCompactRangeOptions;
 import org.apache.hadoop.hdds.utils.db.managed.ManagedDBOptions;
+import org.apache.hadoop.hdds.utils.db.managed.ManagedSlice;
 import org.apache.hadoop.hdds.utils.db.managed.ManagedStatistics;
 import org.apache.hadoop.hdds.utils.db.managed.ManagedTransactionLogIterator;
 import org.apache.hadoop.hdds.utils.db.managed.ManagedWriteOptions;
 import org.apache.ozone.rocksdiff.RocksDBCheckpointDiffer;
 import org.apache.ozone.rocksdiff.RocksDBCheckpointDiffer.RocksDBCheckpointDifferHolder;
+import org.rocksdb.Range;
 import org.rocksdb.RocksDBException;
+import org.rocksdb.TableProperties;
 import org.rocksdb.TransactionLogIterator.BatchResult;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -220,6 +225,15 @@ public void compactTable(String tableName) throws RocksDatabaseException {
     }
   }
 
+  @Override
+  public void compactTable(String tableName, String startKey, String endKey) throws IOException {
+    LOG.info("Starting table compaction for table: {}, range: [{} - {}]", tableName, startKey, endKey);
+    try (ManagedCompactRangeOptions options = new ManagedCompactRangeOptions()) {
+      compactTable(tableName, startKey, endKey, options);
+    }
+    LOG.info("Completed table compaction for table: {}, range: [{} - {}]", tableName, startKey, endKey);
+  }
+
   @Override
   public void compactTable(String tableName, ManagedCompactRangeOptions options) throws RocksDatabaseException {
     RocksDatabase.ColumnFamily columnFamily = db.getColumnFamily(tableName);
@@ -229,6 +243,67 @@ public void compactTable(String tableName, ManagedCompactRangeOptions options) t
     db.compactRange(columnFamily, null, null, options);
   }
 
+  @Override
+  public void compactTable(String tableName, String startKey, String endKey,
+      ManagedCompactRangeOptions options) throws IOException {
+    LOG.info("Starting table compaction with options for table: {}, range: [{} - {}]", tableName, startKey, endKey);
+    ColumnFamily columnFamily = db.getColumnFamily(tableName);
+    if (columnFamily == null) {
+      LOG.error("Table not found for compaction: {}", tableName);
+      throw new IOException("No such table in this DB. TableName : " + tableName);
+    }
+    db.compactRange(columnFamily, StringUtils.string2Bytes(startKey),
+        StringUtils.string2Bytes(endKey), options);
+    LOG.info("Completed table compaction with options for table: {}, range: [{} - {}]", tableName, startKey, endKey);
+  }
+
+  @Override
+  public Map<String, TableProperties> getPropertiesOfTableInRange(String tableName, String startKey,
+      String endKey) throws IOException {
+    LOG.info("Getting table properties for table: {}, range: [{} - {}]", tableName, startKey, endKey);
+    Map<String, TableProperties> result = getPropertiesOfTableInRange(tableName, 
+        Collections.singletonList(new KeyRange(startKey, endKey)));
+    LOG.info("Retrieved {} table properties for table: {}, range: [{} - {}]", 
+        result.size(), tableName, startKey, endKey);
+    return result;
+  }
+
+  @Override
+  public Map<String, TableProperties> getPropertiesOfTableInRange(String tableName, 
+      List<KeyRange> ranges) throws IOException {
+    LOG.info("Getting table properties for table: {}, number of ranges: {}", tableName, ranges.size());
+    ColumnFamily columnFamily = db.getColumnFamily(tableName);
+    if (columnFamily == null) {
+      LOG.error("Table not found for getting properties: {}", tableName);
+      throw new IOException("No such table in this DB. TableName : " + tableName);
+    }
+
+    List<Range> rocksRanges = new ArrayList<>();
+    List<ManagedSlice> managedSlices = new ArrayList<>();
+    try {
+      for (KeyRange t : ranges) {
+        ManagedSlice start = new ManagedSlice(StringUtils.string2Bytes(t.getStartKey()));
+        ManagedSlice end = new ManagedSlice(StringUtils.string2Bytes(t.getEndKey()));
+        managedSlices.add(start);
+        managedSlices.add(end);
+        rocksRanges.add(new Range(start, end));
+      }
+      LOG.info("Converted {} ranges to RocksDB ranges, start to get properties", rocksRanges.size());
+      return db.getPropertiesOfColumnFamilyInRange(columnFamily, rocksRanges);
+    } catch (RocksDatabaseException e) {
+      throw new IOException("Failed to get properties of table in range", e);
+    } finally {
+      // Close all ManagedSlice objects
+      for (ManagedSlice slice : managedSlices) {
+        try {
+          slice.close();
+        } catch (Exception e) {
+          LOG.warn("Failed to close ManagedSlice", e);
+        }
+      }
+    }
+  }
+
   @Override
   public void close() {
     if (metrics != null) {

hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/RocksDatabase.java

@@ -63,7 +63,9 @@
 import org.rocksdb.Holder;
 import org.rocksdb.KeyMayExist;
 import org.rocksdb.LiveFileMetaData;
+import org.rocksdb.Range;
 import org.rocksdb.RocksDBException;
+import org.rocksdb.TableProperties;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -891,6 +893,16 @@ public void deleteFilesNotMatchingPrefix(Map<String, String> prefixPairs) throws
     }
   }
 
+  public Map<String, TableProperties> getPropertiesOfColumnFamilyInRange(ColumnFamily columnFamily,
+      List<Range> ranges) throws RocksDatabaseException {
+    try (UncheckedAutoCloseable ignored = acquire()) {
+      return db.get().getPropertiesOfTablesInRange(columnFamily.getHandle(), ranges);
+    } catch (RocksDBException e) {
+      closeOnError(e);
+      throw toRocksDatabaseException(this, "getPropertiesOfColumnFamilyInRange", e);
+    }
+  }
+
   @Override
   protected void finalize() throws Throwable {
     if (!isClosed()) {

hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/Table.java

@@ -265,6 +265,14 @@ default void cleanupCache(List<Long> epochs) {
     throw new NotImplementedException("cacheIterator is not implemented");
   }
 
+  /**
+   * Return cache iterator maintained for this table.
+   */
+  default Iterator<Map.Entry<CacheKey<KEY>, CacheValue<VALUE>>>
+      cacheIterator(KEY startKey) {
+    throw new NotImplementedException("cacheIterator with startKey is not implemented");
+  }
+
   /**
    * Create the metrics datasource that emits table cache metrics.
    */

hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/TypedTable.java

@@ -436,6 +436,11 @@ public Iterator<Map.Entry<CacheKey<KEY>, CacheValue<VALUE>>> cacheIterator() {
     return cache.iterator();
   }
 
+  @Override
+  public Iterator<Map.Entry<CacheKey<KEY>, CacheValue<VALUE>>> cacheIterator(KEY startKey) {
+    return cache.iterator(startKey);
+  }
+
   @Override
   public TableCacheMetrics createCacheMetrics() {
     return TableCacheMetrics.create(cache, getName());

hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/cache/FullTableCache.java

@@ -53,7 +53,7 @@ public class FullTableCache<KEY, VALUE> implements TableCache<KEY, VALUE> {
   private static final Logger LOG =
       LoggerFactory.getLogger(FullTableCache.class);
 
-  private final Map<CacheKey<KEY>, CacheValue<VALUE>> cache;
+  private final NavigableMap<CacheKey<KEY>, CacheValue<VALUE>> cache;
   private final NavigableMap<Long, Set<CacheKey<KEY>>> epochEntries;
   private final ScheduledExecutorService executorService;
   private final Queue<Long> epochCleanupQueue = new ConcurrentLinkedQueue<>();
@@ -150,6 +150,15 @@ public Iterator<Map.Entry<CacheKey<KEY>, CacheValue<VALUE>>> iterator() {
     return cache.entrySet().iterator();
   }
 
+  @Override
+  public Iterator<Map.Entry<CacheKey<KEY>, CacheValue<VALUE>>> iterator(KEY startKey) {
+    statsRecorder.recordIteration();
+    if (startKey == null) {
+      return cache.entrySet().iterator();
+    }
+    return cache.tailMap(new CacheKey<>(startKey)).entrySet().iterator();
+  }
+
   @VisibleForTesting
   @Override
   public void evictCache(List<Long> epochs) {