smengcl commented on code in PR #11296: URL: https://github.com/apache/ozone/pull/11296#discussion_r4068920969
########## hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/diff/SnapDiffPathResolver.java: ########## @@ -0,0 +1,173 @@ +/* + * 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; + +import static org.apache.hadoop.ozone.OzoneConsts.OM_KEY_PREFIX; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import org.apache.hadoop.hdds.utils.db.managed.ManagedRocksDB; +import org.rocksdb.ColumnFamilyHandle; +import org.rocksdb.RocksDBException; + +/** + * Resolves bucket-relative paths by walking reverse edge links upward from each + * target object id, with an LRU memo for shared ancestors. + */ +final class SnapDiffPathResolver { + + private static final int DEFAULT_LRU_CAPACITY = 4096; + + private final ManagedRocksDB db; + private final ColumnFamilyHandle edgesCf; + private final long bucketObjectId; + private final LinkedHashMap<Long, String> pathCache; + + SnapDiffPathResolver(ManagedRocksDB db, ColumnFamilyHandle edgesCf, long bucketObjectId) { + this.db = db; + this.edgesCf = edgesCf; + this.bucketObjectId = bucketObjectId; + this.pathCache = new LinkedHashMap<Long, String>(DEFAULT_LRU_CAPACITY, 0.75f, true) { + @Override + protected boolean removeEldestEntry(Map.Entry<Long, String> eldest) { + return size() > DEFAULT_LRU_CAPACITY; + } + }; + pathCache.put(bucketObjectId, ""); + } + + String resolvePath(long objectId) throws IOException { + return resolvePath(objectId, null); + } + + List<String> resolvePaths(List<Long> objectIds) throws IOException { + if (objectIds.isEmpty()) { + return Collections.emptyList(); + } + Map<Long, byte[]> prefetchedEdges = prefetchEdgeChains(objectIds); + List<String> paths = new ArrayList<>(objectIds.size()); + for (Long objectId : objectIds) { + paths.add(resolvePath(objectId, prefetchedEdges)); + } + return paths; + } + + private String resolvePath(long objectId, Map<Long, byte[]> prefetchedEdges) throws IOException { + if (objectId == bucketObjectId) { + return ""; + } + String cached = pathCache.get(objectId); + if (cached != null) { + return cached; + } + populatePathCache(objectId, prefetchedEdges); + return pathCache.get(objectId); + } + + /** + * Walks the single-parent chain from {@code objectId} toward the bucket, then + * materializes bucket-relative paths root-to-leaf for every uncached ancestor. + */ + private void populatePathCache(long objectId, Map<Long, byte[]> prefetchedEdges) + throws IOException { + List<Long> chain = new ArrayList<>(); + List<byte[]> linkValues = new ArrayList<>(); + long current = objectId; + while (current != bucketObjectId && !pathCache.containsKey(current)) { + byte[] value = getEdgeValue(current, prefetchedEdges); + if (value == null) { + return; + } + chain.add(current); + linkValues.add(value); + current = SnapDiffJobStore.decodeEdgeLinkParentId(value); + } + String suffix = pathCache.get(current); + if (suffix == null) { + return; Review Comment: The bucket's empty path can be evicted from the 4,096-entry LRU. Once that happens, a walk reaching `bucketObjectId` stops at the loop condition, but this lookup returns `null`, so a valid path is treated as unresolvable and its report entry is dropped. This can be reproduced by resolving 4,100 children under one shared parent in batches, then resolving a separate root-level sibling. The sibling returns `null` despite its edge being present. The bucket root should resolve independently of the cache, e.g. `String suffix = current == bucketObjectId ? "" : pathCache.get(current);`. Please add an eviction regression test. ########## hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/diff/SnapDiffJobStore.java: ########## @@ -262,37 +583,131 @@ public void flushWrites() throws IOException { pendingOps = 0; } + /** + * Starts batched emission of resolved rows to the snap diff report table. + */ + public void beginReportWrite() { + if (reportCfh == null) { + throw new IllegalStateException("Snap diff report column family not configured for job store"); + } + if (reportWriteStarted) { + throw new IllegalStateException("Snap diff report write already started for job " + jobId); + } + reportWriteStarted = true; + reportIndex = 0; + largestReportKey = ""; + } + + /** + * Appends one resolved {@link DiffReportEntry} to the snap diff report table. + */ + public void putReportEntry(DiffReportEntry entry) throws IOException { + if (!reportWriteStarted) { + throw new IllegalStateException("Snap diff report write not started for job " + jobId); + } + appendReportEntry(entry); + } + + /** + * Appends a batch of resolved {@link DiffReportEntry}s to the snap diff report table. + */ + public void putReportEntries(List<DiffReportEntry> entries) throws IOException { + if (!reportWriteStarted) { + throw new IllegalStateException("Snap diff report write not started for job " + jobId); + } + for (DiffReportEntry entry : entries) { + appendReportEntry(entry); + } + } + + /** + * Flushes pending report rows and returns the entry count and largest report key. + */ + public Pair<Long, String> finishReportWrite() throws IOException { + if (!reportWriteStarted) { + throw new IllegalStateException("Snap diff report write not started for job " + jobId); + } + flushWrites(); + reportWriteStarted = false; + return Pair.of(reportIndex, largestReportKey); + } + private byte[] objectIdKeyBuffer(long objectId) { encodeLong(objectIdKeyBuffer, 0, objectId); return objectIdKeyBuffer; } - private byte[] edgeKeyBuffer(long parentId, long objectId) { - encodeLong(edgeKeyBuffer, 0, parentId); - encodeLong(edgeKeyBuffer, Long.BYTES, objectId); - return edgeKeyBuffer; + private byte[] intKeyBuffer(int value) { + encodeInt(intKeyBuffer, 0, value); + return intKeyBuffer; } - private static void encodeLong(byte[] buffer, int offset, long value) { - for (int shift = Long.SIZE - 8; shift >= 0; shift -= 8) { - buffer[offset++] = (byte) (value >>> shift); + private byte[] intValueBuffer(int value) { + byte[] buffer = new byte[Integer.BYTES]; + encodeInt(buffer, 0, value); + return buffer; + } + + private void appendReportEntry(DiffReportEntry entry) throws IOException { + String jobReportKey = getReportKeyForIndex(jobId, entry.getType(), reportIndex++); + batchPut(reportCfh, codecRegistry.asRawData(jobReportKey), codecRegistry.asRawData(entry)); Review Comment: `getReportKeyForIndex` puts the diff-type prefix before the emission index, so RocksDB iteration groups entries by type even after the dependency graph has ordered them. With dependency ordering enabled, consider a new directory `parent` plus a rename `old -> parent/old`. Reading the persisted report returns RENAME before CREATE because the rename prefix sorts first. The destination parent consequently appears after the operation that requires it. For dependency-ordered reports, the report keys need to preserve the computed emission order. Please test the order read back from the report table across different diff types. ########## hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/diff/MergeJoinSnapDiffWriter.java: ########## @@ -0,0 +1,467 @@ +/* + * 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; + +import static org.apache.hadoop.hdfs.protocol.SnapshotDiffReport.DiffType.CREATE; +import static org.apache.hadoop.hdfs.protocol.SnapshotDiffReport.DiffType.DELETE; +import static org.apache.hadoop.hdfs.protocol.SnapshotDiffReport.DiffType.MODIFY; +import static org.apache.hadoop.hdfs.protocol.SnapshotDiffReport.DiffType.RENAME; +import static org.apache.hadoop.ozone.om.snapshot.diff.SnapDiffJobStore.DEFAULT_BATCH_SIZE; +import static org.apache.hadoop.ozone.snapshot.SnapshotDiffReportOzone.getDiffReportEntry; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import org.apache.commons.lang3.tuple.Pair; +import org.apache.hadoop.hdds.utils.db.RocksDatabaseException; +import org.apache.hadoop.hdfs.protocol.SnapshotDiffReport.DiffReportEntry; +import org.apache.hadoop.hdfs.protocol.SnapshotDiffReport.DiffType; +import org.apache.hadoop.ozone.om.snapshot.SnapshotDiffManager; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Stages 2–4 of the optimized full snapshot diff pipeline: merge join and + * classification, top-level delete retention, ancestor backtrack path resolution, + * dependency ordering, and batched report write. + * + * <p>Classified rows are persisted in per-type column families as minimal parent-id + * payloads encoded by {@link SnapDiffJobStore}. Only directory delete/rename object ids + * are held in heap during processing besides path-resolution LRU state. + */ +public final class MergeJoinSnapDiffWriter { + + private static final Logger LOG = LoggerFactory.getLogger(MergeJoinSnapDiffWriter.class); + + private static final DiffType[] FSO_RESOLVE_ORDER = + {MODIFY, RENAME, CREATE}; + + private MergeJoinSnapDiffWriter() { + } + + public static Pair<Long, String> writeReport(SnapshotDiffManager manager, SnapDiffJobStore store, + long bucketObjectId, boolean isFso) throws IOException { + return writeReport(manager, store, bucketObjectId, isFso, false); + } + + /** + * {@code dependencyOrderingEnabled} applies to FSO buckets only; OBS entries are always + * resolved and written directly to the report table. + */ + public static Pair<Long, String> writeReport(SnapshotDiffManager manager, + SnapDiffJobStore store, long bucketObjectId, boolean isFso, + boolean dependencyOrderingEnabled) throws IOException { + store.beginReportWrite(); + if (isFso) { + // Classified CFs hold only objectId + parentId(s). Paths are always resolved before emission. + classifyMergeJoin(store, true); + store.dropListColumnFamilies(); + + SnapDiffPathResolver fromResolver = store.newFromPathResolver(bucketObjectId); + SnapDiffPathResolver toResolver = store.newToPathResolver(bucketObjectId); + boolean[] dependencyOrderingEnabledRef = {dependencyOrderingEnabled}; + int nodeCount = filterAndResolveTopLevelDeletes(manager, store, bucketObjectId, + fromResolver, dependencyOrderingEnabledRef); + store.dropDirectoryIdColumnFamilies(); + store.dropClassifiedColumnFamily(DELETE); + + // Remaining classified entries are resolved and emitted straight to the report table, + // if dependency ordering is disabled. + // Otherwise, entries are emitted into {jobId}-dependency-nodes for ordering. + nodeCount = resolveDiffPaths(store, fromResolver, toResolver, dependencyOrderingEnabledRef, + nodeCount); + for (DiffType diffType : FSO_RESOLVE_ORDER) { + store.dropClassifiedColumnFamily(diffType); + } + store.dropEdgeColumnFamilies(); + + if (dependencyOrderingEnabledRef[0]) { + // Classified entries are ordered and emitted straight to the report table. + orderAndWriteReport(store, nodeCount); + } + store.dropDependencyGraphColumnFamilies(); + store.markTemporaryColumnFamiliesDropped(); + + } else { + classifyMergeJoin(store, false); + // check if dependency graph is required. + store.dropListColumnFamilies(); + } + return store.finishReportWrite(); + } + + private static void classifyMergeJoin(SnapDiffJobStore store, boolean isFso) throws IOException { + try (SnapDiffJobStore.ListIterator newHead = store.newListIterator(); + SnapDiffJobStore.ListIterator oldHead = store.oldListIterator()) { + Map.Entry<Long, byte[]> newEntry = newHead.hasNext() ? newHead.next() : null; + Map.Entry<Long, byte[]> oldEntry = oldHead.hasNext() ? oldHead.next() : null; + + while (newEntry != null || oldEntry != null) { + if (oldEntry == null || (newEntry != null && newEntry.getKey() < oldEntry.getKey())) { + emitCreate(store, newEntry, isFso); + newEntry = newHead.hasNext() ? newHead.next() : null; + } else if (newEntry == null || newEntry.getKey() > oldEntry.getKey()) { + emitDelete(store, oldEntry, isFso); + oldEntry = oldHead.hasNext() ? oldHead.next() : null; + } else { + emitBothPresent(store, newEntry, oldEntry, isFso); + newEntry = newHead.hasNext() ? newHead.next() : null; + oldEntry = oldHead.hasNext() ? oldHead.next() : null; + } + } + } catch (RocksDatabaseException e) { + throw new IOException(e); + } + store.flushWrites(); + } + + private static void emitCreate(SnapDiffJobStore store, Map.Entry<Long, byte[]> newEntry, + boolean isFso) throws IOException { + if (store.isPresentMarker(newEntry.getValue())) { + return; + } + EntryValue value = EntryValue.fromBytes(newEntry.getValue()); + if (isFso) { + store.putClassified(CREATE, newEntry.getKey(), + SnapDiffJobStore.encodeClassifiedParent(value.getParentId())); + } else { + store.putReportEntry(getDiffReportEntry(CREATE, value.getName())); + } + + } + + private static void emitDelete(SnapDiffJobStore store, Map.Entry<Long, byte[]> oldEntry, + boolean isFso) throws IOException { + EntryValue value = EntryValue.fromBytes(oldEntry.getValue()); + if (isFso) { + if (value.isDir()) { + store.addDeletedDirectoryId(oldEntry.getKey()); + } + store.putClassified(DELETE, oldEntry.getKey(), + SnapDiffJobStore.encodeClassifiedParent(value.getParentId())); + } else { + store.putReportEntry(getDiffReportEntry(DELETE, value.getName())); + } + + } + + private static void emitBothPresent(SnapDiffJobStore store, + Map.Entry<Long, byte[]> newEntry, Map.Entry<Long, byte[]> oldEntry, boolean isFso) + throws IOException { + byte[] newBytes = newEntry.getValue(); + if (store.isPresentMarker(newBytes)) { + return; + } + EntryValue newValue = EntryValue.fromBytes(newBytes); + EntryValue oldValue = EntryValue.fromBytes(oldEntry.getValue()); + if (newValue.isDir() != oldValue.isDir()) { + LOG.error("SnapDiff job {} objectId {} has isDir mismatch (new={}, old={})", + store.getJobId(), newEntry.getKey(), newValue.isDir(), oldValue.isDir()); + throw new IOException(String.format( + "Stage 1 invariant violation for job %s objectId %d: isDir mismatch", + store.getJobId(), newEntry.getKey())); + } + boolean pathDiffers = newValue.getParentId() != oldValue.getParentId() + || !newValue.getName().equals(oldValue.getName()); + boolean contentDiffers = !Arrays.equals(newValue.getSignature(), oldValue.getSignature()); + if (pathDiffers) { + if (isFso) { + store.putClassified(RENAME, newEntry.getKey(), SnapDiffJobStore.encodeClassifiedRename( + oldValue.getParentId(), newValue.getParentId())); + // Check should be on oldId so that it is read as this entry from old snapshot has been renamed. + // Even in top-level deletes the ID here is checked against old snapshot namespace. + if (newValue.isDir()) { + store.addRenamedDirectoryId(newEntry.getKey()); + } + } else { + store.putReportEntry(getDiffReportEntry(RENAME, oldValue.getName(), newValue.getName())); + } + } + if (contentDiffers) { + if (isFso) { + store.putClassified(MODIFY, oldEntry.getKey(), + SnapDiffJobStore.encodeClassifiedParent(oldValue.getParentId())); + } else { + store.putReportEntry(getDiffReportEntry(MODIFY, oldValue.getName())); + } + + } + } + + private static int filterAndResolveTopLevelDeletes(SnapshotDiffManager manager, + SnapDiffJobStore store, long bucketObjectId, SnapDiffPathResolver fromResolver, + boolean[] dependencyOrderingEnabled) throws IOException { + Map<Long, Boolean> ancestorMemo = newAncestorMemo(store); + int nodeIndex = 0; + List<Map.Entry<Long, byte[]>> deleteBatch = new ArrayList<>(DEFAULT_BATCH_SIZE); + try (SnapDiffJobStore.ClassifiedIterator deleteIter = store.classifiedIterator(DELETE)) { + while (deleteIter.hasNext()) { + deleteBatch.add(deleteIter.next()); + if (deleteBatch.size() == DEFAULT_BATCH_SIZE) { + nodeIndex = filterAndResolveTopLevelDeleteBatch(manager, store, bucketObjectId, + fromResolver, dependencyOrderingEnabled, ancestorMemo, nodeIndex, deleteBatch); + deleteBatch.clear(); + } + } + } catch (RocksDatabaseException e) { + throw new IOException(e); + } + if (!deleteBatch.isEmpty()) { + nodeIndex = filterAndResolveTopLevelDeleteBatch(manager, store, bucketObjectId, + fromResolver, dependencyOrderingEnabled, ancestorMemo, nodeIndex, deleteBatch); + } + store.flushWrites(); + return dependencyOrderingEnabled[0] ? nodeIndex : 0; + } + + private static Map<Long, Boolean> newAncestorMemo(SnapDiffJobStore store) { + final int capacity = (int) Math.min(store.getMaxInMemoryEntries(), Integer.MAX_VALUE); + return new LinkedHashMap<Long, Boolean>(capacity, 0.75f, true) { + @Override + protected boolean removeEldestEntry(Map.Entry<Long, Boolean> eldest) { + return size() > capacity; + } + }; + } + + @SuppressWarnings("checkstyle:ParameterNumber") + private static int filterAndResolveTopLevelDeleteBatch(SnapshotDiffManager manager, + SnapDiffJobStore store, long bucketObjectId, SnapDiffPathResolver fromResolver, + boolean[] dependencyOrderingEnabled, Map<Long, Boolean> ancestorMemo, int nodeIndex, + List<Map.Entry<Long, byte[]>> deleteBatch) throws IOException { + List<Long> parentObjectIds = new ArrayList<>(deleteBatch.size()); + for (Map.Entry<Long, byte[]> row : deleteBatch) { + parentObjectIds.add(SnapDiffJobStore.decodeClassifiedParent(row.getValue())); + } + boolean[] hasDeletedAncestor = manager.hasDeletedAncestors(parentObjectIds, + store::isDeletedDirectoryId, store::isRenamedDirectoryId, store::multiGetFromParentIds, + bucketObjectId, ancestorMemo); + List<Map.Entry<Long, byte[]>> survivorRows = new ArrayList<>(); + List<Long> survivorObjectIds = new ArrayList<>(); + for (int i = 0; i < deleteBatch.size(); i++) { + if (hasDeletedAncestor[i]) { + continue; + } + Map.Entry<Long, byte[]> row = deleteBatch.get(i); + survivorRows.add(row); + survivorObjectIds.add(row.getKey()); + } + if (survivorRows.isEmpty()) { + return nodeIndex; + } + List<DiffReportEntry> reportEntries = resolveReportEntries(DELETE, survivorObjectIds, + fromResolver, null); + return writeReportEntriesForRows(store, DELETE, survivorRows, reportEntries, + dependencyOrderingEnabled, nodeIndex); + } + + private static int resolveDiffPaths(SnapDiffJobStore store, + SnapDiffPathResolver fromResolver, SnapDiffPathResolver toResolver, + boolean[] dependencyOrderingEnabled, int nodeIndex) throws IOException { + for (DiffType diffType : FSO_RESOLVE_ORDER) { + List<Map.Entry<Long, byte[]>> batch = new ArrayList<>(DEFAULT_BATCH_SIZE); + try (SnapDiffJobStore.ClassifiedIterator iter = store.classifiedIterator(diffType)) { + while (iter.hasNext()) { + batch.add(iter.next()); + if (batch.size() == DEFAULT_BATCH_SIZE) { + nodeIndex = resolveDiffPathBatch(store, diffType, fromResolver, toResolver, + dependencyOrderingEnabled, nodeIndex, batch); + batch.clear(); + } + } + } catch (RocksDatabaseException e) { + throw new IOException(e); + } + if (!batch.isEmpty()) { + nodeIndex = resolveDiffPathBatch(store, diffType, fromResolver, toResolver, + dependencyOrderingEnabled, nodeIndex, batch); + } + } + store.flushWrites(); + return dependencyOrderingEnabled[0] ? nodeIndex : 0; + } + + private static int resolveDiffPathBatch(SnapDiffJobStore store, DiffType diffType, + SnapDiffPathResolver fromResolver, SnapDiffPathResolver toResolver, + boolean[] dependencyOrderingEnabled, int nodeIndex, + List<Map.Entry<Long, byte[]>> batch) throws IOException { + List<Long> objectIds = new ArrayList<>(batch.size()); + for (Map.Entry<Long, byte[]> row : batch) { + objectIds.add(row.getKey()); + } + List<DiffReportEntry> reportEntries = resolveReportEntries(diffType, objectIds, fromResolver, + toResolver); + return writeReportEntriesForRows(store, diffType, batch, reportEntries, + dependencyOrderingEnabled, nodeIndex); + } + + private static List<DiffReportEntry> resolveReportEntries(DiffType diffType, + List<Long> objectIds, SnapDiffPathResolver fromResolver, + SnapDiffPathResolver toResolver) throws IOException { + List<DiffReportEntry> reportEntries = new ArrayList<>(objectIds.size()); + switch (diffType) { + case CREATE: + List<String> toPaths = toResolver.resolvePaths(objectIds); + for (String toPath : toPaths) { + reportEntries.add(toPath != null ? getDiffReportEntry(CREATE, toPath) : null); Review Comment: `FullDiffSequentialReader` populates the reverse edge indexes only for directories, but this resolves each entry's own object ID, including files. File lookups therefore return `null`, and the writer silently drops their report entries. The same issue affects DELETE, MODIFY, and RENAME. This can be reproduced by scanning an empty from-table and a to-table containing one new FSO file through the sequential reader: the writer returns zero entries instead of one CREATE. The existing writer tests manually insert file edges that the reader never produces. Please preserve each file's name and resolve its parent directory, or populate file edges before dropping the candidate lists. A reader-to-writer test would catch this mismatch. ########## hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/diff/MergeJoinSnapDiffWriter.java: ########## @@ -0,0 +1,467 @@ +/* + * 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; + +import static org.apache.hadoop.hdfs.protocol.SnapshotDiffReport.DiffType.CREATE; +import static org.apache.hadoop.hdfs.protocol.SnapshotDiffReport.DiffType.DELETE; +import static org.apache.hadoop.hdfs.protocol.SnapshotDiffReport.DiffType.MODIFY; +import static org.apache.hadoop.hdfs.protocol.SnapshotDiffReport.DiffType.RENAME; +import static org.apache.hadoop.ozone.om.snapshot.diff.SnapDiffJobStore.DEFAULT_BATCH_SIZE; +import static org.apache.hadoop.ozone.snapshot.SnapshotDiffReportOzone.getDiffReportEntry; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import org.apache.commons.lang3.tuple.Pair; +import org.apache.hadoop.hdds.utils.db.RocksDatabaseException; +import org.apache.hadoop.hdfs.protocol.SnapshotDiffReport.DiffReportEntry; +import org.apache.hadoop.hdfs.protocol.SnapshotDiffReport.DiffType; +import org.apache.hadoop.ozone.om.snapshot.SnapshotDiffManager; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Stages 2–4 of the optimized full snapshot diff pipeline: merge join and + * classification, top-level delete retention, ancestor backtrack path resolution, + * dependency ordering, and batched report write. + * + * <p>Classified rows are persisted in per-type column families as minimal parent-id + * payloads encoded by {@link SnapDiffJobStore}. Only directory delete/rename object ids + * are held in heap during processing besides path-resolution LRU state. + */ +public final class MergeJoinSnapDiffWriter { + + private static final Logger LOG = LoggerFactory.getLogger(MergeJoinSnapDiffWriter.class); + + private static final DiffType[] FSO_RESOLVE_ORDER = + {MODIFY, RENAME, CREATE}; + + private MergeJoinSnapDiffWriter() { + } + + public static Pair<Long, String> writeReport(SnapshotDiffManager manager, SnapDiffJobStore store, + long bucketObjectId, boolean isFso) throws IOException { + return writeReport(manager, store, bucketObjectId, isFso, false); + } + + /** + * {@code dependencyOrderingEnabled} applies to FSO buckets only; OBS entries are always + * resolved and written directly to the report table. + */ + public static Pair<Long, String> writeReport(SnapshotDiffManager manager, + SnapDiffJobStore store, long bucketObjectId, boolean isFso, + boolean dependencyOrderingEnabled) throws IOException { + store.beginReportWrite(); + if (isFso) { + // Classified CFs hold only objectId + parentId(s). Paths are always resolved before emission. + classifyMergeJoin(store, true); + store.dropListColumnFamilies(); + + SnapDiffPathResolver fromResolver = store.newFromPathResolver(bucketObjectId); + SnapDiffPathResolver toResolver = store.newToPathResolver(bucketObjectId); + boolean[] dependencyOrderingEnabledRef = {dependencyOrderingEnabled}; + int nodeCount = filterAndResolveTopLevelDeletes(manager, store, bucketObjectId, + fromResolver, dependencyOrderingEnabledRef); + store.dropDirectoryIdColumnFamilies(); + store.dropClassifiedColumnFamily(DELETE); + + // Remaining classified entries are resolved and emitted straight to the report table, + // if dependency ordering is disabled. + // Otherwise, entries are emitted into {jobId}-dependency-nodes for ordering. + nodeCount = resolveDiffPaths(store, fromResolver, toResolver, dependencyOrderingEnabledRef, + nodeCount); + for (DiffType diffType : FSO_RESOLVE_ORDER) { + store.dropClassifiedColumnFamily(diffType); + } + store.dropEdgeColumnFamilies(); + + if (dependencyOrderingEnabledRef[0]) { + // Classified entries are ordered and emitted straight to the report table. + orderAndWriteReport(store, nodeCount); + } + store.dropDependencyGraphColumnFamilies(); + store.markTemporaryColumnFamiliesDropped(); + + } else { + classifyMergeJoin(store, false); + // check if dependency graph is required. + store.dropListColumnFamilies(); + } + return store.finishReportWrite(); + } + + private static void classifyMergeJoin(SnapDiffJobStore store, boolean isFso) throws IOException { + try (SnapDiffJobStore.ListIterator newHead = store.newListIterator(); + SnapDiffJobStore.ListIterator oldHead = store.oldListIterator()) { + Map.Entry<Long, byte[]> newEntry = newHead.hasNext() ? newHead.next() : null; + Map.Entry<Long, byte[]> oldEntry = oldHead.hasNext() ? oldHead.next() : null; + + while (newEntry != null || oldEntry != null) { + if (oldEntry == null || (newEntry != null && newEntry.getKey() < oldEntry.getKey())) { + emitCreate(store, newEntry, isFso); + newEntry = newHead.hasNext() ? newHead.next() : null; + } else if (newEntry == null || newEntry.getKey() > oldEntry.getKey()) { + emitDelete(store, oldEntry, isFso); + oldEntry = oldHead.hasNext() ? oldHead.next() : null; + } else { + emitBothPresent(store, newEntry, oldEntry, isFso); + newEntry = newHead.hasNext() ? newHead.next() : null; + oldEntry = oldHead.hasNext() ? oldHead.next() : null; + } + } + } catch (RocksDatabaseException e) { + throw new IOException(e); + } + store.flushWrites(); + } + + private static void emitCreate(SnapDiffJobStore store, Map.Entry<Long, byte[]> newEntry, + boolean isFso) throws IOException { + if (store.isPresentMarker(newEntry.getValue())) { + return; + } + EntryValue value = EntryValue.fromBytes(newEntry.getValue()); + if (isFso) { + store.putClassified(CREATE, newEntry.getKey(), + SnapDiffJobStore.encodeClassifiedParent(value.getParentId())); + } else { + store.putReportEntry(getDiffReportEntry(CREATE, value.getName())); + } + + } + + private static void emitDelete(SnapDiffJobStore store, Map.Entry<Long, byte[]> oldEntry, + boolean isFso) throws IOException { + EntryValue value = EntryValue.fromBytes(oldEntry.getValue()); + if (isFso) { + if (value.isDir()) { + store.addDeletedDirectoryId(oldEntry.getKey()); + } + store.putClassified(DELETE, oldEntry.getKey(), + SnapDiffJobStore.encodeClassifiedParent(value.getParentId())); + } else { + store.putReportEntry(getDiffReportEntry(DELETE, value.getName())); + } + + } + + private static void emitBothPresent(SnapDiffJobStore store, + Map.Entry<Long, byte[]> newEntry, Map.Entry<Long, byte[]> oldEntry, boolean isFso) + throws IOException { + byte[] newBytes = newEntry.getValue(); + if (store.isPresentMarker(newBytes)) { + return; + } + EntryValue newValue = EntryValue.fromBytes(newBytes); + EntryValue oldValue = EntryValue.fromBytes(oldEntry.getValue()); + if (newValue.isDir() != oldValue.isDir()) { + LOG.error("SnapDiff job {} objectId {} has isDir mismatch (new={}, old={})", + store.getJobId(), newEntry.getKey(), newValue.isDir(), oldValue.isDir()); + throw new IOException(String.format( + "Stage 1 invariant violation for job %s objectId %d: isDir mismatch", + store.getJobId(), newEntry.getKey())); + } + boolean pathDiffers = newValue.getParentId() != oldValue.getParentId() + || !newValue.getName().equals(oldValue.getName()); + boolean contentDiffers = !Arrays.equals(newValue.getSignature(), oldValue.getSignature()); + if (pathDiffers) { + if (isFso) { + store.putClassified(RENAME, newEntry.getKey(), SnapDiffJobStore.encodeClassifiedRename( + oldValue.getParentId(), newValue.getParentId())); + // Check should be on oldId so that it is read as this entry from old snapshot has been renamed. + // Even in top-level deletes the ID here is checked against old snapshot namespace. + if (newValue.isDir()) { + store.addRenamedDirectoryId(newEntry.getKey()); + } + } else { + store.putReportEntry(getDiffReportEntry(RENAME, oldValue.getName(), newValue.getName())); + } + } + if (contentDiffers) { + if (isFso) { + store.putClassified(MODIFY, oldEntry.getKey(), + SnapDiffJobStore.encodeClassifiedParent(oldValue.getParentId())); + } else { + store.putReportEntry(getDiffReportEntry(MODIFY, oldValue.getName())); + } + + } + } + + private static int filterAndResolveTopLevelDeletes(SnapshotDiffManager manager, + SnapDiffJobStore store, long bucketObjectId, SnapDiffPathResolver fromResolver, + boolean[] dependencyOrderingEnabled) throws IOException { + Map<Long, Boolean> ancestorMemo = newAncestorMemo(store); + int nodeIndex = 0; + List<Map.Entry<Long, byte[]>> deleteBatch = new ArrayList<>(DEFAULT_BATCH_SIZE); + try (SnapDiffJobStore.ClassifiedIterator deleteIter = store.classifiedIterator(DELETE)) { + while (deleteIter.hasNext()) { + deleteBatch.add(deleteIter.next()); + if (deleteBatch.size() == DEFAULT_BATCH_SIZE) { + nodeIndex = filterAndResolveTopLevelDeleteBatch(manager, store, bucketObjectId, + fromResolver, dependencyOrderingEnabled, ancestorMemo, nodeIndex, deleteBatch); + deleteBatch.clear(); + } + } + } catch (RocksDatabaseException e) { + throw new IOException(e); + } + if (!deleteBatch.isEmpty()) { + nodeIndex = filterAndResolveTopLevelDeleteBatch(manager, store, bucketObjectId, + fromResolver, dependencyOrderingEnabled, ancestorMemo, nodeIndex, deleteBatch); + } + store.flushWrites(); + return dependencyOrderingEnabled[0] ? nodeIndex : 0; + } + + private static Map<Long, Boolean> newAncestorMemo(SnapDiffJobStore store) { + final int capacity = (int) Math.min(store.getMaxInMemoryEntries(), Integer.MAX_VALUE); + return new LinkedHashMap<Long, Boolean>(capacity, 0.75f, true) { + @Override + protected boolean removeEldestEntry(Map.Entry<Long, Boolean> eldest) { + return size() > capacity; + } + }; + } + + @SuppressWarnings("checkstyle:ParameterNumber") + private static int filterAndResolveTopLevelDeleteBatch(SnapshotDiffManager manager, + SnapDiffJobStore store, long bucketObjectId, SnapDiffPathResolver fromResolver, + boolean[] dependencyOrderingEnabled, Map<Long, Boolean> ancestorMemo, int nodeIndex, + List<Map.Entry<Long, byte[]>> deleteBatch) throws IOException { + List<Long> parentObjectIds = new ArrayList<>(deleteBatch.size()); + for (Map.Entry<Long, byte[]> row : deleteBatch) { + parentObjectIds.add(SnapDiffJobStore.decodeClassifiedParent(row.getValue())); + } + boolean[] hasDeletedAncestor = manager.hasDeletedAncestors(parentObjectIds, + store::isDeletedDirectoryId, store::isRenamedDirectoryId, store::multiGetFromParentIds, + bucketObjectId, ancestorMemo); + List<Map.Entry<Long, byte[]>> survivorRows = new ArrayList<>(); + List<Long> survivorObjectIds = new ArrayList<>(); + for (int i = 0; i < deleteBatch.size(); i++) { + if (hasDeletedAncestor[i]) { + continue; + } + Map.Entry<Long, byte[]> row = deleteBatch.get(i); + survivorRows.add(row); + survivorObjectIds.add(row.getKey()); + } + if (survivorRows.isEmpty()) { + return nodeIndex; + } + List<DiffReportEntry> reportEntries = resolveReportEntries(DELETE, survivorObjectIds, + fromResolver, null); + return writeReportEntriesForRows(store, DELETE, survivorRows, reportEntries, + dependencyOrderingEnabled, nodeIndex); + } + + private static int resolveDiffPaths(SnapDiffJobStore store, + SnapDiffPathResolver fromResolver, SnapDiffPathResolver toResolver, + boolean[] dependencyOrderingEnabled, int nodeIndex) throws IOException { + for (DiffType diffType : FSO_RESOLVE_ORDER) { + List<Map.Entry<Long, byte[]>> batch = new ArrayList<>(DEFAULT_BATCH_SIZE); + try (SnapDiffJobStore.ClassifiedIterator iter = store.classifiedIterator(diffType)) { + while (iter.hasNext()) { + batch.add(iter.next()); + if (batch.size() == DEFAULT_BATCH_SIZE) { + nodeIndex = resolveDiffPathBatch(store, diffType, fromResolver, toResolver, + dependencyOrderingEnabled, nodeIndex, batch); + batch.clear(); + } + } + } catch (RocksDatabaseException e) { + throw new IOException(e); + } + if (!batch.isEmpty()) { + nodeIndex = resolveDiffPathBatch(store, diffType, fromResolver, toResolver, + dependencyOrderingEnabled, nodeIndex, batch); + } + } + store.flushWrites(); + return dependencyOrderingEnabled[0] ? nodeIndex : 0; + } + + private static int resolveDiffPathBatch(SnapDiffJobStore store, DiffType diffType, + SnapDiffPathResolver fromResolver, SnapDiffPathResolver toResolver, + boolean[] dependencyOrderingEnabled, int nodeIndex, + List<Map.Entry<Long, byte[]>> batch) throws IOException { + List<Long> objectIds = new ArrayList<>(batch.size()); + for (Map.Entry<Long, byte[]> row : batch) { + objectIds.add(row.getKey()); + } + List<DiffReportEntry> reportEntries = resolveReportEntries(diffType, objectIds, fromResolver, + toResolver); + return writeReportEntriesForRows(store, diffType, batch, reportEntries, + dependencyOrderingEnabled, nodeIndex); + } + + private static List<DiffReportEntry> resolveReportEntries(DiffType diffType, + List<Long> objectIds, SnapDiffPathResolver fromResolver, + SnapDiffPathResolver toResolver) throws IOException { + List<DiffReportEntry> reportEntries = new ArrayList<>(objectIds.size()); + switch (diffType) { + case CREATE: + List<String> toPaths = toResolver.resolvePaths(objectIds); + for (String toPath : toPaths) { + reportEntries.add(toPath != null ? getDiffReportEntry(CREATE, toPath) : null); + } + break; + case DELETE: + case MODIFY: + List<String> fromPaths = fromResolver.resolvePaths(objectIds); + for (String fromPath : fromPaths) { + reportEntries.add(fromPath != null ? getDiffReportEntry(diffType, fromPath) : null); + } + break; + case RENAME: + List<String> sourcePaths = fromResolver.resolvePaths(objectIds); + List<String> targetPaths = toResolver.resolvePaths(objectIds); + for (int i = 0; i < objectIds.size(); i++) { + String sourcePath = sourcePaths.get(i); + String targetPath = targetPaths.get(i); + if (sourcePath != null && targetPath != null) { + reportEntries.add(getDiffReportEntry(RENAME, sourcePath, targetPath)); + } else { + reportEntries.add(null); + } + } + break; + default: + throw new IllegalArgumentException("Unsupported diff type: " + diffType); + } + return reportEntries; + } + + private static int writeReportEntriesForRows(SnapDiffJobStore store, DiffType diffType, + List<Map.Entry<Long, byte[]>> rows, List<DiffReportEntry> reportEntries, + boolean[] dependencyOrderingEnabled, int nodeIndex) throws IOException { + if (!dependencyOrderingEnabled[0]) { + writeDirectReportEntries(store, diffType, rows, reportEntries); + return 0; + } + + List<SnapDiffDependencyEntry> dependencyEntries = new ArrayList<>(); + for (int i = 0; i < rows.size(); i++) { + DiffReportEntry reportEntry = reportEntries.get(i); + if (reportEntry == null) { + LOG.debug("SnapDiff job {}: dropping {} report entry for unresolvable objectId: {}", + store.getJobId(), diffType.name(), rows.get(i).getKey()); + continue; + } + dependencyEntries.add(buildDependencyEntry(rows.get(i).getKey(), rows.get(i).getValue(), + reportEntry)); + } + if (dependencyEntries.isEmpty()) { + return nodeIndex; + } + if (nodeIndex + dependencyEntries.size() > store.getMaxInMemoryEntries()) { + LOG.error("SnapDiff job {}: dependency node count exceeds limit of {} for job; " + + "falling back to direct report write", + store.getJobId(), store.getMaxInMemoryEntries()); + if (nodeIndex > 0) { + flushUnorderedDependencyNodes(store, nodeIndex); Review Comment: Previously collected dependency nodes may still be buffered in the store's write batch. `flushUnorderedDependencyNodes` reads them directly from RocksDB without first flushing that batch. With dependency ordering enabled and `maxInMemoryEntries = 1`, a directory that is both modified and renamed reproduces this: the MODIFY node is buffered, the RENAME triggers fallback, and the read throws `IOException: Missing dependency node at index 0`. Please flush pending store writes before reading back the accumulated nodes. The existing fallback test exceeds the limit before buffering any nodes, so it misses this case. -- 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]
