HDDS-13912. Modularise Snapshot Delta file computer (Full Diff)

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/utils/IOUtils.java

@@ -17,12 +17,15 @@
 
 package org.apache.hadoop.hdds.utils;
 
+import com.google.common.annotations.VisibleForTesting;
 import jakarta.annotation.Nonnull;
 import java.io.File;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.attribute.BasicFileAttributes;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Properties;
@@ -119,4 +122,15 @@ public static void writePropertiesToFile(File file, Properties properties) throw
     }
     return props;
   }
+
+  /**
+   * Get the INode for file.
+   *
+   * @param file File whose INode is to be retrieved.
+   * @return INode for file.
+   */
+  @VisibleForTesting
+  public static Object getINode(Path file) throws IOException {
+    return Files.readAttributes(file, BasicFileAttributes.class).fileKey();
+  }
 }

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

@@ -17,11 +17,11 @@
 
 package org.apache.ozone.rocksdb.util;
 
+import static org.apache.hadoop.hdds.utils.IOUtils.getINode;
+
 import java.io.IOException;
-import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
-import java.nio.file.attribute.BasicFileAttributes;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -57,7 +57,7 @@ public static Map<Object, SstFileInfo> getSSTFilesWithInodesForComparison(
     Map<Object, SstFileInfo> inodeToSstMap = new HashMap<>();
     for (LiveFileMetaData lfm : liveSSTFilesForCFs) {
       Path sstFilePath = Paths.get(lfm.path(), lfm.fileName());
-      Object inode = Files.readAttributes(sstFilePath, BasicFileAttributes.class).fileKey();
+      Object inode = getINode(sstFilePath);
       inodeToSstMap.put(inode, new SstFileInfo(lfm));
     }
     return inodeToSstMap;

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

@@ -92,7 +92,7 @@ public int hashCode() {
   }
 
   public Path getFilePath(Path directoryPath) {
-    return directoryPath.resolve(fileName + SST_FILE_EXTENSION);
+    return directoryPath.resolve(getFileName() + SST_FILE_EXTENSION);
   }
 
   @Override

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

@@ -17,6 +17,7 @@
 
 package org.apache.hadoop.ozone.om;
 
+import static org.apache.hadoop.hdds.utils.IOUtils.getINode;
 import static org.apache.hadoop.ozone.OzoneConsts.OM_DB_NAME;
 import static org.apache.hadoop.ozone.TestDataUtil.readFully;
 import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_RATIS_SNAPSHOT_MAX_TOTAL_SST_SIZE_KEY;
@@ -361,15 +362,14 @@ private void checkSnapshot(OzoneManager leaderOM, OzoneManager followerOM,
           }
           // If it is a hard link on the leader, it should be a hard
           // link on the follower
-          if (OmSnapshotUtils.getINode(leaderActiveSST)
-              .equals(OmSnapshotUtils.getINode(leaderSnapshotSST))) {
+          if (getINode(leaderActiveSST).equals(getINode(leaderSnapshotSST))) {
             Path followerSnapshotSST =
                 Paths.get(followerSnapshotDir.toString(), fileName);
             Path followerActiveSST =
                 Paths.get(followerActiveDir.toString(), fileName);
             assertEquals(
-                OmSnapshotUtils.getINode(followerActiveSST),
-                OmSnapshotUtils.getINode(followerSnapshotSST),
+                getINode(followerActiveSST),
+                getINode(followerSnapshotSST),
                 "Snapshot sst file is supposed to be a hard link");
             hardLinkCount++;
           }

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

@@ -20,6 +20,7 @@
 import static org.apache.hadoop.hdds.utils.Archiver.includeFile;
 import static org.apache.hadoop.hdds.utils.Archiver.tar;
 import static org.apache.hadoop.hdds.utils.HddsServerUtil.includeRatisSnapshotCompleteFlag;
+import static org.apache.hadoop.hdds.utils.IOUtils.getINode;
 import static org.apache.hadoop.ozone.OzoneConsts.OM_CHECKPOINT_DIR;
 import static org.apache.hadoop.ozone.OzoneConsts.OM_SNAPSHOT_CHECKPOINT_DIR;
 import static org.apache.hadoop.ozone.OzoneConsts.OM_SNAPSHOT_DIR;
@@ -535,8 +536,7 @@ private static Path findLinkPath(Map<String, Map<Path, Path>> files, Path file)
       // Check if the files are hard linked to each other.
       // Note comparison must be done against srcPath, because
       // destPath may only exist on Follower.
-      if (OmSnapshotUtils.getINode(srcPath).equals(
-          OmSnapshotUtils.getINode(file))) {
+      if (getINode(srcPath).equals(getINode(file))) {
         return destPath;
       } else {
         LOG.info("Found non linked sst files with the same name: {}, {}",

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

@@ -17,16 +17,15 @@
 
 package org.apache.hadoop.ozone.om.snapshot;
 
+import static org.apache.hadoop.hdds.utils.IOUtils.getINode;
 import static org.apache.hadoop.ozone.OzoneConsts.OM_CHECKPOINT_DIR;
 
-import com.google.common.annotations.VisibleForTesting;
 import java.io.File;
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
-import java.nio.file.attribute.BasicFileAttributes;
 import java.nio.file.attribute.FileTime;
 import java.util.ArrayList;
 import java.util.List;
@@ -60,17 +59,6 @@ public static String truncateFileName(int truncateLength, Path file) {
     return file.toString().substring(truncateLength);
   }
 
-  /**
-   * Get the INode for file.
-   *
-   * @param file File whose INode is to be retrieved.
-   * @return INode for file.
-   */
-  @VisibleForTesting
-  public static Object getINode(Path file) throws IOException {
-    return Files.readAttributes(file, BasicFileAttributes.class).fileKey();
-  }
-
   /**
    * Returns a string combining the inode (fileKey) and the last modification time (mtime) of the given file.
    * <p>

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/diff/delta/DeltaFileComputer.java

@@ -0,0 +1,50 @@
+/*
+ * 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.ozone.om.snapshot.diff.delta;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.Collection;
+import java.util.Set;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.hadoop.ozone.om.helpers.SnapshotInfo;
+import org.apache.ozone.rocksdb.util.SstFileInfo;
+
+/**
+ * The DeltaFileComputer interface defines a contract for computing delta files
+ * that represent changes between two snapshots. Implementations of this
+ * interface are responsible for determining the modifications made from a
+ * baseline snapshot to a target snapshot in the form of delta files.
+ */
+public interface DeltaFileComputer extends Closeable {
+
+  /**
+   * Retrieves the delta files representing changes between two snapshots for specified tables.
+   *
+   * @param fromSnapshot the baseline snapshot from which changes are computed
+   * @param toSnapshot the target snapshot to which changes are compared
+   * @param tablesToLookup the set of table names to consider when determining changes
+   * @return an {@code Optional} containing a collection of pairs, where each pair consists of a
+   *         {@code Path} representing the delta file and an associated {@code SstFileInfo}, or
+   *         an empty {@code Optional} if no changes are found
+   * @throws IOException if an I/O error occurs while retrieving delta files
+   */
+  Collection<Pair<Path, SstFileInfo>> getDeltaFiles(SnapshotInfo fromSnapshot, SnapshotInfo toSnapshot,
+      Set<String> tablesToLookup) throws IOException;
+}

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/diff/delta/FileLinkDeltaFileComputer.java

@@ -0,0 +1,155 @@
+/*
+ * 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.ozone.om.snapshot.diff.delta;
+
+import static java.nio.file.Files.createDirectories;
+import static org.apache.commons.io.FilenameUtils.getExtension;
+import static org.apache.commons.io.file.PathUtils.deleteDirectory;
+
+import java.io.IOException;
+import java.nio.file.FileAlreadyExistsException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Collection;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.hadoop.hdds.utils.db.TablePrefixInfo;
+import org.apache.hadoop.ozone.om.OMMetadataManager;
+import org.apache.hadoop.ozone.om.OmSnapshot;
+import org.apache.hadoop.ozone.om.OmSnapshotManager;
+import org.apache.hadoop.ozone.om.helpers.SnapshotInfo;
+import org.apache.hadoop.ozone.om.snapshot.OmSnapshotLocalDataManager.ReadableOmSnapshotLocalDataProvider;
+import org.apache.hadoop.ozone.snapshot.SnapshotDiffResponse.SubStatus;
+import org.apache.ozone.rocksdb.util.SstFileInfo;
+import org.apache.ratis.util.function.UncheckedAutoCloseableSupplier;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * The {@code FileLinkDeltaFileComputer} is an abstract class that provides a
+ * base implementation for the {@code DeltaFileComputer} interface. It is
+ * responsible for computing delta files by creating hard links to the
+ * relevant source files in a specified delta directory, enabling a compact
+ * representation of changes between snapshots.
+ *
+ * This class encapsulates the logic for managing snapshots and metadata,
+ * creating hard links for delta representation, and reporting activity
+ * during the computation process.
+ */
+public abstract class FileLinkDeltaFileComputer implements DeltaFileComputer {
+
+  private static final Logger LOG = LoggerFactory.getLogger(FileLinkDeltaFileComputer.class);
+  private final OmSnapshotManager omSnapshotManager;
+  private final OMMetadataManager activeMetadataManager;
+  private final Consumer<SubStatus> activityReporter;
+  private Path deltaDir;
+  private AtomicInteger linkFileCounter = new AtomicInteger(0);
+
+  FileLinkDeltaFileComputer(OmSnapshotManager snapshotManager, OMMetadataManager activeMetadataManager,
+      Path deltaDirPath, Consumer<SubStatus> activityReporter) throws IOException {
+    this.deltaDir = deltaDirPath.toAbsolutePath();
+    this.omSnapshotManager = snapshotManager;
+    this.activityReporter = activityReporter;
+    this.activeMetadataManager = activeMetadataManager;
+    createDirectories(deltaDir);
+  }
+
+  /**
+   * Computes the delta files between two snapshots based on the provided parameters.
+   * The method determines the differences in data between the `fromSnapshot` and
+   * `toSnapshot` and generates a mapping of paths to pairs consisting of a resolved
+   * path and corresponding SST file information.
+   *
+   * @param fromSnapshot the source snapshot from which changes are calculated
+   * @param toSnapshot the target snapshot up to which changes are calculated
+   * @param tablesToLookup a set of table names to filter the tables that should be considered
+   * @param tablePrefixInfo information about table prefixes to apply during computation
+   * @return an Optional containing a map where the key is the delta file path, and the value
+   *         is a pair consisting of a resolved path and the corresponding SST file information.
+   *         Return empty if the delta files could not be computed.
+   * @throws IOException if an I/O error occurs during the computation process
+   */
+  abstract Optional<Map<Path, Pair<Path, SstFileInfo>>> computeDeltaFiles(SnapshotInfo fromSnapshot,
+      SnapshotInfo toSnapshot, Set<String> tablesToLookup, TablePrefixInfo tablePrefixInfo) throws IOException;
+
+  @Override
+  public final Collection<Pair<Path, SstFileInfo>> getDeltaFiles(SnapshotInfo fromSnapshot,
+      SnapshotInfo toSnapshot, Set<String> tablesToLookup) throws IOException {
+    TablePrefixInfo tablePrefixInfo = activeMetadataManager.getTableBucketPrefix(fromSnapshot.getVolumeName(),
+        fromSnapshot.getBucketName());
+    return computeDeltaFiles(fromSnapshot, toSnapshot, tablesToLookup,
+        tablePrefixInfo).map(Map::values).orElseThrow(() -> new IOException(String.format(
+            "Failed to compute delta files for snapshots %s and %s tablesToLookup : %s", fromSnapshot, toSnapshot,
+        tablesToLookup)));
+  }
+
+  void updateActivity(SubStatus status) {
+    activityReporter.accept(status);
+  }
+
+  Path createLink(Path path) throws IOException {
+    Path source = path.toAbsolutePath();
+    Path link;
+    boolean createdLink = false;
+    Path fileName = source.getFileName();
+    if (source.getFileName() == null) {
+      throw new IOException("Unable to create link for path " + source + " since it has no file name");
+    }
+    String extension = getExtension(fileName.toString());
+    extension = StringUtils.isBlank(extension) ? "" : ("." + extension);
+    do {
+      link = deltaDir.resolve(linkFileCounter.incrementAndGet() + extension);
+      try {
+        Files.createLink(link, source);
+        createdLink = true;
+      }  catch (FileAlreadyExistsException ignored) {
+        LOG.info("File for source {} already exists: at {}. Will attempt to create link with a different path", source,
+            link);
+      }
+    } while (!createdLink);
+    return link;
+  }
+
+  ReadableOmSnapshotLocalDataProvider getLocalDataProvider(UUID snapshotId, UUID toResolveSnapshotId)
+      throws IOException {
+    return omSnapshotManager.getSnapshotLocalDataManager().getOmSnapshotLocalData(snapshotId, toResolveSnapshotId);
+  }
+
+  UncheckedAutoCloseableSupplier<OmSnapshot> getSnapshot(SnapshotInfo snapshotInfo) throws IOException {
+    return omSnapshotManager.getActiveSnapshot(snapshotInfo.getVolumeName(), snapshotInfo.getBucketName(),
+        snapshotInfo.getName());
+  }
+
+  OMMetadataManager getActiveMetadataManager() {
+    return activeMetadataManager;
+  }
+
+  @Override
+  public void close() throws IOException {
+    if (deltaDir == null || Files.notExists(deltaDir)) {
+      return;
+    }
+    deleteDirectory(deltaDir);
+  }
+}