smengcl commented on code in PR #11296: URL: https://github.com/apache/ozone/pull/11296#discussion_r4211704615
########## hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/diff/SnapDiffPathResolver.java: ########## @@ -0,0 +1,164 @@ +/* + * 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.LinkedHashMap; +import java.util.List; +import java.util.Map; +import org.apache.hadoop.hdds.utils.db.managed.ManagedRocksDB; +import org.rocksdb.ColumnFamilyHandle; + +/** + * 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 int lruCapacity; + private final LinkedHashMap<Long, String> pathCache; + + SnapDiffPathResolver(ManagedRocksDB db, ColumnFamilyHandle edgesCf, long bucketObjectId) { + this(db, edgesCf, bucketObjectId, DEFAULT_LRU_CAPACITY); + } + + SnapDiffPathResolver(ManagedRocksDB db, ColumnFamilyHandle edgesCf, long bucketObjectId, + int lruCapacity) { + this.db = db; + this.edgesCf = edgesCf; + this.bucketObjectId = bucketObjectId; + this.lruCapacity = lruCapacity > 0 ? lruCapacity : DEFAULT_LRU_CAPACITY; + this.pathCache = new LinkedHashMap<Long, String>(this.lruCapacity, 0.75f, true) { + @Override + protected boolean removeEldestEntry(Map.Entry<Long, String> eldest) { + return size() > SnapDiffPathResolver.this.lruCapacity; + } + }; + } + + 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) { + List<Long> chain = new ArrayList<>(); + List<String> linkValues = new ArrayList<>(); + long current = objectId; + while (current != bucketObjectId && !pathCache.containsKey(current)) { + byte[] value = prefetchedEdges.get(current); + if (value == null) { + return; Review Comment: Prefetch skips paths already in the LRU, but resolving earlier entries in the same batch can evict them. This lookup then treats an edge that was never fetched as missing, and the writer silently drops the affected report entry. This can be reproduced with a one-entry cache: resolve `a`, then resolve `[b, a]`. Resolving `b` evicts `a`, so the batch returns `[b, null]` even though both edges exist. The same issue can occur with a larger cache once it fills. Re-fetching a chain when its edge was not prefetched would preserve correctness: ```diff - private void populatePathCache(long objectId, Map<Long, byte[]> prefetchedEdges) { + private void populatePathCache(long objectId, Map<Long, byte[]> prefetchedEdges) throws IOException { List<Long> chain = new ArrayList<>(); List<String> linkValues = new ArrayList<>(); long current = objectId; while (current != bucketObjectId && !pathCache.containsKey(current)) { + if (!prefetchedEdges.containsKey(current)) { + prefetchedEdges.putAll(prefetchEdgeChains(Collections.singletonList(current))); + } byte[] value = prefetchedEdges.get(current); ``` Please add a regression test for eviction between prefetch and resolution within the same batch. ########## hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/diff/SnapDiffJobStore.java: ########## @@ -262,37 +499,132 @@ public void flushWrites() throws IOException { pendingOps = 0; } + /** + * Starts batched emission of resolved rows to the snap diff report table. + */ + public void beginReportWrite() { + 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); + } + } + + /** + * Appends a batch of resolved and ordered {@link DiffReportEntry}s to the snap diff report table. + */ + public void putOrderedReportEntries(List<DiffReportEntry> entries) throws IOException { + if (!reportWriteStarted) { + throw new IllegalStateException("Snap diff report write not started for job " + jobId); + } + for (DiffReportEntry entry : entries) { + String jobReportKey = getReportKeyForIndex(jobId, leftPad(String.valueOf(reportIndex++), 21, '0')); Review Comment: The ordered keys now preserve emission order, but their leading `0` is rejected by `SnapshotDiffManager.createPageResponse()`, which still accepts only prefixes `1`–`4`. With two ordered report entries and page size `1`, the first page returns token `000000000000000000001`; requesting the second page throws `IOException: Token ... has invalid prefix`. Please update token validation alongside the new key format. A small change in `SnapshotDiffManager.java` is: ```diff - if (!isBlank(index)) { + if (!isBlank(index) && !index.startsWith("0")) { DIFF_TYPE_STRING_MAP.values().stream().filter(index::startsWith).findFirst() - .orElseThrow(() -> new IOException("Token " + index + " has invalid prefix. Valid prefixes: " + .orElseThrow(() -> new IOException("Token " + index + " has invalid prefix. Valid prefixes: 0," + DIFF_TYPE_STRING_MAP.values().stream().map(String::valueOf).collect(Collectors.joining(",")))); } ``` The existing numeric and bounds checks still apply. Please add a test that reads multiple pages of an ordered report through `createPageResponse()`. -- 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]
