diff --git a/hadoop-hdds/rocksdb-checkpoint-differ/src/main/java/org/apache/ozone/rocksdb/util/RdbUtil.java b/hadoop-hdds/rocksdb-checkpoint-differ/src/main/java/org/apache/ozone/rocksdb/util/RdbUtil.java index 48a8ee696c85..97eaa945fdce 100644 --- a/hadoop-hdds/rocksdb-checkpoint-differ/src/main/java/org/apache/ozone/rocksdb/util/RdbUtil.java +++ b/hadoop-hdds/rocksdb-checkpoint-differ/src/main/java/org/apache/ozone/rocksdb/util/RdbUtil.java @@ -19,8 +19,15 @@ import com.google.common.collect.Sets; import java.io.File; +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; +import java.util.Map; import java.util.Set; import java.util.stream.Collectors; import org.apache.hadoop.hdds.StringUtils; @@ -49,4 +56,16 @@ public static Set getSSTFilesForComparison( .map(lfm -> new File(lfm.path(), lfm.fileName()).getPath()) .collect(Collectors.toCollection(HashSet::new)); } + + public static Map getSSTFilesWithInodesForComparison(final ManagedRocksDB rocksDB, List cfs) + throws IOException { + List liveSSTFilesForCFs = getLiveSSTFilesForCFs(rocksDB, cfs); + Map inodeToSstMap = new HashMap<>(); + for (LiveFileMetaData lfm : liveSSTFilesForCFs) { + Path sstFilePath = Paths.get(lfm.path(), lfm.fileName()); + Object inode = Files.readAttributes(sstFilePath, BasicFileAttributes.class).fileKey(); + inodeToSstMap.put(inode, sstFilePath.toString()); + } + return inodeToSstMap; + } } diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/SnapshotDiffManager.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/SnapshotDiffManager.java index 511426885d85..75b4196dd86b 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/SnapshotDiffManager.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/SnapshotDiffManager.java @@ -92,6 +92,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; import java.util.function.BiFunction; +import java.util.stream.Collectors; import java.util.stream.Stream; import org.apache.commons.io.file.PathUtils; import org.apache.commons.lang3.tuple.Pair; @@ -375,6 +376,14 @@ protected Set getSSTFileListForSnapshot(OmSnapshot snapshot, tablesToLookUp); } + @VisibleForTesting + protected Map getSSTFileMapForSnapshot(OmSnapshot snapshot, + List tablesToLookUp) throws IOException { + return RdbUtil.getSSTFilesWithInodesForComparison(((RDBStore)snapshot + .getMetadataManager().getStore()).getDb().getManagedRocksDb(), + tablesToLookUp); + } + /** * Gets the report key for a particular index of snapshot diff job. */ @@ -1203,11 +1212,7 @@ Set getDeltaFiles(OmSnapshot fromSnapshot, .getDb().getManagedRocksDb(); ManagedRocksDB toDB = ((RDBStore)toSnapshot.getMetadataManager().getStore()) .getDb().getManagedRocksDb(); - Set fromSnapshotFiles = getSSTFileListForSnapshot(fromSnapshot, tablesToLookUp); - Set toSnapshotFiles = getSSTFileListForSnapshot(toSnapshot, tablesToLookUp); - Set diffFiles = new HashSet<>(); - diffFiles.addAll(fromSnapshotFiles); - diffFiles.addAll(toSnapshotFiles); + Set diffFiles = getDiffFiles(fromSnapshot, toSnapshot, tablesToLookUp); RocksDiffUtils.filterRelevantSstFiles(diffFiles, tablePrefixes, fromDB, toDB); deltaFiles = Optional.of(diffFiles); } @@ -1217,6 +1222,29 @@ Set getDeltaFiles(OmSnapshot fromSnapshot, toSnapshot.getSnapshotTableKey())); } + private Set getDiffFiles(OmSnapshot fromSnapshot, OmSnapshot toSnapshot, List tablesToLookUp) { + Set diffFiles; + try { + Map fromSnapshotFiles = getSSTFileMapForSnapshot(fromSnapshot, tablesToLookUp); + Map toSnapshotFiles = getSSTFileMapForSnapshot(toSnapshot, tablesToLookUp); + diffFiles = Stream.concat( + fromSnapshotFiles.entrySet().stream() + .filter(e -> !toSnapshotFiles.containsKey(e.getKey())), + toSnapshotFiles.entrySet().stream() + .filter(e -> !fromSnapshotFiles.containsKey(e.getKey()))) + .map(Map.Entry::getValue) + .collect(Collectors.toSet()); + } catch (IOException e) { + // In case of exception during inode read use all files + LOG.error("Exception occurred while populating delta files for snapDiff", e); + LOG.warn("Falling back to full file list comparison, inode-based optimization skipped."); + diffFiles = new HashSet<>(); + diffFiles.addAll(getSSTFileListForSnapshot(fromSnapshot, tablesToLookUp)); + diffFiles.addAll(getSSTFileListForSnapshot(toSnapshot, tablesToLookUp)); + } + return diffFiles; + } + private void validateEstimatedKeyChangesAreInLimits( SstFileSetReader sstFileReader ) throws RocksDBException, IOException { diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/snapshot/TestSnapshotDiffManager.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/snapshot/TestSnapshotDiffManager.java index 6cfb21b2feec..b941000d9421 100644 --- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/snapshot/TestSnapshotDiffManager.java +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/snapshot/TestSnapshotDiffManager.java @@ -1536,7 +1536,7 @@ private void setupMocksForRunningASnapDiff( } @Test - public void testGetDeltaFilesWithFullDiff() throws IOException { + public void testGetDeltaFilesWithFullDiff() throws IOException { SnapshotDiffManager spy = spy(snapshotDiffManager); UUID snap1 = UUID.randomUUID(); OmSnapshot fromSnapshot = getMockedOmSnapshot(snap1); @@ -1545,20 +1545,28 @@ public void testGetDeltaFilesWithFullDiff() throws IOException { Mockito.doAnswer(invocation -> { OmSnapshot snapshot = invocation.getArgument(0); if (snapshot == fromSnapshot) { - return Sets.newHashSet("1", "2", "3"); + Map inodeToFileMap = new HashMap<>(); + inodeToFileMap.put(1, "1.sst"); + inodeToFileMap.put(2, "2.sst"); + inodeToFileMap.put(3, "3.sst"); + return inodeToFileMap; } if (snapshot == toSnapshot) { - return Sets.newHashSet("3", "4", "5"); + Map inodeToFileMap = new HashMap<>(); + inodeToFileMap.put(1, "10.sst"); + inodeToFileMap.put(2, "20.sst"); + inodeToFileMap.put(4, "4.sst"); + return inodeToFileMap; } - return Sets.newHashSet("6", "7", "8"); - }).when(spy).getSSTFileListForSnapshot(Mockito.any(OmSnapshot.class), + return null; + }).when(spy).getSSTFileMapForSnapshot(Mockito.any(OmSnapshot.class), Mockito.anyList()); doNothing().when(spy).recordActivity(any(), any()); doNothing().when(spy).updateProgress(anyString(), anyDouble()); String diffJobKey = snap1 + DELIMITER + snap2; Set deltaFiles = spy.getDeltaFiles(fromSnapshot, toSnapshot, Collections.emptyList(), snapshotInfo, snapshotInfo, true, Collections.emptyMap(), null, diffJobKey); - Assertions.assertEquals(Sets.newHashSet("1", "2", "3", "4", "5"), deltaFiles); + Assertions.assertEquals(Sets.newHashSet("3.sst", "4.sst"), deltaFiles); } @Test