Copilot commented on code in PR #11428:
URL: https://github.com/apache/ozone/pull/11428#discussion_r4266550249
##########
hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/SnapshotDiffManager.java:
##########
@@ -165,13 +166,14 @@ public class SnapshotDiffManager implements
AutoCloseable, SnapshotDiffManagerMX
private static final Logger LOG =
LoggerFactory.getLogger(SnapshotDiffManager.class);
private static final Map<DiffType, String> DIFF_TYPE_STRING_MAP =
- new EnumMap<>(ImmutableMap.of(DELETE, "1", RENAME, "2", CREATE, "3",
MODIFY, "4"));
+ new EnumMap<>(ImmutableMap.of(DELETE, "1", MODIFY, "2", RENAME, "3",
CREATE, "4"));
Review Comment:
Changing `DIFF_TYPE_STRING_MAP` redefines the persisted report-key prefixes
for diff types. Existing DONE jobs written with the old prefix mapping may be
misinterpreted during pagination/token parsing after upgrade (e.g., old '2'
keys that meant RENAME now map to MODIFY). Consider preserving the legacy
mapping for backward compatibility, or explicitly versioning the report-key
layout and supporting both mappings when reading.
##########
hadoop-hdds/rocks-native/src/main/java/org/apache/hadoop/hdds/utils/db/LatestVersionedKWayMergeIterator.java:
##########
@@ -59,41 +60,42 @@ public final class LatestVersionedKWayMergeIterator
implements
private List<MergedKeyValue> emitQueue;
private boolean initialized;
- public static LatestVersionedKWayMergeIterator
overRawSstFiles(Collection<Path> sstFiles) {
- return overRawSstFiles(sstFiles, DEFAULT_READ_AHEAD_SIZE, null);
- }
-
- public static LatestVersionedKWayMergeIterator
overRawSstFiles(Collection<Path> sstFiles,
- int readAheadSizePerFile) {
- return overRawSstFiles(sstFiles, readAheadSizePerFile, null);
- }
-
/**
- * Opens one iterator per SST file and merges them.
+ * Opens one iterator per SST file and merges them within an optional
user-key range.
*
* @param exclusiveMinSequenceNumber when non-null, each source skips
entries with
* sequence {@code <=} this value while advancing; when null, no
entries are skipped
+ * @param fromUserKeyInclusive when non-null and non-empty, each source
seeks to this lower bound
+ * @param toUserKeyExclusive when non-null and non-empty, each source stops
before this upper bound
+ * @apiNote {@code fromUserKeyInclusive} and {@code toUserKeyExclusive} are
wrapped by native
+ * {@link org.apache.hadoop.hdds.utils.db.managed.ManagedSlice}
objects without copying.
+ * Callers must not mutate either array until the returned iterator
is closed.
*/
public static LatestVersionedKWayMergeIterator
overRawSstFiles(Collection<Path> sstFiles,
- int readAheadSizePerFile, Long exclusiveMinSequenceNumber) {
+ int readAheadSizePerFile, Long exclusiveMinSequenceNumber,
+ byte[] fromUserKeyInclusive, byte[] toUserKeyExclusive) {
Objects.requireNonNull(sstFiles, "sstFiles cannot be null");
- ManagedOptions options = new ManagedOptions();
List<ClosableIterator<? extends MergeHead>> sources = new
ArrayList<>(sstFiles.size());
- try {
+ try (ManagedOptions options = new ManagedOptions()) {
for (Path file : sstFiles) {
- sources.add(new RawSstIterator(options, file, readAheadSizePerFile));
+ sources.add(new RawSstIterator(options, file, readAheadSizePerFile,
fromUserKeyInclusive,
+ toUserKeyExclusive));
}
return new LatestVersionedKWayMergeIterator(options, sources,
exclusiveMinSequenceNumber);
} catch (RuntimeException e) {
IOUtils.closeQuietly(sources);
- options.close();
throw e;
}
}
Review Comment:
The `ManagedOptions` is created in a try-with-resources but then passed into
`LatestVersionedKWayMergeIterator` and returned; the try-with-resources will
close `options` before the caller uses the iterator, leaving the iterator
holding a closed native resource. Allocate `ManagedOptions` without
try-with-resources here and let `LatestVersionedKWayMergeIterator.close()`
own/close it (and still close it in the exception path).
##########
hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/diff/SnapDiffJobStore.java:
##########
@@ -39,118 +59,174 @@
* Owns the per-job temporary RocksDB column families and batched writes shared
* between optimized snapshot diff pipeline stages.
*
- * <p>Diff-candidate {@code objectId}s are held in memory while their count is
at
- * most {@code maxInMemoryEntries}; larger sets spill to a temporary column
family.
+ * <p>Diff-candidate {@code objectId}s and DAG directory tombstone table keys
are held in
+ * memory while their count is at most {@code maxInMemoryEntries}; larger sets
spill to a
+ * temporary column family. Deleted and renamed directory object ids collected
during merge
+ * join use the same in-memory limit and spill policy.
*
* <p>All RocksDB puts use raw {@code byte[]} keys and values (the JNI
boundary). Callers
* serialize {@link EntryValue} via {@link EntryValue#toBytes()} before
writing to
* {@code newList}/{@code oldList}.
*
- * <p>This initial version supports the full diff sequential reader ({@link
FullDiffSequentialReader}).
- * DAG diff support extends this store in HDDS-15393.
+ * <p>Temporary column families disable RocksDB auto-compactions and are
dropped once a pipeline
+ * stage no longer reads them. Individual key deletes are avoided so
short-lived CFs do not
+ * accumulate tombstones before drop.
+ *
+ * <p>Shared by the full diff sequential reader ({@link
FullDiffSequentialReader}) and the
+ * DAG diff sequential reader ({@link DagDiffSequentialReader}).
*/
public final class SnapDiffJobStore implements AutoCloseable {
/** Default RocksDB {@code WriteBatch} commit size for job-store puts. */
- public static final int DEFAULT_WRITE_BATCH_SIZE = 1000;
+ public static final int DEFAULT_BATCH_SIZE = 1000;
+
+ /** Reverse edge value header: parentId (8 BE) + nameLen (4 BE). */
+ static final int EDGE_LINK_HEADER_BYTES = Long.BYTES + Integer.BYTES;
+
+ private static final Comparator<byte[]> TOMBSTONE_KEY_COMPARATOR =
+ UnsignedBytes.lexicographicalComparator();
private static final String NEW_LIST_SUFFIX = "-new-list";
private static final String OLD_LIST_SUFFIX = "-old-list";
private static final String CAND_IDS_SUFFIX = "-cand-ids";
+ private static final String DIR_TOMBSTONE_KEYS_SUFFIX =
"-dir-tombstone-keys";
private static final String TO_EDGES_SUFFIX = "-to-edges";
private static final String FROM_EDGES_SUFFIX = "-from-edges";
+ private static final String CLASSIFIED_CREATE_SUFFIX = "-cls-create";
+ private static final String CLASSIFIED_DELETE_SUFFIX = "-cls-delete";
+ private static final String CLASSIFIED_MODIFY_SUFFIX = "-cls-modify";
+ private static final String CLASSIFIED_RENAME_SUFFIX = "-cls-rename";
+ private static final String DEPENDENCY_NODE_SUFFIX = "-dependency-nodes";
+ private static final String DELETED_DIR_IDS_SUFFIX = "-deleted-dir-ids";
+ private static final String RENAMED_DIR_IDS_SUFFIX = "-renamed-dir-ids";
private final ManagedRocksDB db;
+ private final String jobId;
private final boolean fso;
+ private final CodecRegistry codecRegistry;
+ private final ColumnFamilyHandle reportCfh;
private final byte[] presentMarker;
private final ManagedColumnFamilyOptions familyOptions;
private final long maxInMemoryEntries;
+ private final List<ColumnFamilyHandle> temporaryColumnFamilies = new
ArrayList<>();
private ColumnFamilyHandle newListCf;
private ColumnFamilyHandle oldListCf;
private ColumnFamilyHandle toEdgesCf;
private ColumnFamilyHandle fromEdgesCf;
+ private ColumnFamilyHandle classifiedCreateCf;
+ private ColumnFamilyHandle classifiedDeleteCf;
+ private ColumnFamilyHandle classifiedModifyCf;
+ private ColumnFamilyHandle classifiedRenameCf;
+ private ColumnFamilyHandle dependencyNodesCf;
private String diffCandCfName;
private Set<Long> diffCandidates;
private ColumnFamilyHandle diffCandidatesCf;
private boolean diffCandidatesSpilled;
+ private String dirTombstoneKeysCfName;
+ private NavigableSet<byte[]> dirTombstoneKeys;
+ private ColumnFamilyHandle dirTombstoneKeysCf;
+ private boolean dirTombstoneKeysSpilled;
+
+ private Set<Long> deletedDirectoryIds;
+ private Set<Long> renamedDirectoryIds;
+ private String deletedDirectoryIdsCfName;
+ private String renamedDirectoryIdsCfName;
+ private ColumnFamilyHandle deletedDirectoryIdsCf;
+ private ColumnFamilyHandle renamedDirectoryIdsCf;
+ private boolean deletedDirectoryIdsSpilled;
+ private boolean renamedDirectoryIdsSpilled;
+
private final ManagedWriteBatch writeBatch;
private final ManagedWriteOptions writeOptions;
private final int writeBatchSize;
private int pendingOps;
+ private long reportIndex;
+ private String largestReportKey;
+ private boolean reportWriteStarted;
+
/** Reusable big-endian key buffers; safe because RocksDB copies keys on
put/get. */
private final byte[] objectIdKeyBuffer = new byte[Long.BYTES];
- private final byte[] edgeKeyBuffer = new byte[2 * Long.BYTES];
+ private final byte[] intKeyBuffer = new byte[Integer.BYTES];
/** Full diff: shared new/old lists plus FSO edge column families. */
public enum Mode {
- FULL
+ FULL,
+ DAG
}
- private SnapDiffJobStore(ManagedRocksDB db, CodecRegistry codecRegistry,
boolean fso,
- int writeBatchSize, ManagedColumnFamilyOptions familyOptions, long
maxInMemoryEntries)
- throws IOException {
+ @SuppressWarnings("checkstyle:ParameterNumber")
+ private SnapDiffJobStore(ManagedRocksDB db, String jobId, CodecRegistry
codecRegistry, boolean fso,
+ int writeBatchSize, ManagedColumnFamilyOptions familyOptions, long
maxInMemoryEntries,
+ @Nonnull ColumnFamilyHandle reportCfh) throws IOException {
this.db = db;
+ this.jobId = jobId;
this.fso = fso;
+ this.codecRegistry = codecRegistry;
+ this.reportCfh = reportCfh;
this.writeBatchSize = writeBatchSize;
this.familyOptions = familyOptions;
+ this.familyOptions.setDisableAutoCompactions(true);
Review Comment:
`SnapDiffJobStore` mutates the passed-in `ManagedColumnFamilyOptions`
instance by disabling auto-compactions. If the same options object is
shared/reused elsewhere (including for non-temporary CFs), this can
unintentionally affect RocksDB behavior beyond the job-store CFs. Consider
using a dedicated options instance for temporary CFs (copy/clone if supported)
instead of mutating the caller-provided one.
##########
hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/diff/SnapDiffJobStore.java:
##########
@@ -39,118 +59,174 @@
* Owns the per-job temporary RocksDB column families and batched writes shared
* between optimized snapshot diff pipeline stages.
*
- * <p>Diff-candidate {@code objectId}s are held in memory while their count is
at
- * most {@code maxInMemoryEntries}; larger sets spill to a temporary column
family.
+ * <p>Diff-candidate {@code objectId}s and DAG directory tombstone table keys
are held in
+ * memory while their count is at most {@code maxInMemoryEntries}; larger sets
spill to a
+ * temporary column family. Deleted and renamed directory object ids collected
during merge
+ * join use the same in-memory limit and spill policy.
*
* <p>All RocksDB puts use raw {@code byte[]} keys and values (the JNI
boundary). Callers
* serialize {@link EntryValue} via {@link EntryValue#toBytes()} before
writing to
* {@code newList}/{@code oldList}.
*
- * <p>This initial version supports the full diff sequential reader ({@link
FullDiffSequentialReader}).
- * DAG diff support extends this store in HDDS-15393.
+ * <p>Temporary column families disable RocksDB auto-compactions and are
dropped once a pipeline
+ * stage no longer reads them. Individual key deletes are avoided so
short-lived CFs do not
+ * accumulate tombstones before drop.
+ *
+ * <p>Shared by the full diff sequential reader ({@link
FullDiffSequentialReader}) and the
+ * DAG diff sequential reader ({@link DagDiffSequentialReader}).
*/
public final class SnapDiffJobStore implements AutoCloseable {
/** Default RocksDB {@code WriteBatch} commit size for job-store puts. */
- public static final int DEFAULT_WRITE_BATCH_SIZE = 1000;
+ public static final int DEFAULT_BATCH_SIZE = 1000;
+
+ /** Reverse edge value header: parentId (8 BE) + nameLen (4 BE). */
+ static final int EDGE_LINK_HEADER_BYTES = Long.BYTES + Integer.BYTES;
+
+ private static final Comparator<byte[]> TOMBSTONE_KEY_COMPARATOR =
+ UnsignedBytes.lexicographicalComparator();
private static final String NEW_LIST_SUFFIX = "-new-list";
private static final String OLD_LIST_SUFFIX = "-old-list";
private static final String CAND_IDS_SUFFIX = "-cand-ids";
+ private static final String DIR_TOMBSTONE_KEYS_SUFFIX =
"-dir-tombstone-keys";
private static final String TO_EDGES_SUFFIX = "-to-edges";
private static final String FROM_EDGES_SUFFIX = "-from-edges";
+ private static final String CLASSIFIED_CREATE_SUFFIX = "-cls-create";
+ private static final String CLASSIFIED_DELETE_SUFFIX = "-cls-delete";
+ private static final String CLASSIFIED_MODIFY_SUFFIX = "-cls-modify";
+ private static final String CLASSIFIED_RENAME_SUFFIX = "-cls-rename";
+ private static final String DEPENDENCY_NODE_SUFFIX = "-dependency-nodes";
+ private static final String DELETED_DIR_IDS_SUFFIX = "-deleted-dir-ids";
+ private static final String RENAMED_DIR_IDS_SUFFIX = "-renamed-dir-ids";
private final ManagedRocksDB db;
+ private final String jobId;
private final boolean fso;
+ private final CodecRegistry codecRegistry;
+ private final ColumnFamilyHandle reportCfh;
private final byte[] presentMarker;
private final ManagedColumnFamilyOptions familyOptions;
private final long maxInMemoryEntries;
+ private final List<ColumnFamilyHandle> temporaryColumnFamilies = new
ArrayList<>();
private ColumnFamilyHandle newListCf;
private ColumnFamilyHandle oldListCf;
private ColumnFamilyHandle toEdgesCf;
private ColumnFamilyHandle fromEdgesCf;
+ private ColumnFamilyHandle classifiedCreateCf;
+ private ColumnFamilyHandle classifiedDeleteCf;
+ private ColumnFamilyHandle classifiedModifyCf;
+ private ColumnFamilyHandle classifiedRenameCf;
+ private ColumnFamilyHandle dependencyNodesCf;
private String diffCandCfName;
private Set<Long> diffCandidates;
private ColumnFamilyHandle diffCandidatesCf;
private boolean diffCandidatesSpilled;
+ private String dirTombstoneKeysCfName;
+ private NavigableSet<byte[]> dirTombstoneKeys;
+ private ColumnFamilyHandle dirTombstoneKeysCf;
+ private boolean dirTombstoneKeysSpilled;
+
+ private Set<Long> deletedDirectoryIds;
+ private Set<Long> renamedDirectoryIds;
+ private String deletedDirectoryIdsCfName;
+ private String renamedDirectoryIdsCfName;
+ private ColumnFamilyHandle deletedDirectoryIdsCf;
+ private ColumnFamilyHandle renamedDirectoryIdsCf;
+ private boolean deletedDirectoryIdsSpilled;
+ private boolean renamedDirectoryIdsSpilled;
+
private final ManagedWriteBatch writeBatch;
private final ManagedWriteOptions writeOptions;
private final int writeBatchSize;
private int pendingOps;
+ private long reportIndex;
+ private String largestReportKey;
+ private boolean reportWriteStarted;
+
/** Reusable big-endian key buffers; safe because RocksDB copies keys on
put/get. */
private final byte[] objectIdKeyBuffer = new byte[Long.BYTES];
- private final byte[] edgeKeyBuffer = new byte[2 * Long.BYTES];
+ private final byte[] intKeyBuffer = new byte[Integer.BYTES];
/** Full diff: shared new/old lists plus FSO edge column families. */
public enum Mode {
- FULL
+ FULL,
+ DAG
}
- private SnapDiffJobStore(ManagedRocksDB db, CodecRegistry codecRegistry,
boolean fso,
- int writeBatchSize, ManagedColumnFamilyOptions familyOptions, long
maxInMemoryEntries)
- throws IOException {
+ @SuppressWarnings("checkstyle:ParameterNumber")
+ private SnapDiffJobStore(ManagedRocksDB db, String jobId, CodecRegistry
codecRegistry, boolean fso,
+ int writeBatchSize, ManagedColumnFamilyOptions familyOptions, long
maxInMemoryEntries,
+ @Nonnull ColumnFamilyHandle reportCfh) throws IOException {
this.db = db;
+ this.jobId = jobId;
this.fso = fso;
+ this.codecRegistry = codecRegistry;
+ this.reportCfh = reportCfh;
this.writeBatchSize = writeBatchSize;
this.familyOptions = familyOptions;
+ this.familyOptions.setDisableAutoCompactions(true);
this.maxInMemoryEntries = maxInMemoryEntries;
this.presentMarker = codecRegistry.asRawData(Boolean.TRUE);
this.writeBatch = new ManagedWriteBatch();
this.writeOptions = new ManagedWriteOptions();
this.pendingOps = 0;
this.diffCandidates = new HashSet<>();
+ this.dirTombstoneKeys = new TreeSet<>(TOMBSTONE_KEY_COMPARATOR);
+ this.deletedDirectoryIds = new HashSet<>();
+ this.renamedDirectoryIds = new HashSet<>();
}
+ @SuppressWarnings("checkstyle:ParameterNumber")
public static SnapDiffJobStore open(@Nonnull ManagedRocksDB db,
@Nonnull CodecRegistry codecRegistry,
@Nonnull ManagedColumnFamilyOptions familyOptions,
@Nonnull String jobId,
boolean fso,
- @Nonnull Mode mode) throws IOException {
- return open(db, codecRegistry, familyOptions, jobId, fso, mode,
DEFAULT_WRITE_BATCH_SIZE,
- OZONE_OM_SNAPSHOT_DIFF_MAX_IN_MEMORY_ENTRIES_PER_JOB_DEFAULT);
- }
-
- public static SnapDiffJobStore open(@Nonnull ManagedRocksDB db,
- @Nonnull CodecRegistry codecRegistry,
- @Nonnull ManagedColumnFamilyOptions familyOptions,
- @Nonnull String jobId,
- boolean fso,
- @Nonnull Mode mode,
- int writeBatchSize) throws IOException {
- return open(db, codecRegistry, familyOptions, jobId, fso, mode,
writeBatchSize,
- OZONE_OM_SNAPSHOT_DIFF_MAX_IN_MEMORY_ENTRIES_PER_JOB_DEFAULT);
- }
-
- @SuppressWarnings("parameternumber")
- public static SnapDiffJobStore open(@Nonnull ManagedRocksDB db,
- @Nonnull CodecRegistry codecRegistry,
- @Nonnull ManagedColumnFamilyOptions familyOptions,
- @Nonnull String jobId,
- boolean fso,
- @Nonnull Mode mode,
- int writeBatchSize,
- long maxInMemoryEntries) throws IOException {
- if (mode != Mode.FULL) {
- throw new IllegalArgumentException("Unsupported mode: " + mode);
- }
- SnapDiffJobStore store = new SnapDiffJobStore(db, codecRegistry, fso,
writeBatchSize,
- familyOptions, maxInMemoryEntries);
+ @Nonnull ColumnFamilyHandle reportCfh,
+ @Nullable Integer writeBatchSize,
+ @Nullable Long maxInMemoryEntries) throws IOException {
+ int resolvedWriteBatchSize = writeBatchSize != null ? writeBatchSize :
DEFAULT_BATCH_SIZE;
+ long resolvedMaxInMemoryEntries = maxInMemoryEntries != null ?
maxInMemoryEntries
+ : OZONE_OM_SNAPSHOT_DIFF_MAX_IN_MEMORY_ENTRIES_PER_JOB_DEFAULT;
+ SnapDiffJobStore store = new SnapDiffJobStore(db, jobId, codecRegistry,
fso, resolvedWriteBatchSize,
+ familyOptions, resolvedMaxInMemoryEntries, reportCfh);
try {
- store.initColumnFamilies(familyOptions, jobId);
+ store.initColumnFamilies(familyOptions);
return store;
} catch (RocksDBException e) {
store.closeQuietly();
throw new IOException("Failed to open SnapDiff job store for job " +
jobId, e);
}
}
- public boolean isFso() {
- return fso;
+ long getMaxInMemoryEntries() {
+ return maxInMemoryEntries;
+ }
+
+ int getWriteBatchSize() {
+ return writeBatchSize;
+ }
+
+ /** Returns report entries written during a test run. */
+ List<DiffReportEntry> readReportEntriesForTest() throws IOException {
+ List<DiffReportEntry> entries = new ArrayList<>();
+ try (ManagedRocksIterator iterator = new
ManagedRocksIterator(db.get().newIterator(reportCfh))) {
+ iterator.get().seek(codecRegistry.asRawData(jobId));
+ while (iterator.get().isValid()) {
+ String key = codecRegistry.asObject(iterator.get().key(),
String.class);
+ if (!key.startsWith(jobId)) {
+ break;
+ }
+ entries.add(codecRegistry.asObject(iterator.get().value(),
DiffReportEntry.class));
+ iterator.get().next();
+ }
+ }
+ return entries;
}
Review Comment:
This test helper seeks to `jobId` (without the delimiter) and filters with
`key.startsWith(jobId)`, which can accidentally include rows from other jobs
like `job10` when the current jobId is `job1` (lexicographic ordering places
`job10|...` between `job1` and `job1|...`). Seek to `jobId + DELIMITER` and
guard with `startsWith(jobId + DELIMITER)` to avoid cross-job
contamination/flaky tests.
##########
hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/diff/SnapDiffJobStore.java:
##########
@@ -185,13 +263,153 @@ public byte[] getOldList(long objectId) throws
IOException {
return get(oldListCf, objectIdKeyBuffer(objectId));
}
- public boolean hasNewListEntry(long objectId) throws IOException {
- return getNewList(objectId) != null;
+ public String getJobId() {
+ return jobId;
+ }
+
+ /** Returns whether the stored new-list value is a present-marker
(unchanged). */
+ public boolean isPresentMarker(byte[] value) {
+ return value != null && Arrays.equals(value, presentMarker);
+ }
+
+ /** Ascending merge-join iterator over {@code newList}. */
+ public ListIterator newListIterator() throws RocksDatabaseException {
+ return new ListIterator(newListCf);
+ }
+
+ /** Ascending merge-join iterator over {@code oldList}. */
+ public ListIterator oldListIterator() throws RocksDatabaseException {
+ return new ListIterator(oldListCf);
+ }
+
+ /** Builds a from-snapshot path resolver over the reverse edge index. */
+ SnapDiffPathResolver newFromPathResolver(long bucketObjectId) {
+ requireFso();
+ return new SnapDiffPathResolver(db, fromEdgesCf, bucketObjectId,
pathResolverLruCapacity());
+ }
+
+ /** Builds a to-snapshot path resolver over the reverse edge index. */
+ SnapDiffPathResolver newToPathResolver(long bucketObjectId) {
+ requireFso();
+ return new SnapDiffPathResolver(db, toEdgesCf, bucketObjectId,
pathResolverLruCapacity());
+ }
+
+ private int pathResolverLruCapacity() {
+ return (int) maxInMemoryEntries / 2;
Review Comment:
Casting `maxInMemoryEntries` (long) to int before dividing can overflow for
large configured values, producing a negative/incorrect LRU capacity. Clamp to
`Integer.MAX_VALUE` (or perform division in long first, then safely cast) to
ensure a valid capacity.
##########
hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/SnapshotDiffManager.java:
##########
@@ -1399,97 +1403,118 @@ private String resolveBucketRelativePath(boolean
isFSOBucket,
* chain cannot be resolved in the live from-snapshot directory graph, so the
* delete entry can be skipped from the final report.
*
- * @param objectIdToParentId from-snapshot directory parent graph keyed by
stable
- * objectId; must not be the to-snapshot graph
- * @param renamedDirectoryIds objectIds of directories renamed in the diff
(from
- * snapshot objectIds, not destination paths)
+ * @param deletedDirectoryIds membership test for deleted directory object
ids
+ * @param renamedDirectoryIds membership test for renamed directory object
ids
+ * (from snapshot objectIds, not destination paths)
+ * @param ancestorMemo LRU cache of ancestor object ids to deleted-ancestor
results
*/
@VisibleForTesting
- boolean hasDeletedAncestor(long parentObjectId, Set<Long>
deletedDirectoryIds,
- Set<Long> renamedDirectoryIds, Map<Long, Long> objectIdToParentId,
- long bucketObjectId, Map<Long, Boolean> ancestorMemo) {
- Objects.requireNonNull(objectIdToParentId, "objectIdToParentId must not be
null");
+ public boolean[] hasDeletedAncestors(List<Long> parentObjectIds,
+ DirectoryIdCheck deletedDirectoryIds, DirectoryIdCheck
renamedDirectoryIds,
+ ParentIdBatchLookup parentLookup, long bucketObjectId,
+ Map<Long, Boolean> ancestorMemo) throws IOException {
+ Objects.requireNonNull(parentObjectIds, "parentObjectIds must not be
null");
+ Objects.requireNonNull(deletedDirectoryIds, "deletedDirectoryIds must not
be null");
+ Objects.requireNonNull(parentLookup, "parentLookup must not be null");
Objects.requireNonNull(renamedDirectoryIds, "renamedDirectoryIds must not
be null");
-
- if (parentObjectId == bucketObjectId) {
- return false;
- }
- Boolean cached = ancestorMemo.get(parentObjectId);
- if (cached != null) {
- return cached;
+ Objects.requireNonNull(ancestorMemo, "ancestorMemo must not be null");
+
+ int batchSize = parentObjectIds.size();
+ boolean[] hasDeletedAncestor = new boolean[batchSize];
+ long[] currentAncestors = new long[batchSize];
+ boolean[] resolved = new boolean[batchSize];
+ List<List<Long>> ancestorPaths = new ArrayList<>(batchSize);
+ for (int i = 0; i < batchSize; i++) {
+ ancestorPaths.add(new ArrayList<>());
+ currentAncestors[i] = parentObjectIds.get(i);
}
- List<Long> path = new ArrayList<>();
- long current = parentObjectId;
- boolean result;
while (true) {
- if (current == bucketObjectId) {
- result = false;
- break;
- }
- cached = ancestorMemo.get(current);
- if (cached != null) {
- result = cached;
- break;
+ Map<Long, List<Integer>> frontier = new HashMap<>();
+ boolean hasUnresolved = false;
+ for (int i = 0; i < batchSize; i++) {
+ if (resolved[i]) {
+ continue;
+ }
+ hasUnresolved = true;
+ long current = currentAncestors[i];
+ if (current == bucketObjectId) {
+ resolved[i] = true;
+ continue;
+ }
+ Boolean cached = ancestorMemo.get(current);
+ if (cached != null) {
+ resolved[i] = true;
+ hasDeletedAncestor[i] = cached;
+ continue;
+ }
+ if (renamedDirectoryIds.contains(current)) {
+ resolved[i] = true;
+ ancestorMemo.put(current, false);
+ continue;
+ }
+ if (deletedDirectoryIds.contains(current)) {
+ resolved[i] = true;
+ ancestorMemo.put(current, true);
+ hasDeletedAncestor[i] = true;
+ continue;
+ }
+ frontier.computeIfAbsent(current, ignored -> new ArrayList<>()).add(i);
}
- if (renamedDirectoryIds.contains(current)) {
- result = false;
- ancestorMemo.put(current, false);
+
+ if (!hasUnresolved || frontier.isEmpty()) {
break;
}
- if (deletedDirectoryIds.contains(current)) {
- result = true;
- ancestorMemo.put(current, true);
- break;
+
+ List<Long> lookupObjectIds = new ArrayList<>(frontier.keySet());
+ List<Long> nextParents = parentLookup.getParentIds(lookupObjectIds);
+ Objects.requireNonNull(nextParents, "nextParents must not be null");
+ if (nextParents.size() != lookupObjectIds.size()) {
+ throw new IOException("Batch parent lookup returned " +
nextParents.size()
+ + " values for " + lookupObjectIds.size() + " objectIds");
}
- Long nextParent = objectIdToParentId.get(current);
- if (nextParent == null) {
- result = true;
- ancestorMemo.put(current, true);
- break;
+ for (int i = 0; i < lookupObjectIds.size(); i++) {
+ long current = lookupObjectIds.get(i);
+ Long nextParent = nextParents.get(i);
+ List<Integer> positions = frontier.get(current);
+ if (nextParent == null) {
+ ancestorMemo.put(current, true);
+ for (int position : positions) {
+ resolved[position] = true;
+ hasDeletedAncestor[position] = true;
+ }
+ continue;
+ }
+ for (int position : positions) {
+ ancestorPaths.get(position).add(current);
+ currentAncestors[position] = nextParent;
+ }
}
- path.add(current);
- current = nextParent;
}
- for (Long node : path) {
- ancestorMemo.put(node, result);
+
+ for (int i = 0; i < batchSize; i++) {
+ // if hasDeletedAncestor == true only ??
+ for (Long ancestor : ancestorPaths.get(i)) {
+ ancestorMemo.put(ancestor, hasDeletedAncestor[i]);
+ }
}
Review Comment:
The inline comment `// if hasDeletedAncestor == true only ??` reads like an
unresolved question and makes the intended behavior unclear. Please remove it
or replace it with a clear explanation of why all visited ancestors are
memoized with the final result for that position.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
