This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 5c45919aef [core] Add version-aware LocalKvDb range iterator (#8871)
5c45919aef is described below
commit 5c45919aef07f049fba06fb1c97ab6a5cff5e684
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Jul 28 10:14:57 2026 +0800
[core] Add version-aware LocalKvDb range iterator (#8871)
---
.../paimon/lookup/sort/SortLookupStoreReader.java | 9 +
.../apache/paimon/lookup/sort/db/LocalKvDb.java | 348 +++++++++++++++++++++
.../apache/paimon/lookup/sort/db/LsmLevels.java | 100 ++++++
.../sort/db/LocalKvDbAsyncCompactionTest.java | 54 ++++
.../paimon/lookup/sort/db/LocalKvDbTest.java | 200 ++++++++++++
5 files changed, 711 insertions(+)
diff --git
a/paimon-common/src/main/java/org/apache/paimon/lookup/sort/SortLookupStoreReader.java
b/paimon-common/src/main/java/org/apache/paimon/lookup/sort/SortLookupStoreReader.java
index 63b060fee9..ebaebd4be6 100644
---
a/paimon-common/src/main/java/org/apache/paimon/lookup/sort/SortLookupStoreReader.java
+++
b/paimon-common/src/main/java/org/apache/paimon/lookup/sort/SortLookupStoreReader.java
@@ -70,6 +70,15 @@ public class SortLookupStoreReader implements
LookupStoreReader {
return reader.createIterator();
}
+ /**
+ * Close the underlying input stream without invalidating blocks in the
shared cache.
+ *
+ * <p>The reader must not be used after this method returns.
+ */
+ public void closeInput() throws IOException {
+ input.close();
+ }
+
@Override
public void close() throws IOException {
reader.close();
diff --git
a/paimon-common/src/main/java/org/apache/paimon/lookup/sort/db/LocalKvDb.java
b/paimon-common/src/main/java/org/apache/paimon/lookup/sort/db/LocalKvDb.java
index 2c73a87a95..1bf4aa8dfa 100644
---
a/paimon-common/src/main/java/org/apache/paimon/lookup/sort/db/LocalKvDb.java
+++
b/paimon-common/src/main/java/org/apache/paimon/lookup/sort/db/LocalKvDb.java
@@ -26,7 +26,10 @@ import org.apache.paimon.lookup.sort.SortLookupStoreReader;
import org.apache.paimon.lookup.sort.SortLookupStoreWriter;
import org.apache.paimon.memory.MemorySlice;
import org.apache.paimon.options.MemorySize;
+import org.apache.paimon.sst.BlockIterator;
+import org.apache.paimon.sst.SstFileReader;
import org.apache.paimon.utils.BloomFilter;
+import org.apache.paimon.utils.KeyValueIterator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -36,6 +39,7 @@ import javax.annotation.Nullable;
import java.io.Closeable;
import java.io.File;
import java.io.IOException;
+import java.util.AbstractMap;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Comparator;
@@ -43,6 +47,7 @@ import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import java.util.PriorityQueue;
import java.util.TreeMap;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
@@ -125,6 +130,7 @@ public class LocalKvDb implements Closeable {
private final AtomicLong fileSequence;
@Nullable private BulkLoadWriter activeBulkLoadWriter;
+ private int openRangeIterators;
private boolean closed;
private LocalKvDb(
@@ -150,6 +156,7 @@ public class LocalKvDb implements Closeable {
this.readerCache = new HashMap<>();
this.fileSequence = new AtomicLong();
this.activeBulkLoadWriter = null;
+ this.openRangeIterators = 0;
this.closed = false;
LsmCompactor.CompactorFactory compactorFactory =
fileDeleter ->
@@ -218,6 +225,7 @@ public class LocalKvDb implements Closeable {
public void put(byte[] key, byte[] value) throws IOException {
ensureOpen();
ensureNoBulkLoad();
+ ensureNoRangeIterator();
checkCompactionFailure();
if (value.length == 0) {
throw new IllegalArgumentException(
@@ -244,6 +252,7 @@ public class LocalKvDb implements Closeable {
public void delete(byte[] key) throws IOException {
ensureOpen();
ensureNoBulkLoad();
+ ensureNoRangeIterator();
checkCompactionFailure();
MemorySlice wrappedKey = MemorySlice.wrap(key);
byte[] oldValue = memTable.put(wrappedKey, TOMBSTONE);
@@ -298,6 +307,7 @@ public class LocalKvDb implements Closeable {
private BulkLoadWriter createBulkLoadWriter(long expectedEntries) throws
IOException {
ensureOpen();
+ ensureNoRangeIterator();
checkCompactionFailure();
if (activeBulkLoadWriter != null) {
throw new IllegalStateException("Another bulk load is already in
progress.");
@@ -549,6 +559,77 @@ public class LocalKvDb implements Closeable {
return value == null || isTombstone(value) ? null : value;
}
+ /**
+ * Scan live entries in the half-open range [{@code fromInclusive}, {@code
toExclusive}).
+ *
+ * <p>The result is sorted by the configured key comparator. Newer values
and tombstones shadow
+ * older versions in the same way as {@link #get(byte[])}. A null upper
bound scans to the end
+ * of the database.
+ */
+ public List<Map.Entry<byte[], byte[]>> rangeScan(
+ byte[] fromInclusive, @Nullable byte[] toExclusive) throws
IOException {
+ List<Map.Entry<byte[], byte[]>> result = new ArrayList<>();
+ forEachInRange(
+ fromInclusive,
+ toExclusive,
+ (key, value) ->
+ result.add(
+ new AbstractMap.SimpleImmutableEntry<>(
+ key.copyBytes(), value.copyBytes())));
+ return result;
+ }
+
+ /**
+ * Visit live entries in the half-open range [{@code fromInclusive},
{@code toExclusive}).
+ *
+ * <p>Entries are visited in key order with the same shadowing rules as
{@link #get(byte[])}.
+ * The supplied slices are only valid for the duration of the callback and
must be copied if
+ * retained.
+ */
+ public void forEachInRange(
+ byte[] fromInclusive, @Nullable byte[] toExclusive,
RangeEntryConsumer consumer)
+ throws IOException {
+ try (RangeIterator iterator = rangeIterator(fromInclusive,
toExclusive)) {
+ while (iterator.advanceNext()) {
+ consumer.accept(iterator.getKey(), iterator.getValue());
+ }
+ }
+ }
+
+ /**
+ * Create a lazy iterator over live entries in the half-open range [{@code
fromInclusive},
+ * {@code toExclusive}).
+ *
+ * <p>The iterator merges the MemTable and all overlapping SST files in
key order. For duplicate
+ * keys, the newest source wins and tombstones suppress older values. The
iterator must be
+ * closed before modifying or closing the database. Closing releases the
levels read lock so a
+ * concurrent compaction can publish its result.
+ */
+ public RangeIterator rangeIterator(byte[] fromInclusive, @Nullable byte[]
toExclusive)
+ throws IOException {
+ ensureOpen();
+ ensureNoBulkLoad();
+ checkCompactionFailure();
+
+ MemorySlice from = MemorySlice.wrap(fromInclusive);
+ MemorySlice to = toExclusive == null ? null :
MemorySlice.wrap(toExclusive);
+ checkArgument(
+ to == null || keyComparator.compare(from, to) <= 0,
+ "Range start must not be greater than range end.");
+
+ Map<MemorySlice, byte[]> memoryEntries =
+ to == null ? memTable.tailMap(from, true) :
memTable.subMap(from, true, to, false);
+ LsmLevels.RangeSnapshot snapshot = levels.openRangeSnapshot(from, to,
keyComparator);
+ try {
+ RangeIterator iterator = new RangeIterator(snapshot,
memoryEntries, fromInclusive, to);
+ openRangeIterators++;
+ return iterator;
+ } catch (IOException | RuntimeException e) {
+ snapshot.close();
+ throw e;
+ }
+ }
+
//
-------------------------------------------------------------------------
// Flush & Compaction
//
-------------------------------------------------------------------------
@@ -563,6 +644,7 @@ public class LocalKvDb implements Closeable {
public void flush() throws IOException {
ensureOpen();
ensureNoBulkLoad();
+ ensureNoRangeIterator();
checkCompactionFailure();
if (memTable.isEmpty()) {
return;
@@ -594,6 +676,7 @@ public class LocalKvDb implements Closeable {
public void compact() throws IOException {
ensureOpen();
ensureNoBulkLoad();
+ ensureNoRangeIterator();
compaction.fullCompact();
}
@@ -606,6 +689,7 @@ public class LocalKvDb implements Closeable {
if (closed) {
return;
}
+ ensureNoRangeIterator();
closed = true;
IOException failure = null;
@@ -653,6 +737,12 @@ public class LocalKvDb implements Closeable {
return levels.fileCount();
}
+ /** Return the number of readers retained for point lookups. */
+ @VisibleForTesting
+ int getCachedReaderCount() {
+ return readerCache.size();
+ }
+
/** Return the number of SST files at a specific level. */
public int getLevelFileCount(int level) {
return levels.fileCount(level);
@@ -771,6 +861,264 @@ public class LocalKvDb implements Closeable {
return MemorySlice.wrap(Arrays.copyOf(key, key.length));
}
+ private void ensureNoRangeIterator() {
+ if (openRangeIterators > 0) {
+ throw new IllegalStateException(
+ "The database cannot be modified or closed while a range
iterator is open.");
+ }
+ }
+
+ /** Lazy range iterator with newest-version-wins semantics. */
+ public final class RangeIterator
+ implements KeyValueIterator<MemorySlice, MemorySlice>,
AutoCloseable {
+
+ private final LsmLevels.RangeSnapshot snapshot;
+ private final PriorityQueue<RangeSource> sources;
+
+ @Nullable private MemorySlice currentKey;
+ @Nullable private MemorySlice currentValue;
+ private boolean closed;
+
+ private RangeIterator(
+ LsmLevels.RangeSnapshot snapshot,
+ Map<MemorySlice, byte[]> memoryEntries,
+ byte[] fromInclusive,
+ @Nullable MemorySlice toExclusive)
+ throws IOException {
+ this.snapshot = snapshot;
+ this.sources =
+ new PriorityQueue<>(
+ (left, right) -> {
+ int compare =
keyComparator.compare(left.key(), right.key());
+ return compare != 0
+ ? compare
+ : Integer.compare(left.priority(),
right.priority());
+ });
+
+ int priority = 0;
+ advanceAndAdd(new MemoryRangeSource(priority++,
memoryEntries.entrySet().iterator()));
+ for (File file : snapshot.files()) {
+ advanceAndAdd(new SstRangeSource(priority++, file,
fromInclusive, toExclusive));
+ }
+ }
+
+ @Override
+ public boolean advanceNext() throws IOException {
+ if (closed) {
+ return false;
+ }
+
+ currentKey = null;
+ currentValue = null;
+ try {
+ while (!sources.isEmpty()) {
+ RangeSource newest = sources.poll();
+ MemorySlice key = newest.key();
+ MemorySlice value = newest.value();
+ advanceAndAdd(newest);
+
+ while (!sources.isEmpty()
+ && keyComparator.compare(sources.peek().key(),
key) == 0) {
+ advanceAndAdd(sources.poll());
+ }
+
+ if (!isTombstoneSlice(value)) {
+ currentKey = key;
+ currentValue = value;
+ return true;
+ }
+ }
+ close();
+ return false;
+ } catch (IOException | RuntimeException e) {
+ close();
+ throw e;
+ }
+ }
+
+ @Override
+ public MemorySlice getKey() {
+ if (currentKey == null) {
+ throw new IllegalStateException("Range iterator is not
positioned on an entry.");
+ }
+ return currentKey;
+ }
+
+ @Override
+ public MemorySlice getValue() {
+ if (currentValue == null) {
+ throw new IllegalStateException("Range iterator is not
positioned on an entry.");
+ }
+ return currentValue;
+ }
+
+ private void advanceAndAdd(RangeSource source) throws IOException {
+ if (source.advance()) {
+ sources.add(source);
+ }
+ }
+
+ @Override
+ public void close() {
+ if (!closed) {
+ closed = true;
+ sources.clear();
+ currentKey = null;
+ currentValue = null;
+ snapshot.close();
+ openRangeIterators--;
+ }
+ }
+ }
+
+ private interface RangeSource {
+
+ int priority();
+
+ boolean advance() throws IOException;
+
+ MemorySlice key();
+
+ MemorySlice value();
+ }
+
+ private static final class MemoryRangeSource implements RangeSource {
+
+ private final int priority;
+ private final Iterator<Map.Entry<MemorySlice, byte[]>> iterator;
+
+ @Nullable private Map.Entry<MemorySlice, byte[]> current;
+
+ private MemoryRangeSource(int priority,
Iterator<Map.Entry<MemorySlice, byte[]>> iterator) {
+ this.priority = priority;
+ this.iterator = iterator;
+ }
+
+ @Override
+ public int priority() {
+ return priority;
+ }
+
+ @Override
+ public boolean advance() {
+ current = iterator.hasNext() ? iterator.next() : null;
+ return current != null;
+ }
+
+ @Override
+ public MemorySlice key() {
+ return current.getKey();
+ }
+
+ @Override
+ public MemorySlice value() {
+ return MemorySlice.wrap(current.getValue());
+ }
+ }
+
+ private final class SstRangeSource implements RangeSource {
+
+ private final int priority;
+ private final File file;
+ private final byte[] fromInclusive;
+ @Nullable private final MemorySlice toExclusive;
+
+ @Nullable private BlockIterator block;
+ @Nullable private MemorySlice resumeAfterKey;
+ @Nullable private Map.Entry<MemorySlice, MemorySlice> current;
+ private boolean finished;
+
+ private SstRangeSource(
+ int priority, File file, byte[] fromInclusive, @Nullable
MemorySlice toExclusive) {
+ this.priority = priority;
+ this.file = file;
+ this.fromInclusive = Arrays.copyOf(fromInclusive,
fromInclusive.length);
+ this.toExclusive =
+ toExclusive == null ? null :
MemorySlice.wrap(toExclusive.copyBytes());
+ }
+
+ @Override
+ public int priority() {
+ return priority;
+ }
+
+ @Override
+ public boolean advance() throws IOException {
+ while (block == null || !block.hasNext()) {
+ block = loadNextBlock();
+ if (block == null) {
+ current = null;
+ return false;
+ }
+ }
+
+ current = block.next();
+ if (toExclusive != null && keyComparator.compare(current.getKey(),
toExclusive) >= 0) {
+ current = null;
+ finished = true;
+ return false;
+ }
+ if (!block.hasNext()) {
+ resumeAfterKey =
MemorySlice.wrap(current.getKey().copyBytes());
+ }
+ return true;
+ }
+
+ @Nullable
+ private BlockIterator loadNextBlock() throws IOException {
+ if (finished) {
+ return null;
+ }
+
+ SortLookupStoreReader reader = storeFactory.createReader(file);
+ try (Closeable ignored = reader::closeInput) {
+ SstFileReader.SstFileIterator iterator =
reader.createIterator();
+ byte[] seekKey =
+ resumeAfterKey == null ? fromInclusive :
resumeAfterKey.copyBytes();
+ iterator.seekTo(seekKey);
+ boolean skipResumeKey = resumeAfterKey != null;
+ while (true) {
+ BlockIterator nextBlock = iterator.readBatch();
+ if (nextBlock == null) {
+ finished = true;
+ return null;
+ }
+
+ if (skipResumeKey) {
+ if (nextBlock.seekTo(resumeAfterKey)) {
+ nextBlock.next();
+ }
+ skipResumeKey = false;
+ }
+ if (nextBlock.hasNext()) {
+ return nextBlock;
+ }
+ }
+ }
+ }
+
+ @Override
+ public MemorySlice key() {
+ return current.getKey();
+ }
+
+ @Override
+ public MemorySlice value() {
+ return current.getValue();
+ }
+ }
+
+ private static boolean isTombstoneSlice(MemorySlice value) {
+ return value.length() == 0;
+ }
+
+ /** Callback for visiting an entry during a range scan. */
+ @FunctionalInterface
+ public interface RangeEntryConsumer {
+
+ void accept(MemorySlice key, MemorySlice value) throws IOException;
+ }
+
//
-------------------------------------------------------------------------
// Builder
//
-------------------------------------------------------------------------
diff --git
a/paimon-common/src/main/java/org/apache/paimon/lookup/sort/db/LsmLevels.java
b/paimon-common/src/main/java/org/apache/paimon/lookup/sort/db/LsmLevels.java
index 6f13e08dff..dfd9bec755 100644
---
a/paimon-common/src/main/java/org/apache/paimon/lookup/sort/db/LsmLevels.java
+++
b/paimon-common/src/main/java/org/apache/paimon/lookup/sort/db/LsmLevels.java
@@ -119,6 +119,56 @@ class LsmLevels {
}
}
+ /**
+ * Open a snapshot of all SST files which overlap the requested range.
+ *
+ * <p>Files are ordered from newer levels to older levels; Level-0 files
are additionally
+ * ordered newest first. The read lock remains held until the returned
snapshot is closed so a
+ * concurrent compaction cannot delete a file being iterated.
+ */
+ RangeSnapshot openRangeSnapshot(
+ MemorySlice fromInclusive,
+ @Nullable MemorySlice toExclusive,
+ Comparator<MemorySlice> keyComparator) {
+ lock.readLock().lock();
+ boolean success = false;
+ try {
+ List<File> files = new ArrayList<>();
+ for (int level = 0; level < maxLevels; level++) {
+ List<SstFileMetadata> levelFiles = levels.get(level);
+ if (levelFiles.isEmpty()) {
+ continue;
+ }
+
+ if (level == 0) {
+ for (SstFileMetadata metadata : levelFiles) {
+ if (overlapsRange(metadata, fromInclusive,
toExclusive, keyComparator)) {
+ files.add(metadata.getFile());
+ }
+ }
+ continue;
+ }
+
+ int firstFile = findFirstOverlappingFile(levelFiles,
fromInclusive, keyComparator);
+ for (int i = firstFile; i < levelFiles.size(); i++) {
+ SstFileMetadata metadata = levelFiles.get(i);
+ if (toExclusive != null
+ && keyComparator.compare(metadata.getMinKey(),
toExclusive) >= 0) {
+ break;
+ }
+ files.add(metadata.getFile());
+ }
+ }
+ RangeSnapshot snapshot = new RangeSnapshot(files);
+ success = true;
+ return snapshot;
+ } finally {
+ if (!success) {
+ lock.readLock().unlock();
+ }
+ }
+ }
+
List<List<SstFileMetadata>> snapshot() {
lock.readLock().lock();
try {
@@ -265,6 +315,56 @@ class LsmLevels {
return null;
}
+ private static int findFirstOverlappingFile(
+ List<SstFileMetadata> sortedFiles,
+ MemorySlice fromInclusive,
+ Comparator<MemorySlice> keyComparator) {
+ int low = 0;
+ int high = sortedFiles.size();
+ while (low < high) {
+ int mid = low + (high - low) / 2;
+ if (keyComparator.compare(sortedFiles.get(mid).getMaxKey(),
fromInclusive) < 0) {
+ low = mid + 1;
+ } else {
+ high = mid;
+ }
+ }
+ return low;
+ }
+
+ private static boolean overlapsRange(
+ SstFileMetadata metadata,
+ MemorySlice fromInclusive,
+ @Nullable MemorySlice toExclusive,
+ Comparator<MemorySlice> keyComparator) {
+ return keyComparator.compare(metadata.getMaxKey(), fromInclusive) >= 0
+ && (toExclusive == null
+ || keyComparator.compare(metadata.getMinKey(),
toExclusive) < 0);
+ }
+
+ /** SST files in a range protected from concurrent compaction until
closed. */
+ final class RangeSnapshot implements AutoCloseable {
+
+ private final List<File> files;
+ private boolean closed;
+
+ private RangeSnapshot(List<File> files) {
+ this.files = files;
+ }
+
+ List<File> files() {
+ return files;
+ }
+
+ @Override
+ public void close() {
+ if (!closed) {
+ closed = true;
+ lock.readLock().unlock();
+ }
+ }
+ }
+
/** Callback for reading one SST file. */
interface FileLookup {
diff --git
a/paimon-common/src/test/java/org/apache/paimon/lookup/sort/db/LocalKvDbAsyncCompactionTest.java
b/paimon-common/src/test/java/org/apache/paimon/lookup/sort/db/LocalKvDbAsyncCompactionTest.java
index 205b9e51fa..ab94fab9e0 100644
---
a/paimon-common/src/test/java/org/apache/paimon/lookup/sort/db/LocalKvDbAsyncCompactionTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/lookup/sort/db/LocalKvDbAsyncCompactionTest.java
@@ -26,6 +26,9 @@ import org.junit.jupiter.api.io.TempDir;
import java.io.File;
import java.io.IOException;
import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
import java.util.Queue;
import java.util.concurrent.AbstractExecutorService;
import java.util.concurrent.CountDownLatch;
@@ -54,12 +57,48 @@ public class LocalKvDbAsyncCompactionTest {
assertThat(compactionExecutor.numQueuedTasks()).isOne();
assertThat(db.getLevelFileCount(0)).isEqualTo(3);
assertThat(get(db, "shared")).isEqualTo("v3");
+ assertThat(scan(db, "sha", "shb")).containsExactly("shared=v3");
compactionExecutor.runNext();
db.awaitCompaction();
assertThat(db.getLevelFileCount(0)).isZero();
assertThat(get(db, "shared")).isEqualTo("v3");
+ assertThat(scan(db, "sha", "shb")).containsExactly("shared=v3");
+ }
+ }
+
+ @Test
+ void testRangeIteratorPreventsCompactionFromDeletingItsFiles() throws
Exception {
+ ManuallyTriggeredExecutor compactionExecutor = new
ManuallyTriggeredExecutor();
+ File directory = new File(tempDir.toFile(), "range-compaction");
+ try (LocalKvDb db = createDb("range-compaction", compactionExecutor)) {
+ putAndFlush(db, "shared", "v1");
+ putAndFlush(db, "shared", "v2");
+ putAndFlush(db, "shared", "v3");
+
+ LocalKvDb.RangeIterator iterator =
+ db.rangeIterator("sha".getBytes(UTF_8),
"shb".getBytes(UTF_8));
+ ExecutorService compactionRunner =
Executors.newSingleThreadExecutor();
+ Future<?> compactionFuture =
compactionRunner.submit(compactionExecutor::runNext);
+ try {
+ long deadline = System.nanoTime() +
TimeUnit.SECONDS.toNanos(10);
+ while (sstFileCount(directory) < 4 && System.nanoTime() <
deadline) {
+ Thread.yield();
+ }
+
+ assertThat(sstFileCount(directory)).isGreaterThanOrEqualTo(4);
+ assertThat(compactionFuture).isNotDone();
+ assertThat(iterator.advanceNext()).isTrue();
+ assertThat(new String(iterator.getValue().copyBytes(),
UTF_8)).isEqualTo("v3");
+ } finally {
+ iterator.close();
+ compactionFuture.get(10, TimeUnit.SECONDS);
+ compactionRunner.shutdownNow();
+ }
+
+ db.awaitCompaction();
+ assertThat(scan(db, "sha", "shb")).containsExactly("shared=v3");
}
}
@@ -208,6 +247,21 @@ public class LocalKvDbAsyncCompactionTest {
return value == null ? null : new String(value, UTF_8);
}
+ private static List<String> scan(LocalKvDb db, String from, String to)
throws IOException {
+ List<String> result = new ArrayList<>();
+ for (Map.Entry<byte[], byte[]> entry :
+ db.rangeScan(from.getBytes(UTF_8), to.getBytes(UTF_8))) {
+ result.add(
+ new String(entry.getKey(), UTF_8) + "=" + new
String(entry.getValue(), UTF_8));
+ }
+ return result;
+ }
+
+ private static int sstFileCount(File directory) {
+ File[] files = directory.listFiles((ignored, name) ->
name.endsWith(".db"));
+ return files == null ? 0 : files.length;
+ }
+
private static class ManuallyTriggeredExecutor extends
AbstractExecutorService {
private final Queue<Runnable> tasks = new ArrayDeque<>();
diff --git
a/paimon-common/src/test/java/org/apache/paimon/lookup/sort/db/LocalKvDbTest.java
b/paimon-common/src/test/java/org/apache/paimon/lookup/sort/db/LocalKvDbTest.java
index 66174ccf5a..0f49fba43b 100644
---
a/paimon-common/src/test/java/org/apache/paimon/lookup/sort/db/LocalKvDbTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/lookup/sort/db/LocalKvDbTest.java
@@ -27,12 +27,16 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import javax.annotation.Nullable;
+
import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.file.Files;
import java.util.AbstractMap;
import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
@@ -130,6 +134,177 @@ public class LocalKvDbTest {
}
}
+ @Test
+ public void testRangeScanMergesMemTableAndLevels() throws IOException {
+ try (LocalKvDb db = createDb("range-scan-db")) {
+ List<Map.Entry<byte[], byte[]>> initial = new ArrayList<>();
+ initial.add(entry("a-1", "old-1"));
+ initial.add(entry("a-2", "old-2"));
+ initial.add(entry("b-1", "value-b"));
+ db.bulkLoad(initial.iterator(), initial.size());
+
+ putString(db, "a-1", "new-1");
+ db.flush();
+ deleteString(db, "a-2");
+ putString(db, "a-3", "new-3");
+
+ Assertions.assertEquals(
+ Arrays.asList("a-1=new-1", "a-3=new-3"), scanStrings(db,
"a", "b"));
+ Assertions.assertTrue(db.rangeScan("b".getBytes(UTF_8),
"b".getBytes(UTF_8)).isEmpty());
+ Assertions.assertEquals(
+ Collections.singletonList("b-1=value-b"), scanStrings(db,
"b", null));
+
+ db.flush();
+ Assertions.assertEquals(
+ Arrays.asList("a-1=new-1", "a-3=new-3"), scanStrings(db,
"a", "b"));
+ db.compact();
+ Assertions.assertEquals(
+ Arrays.asList("a-1=new-1", "a-3=new-3"), scanStrings(db,
"a", "b"));
+ }
+ }
+
+ @Test
+ public void testForEachInRangeAcrossMemTableAndSingleSst() throws
IOException {
+ try (LocalKvDb db = createDb("range-consumer-db")) {
+ putString(db, "a-1", "value-1");
+ putString(db, "a-2", "value-2");
+ putString(db, "b-1", "value-b");
+
+ Assertions.assertEquals(
+ Arrays.asList("a-1=value-1", "a-2=value-2"),
forEachStrings(db, "a", "b"));
+
+ db.flush();
+ Assertions.assertEquals(
+ Arrays.asList("a-1=value-1", "a-2=value-2"),
forEachStrings(db, "a", "b"));
+
+ putString(db, "a-1", "updated-1");
+ deleteString(db, "a-2");
+ putString(db, "a-3", "value-3");
+ putString(db, "c-1", "value-c");
+ Assertions.assertEquals(
+ Arrays.asList("a-1=updated-1", "a-3=value-3"),
forEachStrings(db, "a", "b"));
+ Assertions.assertEquals(
+ Collections.singletonList("c-1=value-c"),
forEachStrings(db, "c", "d"));
+ }
+ }
+
+ @Test
+ public void testRangeIteratorDeduplicatesVersionsAcrossAllSources() throws
IOException {
+ try (LocalKvDb db = createDb("range-iterator-db")) {
+ db.bulkLoad(
+ Arrays.asList(entry("a", "base-a"), entry("b", "base-b"),
entry("c", "base-c"))
+ .iterator(),
+ 3);
+
+ putString(db, "a", "level-zero-a-1");
+ deleteString(db, "b");
+ putString(db, "d", "level-zero-d");
+ db.flush();
+
+ putString(db, "a", "level-zero-a-2");
+ deleteString(db, "c");
+ putString(db, "e", "level-zero-e");
+ db.flush();
+
+ putString(db, "a", "memtable-a");
+ deleteString(db, "d");
+ putString(db, "f", "memtable-f");
+
+ List<String> result = new ArrayList<>();
+ try (LocalKvDb.RangeIterator iterator =
+ db.rangeIterator("a".getBytes(UTF_8),
"g".getBytes(UTF_8))) {
+ Assertions.assertThrows(
+ IllegalStateException.class, () -> putString(db, "x",
"blocked"));
+ while (iterator.advanceNext()) {
+ result.add(
+ new String(iterator.getKey().copyBytes(), UTF_8)
+ + "="
+ + new
String(iterator.getValue().copyBytes(), UTF_8));
+ }
+ }
+
+ Assertions.assertEquals(
+ Arrays.asList("a=memtable-a", "e=level-zero-e",
"f=memtable-f"), result);
+ }
+ }
+
+ @Test
+ public void testRangeIteratorRejectsBulkLoadWithoutLockUpgrade() throws
IOException {
+ try (LocalKvDb db = createDb("range-iterator-bulk-load-db");
+ LocalKvDb.RangeIterator ignored =
+ db.rangeIterator("a".getBytes(UTF_8),
"b".getBytes(UTF_8))) {
+ List<Map.Entry<byte[], byte[]>> entries =
+ Collections.singletonList(entry("a-1", "value-1"));
+
+ IllegalStateException exception =
+ Assertions.assertThrows(
+ IllegalStateException.class,
+ () -> db.bulkLoad(entries.iterator(),
entries.size()));
+ Assertions.assertTrue(exception.getMessage().contains("range
iterator"));
+ Assertions.assertEquals(0, db.getSstFileCount());
+ }
+ }
+
+ @Test
+ public void testRangeIteratorCrossesSstBlocks() throws IOException {
+ File directory = new File(tempDir.toFile(), "range-iterator-blocks");
+ try (LocalKvDb db =
+ LocalKvDb.builder(directory)
+ .memTableFlushThreshold(1024 * 1024)
+ .blockSize(128)
+ .level0FileNumCompactTrigger(100)
+ .compressOptions(new CompressOptions("none", 1))
+ .build()) {
+ List<Map.Entry<byte[], byte[]>> initial = new ArrayList<>();
+ for (int i = 0; i < 100; i++) {
+ initial.add(entry(String.format("key-%05d", i),
String.format("value-%05d", i)));
+ }
+ db.bulkLoad(initial.iterator(), initial.size());
+
+ putString(db, "key-00020", "updated-20");
+ deleteString(db, "key-00030");
+ db.flush();
+ putString(db, "key-00040", "updated-40");
+
+ List<String> result = scanStrings(db, "key-00010", "key-00050");
+ Assertions.assertEquals(39, result.size());
+ Assertions.assertEquals("key-00010=value-00010", result.get(0));
+ Assertions.assertEquals("key-00020=updated-20", result.get(10));
+ Assertions.assertEquals("key-00031=value-00031", result.get(20));
+ Assertions.assertEquals("key-00040=updated-40", result.get(29));
+ Assertions.assertEquals("key-00049=value-00049", result.get(38));
+ }
+ }
+
+ @Test
+ public void testRangeIteratorDoesNotCacheReadersForOverlappingSsts()
throws IOException {
+ int fileCount = 128;
+ File directory = new File(tempDir.toFile(),
"range-iterator-many-ssts");
+ try (LocalKvDb db =
+ LocalKvDb.builder(directory)
+ .level0FileNumCompactTrigger(fileCount + 1)
+ .compressOptions(new CompressOptions("none", 1))
+ .build()) {
+ for (int i = 0; i < fileCount; i++) {
+ putString(db, String.format("key-%05d", i),
String.format("value-%05d", i));
+ db.flush();
+ }
+ Assertions.assertEquals(fileCount, db.getLevelFileCount(0));
+
+ int entryCount = 0;
+ try (LocalKvDb.RangeIterator iterator =
+ db.rangeIterator("key-00000".getBytes(UTF_8),
"key-99999".getBytes(UTF_8))) {
+ Assertions.assertEquals(0, db.getCachedReaderCount());
+ while (iterator.advanceNext()) {
+ entryCount++;
+ }
+ }
+
+ Assertions.assertEquals(fileCount, entryCount);
+ Assertions.assertEquals(0, db.getCachedReaderCount());
+ }
+ }
+
@Test
public void testFlushToSst() throws IOException {
try (LocalKvDb db = createDb()) {
@@ -1795,6 +1970,31 @@ public class LocalKvDbTest {
return new String(bytes, UTF_8);
}
+ private static List<String> scanStrings(LocalKvDb db, String from,
@Nullable String to)
+ throws IOException {
+ List<String> result = new ArrayList<>();
+ for (Map.Entry<byte[], byte[]> entry :
+ db.rangeScan(from.getBytes(UTF_8), to == null ? null :
to.getBytes(UTF_8))) {
+ result.add(
+ new String(entry.getKey(), UTF_8) + "=" + new
String(entry.getValue(), UTF_8));
+ }
+ return result;
+ }
+
+ private static List<String> forEachStrings(LocalKvDb db, String from,
@Nullable String to)
+ throws IOException {
+ List<String> result = new ArrayList<>();
+ db.forEachInRange(
+ from.getBytes(UTF_8),
+ to == null ? null : to.getBytes(UTF_8),
+ (key, value) ->
+ result.add(
+ new String(key.copyBytes(), UTF_8)
+ + "="
+ + new String(value.copyBytes(),
UTF_8)));
+ return result;
+ }
+
private static void deleteString(LocalKvDb db, String key) throws
IOException {
db.delete(key.getBytes(UTF_8));
}
