jojochuang commented on code in PR #11083: URL: https://github.com/apache/ozone/pull/11083#discussion_r3899492226
########## hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/snapshot/diff/TestFullDiffSequentialReader.java: ########## @@ -0,0 +1,320 @@ +/* + * 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<ColumnFamilyDescriptor> descriptors = Collections.singletonList( + new ColumnFamilyDescriptor(StringUtils.string2Bytes(DEFAULT_COLUMN_FAMILY_NAME), columnFamilyOptions)); + List<ColumnFamilyHandle> 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<byte[], byte[]> 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<byte[], byte[]> 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)); Review Comment: EntryValue is mostly a trivial class except for fromBytes() and toBytes(). There's no test for EntryValue.fromBytes() + EntryValue.toBytes() to verify they are compatible. ########## 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. + * + * <p>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. + * + * <p>Scans iterate the raw snapshot tables ({@code Table<byte[], byte[]>} from + * {@code DBStore#getTable(String)}) so {@link SnapshotDiffValueParser} reads the exact + * persisted protobuf bytes and compare signatures match on-disk layout. + * + * <p>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. + * + * <p>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. + * + * <p>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<byte[], byte[]> fromTable, + Table<byte[], byte[]> 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<byte[], byte[]> fromTable, + Table<byte[], byte[]> toTable, byte[] keyPrefix) throws IOException { + scanToTable(toTable, keyPrefix, true); + scanFromTable(fromTable, keyPrefix, true); + } + + private void scanToTable(Table<byte[], byte[]> table, byte[] keyPrefix, boolean isDir) + throws IOException { + try (Table.KeyValueIterator<byte[], byte[]> 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<byte[], byte[]> table, byte[] keyPrefix, boolean isDir) + throws IOException { + store.flushWrites(); Review Comment: this flushWrites() is probably redundant. scanFromTable() is always called right after scanToTable(), which does a flushWrites() right before exit. ########## hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/diff/SnapDiffJobStore.java: ########## @@ -0,0 +1,370 @@ +/* + * 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.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. + * + * <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>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. + */ +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<Long> diffCandidates; + private ColumnFamilyHandle diffCandidatesCf; + private boolean diffCandidatesSpilled; + + private final ManagedWriteBatch writeBatch; + private final ManagedWriteOptions writeOptions; + private final int writeBatchSize; + private int pendingOps; + + /** 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]; + + /** 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, objectIdKeyBuffer(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, objectIdKeyBuffer(objectId), entryValue); + } + + public void putOldList(long objectId, byte[] entryValue) throws IOException { + batchPut(oldListCf, objectIdKeyBuffer(objectId), entryValue); + } + + public void putToEdge(long parentId, long objectId, byte[] name) throws IOException { + requireFso(); + batchPut(toEdgesCf, edgeKeyBuffer(parentId, objectId), name); + } + + public void putFromEdge(long parentId, long objectId, byte[] name) throws IOException { + requireFso(); + batchPut(fromEdgesCf, edgeKeyBuffer(parentId, objectId), name); + } + + public byte[] getNewList(long objectId) throws IOException { + return get(newListCf, objectIdKeyBuffer(objectId)); + } + + public byte[] getOldList(long objectId) throws IOException { + return get(oldListCf, objectIdKeyBuffer(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, objectIdKeyBuffer(objectId), presentMarker); + return; + } + if (diffCandidates.size() >= maxInMemoryEntries) { + spillDiffCandidates(); + } + if (diffCandidatesSpilled) { + batchPut(diffCandidatesCf, objectIdKeyBuffer(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, objectIdKeyBuffer(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, edgeKeyBuffer(parentId, objectId)); + } + + public byte[] getFromEdgeName(long parentId, long objectId) throws IOException { + requireFso(); + return get(fromEdgesCf, edgeKeyBuffer(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; + } + + private byte[] objectIdKeyBuffer(long objectId) { + encodeLong(objectIdKeyBuffer, 0, objectId); + return objectIdKeyBuffer; + } + + private byte[] edgeKeyBuffer(long parentId, long objectId) { + encodeLong(edgeKeyBuffer, 0, parentId); + encodeLong(edgeKeyBuffer, Long.BYTES, objectId); + return edgeKeyBuffer; + } + + private static void encodeLong(byte[] buffer, int offset, long value) { + for (int shift = Long.SIZE - 8; shift >= 0; shift -= 8) { + buffer[offset++] = (byte) (value >>> shift); + } + } + + 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, objectIdKeyBuffer(objectId), presentMarker); Review Comment: reuse the same byte array is fine. The bytes are copied by ManagedWriteBatch immediately it does not keep the reference. ########## hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/diff/SnapDiffJobStore.java: ########## @@ -0,0 +1,370 @@ +/* + * 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.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. + * + * <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>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. + */ +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<Long> diffCandidates; + private ColumnFamilyHandle diffCandidatesCf; + private boolean diffCandidatesSpilled; + + private final ManagedWriteBatch writeBatch; + private final ManagedWriteOptions writeOptions; + private final int writeBatchSize; + private int pendingOps; + + /** 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]; + + /** 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, objectIdKeyBuffer(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, objectIdKeyBuffer(objectId), entryValue); + } + + public void putOldList(long objectId, byte[] entryValue) throws IOException { + batchPut(oldListCf, objectIdKeyBuffer(objectId), entryValue); + } + + public void putToEdge(long parentId, long objectId, byte[] name) throws IOException { + requireFso(); + batchPut(toEdgesCf, edgeKeyBuffer(parentId, objectId), name); + } + + public void putFromEdge(long parentId, long objectId, byte[] name) throws IOException { + requireFso(); + batchPut(fromEdgesCf, edgeKeyBuffer(parentId, objectId), name); + } + + public byte[] getNewList(long objectId) throws IOException { + return get(newListCf, objectIdKeyBuffer(objectId)); + } + + public byte[] getOldList(long objectId) throws IOException { + return get(oldListCf, objectIdKeyBuffer(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, objectIdKeyBuffer(objectId), presentMarker); + return; + } + if (diffCandidates.size() >= maxInMemoryEntries) { + spillDiffCandidates(); + } + if (diffCandidatesSpilled) { + batchPut(diffCandidatesCf, objectIdKeyBuffer(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, objectIdKeyBuffer(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, edgeKeyBuffer(parentId, objectId)); + } + + public byte[] getFromEdgeName(long parentId, long objectId) throws IOException { + requireFso(); + return get(fromEdgesCf, edgeKeyBuffer(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; + } + + private byte[] objectIdKeyBuffer(long objectId) { + encodeLong(objectIdKeyBuffer, 0, objectId); + return objectIdKeyBuffer; + } + + private byte[] edgeKeyBuffer(long parentId, long objectId) { + encodeLong(edgeKeyBuffer, 0, parentId); + encodeLong(edgeKeyBuffer, Long.BYTES, objectId); + return edgeKeyBuffer; + } + + private static void encodeLong(byte[] buffer, int offset, long value) { + for (int shift = Long.SIZE - 8; shift >= 0; shift -= 8) { + buffer[offset++] = (byte) (value >>> shift); + } + } + + 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, objectIdKeyBuffer(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 { Review Comment: Partial cleanup if close() throws mid-way ########## 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. + * + * <p>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. + * + * <p>Scans iterate the raw snapshot tables ({@code Table<byte[], byte[]>} from + * {@code DBStore#getTable(String)}) so {@link SnapshotDiffValueParser} reads the exact + * persisted protobuf bytes and compare signatures match on-disk layout. + * + * <p>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. + * + * <p>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. + * + * <p>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<byte[], byte[]> fromTable, + Table<byte[], byte[]> 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<byte[], byte[]> fromTable, + Table<byte[], byte[]> toTable, byte[] keyPrefix) throws IOException { + scanToTable(toTable, keyPrefix, true); + scanFromTable(fromTable, keyPrefix, true); + } + + private void scanToTable(Table<byte[], byte[]> table, byte[] keyPrefix, boolean isDir) + throws IOException { + try (Table.KeyValueIterator<byte[], byte[]> 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<byte[], byte[]> table, byte[] keyPrefix, boolean isDir) + throws IOException { + store.flushWrites(); Review Comment: not a big issue as flushWrites is a no-op if pendingWrites == 0 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
