smengcl commented on code in PR #11428: URL: https://github.com/apache/ozone/pull/11428#discussion_r4211185805
########## hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/diff/DagDiffSequentialReader.java: ########## @@ -0,0 +1,253 @@ +/* + * 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 com.google.common.annotations.VisibleForTesting; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import org.apache.hadoop.hdds.utils.db.CodecException; +import org.apache.hadoop.hdds.utils.db.IteratorType; +import org.apache.hadoop.hdds.utils.db.LatestVersionedKWayMergeIterator; +import org.apache.hadoop.hdds.utils.db.LatestVersionedKWayMergeIterator.MergedKeyValue; +import org.apache.hadoop.hdds.utils.db.RocksDatabaseException; +import org.apache.hadoop.hdds.utils.db.Table; +import org.apache.hadoop.ozone.util.ClosableIterator; + +/** + * The DAG-based diff multi-stage sequential read that produces the intermediate structures + * consumed by the later merge-join and path-resolution stages. + * + * <p>Call {@link #scanFileTables} then {@link #scanDirectoryTables} (FSO only) in that order. + * Each method runs the to-side delta SST scan first, then the from-side load for the same + * table pair. Directory stages full-scan {@code fromSnapshot.directoryTable} before + * {@code toSnapshot.directoryTable}. + * + * <p>Scans iterate the raw snapshot tables ({@code Table<byte[], byte[]>} from + * {@code DBStore#getTable(String)}) or raw delta SST files so {@link SnapshotDiffValueParser} reads the exact + * persisted protobuf bytes and compare signatures match on-disk layout. + * + * <p>File/key from-side loads are driven by delta table keys batched at {@code writeBatchSize}. + * Directory {@code DiffCandidateSet} membership is populated only from directory delta SSTs and + * consulted during the from-side directory scan; directory delta tombstones are tracked in a + * separate spilled key set with the same in-memory limit. + * + * <p>Sequence gating ({@code fromSnapshot.dbTxSequenceNumber}) is passed to the K-way merge + * iterator; this reader consumes only entries the iterator emits (tombstones and rows with + * {@code sequence > fromSnapshot.dbTxSequenceNumber}). + */ +public class DagDiffSequentialReader { + + private final SnapDiffJobStore store; + private final long exclusiveMinSequenceNumber; + + /** + * @param store per-job temp column families for this DAG diff job + * @param exclusiveMinSequenceNumber {@code fromSnapshot.dbTxSequenceNumber}; passed to the + * K-way merge iterator + */ + public DagDiffSequentialReader(SnapDiffJobStore store, long exclusiveMinSequenceNumber) { + this.store = store; + this.exclusiveMinSequenceNumber = exclusiveMinSequenceNumber; + } + + /** + * Scans file/key-table delta SSTs with interleaved batched {@code multiGetSkipCache} against + * {@code fromSnapshot.file/keyTable}. + * + * @param deltaSstFiles delta SST paths for the to-side file/key table + * @param fromFileTable raw {@code fromSnapshot}'s file or key table + */ + public void scanFileTables(Collection<Path> deltaSstFiles, + Table<byte[], byte[]> fromFileTable) throws IOException { + if (deltaSstFiles.isEmpty()) { + return; + } + try (LatestVersionedKWayMergeIterator iterator = + LatestVersionedKWayMergeIterator.overRawSstFilesFromSequence(deltaSstFiles, exclusiveMinSequenceNumber)) { + scanFileDeltaEntries(iterator, fromFileTable); + } + } + + /** + * Scans directory-table delta SSTs, then full scans of {@code fromSnapshot.directoryTable} + * and {@code toSnapshot.directoryTable} (FSO only). + * + * @param keyPrefix optional bucket prefix as stored in RocksDB; {@code null} scans the + * full table + * @param deltaSstFiles delta SST paths for the to-side directory table + * @param fromDirectoryTable raw {@code fromSnapshot.directoryTable} + * @param toDirectoryTable raw {@code toSnapshot.directoryTable} + */ + public void scanDirectoryTables(byte[] keyPrefix, Collection<Path> deltaSstFiles, + Table<byte[], byte[]> fromDirectoryTable, + Table<byte[], byte[]> toDirectoryTable) throws IOException { + if (!deltaSstFiles.isEmpty()) { + try (LatestVersionedKWayMergeIterator iterator = + LatestVersionedKWayMergeIterator.overRawSstFilesFromSequence(deltaSstFiles, exclusiveMinSequenceNumber)) { + scanDirectoryDeltaEntries(iterator); + } + } + scanFromDirectoryTable(fromDirectoryTable, keyPrefix); + scanToDirectoryTable(toDirectoryTable, keyPrefix); + } + + @VisibleForTesting + void scanFileTablesFromDelta(ClosableIterator<MergedKeyValue> deltaEntries, + Table<byte[], byte[]> fromFileTable) throws IOException { + scanFileDeltaEntries(deltaEntries, fromFileTable); + } + + @VisibleForTesting + void scanDirectoryTablesFromDelta(ClosableIterator<MergedKeyValue> deltaEntries, byte[] keyPrefix, + Table<byte[], byte[]> fromDirectoryTable, + Table<byte[], byte[]> toDirectoryTable) throws IOException { + scanDirectoryDeltaEntries(deltaEntries); + scanFromDirectoryTable(fromDirectoryTable, keyPrefix); + scanToDirectoryTable(toDirectoryTable, keyPrefix); + } + + private void scanFileDeltaEntries(ClosableIterator<MergedKeyValue> entries, + Table<byte[], byte[]> fromFileTable) throws IOException { + int batchSize = store.getWriteBatchSize(); + List<byte[]> lookupBatch = new ArrayList<>(batchSize); + try { + while (entries.hasNext()) { + MergedKeyValue entry = entries.next(); + lookupBatch.add(entry.getUserKey()); Review Comment: Can we filter both delta scans by the bucket prefix before adding lookup keys or candidates? SST selection only checks whether a file overlaps the bucket’s key range, so a retained SST can contain other buckets’ rows. Those rows enter newList/oldList here and, for OBS, can be emitted directly into the requested bucket’s diff report. Please add coverage with delta entries from two buckets and a non-null prefix. A small possible fix is to pass `keyPrefix` into the reader’s constructor and filter both loops: ```diff +import com.google.common.primitives.Bytes; private final long exclusiveMinSequenceNumber; + private final byte[] keyPrefix; - public DagDiffSequentialReader(SnapDiffJobStore store, long exclusiveMinSequenceNumber) { + public DagDiffSequentialReader(SnapDiffJobStore store, long exclusiveMinSequenceNumber, byte[] keyPrefix) { this.store = store; this.exclusiveMinSequenceNumber = exclusiveMinSequenceNumber; + this.keyPrefix = keyPrefix; } @@ scanFileDeltaEntries MergedKeyValue entry = entries.next(); + if (keyPrefix != null && Bytes.indexOf(entry.getUserKey(), keyPrefix) != 0) { + continue; + } lookupBatch.add(entry.getUserKey()); @@ scanDirectoryDeltaEntries MergedKeyValue entry = entries.next(); + if (keyPrefix != null && Bytes.indexOf(entry.getUserKey(), keyPrefix) != 0) { + continue; + } if (entry.isTombstone()) { ``` Update the manager constructor call to pass `keyPrefix`; existing tests can pass `null`. -- 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]
