HDDS-13929. Modularise Snapshot Delta file computer (Efficient Diff)

hadoop-hdds/rocksdb-checkpoint-differ/src/main/java/org/apache/ozone/rocksdb/util/SstFileSetReader.java

@@ -23,6 +23,7 @@
 import java.io.Closeable;
 import java.io.IOException;
 import java.io.UncheckedIOException;
+import java.nio.file.Path;
 import java.util.Collection;
 import java.util.Comparator;
 import java.util.NoSuchElementException;
@@ -51,11 +52,11 @@
  */
 public class SstFileSetReader {
 
-  private final Collection<String> sstFiles;
+  private final Collection<Path> sstFiles;
 
   private volatile long estimatedTotalKeys = -1;
 
-  public SstFileSetReader(final Collection<String> sstFiles) {
+  public SstFileSetReader(final Collection<Path> sstFiles) {
     this.sstFiles = sstFiles;
   }
 
@@ -77,9 +78,9 @@ public long getEstimatedTotalKeys() throws RocksDBException {
       }
 
       try (ManagedOptions options = new ManagedOptions()) {
-        for (String sstFile : sstFiles) {
+        for (Path sstFile : sstFiles) {
           try (ManagedSstFileReader fileReader = new ManagedSstFileReader(options)) {
-            fileReader.open(sstFile);
+            fileReader.open(sstFile.toAbsolutePath().toString());
             estimatedSize += fileReader.getTableProperties().getNumEntries();
           }
         }
@@ -303,7 +304,7 @@ public int hashCode() {
   private abstract static class MultipleSstFileIterator<T extends Comparable<T>> implements ClosableIterator<T> {
     private final PriorityQueue<HeapEntry<T>> minHeap;
 
-    private MultipleSstFileIterator(Collection<String> sstFiles) {
+    private MultipleSstFileIterator(Collection<Path> sstFiles) {
       this.minHeap = new PriorityQueue<>();
       init();
       initMinHeap(sstFiles);
@@ -313,10 +314,10 @@ private MultipleSstFileIterator(Collection<String> sstFiles) {
 
     protected abstract ClosableIterator<T> getKeyIteratorForFile(String file) throws RocksDBException, IOException;
 
-    private void initMinHeap(Collection<String> files) {
+    private void initMinHeap(Collection<Path> files) {
       try {
-        for (String file : files) {
-          ClosableIterator<T> iterator = getKeyIteratorForFile(file);
+        for (Path file : files) {
+          ClosableIterator<T> iterator = getKeyIteratorForFile(file.toAbsolutePath().toString());
           HeapEntry<T> entry = new HeapEntry<>(iterator);
 
           if (entry.getCurrentKey() != null) {

hadoop-hdds/rocksdb-checkpoint-differ/src/main/java/org/apache/ozone/rocksdiff/RocksDBCheckpointDiffer.java

@@ -741,26 +741,25 @@ private void preconditionChecksForLoadAllCompactionLogs() {
    * exist in backup directory before being involved in compactions),
    * and appends the extension '.sst'.
    */
-  private String getSSTFullPath(String sstFilenameWithoutExtension, Path... dbPaths) {
+  private Path getSSTFullPath(SstFileInfo sstFileInfo, Path... dbPaths) throws IOException {
 
     // Try to locate the SST in the backup dir first
-    final Path sstPathInBackupDir = Paths.get(sstBackupDir, sstFilenameWithoutExtension + SST_FILE_EXTENSION);
+    final Path sstPathInBackupDir = sstFileInfo.getFilePath(Paths.get(sstBackupDir).toAbsolutePath());
     if (Files.exists(sstPathInBackupDir)) {
-      return sstPathInBackupDir.toString();
+      return sstPathInBackupDir.toAbsolutePath();
     }
 
     // SST file does not exist in the SST backup dir, this means the SST file
     // has not gone through any compactions yet and is only available in the
     // src DB directory or destDB directory
     for (Path dbPath : dbPaths) {
-      final Path sstPathInDBDir = dbPath.resolve(sstFilenameWithoutExtension + SST_FILE_EXTENSION);
+      final Path sstPathInDBDir = sstFileInfo.getFilePath(dbPath);
       if (Files.exists(sstPathInDBDir)) {
-        return sstPathInDBDir.toString();
+        return sstPathInDBDir.toAbsolutePath();
       }
     }
 
-    // TODO: More graceful error handling?
-    throw new RuntimeException("Unable to locate SST file: " + sstFilenameWithoutExtension);
+    throw new IOException("Unable to locate SST file: " + sstFileInfo);
   }
 
   /**
@@ -772,15 +771,13 @@ private String getSSTFullPath(String sstFilenameWithoutExtension, Path... dbPath
    * @param dest destination snapshot
    * @param versionMap version map containing the connection between source snapshot version and dest snapshot version.
    * @param tablesToLookup tablesToLookup set of table (column family) names used to restrict which SST files to return.
-   * @param sstFilesDirForSnapDiffJob dir to create hardlinks for SST files
-   *                                 for snapDiff job.
    * @return A list of SST files without extension.
    *         e.g. ["/path/to/sstBackupDir/000050.sst",
    *               "/path/to/sstBackupDir/000060.sst"]
    */
-  public synchronized Optional<List<String>> getSSTDiffListWithFullPath(DifferSnapshotInfo src,
+  public synchronized Optional<Map<Path, SstFileInfo>> getSSTDiffListWithFullPath(DifferSnapshotInfo src,
       DifferSnapshotInfo dest, Map<Integer, Integer> versionMap, TablePrefixInfo prefixInfo,
-      Set<String> tablesToLookup, String sstFilesDirForSnapDiffJob) throws IOException {
+      Set<String> tablesToLookup) throws IOException {
     int srcVersion = src.getMaxVersion();
     if (!versionMap.containsKey(srcVersion)) {
       throw new IOException("No corresponding dest version corresponding srcVersion : " + srcVersion + " in " +
@@ -794,16 +791,15 @@ public synchronized Optional<List<String>> getSSTDiffListWithFullPath(DifferSnap
     // of the sst file names.
     Optional<List<SstFileInfo>> sstDiffList = getSSTDiffList(srcSnapshotVersion, destSnapshotVersion, prefixInfo,
         tablesToLookup, srcVersion == 0);
-
-    return sstDiffList.map(diffList -> diffList.stream()
-        .map(sst -> {
-          String sstFullPath = getSSTFullPath(sst.getFileName(), srcSnapshotVersion.getDbPath(),
-              destSnapshotVersion.getDbPath());
-          Path link = sst.getFilePath(Paths.get(sstFilesDirForSnapDiffJob));
-          Path srcFile = Paths.get(sstFullPath);
-          createLink(link, srcFile);
-          return link.toString();
-        }).collect(Collectors.toList()));
+    if (sstDiffList.isPresent()) {
+      Map<Path, SstFileInfo> sstFileInfoMap = new HashMap<>();
+      for (SstFileInfo sstFileInfo : sstDiffList.get()) {
+        Path sstPath = getSSTFullPath(sstFileInfo, srcSnapshotVersion.getDbPath());
+        sstFileInfoMap.put(sstPath, sstFileInfo);
+      }
+      return Optional.of(sstFileInfoMap);
+    }
+    return Optional.empty();
   }
 
   /**

hadoop-hdds/rocksdb-checkpoint-differ/src/main/java/org/apache/ozone/rocksdiff/RocksDiffUtils.java

@@ -69,12 +69,11 @@ public static <T> Map<T, SstFileInfo> filterRelevantSstFiles(Map<T, SstFileInfo>
 
   /**
    * Filter sst files based on prefixes. The set of sst files to be filtered would be mutated.
-   * @param <T> Type of the key in the map.
    * @param filesToBeFiltered sst files to be filtered.
    * @param tablesToLookup Set of column families to be included in the diff.
    * @param tablePrefixInfo TablePrefixInfo to filter irrelevant SST files.
    */
-  public static <T> Set<SstFileInfo> filterRelevantSstFiles(Set<SstFileInfo> filesToBeFiltered,
+  public static Set<SstFileInfo> filterRelevantSstFiles(Set<SstFileInfo> filesToBeFiltered,
       Set<String> tablesToLookup, TablePrefixInfo tablePrefixInfo) {
     for (Iterator<SstFileInfo> fileIterator = filesToBeFiltered.iterator(); fileIterator.hasNext();) {
       SstFileInfo sstFileInfo = fileIterator.next();

hadoop-hdds/rocksdb-checkpoint-differ/src/test/java/org/apache/ozone/rocksdb/util/TestSstFileSetReader.java

@@ -24,6 +24,7 @@
 import static org.junit.jupiter.api.Assumptions.assumeTrue;
 
 import java.io.File;
+import java.nio.file.Path;
 import java.util.ArrayList;
 import java.util.HashSet;
 import java.util.List;
@@ -74,7 +75,7 @@ class TestSstFileSetReader {
    * @return Absolute path to the created SST file
    * @throws RocksDBException if there's an error during SST file creation
    */
-  private String createRandomSSTFile(TreeMap<String, Integer> keys)
+  private Path createRandomSSTFile(TreeMap<String, Integer> keys)
       throws RocksDBException {
     File file = new File(tempDir, "tmp_sst_file" + fileCounter.incrementAndGet() + ".sst");
 
@@ -94,7 +95,7 @@ private String createRandomSSTFile(TreeMap<String, Integer> keys)
       sstFileWriter.finish();
     }
     assertTrue(file.exists());
-    return file.getAbsolutePath();
+    return file.getAbsoluteFile().toPath();
   }
 
   /**
@@ -121,8 +122,8 @@ private Map<String, Integer> createKeys(int startRange, int endRange) {
    * @return Pair containing the complete sorted key map and list of SST file paths
    * @throws RocksDBException if there's an error during SST file creation
    */
-  private Pair<SortedMap<String, Integer>, List<String>> createDummyData(int numberOfFiles) throws RocksDBException {
-    List<String> files = new ArrayList<>();
+  private Pair<SortedMap<String, Integer>, List<Path>> createDummyData(int numberOfFiles) throws RocksDBException {
+    List<Path> files = new ArrayList<>();
     int numberOfKeysPerFile = 1000;
     TreeMap<String, Integer> keys =
         new TreeMap<>(createKeys(0, numberOfKeysPerFile * numberOfFiles));
@@ -136,7 +137,7 @@ private Pair<SortedMap<String, Integer>, List<String>> createDummyData(int numbe
       cnt += 1;
     }
     for (TreeMap<String, Integer> fileKeys : fileKeysList) {
-      String tmpSSTFile = createRandomSSTFile(fileKeys);
+      Path tmpSSTFile = createRandomSSTFile(fileKeys);
       files.add(tmpSSTFile);
     }
     return Pair.of(keys, files);
@@ -153,8 +154,8 @@ private Pair<SortedMap<String, Integer>, List<String>> createDummyData(int numbe
   @ValueSource(ints = {0, 1, 2, 3, 7, 10})
   public void testGetKeyStream(int numberOfFiles)
       throws RocksDBException {
-    Pair<SortedMap<String, Integer>, List<String>> data = createDummyData(numberOfFiles);
-    List<String> files = data.getRight();
+    Pair<SortedMap<String, Integer>, List<Path>> data = createDummyData(numberOfFiles);
+    List<Path> files = data.getRight();
     SortedMap<String, Integer> keys = data.getLeft();
     // Getting every possible combination of 2 elements from the sampled keys.
     // Reading the sst file lying within the given bounds and
@@ -195,9 +196,9 @@ public void testGetKeyStream(int numberOfFiles)
   public void testGetKeyStreamWithTombstone(int numberOfFiles)
       throws RocksDBException {
     assumeTrue(ManagedRawSSTFileReader.tryLoadLibrary());
-    Pair<SortedMap<String, Integer>, List<String>> data =
+    Pair<SortedMap<String, Integer>, List<Path>> data =
         createDummyData(numberOfFiles);
-    List<String> files = data.getRight();
+    List<Path> files = data.getRight();
     SortedMap<String, Integer> keys = data.getLeft();
     // Getting every possible combination of 2 elements from the sampled keys.
     // Reading the sst file lying within the given bounds and
@@ -237,7 +238,7 @@ public void testMinHeapWithOverlappingSstFiles(int numberOfFiles) throws RocksDB
     assumeTrue(numberOfFiles >= 2);
 
     // Create overlapping SST files with some duplicate keys
-    List<String> files = new ArrayList<>();
+    List<Path> files = new ArrayList<>();
     Map<String, Integer> expectedKeys = new TreeMap<>();
 
     // File 0: keys 0-9 (all valid entries)
@@ -305,7 +306,7 @@ public void testDuplicateKeyHandlingWithLatestFilePrecedence(int numberOfFiles)
       throws RocksDBException {
     assumeTrue(numberOfFiles >= 3);
 
-    List<String> files = new ArrayList<>();
+    List<Path> files = new ArrayList<>();
 
     // All files will contain the same set of keys, but we expect the last file to "win"
     String[] testKeys = {KEY_PREFIX + "duplicate1", KEY_PREFIX + "duplicate2", KEY_PREFIX + "duplicate3"};

hadoop-ozone/ozone-manager/dev-support/findbugsExcludeFile.xml

@@ -16,4 +16,8 @@
    limitations under the License.
 -->
 <FindBugsFilter>
+  <Match>
+    <Class name="org.apache.hadoop.ozone.om.snapshot.diff.delta.TestRDBDifferComputer"/>
+    <Bug pattern="RV_RETURN_VALUE_IGNORED_NO_SIDE_EFFECT" />
+  </Match>
 </FindBugsFilter>

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

@@ -262,10 +262,6 @@ public OmSnapshotManager(OzoneManager ozoneManager) throws IOException {
     }
 
     this.ozoneManager = ozoneManager;
-    RocksDBCheckpointDiffer differ = ozoneManager
-        .getMetadataManager()
-        .getStore()
-        .getRocksDBCheckpointDiffer();
 
     // Soft-limit of lru cache size
     this.softCacheSize = ozoneManager.getConfiguration().getInt(
@@ -312,8 +308,8 @@ public OmSnapshotManager(OzoneManager ozoneManager) throws IOException {
     this.snapshotCache = new SnapshotCache(loader, softCacheSize, ozoneManager.getMetrics(),
         cacheCleanupServiceInterval, compactNonSnapshotDiffTables, ozoneManager.getMetadataManager().getLock());
 
-    this.snapshotDiffManager = new SnapshotDiffManager(snapshotDiffDb, differ,
-        ozoneManager, snapshotLocalDataManager, snapDiffJobCf, snapDiffReportCf,
+    this.snapshotDiffManager = new SnapshotDiffManager(snapshotDiffDb,
+        ozoneManager, snapDiffJobCf, snapDiffReportCf,
         columnFamilyOptions, codecRegistry);
 
     diffCleanupServiceInterval = ozoneManager.getConfiguration()