entry = lookahead;
+ lookahead = null;
+ switch (type) {
+ case KEY_ONLY:
+ return Table.newKeyValue(entry.getKey(), null);
+ case VALUE_ONLY:
+ return Table.newKeyValue(null, entry.getValue());
+ case KEY_AND_VALUE:
+ default:
+ return Table.newKeyValue(entry.getKey(), entry.getValue());
+ }
+ }
+
+ @Override
+ public void removeFromDB() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void close() {
+ }
}
@Override
diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/OMConfigKeys.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/OMConfigKeys.java
index 2990410fbaec..22a1664bca6f 100644
--- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/OMConfigKeys.java
+++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/OMConfigKeys.java
@@ -694,6 +694,13 @@ public final class OMConfigKeys {
OZONE_OM_SNAPSHOT_DIFF_MAX_ALLOWED_KEYS_CHANGED_PER_DIFF_JOB_DEFAULT
= 1_000_000_000L;
+ public static final String
+ OZONE_OM_SNAPSHOT_DIFF_MAX_IN_MEMORY_ENTRIES_PER_JOB
+ = "ozone.om.snapshot.diff.max.in.memory.entries.per.job";
+ public static final long
+ OZONE_OM_SNAPSHOT_DIFF_MAX_IN_MEMORY_ENTRIES_PER_JOB_DEFAULT
+ = 1_000_000L;
+
public static final String OZONE_OM_UPGRADE_QUOTA_RECALCULATE_ENABLE
= "ozone.om.upgrade.quota.recalculate.enabled";
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/diff/EntryValue.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/diff/EntryValue.java
new file mode 100644
index 000000000000..971926d027a1
--- /dev/null
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/diff/EntryValue.java
@@ -0,0 +1,147 @@
+/*
+ * 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 java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.Objects;
+
+/**
+ * Compact intermediate value stored in the {@code newList}/{@code oldList}
+ * column families of the optimized snapshot diff. It keeps only the fields
+ * required to classify a diff entry in the later merge-join stage, avoiding the
+ * cost of holding a full {@code OmKeyInfo}/{@code OmDirectoryInfo}.
+ *
+ * The fields are:
+ *
+ * - {@code parentId} - parent object id (parent directory for FSO).
+ * - {@code name} - leaf name for FSO buckets, full key path for OBS.
+ * - {@code isDir} - whether the entry is a directory.
+ * - {@code signature} - SHA-256 compare signature computed by
+ * {@code SnapshotDiffValueParser} over the meaningful fields.
+ *
+ *
+ * The wire layout is fixed so both the full diff and the DAG diff Stage 1
+ * readers produce identical bytes:
+ *
+ * | parentId (8, big-endian) | isDir (1) | sigLen (4, big-endian) | signature | name (UTF-8, remaining) |
+ *
+ */
+public final class EntryValue {
+
+ private static final int PARENT_ID_BYTES = Long.BYTES;
+ private static final int IS_DIR_BYTES = 1;
+ private static final int SIG_LEN_BYTES = Integer.BYTES;
+ private static final int HEADER_BYTES = PARENT_ID_BYTES + IS_DIR_BYTES + SIG_LEN_BYTES;
+
+ private final long parentId;
+ private final String name;
+ private final boolean isDir;
+ private final byte[] signature;
+
+ public EntryValue(long parentId, String name, boolean isDir, byte[] signature) {
+ this.parentId = parentId;
+ this.name = name == null ? "" : name;
+ this.isDir = isDir;
+ this.signature = signature == null ? new byte[0] : signature;
+ }
+
+ public long getParentId() {
+ return parentId;
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public boolean isDir() {
+ return isDir;
+ }
+
+ public byte[] getSignature() {
+ return Arrays.copyOf(signature, signature.length);
+ }
+
+ /**
+ * Serializes this value to its fixed byte layout.
+ */
+ public byte[] toBytes() {
+ byte[] nameBytes = name.getBytes(StandardCharsets.UTF_8);
+ ByteBuffer buffer = ByteBuffer.allocate(HEADER_BYTES + signature.length + nameBytes.length);
+ buffer.putLong(parentId);
+ buffer.put((byte) (isDir ? 1 : 0));
+ buffer.putInt(signature.length);
+ buffer.put(signature);
+ buffer.put(nameBytes);
+ return buffer.array();
+ }
+
+ /**
+ * Deserializes a value previously produced by {@link #toBytes()}.
+ */
+ public static EntryValue fromBytes(byte[] bytes) {
+ Objects.requireNonNull(bytes, "bytes must not be null");
+ if (bytes.length < HEADER_BYTES) {
+ throw new IllegalArgumentException("EntryValue byte array too short: " + bytes.length);
+ }
+ ByteBuffer buffer = ByteBuffer.wrap(bytes);
+ long parentId = buffer.getLong();
+ boolean isDir = buffer.get() != 0;
+ int sigLen = buffer.getInt();
+ if (sigLen < 0 || sigLen > buffer.remaining()) {
+ throw new IllegalArgumentException("EntryValue has invalid signature length: " + sigLen);
+ }
+ byte[] signature = new byte[sigLen];
+ buffer.get(signature);
+ byte[] nameBytes = new byte[buffer.remaining()];
+ buffer.get(nameBytes);
+ String name = new String(nameBytes, StandardCharsets.UTF_8);
+ return new EntryValue(parentId, name, isDir, signature);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ EntryValue that = (EntryValue) o;
+ return parentId == that.parentId
+ && isDir == that.isDir
+ && name.equals(that.name)
+ && Arrays.equals(signature, that.signature);
+ }
+
+ @Override
+ public int hashCode() {
+ int result = Objects.hash(parentId, name, isDir);
+ result = 31 * result + Arrays.hashCode(signature);
+ return result;
+ }
+
+ @Override
+ public String toString() {
+ return "EntryValue{parentId=" + parentId
+ + ", name='" + name + '\''
+ + ", isDir=" + isDir
+ + ", signatureLen=" + signature.length + '}';
+ }
+}
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/diff/FullDiffSequentialReader.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/diff/FullDiffSequentialReader.java
new file mode 100644
index 000000000000..269d4278e340
--- /dev/null
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/diff/FullDiffSequentialReader.java
@@ -0,0 +1,205 @@
+/*
+ * 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.DEFAULT_OM_UPDATE_ID;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import org.apache.hadoop.hdds.utils.db.CodecException;
+import org.apache.hadoop.hdds.utils.db.IteratorType;
+import org.apache.hadoop.hdds.utils.db.RocksDatabaseException;
+import org.apache.hadoop.hdds.utils.db.Table;
+
+/**
+ * The multi-stage sequential read in FULL diff mode that produces the
+ * intermediate structures consumed by the later merge-join and path-resolution stages.
+ *
+ * Call {@link #scanFileTables} then {@link #scanDirectoryTables} (FSO only)
+ * in that order. Each method runs the to-side scan first, then the from-side
+ * scan for the same table pair.
+ *
+ *
Scans iterate the raw snapshot tables ({@code Table} from
+ * {@code DBStore#getTable(String)}) so {@link SnapshotDiffValueParser} reads the exact
+ * persisted protobuf bytes and compare signatures match on-disk layout.
+ *
+ * Every to-side row is written to {@code newList} once: either a present-marker
+ * (unchanged-marker; membership only) or a full {@link EntryValue} with signature when it
+ * passes the update-id gate. Every from-side row is written to {@code oldList}:
+ * {@code DiffCandidateSet} members store a full {@link EntryValue} with signature; all
+ * other rows store {@code parentId}, {@code name}, and {@code isDir} with an empty
+ * signature.
+ *
+ *
When an update-id gate is supplied (HA OM), to-side gating normally admits
+ * rows with {@code updateID > fromSnapshotDbTxSequenceNumber}. HA OM write paths
+ * are expected to bump {@code updateID} on every meaningful metadata change.
+ * Rows with a missing {@code updateID}, {@code updateID == 0}, or
+ * {@code updateID == DEFAULT_OM_UPDATE_ID} ({@code -1}) are always treated as
+ * candidates as a conservative fallback for legacy or ambiguous rows.
+ *
+ *
When no gate is supplied (non-HA), every to-side row is a candidate and a
+ * compare signature is computed for each.
+ */
+public class FullDiffSequentialReader {
+
+ private final SnapDiffJobStore store;
+ private final long updateIdGate;
+ private final boolean gatingEnabled;
+
+ /**
+ * Non-HA full diff: gating is disabled and every to-side entry is a candidate.
+ */
+ public FullDiffSequentialReader(SnapDiffJobStore store) {
+ this(store, null);
+ }
+
+ /**
+ * @param store per-job temp column families for this full diff job
+ * @param updateIdGate when non-null, enables HA gating using this from-snapshot
+ * transaction index; when null, gating is disabled (non-HA)
+ */
+ public FullDiffSequentialReader(SnapDiffJobStore store, Long updateIdGate) {
+ this.store = store;
+ this.gatingEnabled = updateIdGate != null;
+ this.updateIdGate = updateIdGate != null ? updateIdGate : 0L;
+ }
+
+ /**
+ * Scans {@code toSnapshot.keyTable}/{@code fileTable} then the from-side counterpart.
+ *
+ * @param fromTable raw from-snapshot key/file table
+ * @param toTable raw to-snapshot key/file table
+ * @param keyPrefix optional bucket prefix as stored in RocksDB; {@code null} scans the full table
+ */
+ public void scanFileTables(Table fromTable,
+ Table toTable, byte[] keyPrefix) throws IOException {
+ scanToTable(toTable, keyPrefix, false);
+ scanFromTable(fromTable, keyPrefix, false);
+ }
+
+ /**
+ * Scans {@code toSnapshot.directoryTable} then {@code fromSnapshot.directoryTable}.
+ *
+ * @param fromTable raw from-snapshot directory table
+ * @param toTable raw to-snapshot directory table
+ * @param keyPrefix optional bucket prefix as stored in RocksDB; {@code null} scans the full table
+ */
+ public void scanDirectoryTables(Table fromTable,
+ Table toTable, byte[] keyPrefix) throws IOException {
+ scanToTable(toTable, keyPrefix, true);
+ scanFromTable(fromTable, keyPrefix, true);
+ }
+
+ private void scanToTable(Table table, byte[] keyPrefix, boolean isDir)
+ throws IOException {
+ try (Table.KeyValueIterator iter =
+ table.iterator(keyPrefix, IteratorType.VALUE_ONLY)) {
+ while (iter.hasNext()) {
+ byte[] value = iter.next().getValue();
+ processToSideEntry(value, isDir);
+ }
+ } catch (RocksDatabaseException | CodecException e) {
+ throw new IOException(e);
+ }
+ store.flushWrites();
+ }
+
+ private void scanFromTable(Table table, byte[] keyPrefix, boolean isDir)
+ throws IOException {
+ store.flushWrites();
+ try (Table.KeyValueIterator iter =
+ table.iterator(keyPrefix, IteratorType.VALUE_ONLY)) {
+ while (iter.hasNext()) {
+ byte[] value = iter.next().getValue();
+ processFromSideEntry(value, isDir);
+ }
+ } catch (RocksDatabaseException | CodecException e) {
+ throw new IOException(e);
+ }
+ store.clearDiffCandidates();
+ store.flushWrites();
+ }
+
+ private void processToSideEntry(byte[] value, boolean isDir) throws IOException {
+ SnapshotDiffValueParser.ParsedRequiredInfo info = parseRequired(value, isDir, gatingEnabled);
+ long objectId = info.getObjectId();
+
+ if (isDir) {
+ store.putToEdge(info.getParentId(), objectId, nameBytes(info.getName()));
+ }
+
+ if (isCandidateOnToSide(info)) {
+ store.addDiffCandidate(objectId);
+ byte[] signature = computeSignature(value, isDir);
+ store.putNewList(objectId,
+ new EntryValue(info.getParentId(), info.getName(), isDir, signature).toBytes());
+ } else {
+ store.putNewListPresentMarker(objectId);
+ }
+ }
+
+ private void processFromSideEntry(byte[] value, boolean isDir) throws IOException {
+ SnapshotDiffValueParser.ParsedRequiredInfo info = parseRequired(value, isDir, false);
+ long objectId = info.getObjectId();
+
+ if (isDir) {
+ store.putFromEdge(info.getParentId(), objectId, nameBytes(info.getName()));
+ }
+
+ byte[] oldListValue;
+ if (store.isDiffCandidate(objectId)) {
+ byte[] signature = computeSignature(value, isDir);
+ oldListValue = new EntryValue(info.getParentId(), info.getName(), isDir, signature).toBytes();
+ } else {
+ oldListValue = new EntryValue(info.getParentId(), info.getName(), isDir, null).toBytes();
+ }
+ store.putOldList(objectId, oldListValue);
+ }
+
+ private boolean isCandidateOnToSide(SnapshotDiffValueParser.ParsedRequiredInfo info) {
+ if (!gatingEnabled) {
+ return true;
+ }
+ if (!info.hasUpdateId()) {
+ return true;
+ }
+ long updateId = info.getUpdateId();
+ if (updateId == 0L || updateId == DEFAULT_OM_UPDATE_ID) {
+ return true;
+ }
+ return updateId > updateIdGate;
+ }
+
+ private static SnapshotDiffValueParser.ParsedRequiredInfo parseRequired(byte[] value,
+ boolean isDir, boolean includeUpdateId)
+ throws IOException {
+ return isDir
+ ? SnapshotDiffValueParser.parseDirectoryInfoRequiredFields(value, includeUpdateId)
+ : SnapshotDiffValueParser.parseKeyInfoRequiredFields(value, includeUpdateId);
+ }
+
+ private static byte[] computeSignature(byte[] value, boolean isDir) throws IOException {
+ return isDir
+ ? SnapshotDiffValueParser.computeDirectoryInfoCompareSignature(value)
+ : SnapshotDiffValueParser.computeKeyInfoCompareSignature(value);
+ }
+
+ private static byte[] nameBytes(String name) {
+ return (name == null ? "" : name).getBytes(StandardCharsets.UTF_8);
+ }
+}
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/diff/SnapDiffJobStore.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/diff/SnapDiffJobStore.java
new file mode 100644
index 000000000000..636354bbed30
--- /dev/null
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/diff/SnapDiffJobStore.java
@@ -0,0 +1,354 @@
+/*
+ * 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.om.OMConfigKeys.OZONE_OM_SNAPSHOT_DIFF_MAX_IN_MEMORY_ENTRIES_PER_JOB_DEFAULT;
+import static org.apache.hadoop.ozone.om.snapshot.SnapshotUtils.dropColumnFamilyHandle;
+
+import jakarta.annotation.Nonnull;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.Set;
+import org.apache.hadoop.hdds.StringUtils;
+import org.apache.hadoop.hdds.utils.db.CodecRegistry;
+import org.apache.hadoop.hdds.utils.db.managed.ManagedColumnFamilyOptions;
+import org.apache.hadoop.hdds.utils.db.managed.ManagedRocksDB;
+import org.apache.hadoop.hdds.utils.db.managed.ManagedWriteBatch;
+import org.apache.hadoop.hdds.utils.db.managed.ManagedWriteOptions;
+import org.rocksdb.ColumnFamilyDescriptor;
+import org.rocksdb.ColumnFamilyHandle;
+import org.rocksdb.RocksDBException;
+
+/**
+ * Owns the per-job temporary RocksDB column families and batched writes shared
+ * between optimized snapshot diff pipeline stages.
+ *
+ * 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.
+ *
+ *
This initial version supports the full diff sequential reader ({@link FullDiffSequentialReader}).
+ * DAG diff support extends this store in HDDS-15393.
+ */
+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;
+
+ 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 TO_EDGES_SUFFIX = "-to-edges";
+ private static final String FROM_EDGES_SUFFIX = "-from-edges";
+
+ private final ManagedRocksDB db;
+ private final boolean fso;
+ private final byte[] presentMarker;
+ private final ManagedColumnFamilyOptions familyOptions;
+ private final long maxInMemoryEntries;
+
+ private ColumnFamilyHandle newListCf;
+ private ColumnFamilyHandle oldListCf;
+ private ColumnFamilyHandle toEdgesCf;
+ private ColumnFamilyHandle fromEdgesCf;
+
+ private String diffCandCfName;
+ private Set diffCandidates;
+ private ColumnFamilyHandle diffCandidatesCf;
+ private boolean diffCandidatesSpilled;
+
+ private final ManagedWriteBatch writeBatch;
+ private final ManagedWriteOptions writeOptions;
+ private final int writeBatchSize;
+ private int pendingOps;
+
+ /** Full diff: shared new/old lists plus FSO edge column families. */
+ public enum Mode {
+ FULL
+ }
+
+ private SnapDiffJobStore(ManagedRocksDB db, CodecRegistry codecRegistry, boolean fso,
+ int writeBatchSize, ManagedColumnFamilyOptions familyOptions, long maxInMemoryEntries)
+ throws IOException {
+ this.db = db;
+ this.fso = fso;
+ this.writeBatchSize = writeBatchSize;
+ this.familyOptions = familyOptions;
+ this.maxInMemoryEntries = maxInMemoryEntries;
+ this.presentMarker = codecRegistry.asRawData(Boolean.TRUE);
+ this.writeBatch = new ManagedWriteBatch();
+ this.writeOptions = new ManagedWriteOptions();
+ this.pendingOps = 0;
+ this.diffCandidates = new HashSet<>();
+ }
+
+ 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);
+ try {
+ store.initColumnFamilies(familyOptions, jobId);
+ 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;
+ }
+
+ /** Writes a present-marker for {@code objectId} in {@code newList}. */
+ public void putNewListPresentMarker(long objectId) throws IOException {
+ batchPut(newListCf, objectIdKey(objectId), presentMarker);
+ }
+
+ /** Writes a full diff-candidate {@link EntryValue} for {@code objectId} in {@code newList}. */
+ public void putNewList(long objectId, byte[] entryValue) throws IOException {
+ batchPut(newListCf, objectIdKey(objectId), entryValue);
+ }
+
+ public void putOldList(long objectId, byte[] entryValue) throws IOException {
+ batchPut(oldListCf, objectIdKey(objectId), entryValue);
+ }
+
+ public void putToEdge(long parentId, long objectId, byte[] name) throws IOException {
+ requireFso();
+ batchPut(toEdgesCf, edgeKey(parentId, objectId), name);
+ }
+
+ public void putFromEdge(long parentId, long objectId, byte[] name) throws IOException {
+ requireFso();
+ batchPut(fromEdgesCf, edgeKey(parentId, objectId), name);
+ }
+
+ public byte[] getNewList(long objectId) throws IOException {
+ return get(newListCf, objectIdKey(objectId));
+ }
+
+ public byte[] getOldList(long objectId) throws IOException {
+ return get(oldListCf, objectIdKey(objectId));
+ }
+
+ public boolean hasNewListEntry(long objectId) throws IOException {
+ return getNewList(objectId) != null;
+ }
+
+ public boolean isNewListCandidate(long objectId) throws IOException {
+ byte[] value = getNewList(objectId);
+ return value != null && !Arrays.equals(value, presentMarker);
+ }
+
+ /**
+ * Records a to-side diff candidate {@code objectId}. Retained in memory until
+ * {@code maxInMemoryEntries} is reached, then spilled to a temporary column family.
+ */
+ public void addDiffCandidate(long objectId) throws IOException {
+ if (diffCandidatesSpilled) {
+ batchPut(diffCandidatesCf, objectIdKey(objectId), presentMarker);
+ return;
+ }
+ if (diffCandidates.size() >= maxInMemoryEntries) {
+ spillDiffCandidates();
+ }
+ if (diffCandidatesSpilled) {
+ batchPut(diffCandidatesCf, objectIdKey(objectId), presentMarker);
+ } else {
+ diffCandidates.add(objectId);
+ }
+ }
+
+ /** Returns whether {@code objectId} was gated in as a to-side diff candidate. */
+ public boolean isDiffCandidate(long objectId) throws IOException {
+ if (diffCandidatesSpilled) {
+ return get(diffCandidatesCf, objectIdKey(objectId)) != null;
+ }
+ return diffCandidates.contains(objectId);
+ }
+
+ /** Clears the diff-candidate set after a from-side scan consumes it. */
+ public void clearDiffCandidates() throws IOException {
+ diffCandidates.clear();
+ if (diffCandidatesSpilled) {
+ diffCandidatesCf = dropAndClose(diffCandidatesCf);
+ diffCandidatesSpilled = false;
+ }
+ }
+
+ /** Returns the current in-memory diff-candidate count (for tests and limit wiring). */
+ public int getDiffCandidateCount() {
+ return diffCandidates.size();
+ }
+
+ boolean areDiffCandidatesSpilled() {
+ return diffCandidatesSpilled;
+ }
+
+ public byte[] getToEdgeName(long parentId, long objectId) throws IOException {
+ requireFso();
+ return get(toEdgesCf, edgeKey(parentId, objectId));
+ }
+
+ public byte[] getFromEdgeName(long parentId, long objectId) throws IOException {
+ requireFso();
+ return get(fromEdgesCf, edgeKey(parentId, objectId));
+ }
+
+ public void flushWrites() throws IOException {
+ if (pendingOps == 0) {
+ return;
+ }
+ try {
+ db.get().write(writeOptions, writeBatch);
+ } catch (RocksDBException e) {
+ throw new IOException("Failed to flush SnapDiff job store write batch", e);
+ }
+ writeBatch.clear();
+ pendingOps = 0;
+ }
+
+ public static byte[] objectIdKey(long objectId) {
+ return ByteBuffer.allocate(Long.BYTES).putLong(objectId).array();
+ }
+
+ public static byte[] edgeKey(long parentId, long objectId) {
+ return ByteBuffer.allocate(2 * Long.BYTES).putLong(parentId).putLong(objectId).array();
+ }
+
+ private void initColumnFamilies(ManagedColumnFamilyOptions options, String jobId)
+ throws RocksDBException {
+ newListCf = createColumnFamily(jobId + NEW_LIST_SUFFIX, options);
+ oldListCf = createColumnFamily(jobId + OLD_LIST_SUFFIX, options);
+ diffCandCfName = jobId + CAND_IDS_SUFFIX;
+ if (fso) {
+ toEdgesCf = createColumnFamily(jobId + TO_EDGES_SUFFIX, options);
+ fromEdgesCf = createColumnFamily(jobId + FROM_EDGES_SUFFIX, options);
+ }
+ }
+
+ private void spillDiffCandidates() throws IOException {
+ try {
+ diffCandidatesCf = createColumnFamily(diffCandCfName, familyOptions);
+ } catch (RocksDBException e) {
+ throw new IOException("Failed to create diff candidate column family " + diffCandCfName, e);
+ }
+ for (Long objectId : diffCandidates) {
+ batchPut(diffCandidatesCf, objectIdKey(objectId), presentMarker);
+ }
+ diffCandidates.clear();
+ flushWrites();
+ diffCandidatesSpilled = true;
+ }
+
+ private void batchPut(ColumnFamilyHandle cf, byte[] key, byte[] value) throws IOException {
+ try {
+ writeBatch.put(cf, key, value);
+ } catch (RocksDBException e) {
+ throw new IOException(e);
+ }
+ pendingOps++;
+ if (pendingOps >= writeBatchSize) {
+ flushWrites();
+ }
+ }
+
+ private byte[] get(ColumnFamilyHandle cf, byte[] key) throws IOException {
+ if (cf == null) {
+ return null;
+ }
+ try {
+ return db.get().get(cf, key);
+ } catch (RocksDBException e) {
+ throw new IOException(e);
+ }
+ }
+
+ private ColumnFamilyHandle createColumnFamily(String name, ManagedColumnFamilyOptions options)
+ throws RocksDBException {
+ return db.get().createColumnFamily(
+ new ColumnFamilyDescriptor(StringUtils.string2Bytes(name), options));
+ }
+
+ private void requireFso() {
+ if (!fso) {
+ throw new IllegalStateException("Directory edge column families require an FSO bucket");
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ flushWrites();
+ writeBatch.close();
+ writeOptions.close();
+ newListCf = dropAndClose(newListCf);
+ oldListCf = dropAndClose(oldListCf);
+ diffCandidatesCf = dropAndClose(diffCandidatesCf);
+ toEdgesCf = dropAndClose(toEdgesCf);
+ fromEdgesCf = dropAndClose(fromEdgesCf);
+ }
+
+ private void closeQuietly() {
+ try {
+ close();
+ } catch (IOException ignored) {
+ // best effort while handling a failed open
+ }
+ }
+
+ private ColumnFamilyHandle dropAndClose(ColumnFamilyHandle handle) {
+ if (handle == null) {
+ return null;
+ }
+ dropColumnFamilyHandle(db, handle);
+ handle.close();
+ return null;
+ }
+}
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/SnapshotDiffValueParser.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/diff/SnapshotDiffValueParser.java
similarity index 99%
rename from hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/SnapshotDiffValueParser.java
rename to hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/diff/SnapshotDiffValueParser.java
index b1c09cd082af..15400f533625 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/SnapshotDiffValueParser.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/diff/SnapshotDiffValueParser.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.hadoop.ozone.om.snapshot;
+package org.apache.hadoop.ozone.om.snapshot.diff;
import com.google.protobuf.ByteString;
import com.google.protobuf.CodedInputStream;
diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/snapshot/diff/TestFullDiffSequentialReader.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/snapshot/diff/TestFullDiffSequentialReader.java
new file mode 100644
index 000000000000..5ac9f48a6376
--- /dev/null
+++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/snapshot/diff/TestFullDiffSequentialReader.java
@@ -0,0 +1,299 @@
+/*
+ * 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.hdds.utils.db.DBStoreBuilder.DEFAULT_COLUMN_FAMILY_NAME;
+import static org.apache.hadoop.ozone.OzoneConsts.DEFAULT_OM_UPDATE_ID;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.hadoop.hdds.StringUtils;
+import org.apache.hadoop.hdds.client.RatisReplicationConfig;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor;
+import org.apache.hadoop.hdds.utils.db.CodecException;
+import org.apache.hadoop.hdds.utils.db.CodecRegistry;
+import org.apache.hadoop.hdds.utils.db.InMemoryTestTable;
+import org.apache.hadoop.hdds.utils.db.RocksDatabaseException;
+import org.apache.hadoop.hdds.utils.db.StringCodec;
+import org.apache.hadoop.hdds.utils.db.Table;
+import org.apache.hadoop.hdds.utils.db.managed.ManagedColumnFamilyOptions;
+import org.apache.hadoop.hdds.utils.db.managed.ManagedDBOptions;
+import org.apache.hadoop.hdds.utils.db.managed.ManagedRocksDB;
+import org.apache.hadoop.ozone.om.helpers.OmDirectoryInfo;
+import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.rocksdb.ColumnFamilyDescriptor;
+import org.rocksdb.ColumnFamilyHandle;
+import org.rocksdb.RocksDBException;
+
+/**
+ * Tests the full-diff Stage 1 multi-stage sequential read (HDDS-15394).
+ */
+class TestFullDiffSequentialReader {
+
+ private static final String VOLUME = "vol";
+ private static final String BUCKET = "buck";
+ private static final long BUCKET_OBJECT_ID = 1L;
+
+ @TempDir
+ private static File tempDir;
+ private static ManagedRocksDB db;
+ private static ManagedDBOptions dbOptions;
+ private static ManagedColumnFamilyOptions columnFamilyOptions;
+ private static CodecRegistry codecRegistry;
+ private static final AtomicInteger JOB_ID = new AtomicInteger(0);
+
+ @BeforeAll
+ static void init() throws RocksDBException {
+ dbOptions = new ManagedDBOptions();
+ dbOptions.setCreateIfMissing(true);
+ columnFamilyOptions = new ManagedColumnFamilyOptions();
+ codecRegistry = CodecRegistry.newBuilder().build();
+
+ File dbDir = new File(tempDir, "full-diff-stage1.db");
+ List descriptors = Collections.singletonList(
+ new ColumnFamilyDescriptor(StringUtils.string2Bytes(DEFAULT_COLUMN_FAMILY_NAME), columnFamilyOptions));
+ List handles = new ArrayList<>();
+ db = ManagedRocksDB.open(dbOptions, dbDir.getAbsolutePath(), descriptors, handles);
+ }
+
+ @AfterAll
+ static void teardown() {
+ if (db != null) {
+ db.close();
+ }
+ if (columnFamilyOptions != null) {
+ columnFamilyOptions.close();
+ }
+ if (dbOptions != null) {
+ dbOptions.close();
+ }
+ }
+
+ @Test
+ void testKeyDiffShapesWithGating() throws Exception {
+ long gate = 50L;
+ Table toTable = InMemoryTestTable.forRawBytes();
+ putKey(toTable, "k1", keyInfo("create", 1L, 0L, 60L, 100L));
+ putKey(toTable, "k2", keyInfo("modify", 2L, 0L, 70L, 200L));
+ putKey(toTable, "k3", keyInfo("newname", 3L, 0L, 70L, 100L));
+ putKey(toTable, "k5", keyInfo("unchanged", 5L, 0L, 10L, 100L));
+
+ Table fromTable = InMemoryTestTable.forRawBytes();
+ putKey(fromTable, "k2", keyInfo("modify", 2L, 0L, 10L, 100L));
+ putKey(fromTable, "k3", keyInfo("oldname", 3L, 0L, 10L, 100L));
+ putKey(fromTable, "k4", keyInfo("deleted", 4L, 0L, 10L, 100L));
+ putKey(fromTable, "k5", keyInfo("unchanged", 5L, 0L, 10L, 100L));
+
+ try (SnapDiffJobStore store = newStore(false)) {
+ new FullDiffSequentialReader(store, gate).scanFileTables(fromTable, toTable, null);
+
+ assertTrue(store.isNewListCandidate(1L));
+ assertTrue(store.isNewListCandidate(2L));
+ assertTrue(store.isNewListCandidate(3L));
+ assertFalse(store.hasNewListEntry(4L));
+ assertTrue(store.hasNewListEntry(5L));
+ assertFalse(store.isNewListCandidate(5L));
+
+ assertNull(store.getOldList(1L));
+ assertNotNull(store.getOldList(2L));
+ assertNotNull(store.getOldList(3L));
+ assertNotNull(store.getOldList(4L));
+ assertNotNull(store.getOldList(5L));
+ assertEquals(0, store.getDiffCandidateCount());
+
+ EntryValue unchangedOld = EntryValue.fromBytes(store.getOldList(5L));
+ assertEquals(0, unchangedOld.getSignature().length);
+
+ EntryValue newRename = EntryValue.fromBytes(store.getNewList(3L));
+ EntryValue oldRename = EntryValue.fromBytes(store.getOldList(3L));
+ assertEquals("newname", newRename.getName());
+ assertEquals("oldname", oldRename.getName());
+ assertArrayEquals(newRename.getSignature(), oldRename.getSignature());
+
+ EntryValue newModify = EntryValue.fromBytes(store.getNewList(2L));
+ EntryValue oldModify = EntryValue.fromBytes(store.getOldList(2L));
+ assertFalse(java.util.Arrays.equals(newModify.getSignature(), oldModify.getSignature()));
+ }
+ }
+
+ @Test
+ void testGatingDisabledAdmitsAllToEntries() throws Exception {
+ Table toTable = InMemoryTestTable.forRawBytes();
+ putKey(toTable, "k1", keyInfo("a", 1L, 0L, 5L, 100L));
+ putKey(toTable, "k2", keyInfo("b", 2L, 0L, 5L, 100L));
+
+ Table fromTable = InMemoryTestTable.forRawBytes();
+ putKey(fromTable, "k1", keyInfo("a", 1L, 0L, 5L, 100L));
+
+ try (SnapDiffJobStore store = newStore(false)) {
+ new FullDiffSequentialReader(store).scanFileTables(fromTable, toTable, null);
+ assertTrue(store.isNewListCandidate(1L));
+ assertTrue(store.isNewListCandidate(2L));
+ assertNotNull(store.getOldList(1L));
+ assertEquals(0, store.getDiffCandidateCount());
+ }
+ }
+
+ @Test
+ void testDeleteCandidateHasMetadataWithoutSignature() throws Exception {
+ long gate = 50L;
+ Table toTable = InMemoryTestTable.forRawBytes();
+ putKey(toTable, "k5", keyInfo("unchanged", 5L, 0L, 10L, 100L));
+
+ Table fromTable = InMemoryTestTable.forRawBytes();
+ putKey(fromTable, "k4", keyInfo("deleted", 4L, 0L, 10L, 100L));
+
+ try (SnapDiffJobStore store = newStore(false)) {
+ new FullDiffSequentialReader(store, gate).scanFileTables(fromTable, toTable, null);
+
+ EntryValue deleted = EntryValue.fromBytes(store.getOldList(4L));
+ assertEquals(0, deleted.getSignature().length);
+ assertEquals("deleted", deleted.getName());
+ }
+ }
+
+ @Test
+ void testStaleUpdateIdValuesAreCandidates() throws Exception {
+ long gate = 50L;
+ Table toTable = InMemoryTestTable.forRawBytes();
+ putKey(toTable, "k10", keyInfo("zero", 10L, 0L, 0L, 100L));
+ putKey(toTable, "k11", keyInfo("default-id", 11L, 0L, DEFAULT_OM_UPDATE_ID, 100L));
+
+ Table fromTable = InMemoryTestTable.forRawBytes();
+ putKey(fromTable, "k10", keyInfo("zero", 10L, 0L, 0L, 100L));
+ putKey(fromTable, "k11", keyInfo("default-id", 11L, 0L, DEFAULT_OM_UPDATE_ID, 100L));
+
+ try (SnapDiffJobStore store = newStore(false)) {
+ new FullDiffSequentialReader(store, gate).scanFileTables(fromTable, toTable, null);
+
+ assertTrue(store.isNewListCandidate(10L));
+ assertTrue(store.isNewListCandidate(11L));
+ assertNotNull(store.getOldList(10L));
+ assertNotNull(store.getOldList(11L));
+ assertEquals(0, store.getDiffCandidateCount());
+ }
+ }
+
+ @Test
+ void testDiffCandidatesSpillToRocksDb() throws Exception {
+ Table toTable = InMemoryTestTable.forRawBytes();
+ putKey(toTable, "k1", keyInfo("a", 1L, 0L, 60L, 100L));
+ putKey(toTable, "k2", keyInfo("b", 2L, 0L, 60L, 100L));
+ putKey(toTable, "k3", keyInfo("c", 3L, 0L, 60L, 100L));
+
+ Table fromTable = InMemoryTestTable.forRawBytes();
+ putKey(fromTable, "k1", keyInfo("a", 1L, 0L, 10L, 100L));
+ putKey(fromTable, "k2", keyInfo("b", 2L, 0L, 10L, 100L));
+ putKey(fromTable, "k3", keyInfo("c", 3L, 0L, 10L, 100L));
+
+ try (SnapDiffJobStore store = SnapDiffJobStore.open(db, codecRegistry, columnFamilyOptions,
+ "job" + JOB_ID.incrementAndGet(), false, SnapDiffJobStore.Mode.FULL,
+ SnapDiffJobStore.DEFAULT_WRITE_BATCH_SIZE, 2L)) {
+ new FullDiffSequentialReader(store, 50L).scanFileTables(fromTable, toTable, null);
+
+ assertTrue(store.isNewListCandidate(1L));
+ assertTrue(store.isNewListCandidate(2L));
+ assertTrue(store.isNewListCandidate(3L));
+ assertNotNull(store.getOldList(1L));
+ assertNotNull(store.getOldList(2L));
+ assertNotNull(store.getOldList(3L));
+ assertEquals(0, store.getDiffCandidateCount());
+ }
+ }
+
+ @Test
+ void testFsoDirectoryEdgesPopulated() throws Exception {
+ Table toTable = InMemoryTestTable.forRawBytes();
+ putDir(toTable, "d100", dirInfo("a", 100L, BUCKET_OBJECT_ID, 60L));
+ putDir(toTable, "d101", dirInfo("b", 101L, 100L, 60L));
+
+ Table fromTable = InMemoryTestTable.forRawBytes();
+ putDir(fromTable, "d100", dirInfo("a", 100L, BUCKET_OBJECT_ID, 10L));
+ putDir(fromTable, "d101", dirInfo("b", 101L, 100L, 10L));
+
+ try (SnapDiffJobStore store = newStore(true)) {
+ new FullDiffSequentialReader(store, 50L).scanDirectoryTables(fromTable, toTable, null);
+
+ assertEquals("a", name(store.getToEdgeName(BUCKET_OBJECT_ID, 100L)));
+ assertEquals("b", name(store.getToEdgeName(100L, 101L)));
+ assertEquals("a", name(store.getFromEdgeName(BUCKET_OBJECT_ID, 100L)));
+ assertEquals("b", name(store.getFromEdgeName(100L, 101L)));
+ assertEquals(0, store.getDiffCandidateCount());
+ }
+ }
+
+ private static SnapDiffJobStore newStore(boolean fso) throws IOException {
+ return SnapDiffJobStore.open(db, codecRegistry, columnFamilyOptions,
+ "job" + JOB_ID.incrementAndGet(), fso, SnapDiffJobStore.Mode.FULL);
+ }
+
+ private static void putKey(Table table, String key, OmKeyInfo keyInfo)
+ throws CodecException, RocksDatabaseException {
+ table.put(tableKey(key), OmKeyInfo.getKeyTableCodec().toPersistedFormat(keyInfo));
+ }
+
+ private static void putDir(Table table, String key, OmDirectoryInfo dirInfo)
+ throws CodecException, RocksDatabaseException {
+ table.put(tableKey(key), OmDirectoryInfo.getCodec().toPersistedFormat(dirInfo));
+ }
+
+ private static byte[] tableKey(String key) throws CodecException {
+ return StringCodec.get().toPersistedFormat(key);
+ }
+
+ private static OmKeyInfo keyInfo(String keyName, long objectId, long parentId, long updateId, long dataSize) {
+ return new OmKeyInfo.Builder()
+ .setVolumeName(VOLUME)
+ .setBucketName(BUCKET)
+ .setKeyName(keyName)
+ .setReplicationConfig(RatisReplicationConfig.getInstance(ReplicationFactor.ONE))
+ .setObjectID(objectId)
+ .setParentObjectID(parentId)
+ .setUpdateID(updateId)
+ .setDataSize(dataSize)
+ .build();
+ }
+
+ private static OmDirectoryInfo dirInfo(String name, long objectId, long parentId, long updateId) {
+ return OmDirectoryInfo.newBuilder()
+ .setName(name)
+ .setObjectID(objectId)
+ .setParentObjectID(parentId)
+ .setUpdateID(updateId)
+ .build();
+ }
+
+ private static String name(byte[] value) {
+ return value == null ? null : new String(value, StandardCharsets.UTF_8);
+ }
+}
diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/snapshot/TestSnapshotDiffValueParser.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/snapshot/diff/TestSnapshotDiffValueParser.java
similarity index 99%
rename from hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/snapshot/TestSnapshotDiffValueParser.java
rename to hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/snapshot/diff/TestSnapshotDiffValueParser.java
index 95e268ca0b35..951ce6c57721 100644
--- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/snapshot/TestSnapshotDiffValueParser.java
+++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/snapshot/diff/TestSnapshotDiffValueParser.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.hadoop.ozone.om.snapshot;
+package org.apache.hadoop.ozone.om.snapshot.diff;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;