This is an automated email from the ASF dual-hosted git repository.
SaketaChalamchala pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/master by this push:
new 4f5ae5454fe HDDS-15388. K way merge iterator over native SST file
reader to emit latest tombstone and latest value of a key. (#10484)
4f5ae5454fe is described below
commit 4f5ae5454fef0ecb9fbec8750af6204f4ee28540
Author: SaketaChalamchala <[email protected]>
AuthorDate: Fri Jul 24 13:06:44 2026 -0700
HDDS-15388. K way merge iterator over native SST file reader to emit latest
tombstone and latest value of a key. (#10484)
---
.../dev-support/findbugsExcludeFile.xml | 9 +
.../utils/db/LatestVersionedKWayMergeIterator.java | 504 +++++++++++++++++++++
.../db/TestLatestVersionedKWayMergeIterator.java | 253 +++++++++++
...estLatestVersionedKWayMergeIteratorOverSst.java | 204 +++++++++
.../hdds/utils/db/TestRawSstFileRecords.java | 121 +++++
5 files changed, 1091 insertions(+)
diff --git a/hadoop-hdds/rocks-native/dev-support/findbugsExcludeFile.xml
b/hadoop-hdds/rocks-native/dev-support/findbugsExcludeFile.xml
index 40d78d0cd6c..9b97abe8e46 100644
--- a/hadoop-hdds/rocks-native/dev-support/findbugsExcludeFile.xml
+++ b/hadoop-hdds/rocks-native/dev-support/findbugsExcludeFile.xml
@@ -15,4 +15,13 @@
limitations under the License.
-->
<FindBugsFilter>
+ <!-- MergedKeyValue owns key/value arrays copied once at emit time; callers
must not mutate. -->
+ <Match>
+ <Class
name="org.apache.hadoop.hdds.utils.db.LatestVersionedKWayMergeIterator$MergedKeyValue"/>
+ <Or>
+ <Method name="getUserKey"/>
+ <Method name="getValue"/>
+ </Or>
+ <Bug pattern="EI_EXPOSE_REP"/>
+ </Match>
</FindBugsFilter>
diff --git
a/hadoop-hdds/rocks-native/src/main/java/org/apache/hadoop/hdds/utils/db/LatestVersionedKWayMergeIterator.java
b/hadoop-hdds/rocks-native/src/main/java/org/apache/hadoop/hdds/utils/db/LatestVersionedKWayMergeIterator.java
new file mode 100644
index 00000000000..fc072c7fde6
--- /dev/null
+++
b/hadoop-hdds/rocks-native/src/main/java/org/apache/hadoop/hdds/utils/db/LatestVersionedKWayMergeIterator.java
@@ -0,0 +1,504 @@
+/*
+ * 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.hdds.utils.db;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.primitives.UnsignedLong;
+import java.nio.ByteBuffer;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.NoSuchElementException;
+import java.util.Objects;
+import java.util.PriorityQueue;
+import org.apache.hadoop.hdds.utils.IOUtils;
+import org.apache.hadoop.hdds.utils.db.managed.ManagedOptions;
+import org.apache.hadoop.ozone.util.ClosableIterator;
+
+/**
+ * K-way merge over RocksDB SST files for snapshot diff.
+ * <p>
+ * A single min-heap orders source heads by user key. For each user key, all
versions
+ * are drained and the latest value and latest tombstone are tracked, then
snapshot-diff
+ * emit rules apply: emit the latest tombstone and/or latest value, including
both when
+ * a delete is followed by a newer recreate.
+ * <p>
+ * When constructed with an exclusive minimum sequence number {@code S}, each
SST
+ * source skips entries whose sequence is {@code <= S} while advancing.
+ */
+public final class LatestVersionedKWayMergeIterator implements
+ ClosableIterator<LatestVersionedKWayMergeIterator.MergedKeyValue> {
+
+ /** RocksDB {@code ValueType::kTypeValue}. */
+ public static final int ROCKS_TYPE_VALUE = 1;
+ private static final int DEFAULT_READ_AHEAD_SIZE = 2 * 1024 * 1024;
+
+ private final ManagedOptions options;
+ private final List<ClosableIterator<? extends MergeHead>> iterators;
+ private final Long exclusiveMinSequenceNumber;
+
+ private final PriorityQueue<HeapEntry> sourceHeads;
+
+ private List<MergedKeyValue> emitQueue;
+ private boolean initialized;
+
+ public static LatestVersionedKWayMergeIterator
overRawSstFiles(Collection<Path> sstFiles) {
+ return overRawSstFiles(sstFiles, DEFAULT_READ_AHEAD_SIZE, null);
+ }
+
+ public static LatestVersionedKWayMergeIterator
overRawSstFiles(Collection<Path> sstFiles,
+ int readAheadSizePerFile) {
+ return overRawSstFiles(sstFiles, readAheadSizePerFile, null);
+ }
+
+ /**
+ * Opens one iterator per SST file and merges them.
+ *
+ * @param exclusiveMinSequenceNumber when non-null, each source skips
entries with
+ * sequence {@code <=} this value while advancing; when null, no
entries are skipped
+ */
+ public static LatestVersionedKWayMergeIterator
overRawSstFiles(Collection<Path> sstFiles,
+ int readAheadSizePerFile, Long exclusiveMinSequenceNumber) {
+ Objects.requireNonNull(sstFiles, "sstFiles cannot be null");
+ ManagedOptions options = new ManagedOptions();
+ List<ClosableIterator<? extends MergeHead>> sources = new
ArrayList<>(sstFiles.size());
+ try {
+ for (Path file : sstFiles) {
+ sources.add(new RawSstIterator(options, file, readAheadSizePerFile));
+ }
+ return new LatestVersionedKWayMergeIterator(options, sources,
exclusiveMinSequenceNumber);
+ } catch (RuntimeException e) {
+ IOUtils.closeQuietly(sources);
+ options.close();
+ throw e;
+ }
+ }
+
+ public static LatestVersionedKWayMergeIterator
overRawSstFilesFromSequence(Collection<Path> sstFiles,
+ long exclusiveMinSequenceNumber) {
+ return overRawSstFiles(sstFiles, DEFAULT_READ_AHEAD_SIZE,
exclusiveMinSequenceNumber);
+ }
+
+ @VisibleForTesting
+ public static LatestVersionedKWayMergeIterator forTest(
+ List<ClosableIterator<MergedKeyValue>> iterators, Long
exclusiveMinSequenceNumber) {
+ List<ClosableIterator<? extends MergeHead>> sources = new
ArrayList<>(iterators.size());
+ sources.addAll(iterators);
+ return new LatestVersionedKWayMergeIterator(null, sources,
exclusiveMinSequenceNumber);
+ }
+
+ private LatestVersionedKWayMergeIterator(
+ ManagedOptions options,
+ List<ClosableIterator<? extends MergeHead>> iterators,
+ Long exclusiveMinSequenceNumber) {
+ this.options = options;
+ this.iterators = new ArrayList<>(Objects.requireNonNull(iterators,
"iterators cannot be null"));
+ this.exclusiveMinSequenceNumber = exclusiveMinSequenceNumber;
+ this.sourceHeads = new PriorityQueue<>(Math.max(this.iterators.size(), 1));
+ this.emitQueue = new ArrayList<>();
+ }
+
+ @Override
+ public boolean hasNext() {
+ if (!emitQueue.isEmpty()) {
+ return true;
+ }
+ return advance();
+ }
+
+ @Override
+ public MergedKeyValue next() {
+ if (!hasNext()) {
+ throw new NoSuchElementException("No more elements found.");
+ }
+ return emitQueue.remove(0);
+ }
+
+ private boolean advance() {
+ if (!initialized) {
+ initHeap();
+ initialized = true;
+ }
+
+ while (emitQueue.isEmpty() && !sourceHeads.isEmpty()) {
+ processNextUserKey();
+ }
+
+ return !emitQueue.isEmpty();
+ }
+
+ private void processNextUserKey() {
+ if (sourceHeads.isEmpty()) {
+ return;
+ }
+
+ byte[] nextKey = sourceHeads.peek().current.getUserKey();
+ MergeHead latestValue = null;
+ long latestValueSeq = -1L;
+ MergeHead latestTombstone = null;
+ long latestTombstoneSeq = -1L;
+
+ while (heapHasUserKey(nextKey)) {
+ List<HeapEntry> polled = new ArrayList<>();
+ while (heapHasUserKey(nextKey)) {
+ HeapEntry entry = sourceHeads.poll();
+ MergeHead head = entry.current;
+ if (head.isTombstone()) {
+ if (head.getSequence() > latestTombstoneSeq) {
+ latestTombstone = head;
+ latestTombstoneSeq = head.getSequence();
+ }
+ } else if (head.getSequence() > latestValueSeq) {
+ latestValue = head;
+ latestValueSeq = head.getSequence();
+ }
+ polled.add(entry);
+ }
+ for (HeapEntry entry : polled) {
+ if (entry.current == latestValue && entry.current instanceof
RawSstHeapHead) {
+ ((RawSstHeapHead) entry.current).snapshotValue();
+ }
+ entry.advance();
+ if (entry.current != null) {
+ sourceHeads.offer(entry);
+ }
+ }
+ }
+
+ emitForUserKey(latestValue, latestTombstone);
+ }
+
+ private void emitForUserKey(MergeHead latestValue, MergeHead
latestTombstone) {
+ if (latestValue != null && latestTombstone != null) {
+ if (latestValue.getSequence() > latestTombstone.getSequence()) {
+ emitQueue.add(latestTombstone.toMergedKeyValue());
+ emitQueue.add(latestValue.toMergedKeyValue());
+ } else {
+ emitQueue.add(latestTombstone.toMergedKeyValue());
+ }
+ } else if (latestValue != null) {
+ emitQueue.add(latestValue.toMergedKeyValue());
+ } else if (latestTombstone != null) {
+ emitQueue.add(latestTombstone.toMergedKeyValue());
+ }
+ }
+
+ private boolean heapHasUserKey(byte[] userKey) {
+ return !sourceHeads.isEmpty()
+ && compareUserKeys(sourceHeads.peek().current.getUserKey(), userKey)
== 0;
+ }
+
+ private void initHeap() {
+ for (int idx = 0; idx < iterators.size(); idx++) {
+ ClosableIterator<? extends MergeHead> iterator = iterators.get(idx);
+ HeapEntry entry = new HeapEntry(idx, iterator);
+ if (entry.current != null) {
+ sourceHeads.offer(entry);
+ }
+ }
+ }
+
+ @Override
+ public void close() {
+ IOUtils.closeQuietly(iterators);
+ if (options != null) {
+ options.close();
+ }
+ }
+
+ private static int compareUserKeys(byte[] left, byte[] right) {
+ if (left == right) {
+ return 0;
+ }
+ if (left == null) {
+ return -1;
+ }
+ if (right == null) {
+ return 1;
+ }
+ int minLength = Math.min(left.length, right.length);
+ for (int i = 0; i < minLength; i++) {
+ int l = left[i] & 0xff;
+ int r = right[i] & 0xff;
+ if (l != r) {
+ return Integer.compare(l, r);
+ }
+ }
+ return Integer.compare(left.length, right.length);
+ }
+
+ private static int compareHeapOrder(MergeHead left, int leftIndex,
+ MergeHead right, int rightIndex) {
+ int keyCompare = compareUserKeys(left.getUserKey(), right.getUserKey());
+ if (keyCompare != 0) {
+ return keyCompare;
+ }
+ int seqCompare = Long.compare(right.getSequence(), left.getSequence());
+ if (seqCompare != 0) {
+ return seqCompare;
+ }
+ int tombstoneCompare = Boolean.compare(right.isTombstone(),
left.isTombstone());
+ if (tombstoneCompare != 0) {
+ return tombstoneCompare;
+ }
+ return Integer.compare(leftIndex, rightIndex);
+ }
+
+ private interface MergeHead {
+ byte[] getUserKey();
+
+ long getSequence();
+
+ boolean isTombstone();
+
+ MergedKeyValue toMergedKeyValue();
+ }
+
+ private static final class RawSstHeapHead implements MergeHead {
+ private final byte[] key;
+ private final long sequence;
+ private final int type;
+ private final CodecBuffer valueBuffer;
+ private byte[] snapshottedValue;
+
+ RawSstHeapHead(byte[] key, long sequence, int type, CodecBuffer
valueBuffer) {
+ this.key = Objects.requireNonNull(key, "key cannot be null");
+ this.sequence = sequence;
+ this.type = type;
+ this.valueBuffer = valueBuffer;
+ }
+
+ void snapshotValue() {
+ if (snapshottedValue != null || isTombstone() || valueBuffer == null) {
+ return;
+ }
+ snapshottedValue = copyBuffer(valueBuffer);
+ }
+
+ @Override
+ public byte[] getUserKey() {
+ return key;
+ }
+
+ @Override
+ public long getSequence() {
+ return sequence;
+ }
+
+ @Override
+ public boolean isTombstone() {
+ return type != ROCKS_TYPE_VALUE;
+ }
+
+ @Override
+ public MergedKeyValue toMergedKeyValue() {
+ byte[] value = snapshottedValue;
+ if (value == null && !isTombstone() && valueBuffer != null) {
+ value = copyBuffer(valueBuffer);
+ }
+ return new MergedKeyValue(key, UnsignedLong.fromLongBits(sequence),
type, value);
+ }
+ }
+
+ private final class HeapEntry implements Comparable<HeapEntry> {
+ private final int index;
+ private final ClosableIterator<? extends MergeHead> iterator;
+ private MergeHead current;
+
+ private HeapEntry(int index, ClosableIterator<? extends MergeHead>
iterator) {
+ this.index = index;
+ this.iterator = iterator;
+ advance();
+ }
+
+ private void advance() {
+ while (true) {
+ if (!iterator.hasNext()) {
+ current = null;
+ iterator.close();
+ return;
+ }
+ MergeHead next = iterator.next();
+ if (exclusiveMinSequenceNumber == null
+ || next.getSequence() > exclusiveMinSequenceNumber) {
+ current = next;
+ return;
+ }
+ }
+ }
+
+ @Override
+ public int compareTo(HeapEntry other) {
+ return compareHeapOrder(
+ this.current, this.index, other.current, other.index);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof HeapEntry)) {
+ return false;
+ }
+ HeapEntry other = (HeapEntry) o;
+ return index == other.index;
+ }
+
+ @Override
+ public int hashCode() {
+ return index;
+ }
+ }
+
+ private static final class RawSstIterator implements
ClosableIterator<RawSstHeapHead> {
+ private final ManagedRawSSTFileReader reader;
+ private final
ManagedRawSSTFileIterator<ManagedRawSSTFileIterator.KeyValue> iterator;
+ private boolean closed;
+
+ private RawSstIterator(ManagedOptions options, Path file, int
readAheadSize) {
+ ManagedRawSSTFileReader openedReader = new ManagedRawSSTFileReader(
+ options, file.toAbsolutePath().toString(), readAheadSize);
+ ManagedRawSSTFileIterator<ManagedRawSSTFileIterator.KeyValue>
openedIterator;
+ try {
+ openedIterator = openedReader.newIterator(
+ kv -> kv, null, null, IteratorType.KEY_AND_VALUE);
+ } catch (RuntimeException e) {
+ openedReader.close();
+ throw e;
+ }
+ this.reader = openedReader;
+ this.iterator = openedIterator;
+ }
+
+ @Override
+ public boolean hasNext() {
+ return iterator.hasNext();
+ }
+
+ @Override
+ public RawSstHeapHead next() {
+ ManagedRawSSTFileIterator.KeyValue keyValue = iterator.next();
+ int type = keyValue.getType();
+ CodecBuffer valueBuffer = type == ROCKS_TYPE_VALUE ? keyValue.getValue()
: null;
+ return new RawSstHeapHead(
+ copyBuffer(keyValue.getKey()),
+ keyValue.getSequence().longValue(),
+ type,
+ valueBuffer);
+ }
+
+ @Override
+ public void close() {
+ if (closed) {
+ return;
+ }
+ closed = true;
+ iterator.close();
+ reader.close();
+ }
+ }
+
+ /**
+ * A merged RocksDB internal key after applying snapshot-diff emit rules
across
+ * multiple SST sources.
+ * <p>
+ * Each instance represents one emitted version for a user key: either a
value
+ * ({@link #ROCKS_TYPE_VALUE}) or a tombstone (any other RocksDB value type).
+ * When the latest value has a higher sequence than the latest tombstone for
+ * the same user key, both are emitted (delete followed by recreate).
Otherwise
+ * only the winning tombstone or value is emitted.
+ * <p>
+ * Key and value arrays are owned by this object. Callers must treat returned
+ * {@code byte[]} references as read-only; snapshot-diff integration should
+ * consume them once without retaining mutable aliases.
+ */
+ public static final class MergedKeyValue implements MergeHead {
+ private final byte[] key;
+ private final UnsignedLong sequence;
+ private final int type;
+ private final byte[] value;
+
+ /** Creates a merged entry for unit tests. */
+ @VisibleForTesting
+ public static MergedKeyValue of(byte[] key, long sequence, int type,
byte[] value) {
+ return new MergedKeyValue(
+ Arrays.copyOf(key, key.length),
+ UnsignedLong.fromLongBits(sequence),
+ type,
+ value == null ? null : Arrays.copyOf(value, value.length));
+ }
+
+ private MergedKeyValue(byte[] key, UnsignedLong sequence, int type, byte[]
value) {
+ this.key = key;
+ this.sequence = Objects.requireNonNull(sequence, "sequence cannot be
null");
+ this.type = type;
+ this.value = value;
+ }
+
+ /** Returns the user key bytes shared by all versions merged for this
emit. */
+ @Override
+ public byte[] getUserKey() {
+ return key;
+ }
+
+ /** Returns the RocksDB sequence number as a signed {@code long}. */
+ @Override
+ public long getSequence() {
+ return sequence.longValue();
+ }
+
+ /** Returns the RocksDB internal value type for this record. */
+ public int getValueType() {
+ return type;
+ }
+
+ /** Returns the RocksDB sequence number. */
+ public UnsignedLong getSequenceNumber() {
+ return sequence;
+ }
+
+ /** Returns the value bytes, or {@code null} for tombstones. */
+ public byte[] getValue() {
+ return value;
+ }
+
+ /** Returns {@code true} when this record is a tombstone rather than a
value. */
+ @Override
+ public boolean isTombstone() {
+ return type != ROCKS_TYPE_VALUE;
+ }
+
+ @Override
+ public MergedKeyValue toMergedKeyValue() {
+ return this;
+ }
+ }
+
+ private static byte[] copyBuffer(CodecBuffer buffer) {
+ if (buffer == null) {
+ return null;
+ }
+ ByteBuffer byteBuffer = buffer.asReadOnlyByteBuffer();
+ byte[] bytes = new byte[byteBuffer.remaining()];
+ byteBuffer.get(bytes);
+ return bytes;
+ }
+}
diff --git
a/hadoop-hdds/rocks-native/src/test/java/org/apache/hadoop/hdds/utils/db/TestLatestVersionedKWayMergeIterator.java
b/hadoop-hdds/rocks-native/src/test/java/org/apache/hadoop/hdds/utils/db/TestLatestVersionedKWayMergeIterator.java
new file mode 100644
index 00000000000..76c79362a46
--- /dev/null
+++
b/hadoop-hdds/rocks-native/src/test/java/org/apache/hadoop/hdds/utils/db/TestLatestVersionedKWayMergeIterator.java
@@ -0,0 +1,253 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hdds.utils.db;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import
org.apache.hadoop.hdds.utils.db.LatestVersionedKWayMergeIterator.MergedKeyValue;
+import org.apache.hadoop.ozone.util.ClosableIterator;
+import org.junit.jupiter.api.Named;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+class TestLatestVersionedKWayMergeIterator {
+
+ private static Stream<Arguments> mergeScenarios() {
+ return Stream.of(
+ Named.of("recreate: tombstone then newer value emits both",
+ scenario(
+ expected(kv("k1", 2, 0, null), kv("k1", 3, 1, "v3")),
+ source(kv("k1", 1, 1, "v1"), kv("k1", 2, 0, null)),
+ source(kv("k1", 3, 1, "v3")))),
+ Named.of("delete only: latest tombstone wins",
+ scenario(
+ expected(kv("k1", 5, 0, null)),
+ source(kv("k1", 1, 1, "v1")),
+ source(kv("k1", 5, 0, null)))),
+ Named.of("value only: latest value wins",
+ scenario(
+ expected(kv("k1", 5, 1, "v5")),
+ source(kv("k1", 1, 1, "v1")),
+ source(kv("k1", 5, 1, "v5")))),
+ Named.of("three-file worked example",
+ scenario(
+ expected(kv("k1", 30, 1, "v30"), kv("k2", 15, 0, null),
kv("k2", 25, 1, "v25")),
+ source(kv("k1", 10, 1, "v10"), kv("k1", 5, 1, "v5"), kv("k2",
20, 1, "v20")),
+ source(kv("k1", 3, 1, "v3"), kv("k1", 15, 1, "v15"), kv("k2",
15, 0, null)),
+ source(kv("k1", 1, 1, "v1"), kv("k1", 30, 1, "v30"), kv("k2",
25, 1, "v25")))),
+ Named.of("multi-key: recreate on k1, delete-only on k2",
+ scenario(
+ expected(kv("k1", 3, 0, null), kv("k1", 10, 1, "v10"),
kv("k2", 15, 0, null)),
+ source(kv("k1", 10, 1, "v10"), kv("k1", 3, 0, null), kv("k2",
10, 1, "v10")),
+ source(kv("k2", 15, 0, null)))),
+ Named.of("duplicate tombstones deduped to highest sequence",
+ scenario(
+ expected(kv("k1", 7, 0, null)),
+ source(kv("k1", 4, 0, null), kv("k1", 2, 0, null)),
+ source(kv("k1", 7, 0, null)))),
+ Named.of("multiple recreate cycles on same key",
+ scenario(
+ expected(kv("k1", 4, 0, null), kv("k1", 6, 1, "v6")),
+ source(
+ kv("k1", 1, 1, "v1"),
+ kv("k1", 2, 0, null),
+ kv("k1", 3, 1, "v3"),
+ kv("k1", 4, 0, null),
+ kv("k1", 5, 1, "v5"),
+ kv("k1", 6, 1, "v6")))),
+ Named.of("interleaved keys preserve user-key order",
+ scenario(
+ expected(kv("a", 1, 1, "a1"), kv("b", 2, 1, "b1"), kv("c", 3,
1, "c1")),
+ source(kv("a", 1, 1, "a1"), kv("c", 3, 1, "c1")),
+ source(kv("b", 2, 1, "b1")))),
+ Named.of("empty source files are ignored",
+ scenario(
+ expected(kv("k1", 2, 1, "v2")),
+ source(kv("k1", 1, 1, "v1")),
+ source(),
+ source(kv("k1", 2, 1, "v2"))))
+ ).map(Arguments::of);
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("mergeScenarios")
+ void testMergeMatchesExpectedOutput(MergeScenario scenario) {
+ assertResultsEqual(scenario.expected, merge(scenario.sources));
+ }
+
+ @Test
+ void testExclusiveMinSequenceNumberFiltersPerKey() {
+ long exclusiveMinSequenceNumber = 5L;
+ List<List<MergedKeyValue>> sources = Arrays.asList(
+ Arrays.asList(
+ kv("k1", 7, 1, "v7"),
+ kv("k2", 1, 1, "v1"),
+ kv("k3", 6, 0, null),
+ kv("k4", 10, 1, "v10")),
+ Arrays.asList(
+ kv("k1", 3, 0, null),
+ kv("k2", 3, 0, null),
+ kv("k3", 8, 1, "v8")));
+ List<MergedKeyValue> expected = Arrays.asList(
+ kv("k1", 7, 1, "v7"),
+ kv("k3", 6, 0, null),
+ kv("k3", 8, 1, "v8"),
+ kv("k4", 10, 1, "v10"));
+
+ assertResultsEqual(expected, merge(sources, exclusiveMinSequenceNumber));
+ }
+
+ private static List<MergedKeyValue> merge(List<List<MergedKeyValue>>
sources) {
+ return merge(sources, null);
+ }
+
+ private static List<MergedKeyValue> merge(List<List<MergedKeyValue>> sources,
+ Long exclusiveMinSequenceNumber) {
+ List<ClosableIterator<MergedKeyValue>> iterators = sources.stream()
+ .map(ListIterator::new)
+ .collect(Collectors.toList());
+
+ List<MergedKeyValue> results = new ArrayList<>();
+ try (LatestVersionedKWayMergeIterator iterator =
+ LatestVersionedKWayMergeIterator.forTest(iterators,
exclusiveMinSequenceNumber)) {
+ while (iterator.hasNext()) {
+ results.add(iterator.next());
+ }
+ }
+ return results;
+ }
+
+ private static void assertResultsEqual(List<MergedKeyValue> expected,
List<MergedKeyValue> actual) {
+ assertEquals(expected.size(), actual.size(),
+ () -> "expected=" + describe(expected) + " actual=" +
describe(actual));
+
+ for (int i = 0; i < expected.size(); i++) {
+ MergedKeyValue exp = expected.get(i);
+ MergedKeyValue act = actual.get(i);
+ assertArrayEquals(exp.getUserKey(), act.getUserKey(), "key mismatch at
index " + i);
+ assertEquals(exp.getSequence(), act.getSequence(), "sequence mismatch at
index " + i);
+ assertEquals(exp.getValueType(), act.getValueType(), "type mismatch at
index " + i);
+ if (exp.getValue() == null) {
+ assertNull(act.getValue(), "value should be null at index " + i);
+ } else {
+ assertArrayEquals(exp.getValue(), act.getValue(), "value mismatch at
index " + i);
+ }
+ }
+ }
+
+ private static String describe(List<MergedKeyValue> entries) {
+ StringBuilder sb = new StringBuilder("[");
+ for (MergedKeyValue entry : entries) {
+ sb.append('{')
+ .append(asString(entry.getUserKey()))
+ .append(", seq=").append(entry.getSequence())
+ .append(", type=").append(entry.getValueType())
+ .append("} ");
+ }
+ return sb.append(']').toString();
+ }
+
+ private static MergeScenario scenario(Expected expected, Source... sources) {
+ List<List<MergedKeyValue>> sourceKvs = new ArrayList<>();
+ for (Source source : sources) {
+ sourceKvs.add(source.entries);
+ }
+ return new MergeScenario(sourceKvs, expected.entries);
+ }
+
+ private static Expected expected(MergedKeyValue... entries) {
+ return new Expected(Arrays.asList(entries));
+ }
+
+ private static Source source(MergedKeyValue... entries) {
+ return new Source(Arrays.asList(entries));
+ }
+
+ private static MergedKeyValue kv(String key, long sequence, int type, String
value) {
+ return MergedKeyValue.of(
+ key.getBytes(StandardCharsets.UTF_8),
+ sequence,
+ type,
+ value == null ? null : value.getBytes(StandardCharsets.UTF_8));
+ }
+
+ private static String asString(byte[] bytes) {
+ return new String(bytes, StandardCharsets.UTF_8);
+ }
+
+ private static final class MergeScenario {
+ private final List<List<MergedKeyValue>> sources;
+ private final List<MergedKeyValue> expected;
+
+ private MergeScenario(List<List<MergedKeyValue>> sources,
+ List<MergedKeyValue> expected) {
+ this.sources = sources;
+ this.expected = expected;
+ }
+ }
+
+ private static final class Expected {
+ private final List<MergedKeyValue> entries;
+
+ private Expected(List<MergedKeyValue> entries) {
+ this.entries = entries;
+ }
+ }
+
+ private static final class Source {
+ private final List<MergedKeyValue> entries;
+
+ private Source(List<MergedKeyValue> entries) {
+ this.entries = entries;
+ }
+ }
+
+ private static final class ListIterator implements
ClosableIterator<MergedKeyValue> {
+ private final Iterator<MergedKeyValue> iterator;
+
+ private ListIterator(List<MergedKeyValue> entries) {
+ this.iterator = entries.iterator();
+ }
+
+ @Override
+ public boolean hasNext() {
+ return iterator.hasNext();
+ }
+
+ @Override
+ public MergedKeyValue next() {
+ return iterator.next();
+ }
+
+ @Override
+ public void close() {
+ // Nothing to close.
+ }
+ }
+}
diff --git
a/hadoop-hdds/rocks-native/src/test/java/org/apache/hadoop/hdds/utils/db/TestLatestVersionedKWayMergeIteratorOverSst.java
b/hadoop-hdds/rocks-native/src/test/java/org/apache/hadoop/hdds/utils/db/TestLatestVersionedKWayMergeIteratorOverSst.java
new file mode 100644
index 00000000000..3b7c9e4bed0
--- /dev/null
+++
b/hadoop-hdds/rocks-native/src/test/java/org/apache/hadoop/hdds/utils/db/TestLatestVersionedKWayMergeIteratorOverSst.java
@@ -0,0 +1,204 @@
+/*
+ * 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.hdds.utils.db;
+
+import static
org.apache.hadoop.hdds.utils.NativeConstants.ROCKS_TOOLS_NATIVE_PROPERTY;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.apache.hadoop.hdds.utils.NativeLibraryNotLoadedException;
+import
org.apache.hadoop.hdds.utils.db.LatestVersionedKWayMergeIterator.MergedKeyValue;
+import org.apache.hadoop.hdds.utils.db.TestRawSstFileRecords.SourceRecord;
+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.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.EnabledIfSystemProperty;
+import org.junit.jupiter.api.io.TempDir;
+import org.rocksdb.ColumnFamilyDescriptor;
+import org.rocksdb.ColumnFamilyHandle;
+import org.rocksdb.FlushOptions;
+import org.rocksdb.RocksDB;
+import org.rocksdb.RocksDBException;
+
+/**
+ * End-to-end test over real SST files produced by RocksDB flush.
+ * <p>
+ * {@link org.apache.hadoop.hdds.utils.db.managed.ManagedSstFileWriter} is not
used here because
+ * it stores sequence number 0 on every key and rejects duplicate user keys
per file. Flushing a
+ * real RocksDB after each logical source batch yields SST files with global
sequence numbers.
+ * Memtable flushes retain only the latest value per user key within each SST;
competing versions
+ * appear across separate flushed files, matching production snapshot-diff
inputs.
+ */
+@EnabledIfSystemProperty(named = ROCKS_TOOLS_NATIVE_PROPERTY, matches = "true")
+class TestLatestVersionedKWayMergeIteratorOverSst {
+
+ @TempDir
+ private Path tempDir;
+
+ @BeforeAll
+ static void loadNativeLibrary() throws NativeLibraryNotLoadedException {
+ ManagedRawSSTFileReader.loadLibrary();
+ }
+
+ @Test
+ void testWorkedExampleOverRealSstFiles() throws Exception {
+ // Mirrors the unit-test "three-file worked example": cross-file k-way
merge with competing
+ // versions, k1 latest-value-only, k2 delete-then-recreate across files.
+ Path dbDir = tempDir.resolve("worked-example-db");
+ Files.createDirectories(dbDir);
+ Set<String> knownSstFiles = new HashSet<>();
+ List<Path> sstFiles = new ArrayList<>(3);
+
+ try (ManagedDBOptions dbOptions = new ManagedDBOptions();
+ ManagedColumnFamilyOptions cfOptions = new
ManagedColumnFamilyOptions();
+ FlushOptions flushOptions = new FlushOptions()) {
+ dbOptions.setCreateIfMissing(true);
+ List<ColumnFamilyDescriptor> columnFamilyDescriptors =
Collections.singletonList(
+ new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY,
cfOptions));
+ List<ColumnFamilyHandle> columnFamilyHandles = new ArrayList<>();
+ try (ManagedRocksDB db = ManagedRocksDB.open(
+ dbOptions, dbDir.toString(), columnFamilyDescriptors,
columnFamilyHandles);
+ ColumnFamilyHandle cf = columnFamilyHandles.get(0)) {
+
+ // Source A: latest k1 wins within the memtable before flush, plus k2.
+ rocksPut(db, cf, "k1", "v5");
+ rocksPut(db, cf, "k1", "v10");
+ rocksPut(db, cf, "k2", "v20");
+ sstFiles.add(flushAndCopySst(db, dbDir, cf, flushOptions,
knownSstFiles, "a"));
+
+ // Source B: competing k1 version and a k2 tombstone.
+ rocksPut(db, cf, "k1", "v3");
+ rocksPut(db, cf, "k1", "v15");
+ rocksDelete(db, cf, "k2");
+ sstFiles.add(flushAndCopySst(db, dbDir, cf, flushOptions,
knownSstFiles, "b"));
+
+ // Source C: winning k1/k2 values.
+ rocksPut(db, cf, "k1", "v1");
+ rocksPut(db, cf, "k1", "v30");
+ rocksPut(db, cf, "k2", "v25");
+ sstFiles.add(flushAndCopySst(db, dbDir, cf, flushOptions,
knownSstFiles, "c"));
+ }
+ }
+
+ List<List<SourceRecord>> perSource =
TestRawSstFileRecords.readFiles(sstFiles);
+ long k1VersionsAcrossFiles = perSource.stream()
+ .flatMap(List::stream)
+ .filter(record -> Arrays.equals(record.getUserKey(), keyBytes("k1")))
+ .count();
+ assertEquals(3, k1VersionsAcrossFiles,
+ "each flushed SST should contribute one surviving k1 version");
+ long distinctK1Sequences = perSource.stream()
+ .flatMap(List::stream)
+ .filter(record -> Arrays.equals(record.getUserKey(), keyBytes("k1")))
+ .mapToLong(SourceRecord::getSequence)
+ .distinct()
+ .count();
+ assertEquals(3, distinctK1Sequences,
+ "k1 versions across SST files should carry distinct RocksDB sequence
numbers");
+
+ List<MergedKeyValue> actual = mergeSstFiles(sstFiles.toArray(new Path[0]));
+ assertEquals(3, actual.size(), "expected k1 winner plus k2 tombstone and
recreate value");
+
+ MergedKeyValue k1Winner = actual.get(0);
+ assertArrayEquals(keyBytes("k1"), k1Winner.getUserKey());
+ assertEquals(LatestVersionedKWayMergeIterator.ROCKS_TYPE_VALUE,
k1Winner.getValueType());
+ assertArrayEquals(valueBytes("v30"), k1Winner.getValue());
+
+ MergedKeyValue k2Tombstone = actual.get(1);
+ assertArrayEquals(keyBytes("k2"), k2Tombstone.getUserKey());
+ assertNotEquals(LatestVersionedKWayMergeIterator.ROCKS_TYPE_VALUE,
k2Tombstone.getValueType());
+
+ MergedKeyValue k2Value = actual.get(2);
+ assertArrayEquals(keyBytes("k2"), k2Value.getUserKey());
+ assertEquals(LatestVersionedKWayMergeIterator.ROCKS_TYPE_VALUE,
k2Value.getValueType());
+ assertArrayEquals(valueBytes("v25"), k2Value.getValue());
+ assertTrue(k2Value.getSequence() > k2Tombstone.getSequence(),
+ "recreate value must be newer than the tombstone");
+ }
+
+ private Path flushAndCopySst(ManagedRocksDB db, Path dbDir,
ColumnFamilyHandle cf,
+ FlushOptions flushOptions, Set<String> knownSstFiles, String label)
+ throws RocksDBException, IOException {
+ db.get().flush(flushOptions, cf);
+ Path newSst = findNewSstFile(dbDir, knownSstFiles);
+ Path dest = tempDir.resolve(label + "-" + newSst.getFileName());
+ Files.copy(newSst, dest);
+ return dest;
+ }
+
+ private static Path findNewSstFile(Path dbDir, Set<String> knownSstFiles)
throws IOException {
+ try (Stream<Path> sstPaths = Files.list(dbDir)) {
+ List<Path> newFiles = sstPaths
+ .filter(path -> path.getFileName().toString().endsWith(".sst"))
+ .filter(path -> knownSstFiles.add(path.getFileName().toString()))
+ .sorted()
+ .collect(Collectors.toList());
+ if (newFiles.size() != 1) {
+ throw new IllegalStateException(
+ "Expected exactly one new SST file under " + dbDir + ", found " +
newFiles);
+ }
+ return newFiles.get(0);
+ }
+ }
+
+ private List<MergedKeyValue> mergeSstFiles(Path... sstFiles) throws
Exception {
+ List<MergedKeyValue> results = new ArrayList<>();
+ try (LatestVersionedKWayMergeIterator iterator =
+
LatestVersionedKWayMergeIterator.overRawSstFiles(Arrays.asList(sstFiles))) {
+ while (iterator.hasNext()) {
+ results.add(iterator.next());
+ }
+ }
+ return results;
+ }
+
+ private static void rocksPut(ManagedRocksDB db, ColumnFamilyHandle cf,
String key, String value)
+ throws RocksDBException {
+ db.get().put(cf, keyBytes(key), valueBytes(value));
+ }
+
+ private static void rocksDelete(ManagedRocksDB db, ColumnFamilyHandle cf,
String key)
+ throws RocksDBException {
+ db.get().delete(cf, keyBytes(key));
+ }
+
+ private static byte[] keyBytes(String key) {
+ return key.getBytes(StandardCharsets.UTF_8);
+ }
+
+ private static byte[] valueBytes(String value) {
+ return value.getBytes(StandardCharsets.UTF_8);
+ }
+
+}
diff --git
a/hadoop-hdds/rocks-native/src/test/java/org/apache/hadoop/hdds/utils/db/TestRawSstFileRecords.java
b/hadoop-hdds/rocks-native/src/test/java/org/apache/hadoop/hdds/utils/db/TestRawSstFileRecords.java
new file mode 100644
index 00000000000..cd2cdc3abf1
--- /dev/null
+++
b/hadoop-hdds/rocks-native/src/test/java/org/apache/hadoop/hdds/utils/db/TestRawSstFileRecords.java
@@ -0,0 +1,121 @@
+/*
+ * 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.hdds.utils.db;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import org.apache.hadoop.hdds.utils.db.managed.ManagedOptions;
+
+/**
+ * Test helper that reads all key versions from raw SST files.
+ */
+final class TestRawSstFileRecords {
+
+ private static final int DEFAULT_READ_AHEAD_SIZE = 2 * 1024 * 1024;
+
+ private TestRawSstFileRecords() {
+ }
+
+ static List<SourceRecord> readFile(Path sstFile) throws IOException {
+ return readFile(sstFile, DEFAULT_READ_AHEAD_SIZE);
+ }
+
+ static List<SourceRecord> readFile(Path sstFile, int readAheadSize) throws
IOException {
+ List<SourceRecord> records = new ArrayList<>();
+ try (ManagedOptions options = new ManagedOptions();
+ ManagedRawSSTFileReader reader = new ManagedRawSSTFileReader(
+ options, sstFile.toAbsolutePath().toString(), readAheadSize);
+ ManagedRawSSTFileIterator<ManagedRawSSTFileIterator.KeyValue>
iterator =
+ reader.newIterator(kv -> kv, null, null,
IteratorType.KEY_AND_VALUE)) {
+ while (iterator.hasNext()) {
+ ManagedRawSSTFileIterator.KeyValue kv = iterator.next();
+ byte[] key = copyBuffer(kv.getKey());
+ byte[] value = kv.getType() ==
LatestVersionedKWayMergeIterator.ROCKS_TYPE_VALUE
+ ? copyBuffer(kv.getValue()) : null;
+ records.add(new SourceRecord(key, kv.getSequence().longValue(),
kv.getType(), value));
+ }
+ }
+ return records;
+ }
+
+ static List<List<SourceRecord>> readFiles(List<Path> sstFiles) throws
IOException {
+ return readFiles(sstFiles, DEFAULT_READ_AHEAD_SIZE);
+ }
+
+ static List<List<SourceRecord>> readFiles(List<Path> sstFiles, int
readAheadSize)
+ throws IOException {
+ List<List<SourceRecord>> perSource = new ArrayList<>(sstFiles.size());
+ for (Path sstFile : sstFiles) {
+ perSource.add(readFile(sstFile, readAheadSize));
+ }
+ return perSource;
+ }
+
+ private static byte[] copyBuffer(CodecBuffer buffer) {
+ if (buffer == null) {
+ return null;
+ }
+ ByteBuffer byteBuffer = buffer.asReadOnlyByteBuffer();
+ byte[] bytes = new byte[byteBuffer.remaining()];
+ byteBuffer.get(bytes);
+ return bytes;
+ }
+
+ static final class SourceRecord {
+ private final byte[] userKey;
+ private final long sequence;
+ private final int type;
+ private final byte[] value;
+
+ SourceRecord(byte[] userKey, long sequence, int type, byte[] value) {
+ this.userKey = userKey;
+ this.sequence = sequence;
+ this.type = type;
+ this.value = value;
+ }
+
+ byte[] getUserKey() {
+ return userKey;
+ }
+
+ long getSequence() {
+ return sequence;
+ }
+
+ int getType() {
+ return type;
+ }
+
+ byte[] getValue() {
+ return value;
+ }
+
+ @Override
+ public String toString() {
+ return "SourceRecord{key=" + Arrays.toString(userKey)
+ + ", seq=" + sequence
+ + ", type=" + type
+ + ", value=" + (value == null ? null : Arrays.toString(value))
+ + '}';
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
