From 5b06d497ddda3e0ff28c04603d9c3ae6d51bb434 Mon Sep 17 00:00:00 2001 From: SaketaChalamchala Date: Wed, 19 Aug 2026 14:47:32 -0700 Subject: [PATCH 1/2] HDDS-15394. Moved SnapshotDiffValueParser to om.snapshot.diff package. --- .../ozone/om/snapshot/{ => diff}/SnapshotDiffValueParser.java | 2 +- .../om/snapshot/{ => diff}/TestSnapshotDiffValueParser.java | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) rename hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/{ => diff}/SnapshotDiffValueParser.java (99%) rename hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/snapshot/{ => diff}/TestSnapshotDiffValueParser.java (99%) 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/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; From bc63f42b1a72e69c17afb9a97acff240c17a124d Mon Sep 17 00:00:00 2001 From: SaketaChalamchala Date: Fri, 21 Aug 2026 14:11:40 -0700 Subject: [PATCH 2/2] HDDS-15394. Full Diff Sequential reader of to and from snapshots. --- .../src/main/resources/ozone-default.xml | 10 + .../hdds/utils/db/InMemoryTestTable.java | 132 ++++++- .../apache/hadoop/ozone/om/OMConfigKeys.java | 7 + .../ozone/om/snapshot/diff/EntryValue.java | 147 ++++++++ .../diff/FullDiffSequentialReader.java | 205 ++++++++++ .../om/snapshot/diff/SnapDiffJobStore.java | 354 ++++++++++++++++++ .../diff/TestFullDiffSequentialReader.java | 299 +++++++++++++++ 7 files changed, 1151 insertions(+), 3 deletions(-) create mode 100644 hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/diff/EntryValue.java create mode 100644 hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/diff/FullDiffSequentialReader.java create mode 100644 hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/diff/SnapDiffJobStore.java create mode 100644 hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/snapshot/diff/TestFullDiffSequentialReader.java diff --git a/hadoop-hdds/common/src/main/resources/ozone-default.xml b/hadoop-hdds/common/src/main/resources/ozone-default.xml index 9528ca27fc21..248119303b29 100644 --- a/hadoop-hdds/common/src/main/resources/ozone-default.xml +++ b/hadoop-hdds/common/src/main/resources/ozone-default.xml @@ -4954,6 +4954,16 @@ + + ozone.om.snapshot.diff.max.in.memory.entries.per.job + 1000000 + OZONE, OM + + Maximum number of diff-candidate object IDs a snapshot diff job may retain + in memory before spilling to a temporary RocksDB column family. + + + hdds.secret.key.file.name secret_keys.json diff --git a/hadoop-hdds/framework/src/test/java/org/apache/hadoop/hdds/utils/db/InMemoryTestTable.java b/hadoop-hdds/framework/src/test/java/org/apache/hadoop/hdds/utils/db/InMemoryTestTable.java index 2f5b8be7f299..a462c9475bc0 100644 --- a/hadoop-hdds/framework/src/test/java/org/apache/hadoop/hdds/utils/db/InMemoryTestTable.java +++ b/hadoop-hdds/framework/src/test/java/org/apache/hadoop/hdds/utils/db/InMemoryTestTable.java @@ -17,11 +17,14 @@ package org.apache.hadoop.hdds.utils.db; +import com.google.common.primitives.UnsignedBytes; import java.io.File; import java.util.Collections; +import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.NavigableMap; +import java.util.NoSuchElementException; import java.util.concurrent.ConcurrentSkipListMap; import org.apache.hadoop.hdds.utils.MetadataKeyFilters.KeyPrefixFilter; @@ -45,11 +48,26 @@ public InMemoryTestTable(String name) { } public InMemoryTestTable(Map map, String name) { - this.map = new ConcurrentSkipListMap<>(map); - this.map.putAll(map); + this(new ConcurrentSkipListMap<>(map), name); + } + + private InMemoryTestTable(NavigableMap map, String name) { + this.map = map; this.name = name; } + /** Raw {@code byte[]}/{@code byte[]} table with unsigned lexicographic key order. */ + public static InMemoryTestTable forRawBytes() { + return new InMemoryTestTable<>( + new ConcurrentSkipListMap<>(UnsignedBytes.lexicographicalComparator()), "raw"); + } + + /** Raw {@code byte[]}/{@code byte[]} table with unsigned lexicographic key order. */ + public static InMemoryTestTable forRawBytes(String name) { + return new InMemoryTestTable<>( + new ConcurrentSkipListMap<>(UnsignedBytes.lexicographicalComparator()), name); + } + @Override public void put(KEY key, VALUE value) { map.put(key, value); @@ -102,7 +120,115 @@ public void clear() { @Override public KeyValueIterator iterator(KEY prefix, IteratorType type) { - throw new UnsupportedOperationException(); + if (prefix instanceof byte[]) { + return new InMemoryKeyValueIterator<>(map, type, (byte[]) prefix); + } + NavigableMap view; + if (prefix == null) { + view = map; + } else if (prefix instanceof String) { + String endPrefix = (String) prefix + Character.MAX_VALUE; + view = map.subMap(prefix, true, (KEY) endPrefix, false); + } else { + view = map.tailMap(prefix, true); + } + return new InMemoryKeyValueIterator<>(view, type, null); + } + + private static final class InMemoryKeyValueIterator + implements KeyValueIterator { + private final Iterator> entries; + private final IteratorType type; + private final byte[] bytePrefix; + private Map.Entry lookahead; + + private InMemoryKeyValueIterator(NavigableMap map, IteratorType type, + byte[] bytePrefix) { + if (bytePrefix == null || bytePrefix.length == 0) { + this.entries = map.entrySet().iterator(); + } else { + this.entries = map.tailMap((KEY) bytePrefix, true).entrySet().iterator(); + } + this.type = type; + this.bytePrefix = bytePrefix; + } + + private boolean startsWithPrefix(byte[] key) { + if (bytePrefix == null || bytePrefix.length == 0) { + return true; + } + if (key == null || key.length < bytePrefix.length) { + return false; + } + for (int i = 0; i < bytePrefix.length; i++) { + if (key[i] != bytePrefix[i]) { + return false; + } + } + return true; + } + + private Map.Entry advance() { + while (entries.hasNext()) { + Map.Entry entry = entries.next(); + if (!(entry.getKey() instanceof byte[]) + || startsWithPrefix((byte[]) entry.getKey())) { + return entry; + } + } + return null; + } + + @Override + public void seekToFirst() { + throw new UnsupportedOperationException(); + } + + @Override + public void seekToLast() { + throw new UnsupportedOperationException(); + } + + @Override + public KeyValue seek(KEY key) { + throw new UnsupportedOperationException(); + } + + @Override + public boolean hasNext() { + if (lookahead != null) { + return true; + } + lookahead = advance(); + return lookahead != null; + } + + @Override + public KeyValue next() { + if (!hasNext()) { + throw new NoSuchElementException(); + } + Map.Entry 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/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); + } +}