You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by ya...@apache.org on 2026/10/11 12:51:30 UTC
(fluss) branch main updated: [kv] Keep KV snapshots readable by FRocksDB 6.20.3 for uniform keys (#4598)
This is an automated email from the ASF dual-hosted git repository.
platinumhamburg pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new 9ddf6338b [kv] Keep KV snapshots readable by FRocksDB 6.20.3 for uniform keys (#4598)
9ddf6338b is described below
commit 9ddf6338b605b2a82457a9f2edb73c706dc55586
Author: Muhammet Orazov <91...@users.noreply.github.com>
AuthorDate: Sun Oct 11 14:51:25 2026 +0200
[kv] Keep KV snapshots readable by FRocksDB 6.20.3 for uniform keys (#4598)
RocksJava 11.8.1 defaults uniformCvThreshold to 0.2, so RocksDB sets an
is_uniform bit in the footer of index blocks with evenly spaced keys.
FRocksDB 6.20.3 reads that bit as part of the restart count and fails
the scan with "bad block contents". Pin the threshold to -1 next to the
existing format_version and checksum pins.
Extend the FRocksDB compatibility test with a snapshot of sequential
big-endian keys, scanned in full by the FRocksDB reader.
Closes #4597.
---
.../kv/rocksdb/RocksDBResourceContainer.java | 3 +-
.../server/kv/snapshot/FrocksDBSnapshotReader.java | 22 ++++++++
.../kv/snapshot/RocksIncrementalSnapshotTest.java | 62 ++++++++++++++++------
3 files changed, 70 insertions(+), 17 deletions(-)
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/rocksdb/RocksDBResourceContainer.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/rocksdb/RocksDBResourceContainer.java
index 46bc190a1..a06ef041b 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/kv/rocksdb/RocksDBResourceContainer.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/rocksdb/RocksDBResourceContainer.java
@@ -358,7 +358,8 @@ public class RocksDBResourceContainer implements AutoCloseable {
// Keep snapshots readable by FRocksDB 6.20.3.
blockBasedTableConfig
.setFormatVersion(FROCKSDB_COMPATIBLE_FORMAT_VERSION)
- .setChecksumType(ChecksumType.kCRC32c);
+ .setChecksumType(ChecksumType.kCRC32c)
+ .setUniformCvThreshold(-1);
blockBasedTableConfig.setBlockSize(
internalGetOption(ConfigOptions.KV_BLOCK_SIZE).getBytes());
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/FrocksDBSnapshotReader.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/FrocksDBSnapshotReader.java
index 3a791081b..6517107fa 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/FrocksDBSnapshotReader.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/FrocksDBSnapshotReader.java
@@ -19,6 +19,7 @@ package org.apache.fluss.server.kv.snapshot;
import org.rocksdb.Options;
import org.rocksdb.RocksDB;
+import org.rocksdb.RocksIterator;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
@@ -26,6 +27,9 @@ import java.util.Arrays;
/** Reads a RocksDB snapshot in an isolated process using the legacy FRocksDB JNI dependency. */
public final class FrocksDBSnapshotReader {
+ /** Switches the reader to a full scan that expects the given number of rows. */
+ static final String COUNT_ARG = "--count";
+
private FrocksDBSnapshotReader() {}
/** Opens a snapshot and verifies the key/value pairs supplied after the database path. */
@@ -38,6 +42,10 @@ public final class FrocksDBSnapshotReader {
RocksDB.loadLibrary();
try (Options options = new Options();
RocksDB rocksDB = RocksDB.openReadOnly(options, args[0])) {
+ if (COUNT_ARG.equals(args[1])) {
+ assertRowCount(rocksDB, Long.parseLong(args[2]));
+ return;
+ }
for (int i = 1; i < args.length; i += 2) {
byte[] actualValue = rocksDB.get(args[i].getBytes(StandardCharsets.UTF_8));
byte[] expectedValue = args[i + 1].getBytes(StandardCharsets.UTF_8);
@@ -50,4 +58,18 @@ public final class FrocksDBSnapshotReader {
}
}
}
+
+ private static void assertRowCount(RocksDB rocksDB, long expectedRows) throws Exception {
+ long rows = 0;
+ try (RocksIterator iterator = rocksDB.newIterator()) {
+ for (iterator.seekToFirst(); iterator.isValid(); iterator.next()) {
+ rows++;
+ }
+ iterator.status();
+ }
+ if (rows != expectedRows) {
+ throw new AssertionError(
+ String.format("Expected %d rows but scanned %d", expectedRows, rows));
+ }
+ }
}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/RocksIncrementalSnapshotTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/RocksIncrementalSnapshotTest.java
index d5b1a7593..5889ceb30 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/RocksIncrementalSnapshotTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/RocksIncrementalSnapshotTest.java
@@ -37,6 +37,7 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
import org.junit.jupiter.api.io.TempDir;
+import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.nio.file.Path;
import java.util.Collection;
@@ -162,6 +163,43 @@ class RocksIncrementalSnapshotTest {
@Test
void testSnapshotCanBeReadByFrocksDB(
@TempDir Path snapshotBaseDir, @TempDir Path snapshotDownDir) throws Exception {
+ RocksDB rocksDB = rocksDBExtension.getRocksDb();
+ rocksDB.put(
+ "key1".getBytes(StandardCharsets.UTF_8), "val1".getBytes(StandardCharsets.UTF_8));
+ rocksDB.put(
+ "key2".getBytes(StandardCharsets.UTF_8), "val2".getBytes(StandardCharsets.UTF_8));
+
+ assertSnapshotReadableByFrocksDB(
+ snapshotBaseDir, snapshotDownDir, "key1", "val1", "key2", "val2");
+ }
+
+ /**
+ * Sequential big-endian long keys, like encoded integer primary keys, make the index blocks
+ * hold evenly spaced keys. RocksDB 11 then marks those blocks "uniform" in the block footer
+ * unless uniform_cv_threshold is -1, and FRocksDB fails the scan with "bad block contents".
+ */
+ @Test
+ void testSnapshotWithUniformKeysCanBeReadByFrocksDB(
+ @TempDir Path snapshotBaseDir, @TempDir Path snapshotDownDir) throws Exception {
+ int numRows = 10_000;
+ RocksDB rocksDB = rocksDBExtension.getRocksDb();
+ for (int i = 0; i < numRows; i++) {
+ rocksDB.put(longKey(i), longKey(i));
+ }
+
+ assertSnapshotReadableByFrocksDB(
+ snapshotBaseDir,
+ snapshotDownDir,
+ FrocksDBSnapshotReader.COUNT_ARG,
+ String.valueOf(numRows));
+ }
+
+ private static byte[] longKey(long value) {
+ return ByteBuffer.allocate(Long.BYTES).putLong(value).array();
+ }
+
+ private void assertSnapshotReadableByFrocksDB(
+ Path snapshotBaseDir, Path snapshotDownDir, String... readerArgs) throws Exception {
FsPath testingTabletDir = FsPath.fromLocalFile(snapshotBaseDir.toFile());
SnapshotLocation snapshotLocation =
new SnapshotLocation(
@@ -172,14 +210,6 @@ class RocksIncrementalSnapshotTest {
try (CloseableRegistry closeableRegistry = new CloseableRegistry();
RocksIncrementalSnapshot incrementalSnapshot = createIncrementalSnapshot()) {
- RocksDB rocksDB = rocksDBExtension.getRocksDb();
- rocksDB.put(
- "key1".getBytes(StandardCharsets.UTF_8),
- "val1".getBytes(StandardCharsets.UTF_8));
- rocksDB.put(
- "key2".getBytes(StandardCharsets.UTF_8),
- "val2".getBytes(StandardCharsets.UTF_8));
-
KvSnapshotHandle snapshotHandle =
snapshot(1L, incrementalSnapshot, snapshotLocation, closeableRegistry);
incrementalSnapshot.notifySnapshotComplete(1L);
@@ -190,16 +220,16 @@ class RocksIncrementalSnapshotTest {
snapshotDataDownloader.transferAllDataToDirectory(
new KvSnapshotDownloadSpec(snapshotHandle, restoredDbPath), closeableRegistry);
+ TestProcessBuilder frocksDBReaderBuilder =
+ new TestProcessBuilder(FrocksDBSnapshotReader.class.getName())
+ .addMainClassArg(restoredDbPath.toString());
+ for (String readerArg : readerArgs) {
+ frocksDBReaderBuilder.addMainClassArg(readerArg);
+ }
+
TestProcessBuilder.TestProcess frocksDBReader = null;
try {
- frocksDBReader =
- new TestProcessBuilder(FrocksDBSnapshotReader.class.getName())
- .addMainClassArg(restoredDbPath.toString())
- .addMainClassArg("key1")
- .addMainClassArg("val1")
- .addMainClassArg("key2")
- .addMainClassArg("val2")
- .start();
+ frocksDBReader = frocksDBReaderBuilder.start();
boolean exited = frocksDBReader.getProcess().waitFor(1, TimeUnit.MINUTES);
assertThat(exited)