HDDS-15394. Sequential Reader of from and to snapshots for full snapshot diff.

hadoop-hdds/common/src/main/resources/ozone-default.xml

@@ -4954,6 +4954,16 @@
     </description>
   </property>
 
+  <property>
+    <name>ozone.om.snapshot.diff.max.in.memory.entries.per.job</name>
+    <value>1000000</value>
+    <tag>OZONE, OM</tag>
+    <description>
+      Maximum number of diff-candidate object IDs a snapshot diff job may retain
+      in memory before spilling to a temporary RocksDB column family.
+    </description>
+  </property>
+
   <property>
     <name>hdds.secret.key.file.name</name>
     <value>secret_keys.json</value>

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<KEY, VALUE> map, String name) {
-    this.map = new ConcurrentSkipListMap<>(map);
-    this.map.putAll(map);
+    this(new ConcurrentSkipListMap<>(map), name);
+  }
+
+  private InMemoryTestTable(NavigableMap<KEY, VALUE> map, String name) {
+    this.map = map;
     this.name = name;
   }
 
+  /** Raw {@code byte[]}/{@code byte[]} table with unsigned lexicographic key order. */
+  public static InMemoryTestTable<byte[], byte[]> forRawBytes() {
+    return new InMemoryTestTable<>(
+        new ConcurrentSkipListMap<>(UnsignedBytes.lexicographicalComparator()), "raw");
+  }
+
+  /** Raw {@code byte[]}/{@code byte[]} table with unsigned lexicographic key order. */
+  public static InMemoryTestTable<byte[], byte[]> 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<KEY, VALUE> iterator(KEY prefix, IteratorType type) {
-    throw new UnsupportedOperationException();
+    if (prefix instanceof byte[]) {
+      return new InMemoryKeyValueIterator<>(map, type, (byte[]) prefix);
+    }
+    NavigableMap<KEY, VALUE> 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<KEY, VALUE>
+      implements KeyValueIterator<KEY, VALUE> {
+    private final Iterator<Map.Entry<KEY, VALUE>> entries;
+    private final IteratorType type;
+    private final byte[] bytePrefix;
+    private Map.Entry<KEY, VALUE> lookahead;
+
+    private InMemoryKeyValueIterator(NavigableMap<KEY, VALUE> 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<KEY, VALUE> advance() {
+      while (entries.hasNext()) {
+        Map.Entry<KEY, VALUE> 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<KEY, VALUE> seek(KEY key) {
+      throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public boolean hasNext() {
+      if (lookahead != null) {
+        return true;
+      }
+      lookahead = advance();
+      return lookahead != null;
+    }
+
+    @Override
+    public KeyValue<KEY, VALUE> next() {
+      if (!hasNext()) {
+        throw new NoSuchElementException();
+      }
+      Map.Entry<KEY, VALUE> 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

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";
 

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}.
+ *
+ * <p>The fields are:
+ * <ul>
+ *   <li>{@code parentId} - parent object id (parent directory for FSO).</li>
+ *   <li>{@code name} - leaf name for FSO buckets, full key path for OBS.</li>
+ *   <li>{@code isDir} - whether the entry is a directory.</li>
+ *   <li>{@code signature} - SHA-256 compare signature computed by
+ *       {@code SnapshotDiffValueParser} over the meaningful fields.</li>
+ * </ul>
+ *
+ * <p>The wire layout is fixed so both the full diff and the DAG diff Stage 1
+ * readers produce identical bytes:
+ * <pre>
+ *   | parentId (8, big-endian) | isDir (1) | sigLen (4, big-endian) | signature | name (UTF-8, remaining) |
+ * </pre>
+ */
+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 + '}';
+  }
+}