You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@ozone.apache.org by "SaketaChalamchala (via GitHub)" <gi...@apache.org> on 2026/10/07 02:35:33 UTC
[PR] HDDS-15393. Sequential reader of delta SST files for DAG based snapshot diff. [ozone]
SaketaChalamchala opened a new pull request, #11428:
URL: https://github.com/apache/ozone/pull/11428
## What changes were proposed in this pull request?
Developed with the help of Cursor AI.
This PR is intended as the foundation for a more efficient snapshot diff ([HDDS-9154](https://issues.apache.org/jira/browse/HDDS-9154)).
This PR needs to be merged after https://github.com/apache/ozone/pull/11296
Baseline DAG based diff uses the raw SST iterator to scan the delta set of SST files between from and to snapshots. The RocksDB keys scanned from the delta are then looked up in from and to snapshots using random point lookups to create diff candidate sets. These diff candidate sets from both sides are then compared to generate the diff report.
The new `DagDiffSequentialReader` mirrors `FullDiffSequentialReader` in what it produces (`newList`, `oldList`, from/to reverse-edge indexes, and directory diff-candidate tracking) while sourcing the to-side from delta SST K-way merges instead of full table scans.
Details:
For file/key tables,
1. K-way merge over to-side delta SST files with sequence gating (`fromSnapshot.dbTxSequenceNumber`) delegated to `LatestVersionedKWayMergeIterator` to populate the `newList`. This eliminates the random lookups on the toSnapshot file/key tables.
2. Interleaved batching at `SnapDiffJobStore.getWriteBatchSize()`: delta keys are accumulated and looked up via `multiGetSkipCache` against the raw fromSnapshot table to populate `oldList` instead of individual random point lookups.
3. Operates on raw `Table<byte[], byte[]>` handles and persisted protobuf bytes via `SnapshotDiffValueParser` (no re-encoding through typed helpers).
For directory tables (FSO only),
1. Similar to the fileTable scan K-way merge over to-side delta SST files with sequence gating is used to populate the `newList`. The diff candidates from this scan are tracked.
2. Full sequential scan of `fromSnapshot.directoryTable` builds `fromEdges` and `oldList` is populated for diff candidates tracked in 1.
3. Full sequential scan of `toSnapshot.directoryTable` builds `toEdges`.
- **DagDiffSequentialReader** owns sequential scanning of delta files and tables.
- **SnapDiffJobStore** owns per-job temporary RocksDB column families:
## What is the link to the Apache JIRA
https://issues.apache.org/jira/browse/HDDS-15393
## How was this patch tested?
Unit Tests.
--
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: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-15393. Sequential reader of delta SST files for DAG based snapshot diff. [ozone]
Posted by "smengcl (via GitHub)" <gi...@apache.org>.
smengcl merged PR #11428:
URL: https://github.com/apache/ozone/pull/11428
--
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: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-15393. Sequential reader of delta SST files for DAG based snapshot diff. [ozone]
Posted by "SaketaChalamchala (via GitHub)" <gi...@apache.org>.
SaketaChalamchala commented on code in PR #11428:
URL: https://github.com/apache/ozone/pull/11428#discussion_r4285220206
##########
hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/diff/SnapDiffJobStore.java:
##########
@@ -710,6 +781,7 @@ public void close() throws IOException {
while (!temporaryColumnFamilies.isEmpty()) {
dropAndClose(temporaryColumnFamilies.get(0));
}
+ closeDeeply(tmpCfFamilyOptions);
Review Comment:
Done.
--
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: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-15393. Sequential reader of delta SST files for DAG based snapshot diff. [ozone]
Posted by "smengcl (via GitHub)" <gi...@apache.org>.
smengcl commented on PR #11428:
URL: https://github.com/apache/ozone/pull/11428#issuecomment-6227755663
Thanks @SaketaChalamchala for the patch.
--
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: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-15393. Sequential reader of delta SST files for DAG based snapshot diff. [ozone]
Posted by "smengcl (via GitHub)" <gi...@apache.org>.
smengcl commented on code in PR #11428:
URL: https://github.com/apache/ozone/pull/11428#discussion_r4279153058
##########
hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/diff/SnapDiffJobStore.java:
##########
@@ -710,6 +781,7 @@ public void close() throws IOException {
while (!temporaryColumnFamilies.isEmpty()) {
dropAndClose(temporaryColumnFamilies.get(0));
}
+ closeDeeply(tmpCfFamilyOptions);
Review Comment:
Can we close only the copied column-family options here? RocksDB's `ColumnFamilyOptions` copy constructor retains the same `tableFormatConfig` object. If the supplied options use a `ManagedBlockBasedTableConfig` (as `DBProfile.SSD/TEST` do), `closeDeeply` also closes the caller's shared block-cache and Bloom-filter handles, leaving those handles unusable even though the caller still owns its options.
I reproduced this with `SnapDiffJobStore.open()` on this head: after closing the store, the original table config is closed and both caller-owned handles have `isOwningHandle() == false`. Closing just the copied options preserves them. Please cover this ownership case in a regression test.
```diff
-import static org.apache.hadoop.hdds.utils.db.managed.ManagedColumnFamilyOptions.closeDeeply;
@@
- closeDeeply(tmpCfFamilyOptions);
+ tmpCfFamilyOptions.close();
```
--
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: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-15393. Sequential reader of delta SST files for DAG based snapshot diff. [ozone]
Posted by "smengcl (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-15393. Sequential reader of delta SST files for DAG based snapshot diff. [ozone]
Posted by "SaketaChalamchala (via GitHub)" <gi...@apache.org>.
SaketaChalamchala commented on code in PR #11428:
URL: https://github.com/apache/ozone/pull/11428#discussion_r4252661075
##########
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:
I updated the K way iterator to accept the bucket prefix bounds.
##########
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:
I updated the K way iterator to accept the bucket prefix bounds. PTAL.
--
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: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-15393. Sequential reader of delta SST files for DAG based snapshot diff. [ozone]
Posted by "smengcl (via GitHub)" <gi...@apache.org>.
smengcl commented on code in PR #11428:
URL: https://github.com/apache/ozone/pull/11428#discussion_r4255818480
##########
hadoop-hdds/rocks-native/src/main/java/org/apache/hadoop/hdds/utils/db/LatestVersionedKWayMergeIterator.java:
##########
@@ -407,12 +417,27 @@ public RawSstHeapHead next() {
@Override
public void close() {
- if (closed) {
- return;
+ if (iterator != null) {
+ iterator.close();
+ }
+ if (reader != null) {
+ reader.close();
Review Comment:
Can we restore the closed guard here? HeapEntry.advance() closes each source when it is exhausted, and LatestVersionedKWayMergeIterator.close() closes it again during cleanup. ManagedRawSSTFileReader.close() unconditionally deletes the native reader, so removing the guard allows a double-free on normal scan completion and can abort the JVM.
Please verify exhaustion followed by close over a real SST.
Small suggested diff:
```diff
private final ManagedSlice fromSlice;
private final ManagedSlice toSlice;
+ private boolean closed;
@@
public void close() {
+ if (closed) {
+ return;
+ }
+ closed = true;
if (iterator != null) {
iterator.close();
```
--
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: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-15393. Sequential reader of delta SST files for DAG based snapshot diff. [ozone]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org