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 bcf2c2b274 [core] Remove RocksDB lookup state (#8886)
bcf2c2b274 is described below
commit bcf2c2b274c76343077c22fdde82a51a77f49a39
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Jul 28 16:52:57 2026 +0800
[core] Remove RocksDB lookup state (#8886)
---
docs/docs/maintenance/configurations.mdx | 7 -
docs/docs/primary-key-table/data-distribution.md | 2 +-
docs/generated/core_configuration.html | 14 +-
docs/generated/rocksdb_configuration.html | 138 ---------
.../main/java/org/apache/paimon/CoreOptions.java | 15 +-
paimon-benchmark/paimon-micro-benchmarks/pom.xml | 6 -
.../benchmark/lookup/LocalKvDbBenchmark.java | 296 +++----------------
paimon-core/pom.xml | 13 -
.../paimon/crosspartition/GlobalIndexAssigner.java | 10 +-
.../paimon/lookup/local/LocalKvStateFactory.java | 6 +-
.../paimon/lookup/rocksdb/RocksDBBulkLoader.java | 138 ---------
.../paimon/lookup/rocksdb/RocksDBListState.java | 77 -----
.../paimon/lookup/rocksdb/RocksDBOptions.java | 325 ---------------------
.../paimon/lookup/rocksdb/RocksDBSetState.java | 128 --------
.../apache/paimon/lookup/rocksdb/RocksDBState.java | 140 ---------
.../paimon/lookup/rocksdb/RocksDBStateFactory.java | 214 --------------
.../paimon/lookup/rocksdb/RocksDBValueState.java | 94 ------
.../lookup/rocksdb/RocksDBStateFactoryTest.java | 167 -----------
.../configuration/ConfigOptionsDocGenerator.java | 1 -
paimon-flink/paimon-flink-1.17/pom.xml | 7 -
.../flink/lookup/FileStoreLookupFunction.java | 4 +-
.../paimon/flink/lookup/PartitionRefresher.java | 2 +-
.../lookup/DynamicPartitionLevelLoaderTest.java | 3 +-
.../flink/lookup/FileStoreLookupFunctionTest.java | 5 +-
.../paimon/flink/lookup/RocksDBListStateTest.java | 102 -------
25 files changed, 74 insertions(+), 1840 deletions(-)
diff --git a/docs/docs/maintenance/configurations.mdx
b/docs/docs/maintenance/configurations.mdx
index e2cb83d8d8..29feaeacd7 100644
--- a/docs/docs/maintenance/configurations.mdx
+++ b/docs/docs/maintenance/configurations.mdx
@@ -14,7 +14,6 @@ import flinkConnectorConfigurationHtml from
'@site/generated/flink_connector_con
import sparkCatalogConfigurationHtml from
'@site/generated/spark_catalog_configuration.html';
import sparkConnectorConfigurationHtml from
'@site/generated/spark_connector_configuration.html';
import orcConfigurationHtml from '@site/generated/orc_configuration.html';
-import rocksdbConfigurationHtml from
'@site/generated/rocksdb_configuration.html';
<!--
Licensed to the Apache Software Foundation (ASF) under one
@@ -94,9 +93,3 @@ Spark connector options for paimon.
### ORC Options
<ConfigTable html={orcConfigurationHtml} />
-
-### RocksDB Options
-
-The following options allow users to finely adjust RocksDB for better
performance. You can either specify them in table properties or in dynamic
table hints.
-
-<ConfigTable html={rocksdbConfigurationHtml} />
diff --git a/docs/docs/primary-key-table/data-distribution.md
b/docs/docs/primary-key-table/data-distribution.md
index a7cf7aa16d..a7de95c838 100644
--- a/docs/docs/primary-key-table/data-distribution.md
+++ b/docs/docs/primary-key-table/data-distribution.md
@@ -120,7 +120,7 @@ Performance: For tables with a large amount of data, there
will be a significant
initialization takes a long time.
If your upsert does not rely on too old data, you can consider configuring
index TTL to reduce Index and initialization time:
-- `'cross-partition-upsert.index-ttl'`: The TTL in rocksdb index and
initialization, this can avoid maintaining too many
+- `'cross-partition-upsert.index-ttl'`: The TTL in local index and
initialization, this can avoid maintaining too many
indexes and lead to worse and worse performance.
You can also use Cross Partitions Upsert with bucket (N > 0) or bucket (-2),
in these modes, there is no global index to
diff --git a/docs/generated/core_configuration.html
b/docs/generated/core_configuration.html
index 7bc238d4ff..a5a2412092 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -480,7 +480,7 @@ under the License.
<td><h5>cross-partition-upsert.index-ttl</h5></td>
<td style="word-wrap: break-word;">(none)</td>
<td>Duration</td>
- <td>The TTL in rocksdb index for cross partition upsert (primary
keys not contain all partition fields), this can avoid maintaining too many
indexes and lead to worse and worse performance, but please note that this may
also cause data duplication.</td>
+ <td>The TTL in local index for cross partition upsert (primary
keys not contain all partition fields), this can avoid maintaining too many
indexes and lead to worse and worse performance, but please note that this may
also cause data duplication.</td>
</tr>
<tr>
<td><h5>data-evolution.compaction.rewrite-row-ids</h5></td>
@@ -921,6 +921,12 @@ Mainly to resolve data skew on primary keys. We recommend
starting with 64 mb wh
<td>MemorySize</td>
<td>Max memory size for lookup cache.</td>
</tr>
+ <tr>
+ <td><h5>lookup.cache-rows</h5></td>
+ <td style="word-wrap: break-word;">10000</td>
+ <td>Long</td>
+ <td>The maximum number of rows to store in the cache.</td>
+ </tr>
<tr>
<td><h5>lookup.cache-spill-compression</h5></td>
<td style="word-wrap: break-word;">"zstd"</td>
@@ -945,6 +951,12 @@ Mainly to resolve data skew on primary keys. We recommend
starting with 64 mb wh
<td>Double</td>
<td>The fraction of cache memory that is reserved for
high-priority data like index, filter.</td>
</tr>
+ <tr>
+ <td><h5>lookup.continuous.discovery-interval</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Duration</td>
+ <td>The discovery interval of lookup continuous reading. This is
used as an SQL hint. If it's not configured, the lookup function will fallback
to 'continuous.discovery-interval'.</td>
+ </tr>
<tr>
<td><h5>lookup.hash-load-factor</h5></td>
<td style="word-wrap: break-word;">0.75</td>
diff --git a/docs/generated/rocksdb_configuration.html
b/docs/generated/rocksdb_configuration.html
deleted file mode 100644
index d24f2f379f..0000000000
--- a/docs/generated/rocksdb_configuration.html
+++ /dev/null
@@ -1,138 +0,0 @@
-<!--
-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.
--->
-<table class="configuration table table-bordered">
- <thead>
- <tr>
- <th class="text-left" style="width: 20%">Key</th>
- <th class="text-left" style="width: 15%">Default</th>
- <th class="text-left" style="width: 10%">Type</th>
- <th class="text-left" style="width: 55%">Description</th>
- </tr>
- </thead>
- <tbody>
- <tr>
- <td><h5>lookup.cache-rows</h5></td>
- <td style="word-wrap: break-word;">10000</td>
- <td>Long</td>
- <td>The maximum number of rows to store in the cache.</td>
- </tr>
- <tr>
- <td><h5>lookup.continuous.discovery-interval</h5></td>
- <td style="word-wrap: break-word;">(none)</td>
- <td>Duration</td>
- <td>The discovery interval of lookup continuous reading. This is
used as an SQL hint. If it's not configured, the lookup function will fallback
to 'continuous.discovery-interval'.</td>
- </tr>
- <tr>
- <td><h5>rocksdb.block.blocksize</h5></td>
- <td style="word-wrap: break-word;">4 kb</td>
- <td>MemorySize</td>
- <td>The approximate size (in bytes) of user data packed per block.
The default blocksize is '4KB'.</td>
- </tr>
- <tr>
- <td><h5>rocksdb.block.cache-size</h5></td>
- <td style="word-wrap: break-word;">128 mb</td>
- <td>MemorySize</td>
- <td>The amount of the cache for data blocks in RocksDB.</td>
- </tr>
- <tr>
- <td><h5>rocksdb.block.metadata-blocksize</h5></td>
- <td style="word-wrap: break-word;">4 kb</td>
- <td>MemorySize</td>
- <td>Approximate size of partitioned metadata packed per block.
Currently applied to indexes block when partitioned index/filters option is
enabled. The default blocksize is '4KB'.</td>
- </tr>
- <tr>
- <td><h5>rocksdb.bloom-filter.bits-per-key</h5></td>
- <td style="word-wrap: break-word;">10.0</td>
- <td>Double</td>
- <td>Bits per key that bloom filter will use, this only take effect
when bloom filter is used. The default value is 10.0.</td>
- </tr>
- <tr>
- <td><h5>rocksdb.bloom-filter.block-based-mode</h5></td>
- <td style="word-wrap: break-word;">false</td>
- <td>Boolean</td>
- <td>If true, RocksDB will use block-based filter instead of full
filter, this only take effect when bloom filter is used. The default value is
'false'.</td>
- </tr>
- <tr>
- <td><h5>rocksdb.compaction.level.max-size-level-base</h5></td>
- <td style="word-wrap: break-word;">256 mb</td>
- <td>MemorySize</td>
- <td>The upper-bound of the total size of level base files in
bytes. The default value is '256MB'.</td>
- </tr>
- <tr>
- <td><h5>rocksdb.compaction.level.target-file-size-base</h5></td>
- <td style="word-wrap: break-word;">64 mb</td>
- <td>MemorySize</td>
- <td>The target file size for compaction, which determines a
level-1 file size. The default value is '64MB'.</td>
- </tr>
- <tr>
- <td><h5>rocksdb.compaction.level.use-dynamic-size</h5></td>
- <td style="word-wrap: break-word;">false</td>
- <td>Boolean</td>
- <td>If true, RocksDB will pick target size of each level
dynamically. From an empty DB, RocksDB would make last level the base level,
which means merging L0 data into the last level, until it exceeds
max_bytes_for_level_base. And then repeat this process for second last level
and so on. The default value is 'false'. For more information, please refer to
<a
href="https://github.com/facebook/rocksdb/wiki/Leveled-Compaction#level_compaction_dynamic_level_bytes-is-true";>RocksDB's
[...]
- </tr>
- <tr>
- <td><h5>rocksdb.compaction.style</h5></td>
- <td style="word-wrap: break-word;">LEVEL</td>
- <td><p>Enum</p></td>
- <td>The specified compaction style for DB. Candidate compaction
style is LEVEL, FIFO, UNIVERSAL or NONE, and Flink chooses 'LEVEL' as default
style.<br /><br />Possible
values:<ul><li>"LEVEL"</li><li>"UNIVERSAL"</li><li>"FIFO"</li><li>"NONE"</li></ul></td>
- </tr>
- <tr>
- <td><h5>rocksdb.compression.type</h5></td>
- <td style="word-wrap: break-word;">LZ4_COMPRESSION</td>
- <td><p>Enum</p></td>
- <td>The compression type.<br /><br />Possible
values:<ul><li>"NO_COMPRESSION"</li><li>"SNAPPY_COMPRESSION"</li><li>"ZLIB_COMPRESSION"</li><li>"BZLIB2_COMPRESSION"</li><li>"LZ4_COMPRESSION"</li><li>"LZ4HC_COMPRESSION"</li><li>"XPRESS_COMPRESSION"</li><li>"ZSTD_COMPRESSION"</li><li>"DISABLE_COMPRESSION_OPTION"</li></ul></td>
- </tr>
- <tr>
- <td><h5>rocksdb.files.open</h5></td>
- <td style="word-wrap: break-word;">-1</td>
- <td>Integer</td>
- <td>The maximum number of open files (per stateful operator) that
can be used by the DB, '-1' means no limit. The default value is '-1'.</td>
- </tr>
- <tr>
- <td><h5>rocksdb.thread.num</h5></td>
- <td style="word-wrap: break-word;">2</td>
- <td>Integer</td>
- <td>The maximum number of concurrent background flush and
compaction jobs (per stateful operator). The default value is '2'.</td>
- </tr>
- <tr>
- <td><h5>rocksdb.use-bloom-filter</h5></td>
- <td style="word-wrap: break-word;">false</td>
- <td>Boolean</td>
- <td>If true, every newly created SST file will contain a Bloom
filter. It is disabled by default.</td>
- </tr>
- <tr>
- <td><h5>rocksdb.writebuffer.count</h5></td>
- <td style="word-wrap: break-word;">2</td>
- <td>Integer</td>
- <td>The maximum number of write buffers that are built up in
memory. The default value is '2'.</td>
- </tr>
- <tr>
- <td><h5>rocksdb.writebuffer.number-to-merge</h5></td>
- <td style="word-wrap: break-word;">1</td>
- <td>Integer</td>
- <td>The minimum number of write buffers that will be merged
together before writing to storage. The default value is '1'.</td>
- </tr>
- <tr>
- <td><h5>rocksdb.writebuffer.size</h5></td>
- <td style="word-wrap: break-word;">64 mb</td>
- <td>MemorySize</td>
- <td>The amount of data built up in memory (backed by an unsorted
log on disk) before converting to a sorted on-disk files. The default
writebuffer size is '64MB'.</td>
- </tr>
- </tbody>
-</table>
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
index ec95ab6ec5..1a8bd6755e 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -1498,6 +1498,19 @@ public class CoreOptions implements Serializable {
.defaultValue(0.75F)
.withDescription("The index load factor for lookup.");
+ public static final ConfigOption<Long> LOOKUP_CACHE_ROWS =
+ key("lookup.cache-rows")
+ .longType()
+ .defaultValue(10_000L)
+ .withDescription("The maximum number of rows to store in
the cache.");
+
+ public static final ConfigOption<Duration>
LOOKUP_CONTINUOUS_DISCOVERY_INTERVAL =
+ key("lookup.continuous.discovery-interval")
+ .durationType()
+ .noDefaultValue()
+ .withDescription(
+ "The discovery interval of lookup continuous
reading. This is used as an SQL hint. If it's not configured, the lookup
function will fallback to 'continuous.discovery-interval'.");
+
public static final ConfigOption<Duration> LOOKUP_CACHE_FILE_RETENTION =
key("lookup.cache-file-retention")
.durationType()
@@ -2068,7 +2081,7 @@ public class CoreOptions implements Serializable {
.durationType()
.noDefaultValue()
.withDescription(
- "The TTL in rocksdb index for cross partition
upsert (primary keys not contain all partition fields), "
+ "The TTL in local index for cross partition upsert
(primary keys not contain all partition fields), "
+ "this can avoid maintaining too many
indexes and lead to worse and worse performance, "
+ "but please note that this may also
cause data duplication.");
diff --git a/paimon-benchmark/paimon-micro-benchmarks/pom.xml
b/paimon-benchmark/paimon-micro-benchmarks/pom.xml
index 7677e57d7e..7a8feb0b01 100644
--- a/paimon-benchmark/paimon-micro-benchmarks/pom.xml
+++ b/paimon-benchmark/paimon-micro-benchmarks/pom.xml
@@ -161,12 +161,6 @@ under the License.
<scope>test</scope>
</dependency>
- <dependency>
- <groupId>com.ververica</groupId>
- <artifactId>frocksdbjni</artifactId>
- <version>6.20.3-ververica-2.0</version>
- <scope>test</scope>
- </dependency>
</dependencies>
<profiles>
diff --git
a/paimon-benchmark/paimon-micro-benchmarks/src/test/java/org/apache/paimon/benchmark/lookup/LocalKvDbBenchmark.java
b/paimon-benchmark/paimon-micro-benchmarks/src/test/java/org/apache/paimon/benchmark/lookup/LocalKvDbBenchmark.java
index b02971f60f..3deff8606f 100644
---
a/paimon-benchmark/paimon-micro-benchmarks/src/test/java/org/apache/paimon/benchmark/lookup/LocalKvDbBenchmark.java
+++
b/paimon-benchmark/paimon-micro-benchmarks/src/test/java/org/apache/paimon/benchmark/lookup/LocalKvDbBenchmark.java
@@ -32,9 +32,6 @@ import org.apache.paimon.lookup.ListState;
import org.apache.paimon.lookup.SetState;
import org.apache.paimon.lookup.StateFactory;
import org.apache.paimon.lookup.local.LocalKvStateFactory;
-import org.apache.paimon.lookup.rocksdb.RocksDBBulkLoader;
-import org.apache.paimon.lookup.rocksdb.RocksDBOptions;
-import org.apache.paimon.lookup.rocksdb.RocksDBStateFactory;
import org.apache.paimon.lookup.sort.db.LocalKvDb;
import org.apache.paimon.options.MemorySize;
import org.apache.paimon.options.Options;
@@ -44,13 +41,8 @@ import org.apache.paimon.utils.FileIOUtils;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
-import org.rocksdb.CompressionType;
-import org.rocksdb.RocksDB;
-import org.rocksdb.RocksDBException;
-import org.rocksdb.WriteOptions;
import java.io.ByteArrayOutputStream;
-import java.io.Closeable;
import java.io.File;
import java.io.IOException;
import java.nio.file.Path;
@@ -155,47 +147,6 @@ public class LocalKvDbBenchmark {
assertThat(checksum).isNotZero();
}
- @Test
- public void testPointLookupComparison() throws IOException {
- File localDirectory = new File(tempDir.toFile(),
"point-lookup-local-kv-db");
- File rocksDirectory = new File(tempDir.toFile(), "point-lookup-rocks");
- byte[][] hitKeys = queryKeys(false);
- byte[][] missKeys = queryKeys(true);
-
- try (LocalKvDb localDb = createDb(localDirectory);
- RocksDbHandle rocksDb = createRocksDb(rocksDirectory)) {
- long localLoadStart = System.nanoTime();
- localDb.bulkLoad(entries(RECORD_COUNT), RECORD_COUNT);
- long localLoadNanos = System.nanoTime() - localLoadStart;
- long rocksLoadStart = System.nanoTime();
- rocksDb.bulkLoad(entries(RECORD_COUNT));
- long rocksLoadNanos = System.nanoTime() - rocksLoadStart;
- System.out.printf(
- Locale.ROOT,
- "Build: local-kv-db=%.1f ms / %.2f MB, rocks=%.1f ms /
%.2f MB%n",
- localLoadNanos / 1_000_000.0,
- directorySize(localDirectory) / (1024.0 * 1024.0),
- rocksLoadNanos / 1_000_000.0,
- directorySize(rocksDirectory) / (1024.0 * 1024.0));
-
- Benchmark benchmark =
- new Benchmark(
- "local-kv-db-vs-rocks-point-lookup-" +
benchmarkDescription(),
- OPERATION_COUNT)
- .setNumWarmupIters(2)
- .setOutputPerIteration(true);
- benchmark.addCase("local-kv-db-hit", 5, () -> lookup(localDb,
hitKeys, false));
- benchmark.addCase("rocks-hit", 5, () -> lookup(rocksDb.db,
hitKeys, false));
- benchmark.addCase("local-kv-db-miss", 5, () -> lookup(localDb,
missKeys, true));
- benchmark.addCase("rocks-miss", 5, () -> lookup(rocksDb.db,
missKeys, true));
- benchmark.run();
- } finally {
- FileIOUtils.deleteDirectoryQuietly(localDirectory);
- FileIOUtils.deleteDirectoryQuietly(rocksDirectory);
- }
- assertThat(checksum).isNotZero();
- }
-
@Test
public void testClusteringBlockSizeComparison() throws IOException {
int cacheSizeMb =
intProperty("local-kv-db.benchmark.clustering-cache-size-mb", 32);
@@ -299,120 +250,57 @@ public class LocalKvDbBenchmark {
}
@Test
- public void testMixedReadWriteComparison() throws IOException {
- File localDirectory = new File(tempDir.toFile(),
"mixed-read-write-local-kv-db");
- File rocksDirectory = new File(tempDir.toFile(),
"mixed-read-write-rocks");
- byte[][] hitKeys = queryKeys(false);
- byte[][] values = updateValues();
-
- try (LocalKvDb localDb = createDb(localDirectory);
- RocksDbHandle rocksDb = createRocksDb(rocksDirectory)) {
- localDb.bulkLoad(entries(RECORD_COUNT), RECORD_COUNT);
- rocksDb.bulkLoad(entries(RECORD_COUNT));
-
- Benchmark benchmark =
- new Benchmark(
- "local-kv-db-vs-rocks-mixed-read-write-"
- + benchmarkDescription(),
- OPERATION_COUNT)
- .setNumWarmupIters(2)
- .setOutputPerIteration(true);
- benchmark.addCase(
- "local-kv-db-50-percent-put",
- 5,
- () -> mixedReadWrite(localDb, hitKeys, values));
- benchmark.addCase(
- "rocks-50-percent-put", 5, () -> mixedReadWrite(rocksDb,
hitKeys, values));
- benchmark.run();
- } finally {
- FileIOUtils.deleteDirectoryQuietly(localDirectory);
- FileIOUtils.deleteDirectoryQuietly(rocksDirectory);
- }
- assertThat(checksum).isNotZero();
- }
+ public void testStateFanOut() throws IOException {
+ benchmarkListState("list-warmup");
+ benchmarkListBulkLoad("list-bulk-warmup");
+ benchmarkSetState("set-warmup");
- @Test
- public void testStateFanOutComparison() throws IOException {
- benchmarkListState("local-list-warmup", true);
- benchmarkListState("rocks-list-warmup", false);
- benchmarkListBulkLoad("local-list-bulk-warmup", true);
- benchmarkListBulkLoad("rocks-list-bulk-warmup", false);
- benchmarkSetState("local-set-warmup", true);
- benchmarkSetState("rocks-set-warmup", false);
-
- StateBenchmarkResult localList = benchmarkListState("local-list",
true);
- StateBenchmarkResult rocksList = benchmarkListState("rocks-list",
false);
- StateBenchmarkResult localListBulk =
benchmarkListBulkLoad("local-list-bulk", true);
- StateBenchmarkResult rocksListBulk =
benchmarkListBulkLoad("rocks-list-bulk", false);
- StateBenchmarkResult localSet = benchmarkSetState("local-set", true);
- StateBenchmarkResult rocksSet = benchmarkSetState("rocks-set", false);
+ StateBenchmarkResult list = benchmarkListState("list");
+ StateBenchmarkResult listBulk = benchmarkListBulkLoad("list-bulk");
+ StateBenchmarkResult set = benchmarkSetState("set");
System.out.printf(
Locale.ROOT,
"ListState (%d keys x %d values, local cache=%s):%n"
- + " local-kv-db add=%.1f ms, get=%.1f ms, cached=%.1f
ms, close=%.1f ms, size=%.2f MB%n"
- + " rocks add=%.1f ms, get=%.1f ms, cached=%.1f
ms, close=%.1f ms, size=%.2f MB%n",
+ + " add=%.1f ms, get=%.1f ms, cached=%.1f ms,
close=%.1f ms, size=%.2f MB%n",
STATE_KEY_COUNT,
STATE_FAN_OUT,
CACHE_OFF_HEAP ? "off-heap" : "heap",
- localList.updateNanos / 1_000_000.0,
- localList.readNanos / 1_000_000.0,
- localList.cachedReadNanos / 1_000_000.0,
- localList.closeNanos / 1_000_000.0,
- localList.directoryBytes / (1024.0 * 1024.0),
- rocksList.updateNanos / 1_000_000.0,
- rocksList.readNanos / 1_000_000.0,
- rocksList.cachedReadNanos / 1_000_000.0,
- rocksList.closeNanos / 1_000_000.0,
- rocksList.directoryBytes / (1024.0 * 1024.0));
+ list.updateNanos / 1_000_000.0,
+ list.readNanos / 1_000_000.0,
+ list.cachedReadNanos / 1_000_000.0,
+ list.closeNanos / 1_000_000.0,
+ list.directoryBytes / (1024.0 * 1024.0));
System.out.printf(
Locale.ROOT,
"ListState bulk load (%d keys x %d values):%n"
- + " local-kv-db load=%.1f ms, get=%.1f ms,
cached=%.1f ms, close=%.1f ms, size=%.2f MB%n"
- + " rocks load=%.1f ms, get=%.1f ms,
cached=%.1f ms, close=%.1f ms, size=%.2f MB%n",
+ + " load=%.1f ms, get=%.1f ms, cached=%.1f ms,
close=%.1f ms, size=%.2f MB%n",
STATE_KEY_COUNT,
STATE_FAN_OUT,
- localListBulk.updateNanos / 1_000_000.0,
- localListBulk.readNanos / 1_000_000.0,
- localListBulk.cachedReadNanos / 1_000_000.0,
- localListBulk.closeNanos / 1_000_000.0,
- localListBulk.directoryBytes / (1024.0 * 1024.0),
- rocksListBulk.updateNanos / 1_000_000.0,
- rocksListBulk.readNanos / 1_000_000.0,
- rocksListBulk.cachedReadNanos / 1_000_000.0,
- rocksListBulk.closeNanos / 1_000_000.0,
- rocksListBulk.directoryBytes / (1024.0 * 1024.0));
+ listBulk.updateNanos / 1_000_000.0,
+ listBulk.readNanos / 1_000_000.0,
+ listBulk.cachedReadNanos / 1_000_000.0,
+ listBulk.closeNanos / 1_000_000.0,
+ listBulk.directoryBytes / (1024.0 * 1024.0));
System.out.printf(
Locale.ROOT,
"SetState (%d keys x %d values):%n"
- + " local-kv-db add=%.1f ms, get=%.1f ms, cached=%.1f
ms, close=%.1f ms, size=%.2f MB%n"
- + " rocks add=%.1f ms, get=%.1f ms, cached=%.1f
ms, close=%.1f ms, size=%.2f MB%n",
+ + " add=%.1f ms, get=%.1f ms, cached=%.1f ms,
close=%.1f ms, size=%.2f MB%n",
STATE_KEY_COUNT,
STATE_FAN_OUT,
- localSet.updateNanos / 1_000_000.0,
- localSet.readNanos / 1_000_000.0,
- localSet.cachedReadNanos / 1_000_000.0,
- localSet.closeNanos / 1_000_000.0,
- localSet.directoryBytes / (1024.0 * 1024.0),
- rocksSet.updateNanos / 1_000_000.0,
- rocksSet.readNanos / 1_000_000.0,
- rocksSet.cachedReadNanos / 1_000_000.0,
- rocksSet.closeNanos / 1_000_000.0,
- rocksSet.directoryBytes / (1024.0 * 1024.0));
-
- checksum +=
- localList.checksum
- + rocksList.checksum
- + localListBulk.checksum
- + rocksListBulk.checksum
- + localSet.checksum
- + rocksSet.checksum;
+ set.updateNanos / 1_000_000.0,
+ set.readNanos / 1_000_000.0,
+ set.cachedReadNanos / 1_000_000.0,
+ set.closeNanos / 1_000_000.0,
+ set.directoryBytes / (1024.0 * 1024.0));
+
+ checksum += list.checksum + listBulk.checksum + set.checksum;
assertThat(checksum).isNotZero();
}
- private StateBenchmarkResult benchmarkListState(String name, boolean
local) throws IOException {
+ private StateBenchmarkResult benchmarkListState(String name) throws
IOException {
File directory = new File(tempDir.toFile(), name);
- StateFactory factory = createStateFactory(directory, local);
+ StateFactory factory = createStateFactory(directory);
long updateNanos = 0;
long readNanos = 0;
long cachedReadNanos = 0;
@@ -467,9 +355,9 @@ public class LocalKvDbBenchmark {
updateNanos, readNanos, cachedReadNanos, closeNanos,
directoryBytes, localChecksum);
}
- private StateBenchmarkResult benchmarkSetState(String name, boolean local)
throws IOException {
+ private StateBenchmarkResult benchmarkSetState(String name) throws
IOException {
File directory = new File(tempDir.toFile(), name);
- StateFactory factory = createStateFactory(directory, local);
+ StateFactory factory = createStateFactory(directory);
long updateNanos = 0;
long readNanos = 0;
long cachedReadNanos = 0;
@@ -524,10 +412,9 @@ public class LocalKvDbBenchmark {
updateNanos, readNanos, cachedReadNanos, closeNanos,
directoryBytes, localChecksum);
}
- private StateBenchmarkResult benchmarkListBulkLoad(String name, boolean
local)
- throws IOException {
+ private StateBenchmarkResult benchmarkListBulkLoad(String name) throws
IOException {
File directory = new File(tempDir.toFile(), name);
- StateFactory factory = createStateFactory(directory, local);
+ StateFactory factory = createStateFactory(directory);
long loadNanos = 0;
long readNanos = 0;
long cachedReadNanos = 0;
@@ -589,18 +476,13 @@ public class LocalKvDbBenchmark {
loadNanos, readNanos, cachedReadNanos, closeNanos,
directoryBytes, localChecksum);
}
- private StateFactory createStateFactory(File directory, boolean local)
throws IOException {
+ private StateFactory createStateFactory(File directory) {
Options options = new Options();
options.set(
CoreOptions.LOOKUP_CACHE_MAX_MEMORY_SIZE,
MemorySize.ofMebiBytes(CACHE_SIZE_MB));
options.set(CoreOptions.LOOKUP_CACHE_SPILL_COMPRESSION, COMPRESSION);
- options.set(RocksDBOptions.BLOCK_CACHE_SIZE,
MemorySize.ofMebiBytes(CACHE_SIZE_MB));
- options.set(RocksDBOptions.COMPRESSION_TYPE, rocksCompression());
- if (local) {
- return new LocalKvStateFactory(
- directory.getAbsolutePath(), options, null, null,
CACHE_OFF_HEAP);
- }
- return new RocksDBStateFactory(directory.getAbsolutePath(), options,
null);
+ return new LocalKvStateFactory(
+ directory.getAbsolutePath(), options, null, null,
CACHE_OFF_HEAP);
}
private void mixedReadWrite(LocalKvDb db, byte[][] keys, byte[][] values) {
@@ -623,26 +505,6 @@ public class LocalKvDbBenchmark {
}
}
- private void mixedReadWrite(RocksDbHandle handle, byte[][] keys, byte[][]
values) {
- try {
- long localChecksum = 0;
- for (int i = 0; i < OPERATION_COUNT; i++) {
- byte[] key = keys[i];
- if ((i & 1) == 0) {
- handle.db.put(handle.writeOptions, key, values[i &
(values.length - 1)]);
- } else {
- byte[] result = handle.db.get(key);
- if (result != null) {
- localChecksum += result[0] & 0xff;
- }
- }
- }
- checksum += localChecksum;
- } catch (RocksDBException e) {
- throw new RuntimeException(e);
- }
- }
-
private static byte[][] updateValues() {
byte[][] values = new byte[1024][];
for (int i = 0; i < values.length; i++) {
@@ -672,27 +534,6 @@ public class LocalKvDbBenchmark {
}
}
- private void lookup(RocksDB db, byte[][] keys, boolean expectMiss) {
- try {
- long localChecksum = 0;
- for (byte[] key : keys) {
- byte[] value = db.get(key);
- if (expectMiss) {
- if (value != null) {
- throw new IllegalStateException("Expected lookup
miss.");
- }
- } else if (value == null) {
- throw new IllegalStateException("Expected lookup hit.");
- } else {
- localChecksum += value[0] & 0xff;
- }
- }
- checksum += localChecksum;
- } catch (RocksDBException e) {
- throw new RuntimeException(e);
- }
- }
-
private LocalKvDb createDb(File directory) {
LocalKvDb.Builder builder =
LocalKvDb.builder(directory)
@@ -729,37 +570,6 @@ public class LocalKvDbBenchmark {
: new CacheManager(cacheSize, 0);
}
- private RocksDbHandle createRocksDb(File directory) throws IOException {
- Options options = new Options();
- options.set(RocksDBOptions.WRITE_BUFFER_SIZE,
MemorySize.ofMebiBytes(MEMTABLE_SIZE_MB));
- options.set(RocksDBOptions.TARGET_FILE_SIZE_BASE,
MemorySize.ofMebiBytes(SST_FILE_SIZE_MB));
- options.set(RocksDBOptions.BLOCK_SIZE,
MemorySize.ofKibiBytes(BLOCK_SIZE_KB));
- options.set(RocksDBOptions.BLOCK_CACHE_SIZE,
MemorySize.ofMebiBytes(CACHE_SIZE_MB));
- options.set(RocksDBOptions.COMPRESSION_TYPE, rocksCompression());
- if (BLOOM_FILTER_FPP > 0) {
- options.set(RocksDBOptions.USE_BLOOM_FILTER, true);
- options.set(
- RocksDBOptions.BLOOM_FILTER_BITS_PER_KEY,
- -Math.log(BLOOM_FILTER_FPP) / (Math.log(2) * Math.log(2)));
- }
- return new RocksDbHandle(
- new RocksDBStateFactory(directory.getAbsolutePath(), options,
null));
- }
-
- private static CompressionType rocksCompression() {
- switch (COMPRESSION.toLowerCase()) {
- case "none":
- return CompressionType.NO_COMPRESSION;
- case "lz4":
- return CompressionType.LZ4_COMPRESSION;
- case "zstd":
- return CompressionType.ZSTD_COMPRESSION;
- default:
- throw new IllegalArgumentException(
- "Unsupported RocksDB compression for benchmark: " +
COMPRESSION);
- }
- }
-
private byte[][] queryKeys(boolean missing) {
byte[][] keys = new byte[OPERATION_COUNT][];
long random = missing ? 0xcafebabeL : 0x5deece66dL;
@@ -912,40 +722,6 @@ public class LocalKvDbBenchmark {
BLOOM_FILTER_FPP > 0 ? Double.toString(BLOOM_FILTER_FPP) :
"disabled");
}
- private static class RocksDbHandle implements Closeable {
-
- private final RocksDBStateFactory factory;
- private final RocksDB db;
- private final WriteOptions writeOptions;
-
- private RocksDbHandle(RocksDBStateFactory factory) {
- this.factory = factory;
- this.db = factory.db();
- this.writeOptions = new WriteOptions().setDisableWAL(true);
- }
-
- private void bulkLoad(Iterator<Map.Entry<byte[], byte[]>> entries)
throws IOException {
- RocksDBBulkLoader loader =
- new RocksDBBulkLoader(
- db, factory.options(),
db.getDefaultColumnFamily(), factory.path());
- try {
- while (entries.hasNext()) {
- Map.Entry<byte[], byte[]> entry = entries.next();
- loader.write(entry.getKey(), entry.getValue());
- }
- } catch (RocksDBBulkLoader.WriteException e) {
- throw new IOException(e);
- }
- loader.finish();
- }
-
- @Override
- public void close() throws IOException {
- writeOptions.close();
- factory.close();
- }
- }
-
private static class StateBenchmarkResult {
private final long updateNanos;
diff --git a/paimon-core/pom.xml b/paimon-core/pom.xml
index 8cbcc0ca47..e868a6a3a1 100644
--- a/paimon-core/pom.xml
+++ b/paimon-core/pom.xml
@@ -31,10 +31,6 @@ under the License.
<artifactId>paimon-core</artifactId>
<name>Paimon : Core</name>
- <properties>
- <frocksdbjni.version>6.20.3-ververica-2.0</frocksdbjni.version>
- </properties>
-
<dependencies>
<dependency>
<groupId>org.apache.paimon</groupId>
@@ -83,15 +79,6 @@ under the License.
</exclusions>
</dependency>
- <!-- flink dependencies -->
-
- <dependency>
- <groupId>com.ververica</groupId>
- <artifactId>frocksdbjni</artifactId>
- <version>${frocksdbjni.version}</version>
- <scope>provided</scope>
- </dependency>
-
<!-- test dependencies -->
<dependency>
diff --git
a/paimon-core/src/main/java/org/apache/paimon/crosspartition/GlobalIndexAssigner.java
b/paimon-core/src/main/java/org/apache/paimon/crosspartition/GlobalIndexAssigner.java
index 6e5d4b6e51..eb56a537fc 100644
---
a/paimon-core/src/main/java/org/apache/paimon/crosspartition/GlobalIndexAssigner.java
+++
b/paimon-core/src/main/java/org/apache/paimon/crosspartition/GlobalIndexAssigner.java
@@ -34,7 +34,6 @@ import org.apache.paimon.lookup.StateUtils;
import org.apache.paimon.lookup.ValueBulkLoader;
import org.apache.paimon.lookup.ValueState;
import org.apache.paimon.lookup.local.LocalKvStateFactory;
-import org.apache.paimon.lookup.rocksdb.RocksDBOptions;
import org.apache.paimon.memory.HeapMemorySegmentPool;
import org.apache.paimon.options.MemorySize;
import org.apache.paimon.options.Options;
@@ -74,7 +73,7 @@ import java.util.function.BiConsumer;
import java.util.function.Function;
import java.util.stream.IntStream;
-import static org.apache.paimon.lookup.rocksdb.RocksDBOptions.BLOCK_CACHE_SIZE;
+import static org.apache.paimon.CoreOptions.LOOKUP_CACHE_ROWS;
import static org.apache.paimon.utils.Preconditions.checkArgument;
/** Assign UPDATE_BEFORE and bucket for the input record, output record with
bucket. */
@@ -148,10 +147,7 @@ public class GlobalIndexAssigner implements Serializable,
Closeable {
Options options = coreOptions.toConfiguration();
Options stateOptions = Options.fromMap(new HashMap<>(options.toMap()));
// we should avoid too small memory
- long configuredCache =
- options.contains(CoreOptions.LOOKUP_CACHE_MAX_MEMORY_SIZE)
- ? coreOptions.lookupCacheMaxMemory().getBytes()
- : options.get(BLOCK_CACHE_SIZE).getBytes();
+ long configuredCache = coreOptions.lookupCacheMaxMemory().getBytes();
long cacheMemory = Math.max(offHeapMemory, configuredCache);
stateOptions.set(CoreOptions.LOOKUP_CACHE_MAX_MEMORY_SIZE, new
MemorySize(cacheMemory));
this.stateFactory =
@@ -167,7 +163,7 @@ public class GlobalIndexAssigner implements Serializable,
Closeable {
INDEX_NAME,
new RowCompactedSerializer(keyType),
new PositiveIntIntSerializer(),
- options.get(RocksDBOptions.LOOKUP_CACHE_ROWS));
+ options.get(LOOKUP_CACHE_ROWS));
this.partMapping = new IDMapping<>(BinaryRow::copy);
this.bucketAssigner = new BucketAssigner();
diff --git
a/paimon-core/src/main/java/org/apache/paimon/lookup/local/LocalKvStateFactory.java
b/paimon-core/src/main/java/org/apache/paimon/lookup/local/LocalKvStateFactory.java
index f6f538ad0d..b3041a988f 100644
---
a/paimon-core/src/main/java/org/apache/paimon/lookup/local/LocalKvStateFactory.java
+++
b/paimon-core/src/main/java/org/apache/paimon/lookup/local/LocalKvStateFactory.java
@@ -25,7 +25,6 @@ import org.apache.paimon.lookup.ListState;
import org.apache.paimon.lookup.SetState;
import org.apache.paimon.lookup.StateFactory;
import org.apache.paimon.lookup.ValueState;
-import org.apache.paimon.lookup.rocksdb.RocksDBOptions;
import org.apache.paimon.lookup.sort.db.LocalKvDb;
import org.apache.paimon.options.MemorySize;
import org.apache.paimon.options.Options;
@@ -86,10 +85,7 @@ public class LocalKvStateFactory implements StateFactory {
}
this.coreOptions = new CoreOptions(options);
this.options = options;
- MemorySize cacheMemory =
- options.contains(CoreOptions.LOOKUP_CACHE_MAX_MEMORY_SIZE)
- ? coreOptions.lookupCacheMaxMemory()
- : options.get(RocksDBOptions.BLOCK_CACHE_SIZE);
+ MemorySize cacheMemory = coreOptions.lookupCacheMaxMemory();
this.cacheManager =
offHeapCache
? CacheManager.createOffHeap(
diff --git
a/paimon-core/src/main/java/org/apache/paimon/lookup/rocksdb/RocksDBBulkLoader.java
b/paimon-core/src/main/java/org/apache/paimon/lookup/rocksdb/RocksDBBulkLoader.java
deleted file mode 100644
index 8278376f73..0000000000
---
a/paimon-core/src/main/java/org/apache/paimon/lookup/rocksdb/RocksDBBulkLoader.java
+++ /dev/null
@@ -1,138 +0,0 @@
-/*
- * 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.paimon.lookup.rocksdb;
-
-import org.apache.paimon.lookup.ListBulkLoader;
-import org.apache.paimon.lookup.ValueBulkLoader;
-import org.apache.paimon.utils.ListDelimitedSerializer;
-
-import org.rocksdb.ColumnFamilyHandle;
-import org.rocksdb.EnvOptions;
-import org.rocksdb.IngestExternalFileOptions;
-import org.rocksdb.Options;
-import org.rocksdb.RocksDB;
-import org.rocksdb.RocksDBException;
-import org.rocksdb.SstFileWriter;
-import org.rocksdb.TtlDB;
-
-import java.io.File;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.UUID;
-
-/** Bulk loader for RocksDB. */
-public class RocksDBBulkLoader implements ValueBulkLoader, ListBulkLoader {
-
- private final String uuid = UUID.randomUUID().toString();
- private final ListDelimitedSerializer listSerializer = new
ListDelimitedSerializer();
-
- private final ColumnFamilyHandle columnFamily;
- private final String path;
- private final RocksDB db;
- private final boolean isTtlEnabled;
- private final Options options;
- private final List<String> files = new ArrayList<>();
- private final int currentTimeSeconds;
-
- private SstFileWriter writer = null;
- private int sstIndex = 0;
- private long recordNum = 0;
-
- public RocksDBBulkLoader(
- RocksDB db, Options options, ColumnFamilyHandle columnFamily,
String path) {
- this.db = db;
- this.isTtlEnabled = db instanceof TtlDB;
- this.options = options;
- this.columnFamily = columnFamily;
- this.path = path;
- this.currentTimeSeconds = (int) (System.currentTimeMillis() / 1000);
- }
-
- @Override
- public void write(byte[] key, byte[] value) throws WriteException {
- try {
- if (writer == null) {
- writer = new SstFileWriter(new EnvOptions(), options);
- String path = new File(this.path, "sst-" + uuid + "-" +
(sstIndex++)).getPath();
- writer.open(path);
- files.add(path);
- }
-
- if (isTtlEnabled) {
- value = appendTimestamp(value);
- }
-
- try {
- writer.put(key, value);
- } catch (RocksDBException e) {
- throw new WriteException(e);
- }
-
- recordNum++;
- if (recordNum % 1000 == 0 && writer.fileSize() >=
options.targetFileSizeBase()) {
- writer.finish();
- writer.close();
- writer = null;
- recordNum = 0;
- }
- } catch (RocksDBException e) {
- throw new RuntimeException(e);
- }
- }
-
- @Override
- public void write(byte[] key, List<byte[]> value) throws WriteException {
- byte[] bytes;
- try {
- bytes = listSerializer.serializeList(value);
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- write(key, bytes);
- }
-
- private byte[] appendTimestamp(byte[] value) {
- byte[] newValue = new byte[value.length + 4];
- System.arraycopy(value, 0, newValue, 0, value.length);
- newValue[value.length] = (byte) (currentTimeSeconds & 0xff);
- newValue[value.length + 1] = (byte) ((currentTimeSeconds >> 8) & 0xff);
- newValue[value.length + 2] = (byte) ((currentTimeSeconds >> 16) &
0xff);
- newValue[value.length + 3] = (byte) ((currentTimeSeconds >> 24) &
0xff);
- return newValue;
- }
-
- @Override
- public void finish() {
- try {
- if (writer != null) {
- writer.finish();
- writer.close();
- }
-
- if (files.size() > 0) {
- IngestExternalFileOptions ingestOptions = new
IngestExternalFileOptions();
- db.ingestExternalFile(columnFamily, files, ingestOptions);
- ingestOptions.close();
- }
- } catch (RocksDBException e) {
- throw new RuntimeException(e);
- }
- }
-}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/lookup/rocksdb/RocksDBListState.java
b/paimon-core/src/main/java/org/apache/paimon/lookup/rocksdb/RocksDBListState.java
deleted file mode 100644
index 70716c40cb..0000000000
---
a/paimon-core/src/main/java/org/apache/paimon/lookup/rocksdb/RocksDBListState.java
+++ /dev/null
@@ -1,77 +0,0 @@
-/*
- * 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.paimon.lookup.rocksdb;
-
-import org.apache.paimon.data.serializer.Serializer;
-import org.apache.paimon.lookup.ListState;
-import org.apache.paimon.utils.ListDelimitedSerializer;
-
-import org.rocksdb.ColumnFamilyHandle;
-import org.rocksdb.RocksDBException;
-
-import java.io.IOException;
-import java.util.Collections;
-import java.util.List;
-
-/** RocksDB state for key -> List of value. */
-public class RocksDBListState<K, V> extends RocksDBState<K, V, List<V>>
implements ListState<K, V> {
-
- private final ListDelimitedSerializer listSerializer = new
ListDelimitedSerializer();
-
- public RocksDBListState(
- RocksDBStateFactory stateFactory,
- ColumnFamilyHandle columnFamily,
- Serializer<K> keySerializer,
- Serializer<V> valueSerializer,
- long lruCacheSize) {
- super(stateFactory, columnFamily, keySerializer, valueSerializer,
lruCacheSize);
- }
-
- @Override
- public void add(K key, V value) throws IOException {
- byte[] keyBytes = serializeKey(key);
- byte[] valueBytes = serializeValue(value);
- try {
- db.merge(columnFamily, writeOptions, keyBytes, valueBytes);
- } catch (RocksDBException e) {
- throw new IOException(e);
- }
- cache.invalidate(wrap(keyBytes));
- }
-
- @Override
- public List<V> get(K key) throws IOException {
- byte[] keyBytes = serializeKey(key);
- return cache.get(
- wrap(keyBytes),
- k -> {
- byte[] valueBytes;
- try {
- valueBytes = db.get(columnFamily, keyBytes);
- } catch (RocksDBException e) {
- throw new RuntimeException(e);
- }
- List<V> rows = listSerializer.deserializeList(valueBytes,
valueSerializer);
- if (rows == null) {
- return Collections.emptyList();
- }
- return rows;
- });
- }
-}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/lookup/rocksdb/RocksDBOptions.java
b/paimon-core/src/main/java/org/apache/paimon/lookup/rocksdb/RocksDBOptions.java
deleted file mode 100644
index 2cde6d6e7b..0000000000
---
a/paimon-core/src/main/java/org/apache/paimon/lookup/rocksdb/RocksDBOptions.java
+++ /dev/null
@@ -1,325 +0,0 @@
-/*
- * 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.paimon.lookup.rocksdb;
-
-import org.apache.paimon.annotation.Documentation;
-import org.apache.paimon.options.ConfigOption;
-import org.apache.paimon.options.MemorySize;
-import org.apache.paimon.options.Options;
-import org.apache.paimon.options.description.Description;
-
-import org.rocksdb.BlockBasedTableConfig;
-import org.rocksdb.BloomFilter;
-import org.rocksdb.ColumnFamilyOptions;
-import org.rocksdb.CompactionStyle;
-import org.rocksdb.CompressionType;
-import org.rocksdb.DBOptions;
-import org.rocksdb.InfoLogLevel;
-import org.rocksdb.PlainTableConfig;
-import org.rocksdb.TableFormatConfig;
-
-import java.time.Duration;
-
-import static org.apache.paimon.options.ConfigOptions.key;
-import static org.apache.paimon.options.description.LinkElement.link;
-import static org.apache.paimon.options.description.TextElement.code;
-import static org.rocksdb.CompactionStyle.FIFO;
-import static org.rocksdb.CompactionStyle.LEVEL;
-import static org.rocksdb.CompactionStyle.NONE;
-import static org.rocksdb.CompactionStyle.UNIVERSAL;
-import static org.rocksdb.CompressionType.LZ4_COMPRESSION;
-import static org.rocksdb.InfoLogLevel.INFO_LEVEL;
-
-/** Options for rocksdb. Copied from flink {@code RocksDBConfigurableOptions}.
*/
-public class RocksDBOptions {
-
- public static final ConfigOption<Long> LOOKUP_CACHE_ROWS =
- key("lookup.cache-rows")
- .longType()
- .defaultValue(10_000L)
- .withDescription("The maximum number of rows to store in
the cache.");
-
- public static final ConfigOption<Duration>
LOOKUP_CONTINUOUS_DISCOVERY_INTERVAL =
- key("lookup.continuous.discovery-interval")
- .durationType()
- .noDefaultValue()
- .withDescription(
- "The discovery interval of lookup continuous
reading. This is used as an SQL hint. If it's not configured, the lookup
function will fallback to 'continuous.discovery-interval'.");
-
- //
--------------------------------------------------------------------------
- // Provided configurable DBOptions within Flink
- //
--------------------------------------------------------------------------
-
- public static final ConfigOption<Integer> MAX_BACKGROUND_THREADS =
- key("rocksdb.thread.num")
- .intType()
- .defaultValue(2)
- .withDescription(
- "The maximum number of concurrent background flush
and compaction jobs (per stateful operator). "
- + "The default value is '2'.");
-
- public static final ConfigOption<Integer> MAX_OPEN_FILES =
- key("rocksdb.files.open")
- .intType()
- .defaultValue(-1)
- .withDescription(
- "The maximum number of open files (per stateful
operator) that can be used by the DB, '-1' means no limit. "
- + "The default value is '-1'.");
-
- @Documentation.ExcludeFromDocumentation("Internal use only")
- public static final ConfigOption<MemorySize> LOG_MAX_FILE_SIZE =
- key("rocksdb.log.max-file-size")
- .memoryType()
- .defaultValue(MemorySize.parse("25mb"))
- .withDescription(
- "The maximum size of RocksDB's file used for
information logging. "
- + "If the log files becomes larger than
this, a new file will be created. "
- + "If 0, all logs will be written to one
log file. "
- + "The default maximum file size is
'25MB'. ");
-
- @Documentation.ExcludeFromDocumentation("Internal use only")
- public static final ConfigOption<Integer> LOG_FILE_NUM =
- key("rocksdb.log.file-num")
- .intType()
- .defaultValue(4)
- .withDescription(
- "The maximum number of files RocksDB should keep
for information logging (Default setting: 4).");
-
- @Documentation.ExcludeFromDocumentation("Internal use only")
- public static final ConfigOption<String> LOG_DIR =
- key("rocksdb.log.dir")
- .stringType()
- .noDefaultValue()
- .withDescription(
- "The directory for RocksDB's information logging
files. "
- + "If empty (Flink default setting), log
files will be in the same directory as the Flink log. "
- + "If non-empty, this directory will be
used and the data directory's absolute path will be used as the prefix of the
log file name.");
-
- @Documentation.ExcludeFromDocumentation("Internal use only")
- public static final ConfigOption<InfoLogLevel> LOG_LEVEL =
- key("rocksdb.log.level")
- .enumType(InfoLogLevel.class)
- .defaultValue(INFO_LEVEL)
- .withDescription(
- Description.builder()
- .text(
- "The specified information logging
level for RocksDB. "
- + "If unset, Flink will
use %s.",
- code(INFO_LEVEL.name()))
- .linebreak()
- .text(
- "Note: RocksDB info logs will not
be written to the TaskManager logs and there "
- + "is no rolling strategy,
unless you configure %s, %s, and %s accordingly. "
- + "Without a rolling
strategy, long-running tasks may lead to uncontrolled "
- + "disk space usage if
configured with increased log levels!",
- code(LOG_DIR.key()),
- code(LOG_MAX_FILE_SIZE.key()),
- code(LOG_FILE_NUM.key()))
- .linebreak()
- .text(
- "There is no need to modify the
RocksDB log level, unless for troubleshooting RocksDB.")
- .build());
-
- //
--------------------------------------------------------------------------
- // Provided configurable ColumnFamilyOptions within Flink
- //
--------------------------------------------------------------------------
-
- public static final ConfigOption<CompressionType> COMPRESSION_TYPE =
- key("rocksdb.compression.type")
- .enumType(CompressionType.class)
- .defaultValue(LZ4_COMPRESSION)
- .withDescription("The compression type.");
-
- public static final ConfigOption<CompactionStyle> COMPACTION_STYLE =
- key("rocksdb.compaction.style")
- .enumType(CompactionStyle.class)
- .defaultValue(LEVEL)
- .withDescription(
- String.format(
- "The specified compaction style for DB.
Candidate compaction style is %s, %s, %s or %s, "
- + "and Flink chooses '%s' as
default style.",
- LEVEL.name(),
- FIFO.name(),
- UNIVERSAL.name(),
- NONE.name(),
- LEVEL.name()));
-
- public static final ConfigOption<Boolean> USE_DYNAMIC_LEVEL_SIZE =
- key("rocksdb.compaction.level.use-dynamic-size")
- .booleanType()
- .defaultValue(false)
- .withDescription(
- Description.builder()
- .text(
- "If true, RocksDB will pick target
size of each level dynamically. From an empty DB, ")
- .text(
- "RocksDB would make last level the
base level, which means merging L0 data into the last level, ")
- .text(
- "until it exceeds
max_bytes_for_level_base. And then repeat this process for second last level
and so on. ")
- .text("The default value is 'false'. ")
- .text(
- "For more information, please
refer to %s",
- link(
-
"https://github.com/facebook/rocksdb/wiki/Leveled-Compaction#level_compaction_dynamic_level_bytes-is-true";,
- "RocksDB's doc."))
- .build());
-
- public static final ConfigOption<MemorySize> TARGET_FILE_SIZE_BASE =
- key("rocksdb.compaction.level.target-file-size-base")
- .memoryType()
- .defaultValue(MemorySize.parse("64mb"))
- .withDescription(
- "The target file size for compaction, which
determines a level-1 file size. "
- + "The default value is '64MB'.");
-
- public static final ConfigOption<MemorySize> MAX_SIZE_LEVEL_BASE =
- key("rocksdb.compaction.level.max-size-level-base")
- .memoryType()
- .defaultValue(MemorySize.parse("256mb"))
- .withDescription(
- "The upper-bound of the total size of level base
files in bytes. "
- + "The default value is '256MB'.");
-
- public static final ConfigOption<MemorySize> WRITE_BUFFER_SIZE =
- key("rocksdb.writebuffer.size")
- .memoryType()
- .defaultValue(MemorySize.parse("64mb"))
- .withDescription(
- "The amount of data built up in memory (backed by
an unsorted log on disk) "
- + "before converting to a sorted on-disk
files. The default writebuffer size is '64MB'.");
-
- public static final ConfigOption<Integer> MAX_WRITE_BUFFER_NUMBER =
- key("rocksdb.writebuffer.count")
- .intType()
- .defaultValue(2)
- .withDescription(
- "The maximum number of write buffers that are
built up in memory. "
- + "The default value is '2'.");
-
- public static final ConfigOption<Integer> MIN_WRITE_BUFFER_NUMBER_TO_MERGE
=
- key("rocksdb.writebuffer.number-to-merge")
- .intType()
- .defaultValue(1)
- .withDescription(
- "The minimum number of write buffers that will be
merged together before writing to storage. "
- + "The default value is '1'.");
-
- public static final ConfigOption<MemorySize> BLOCK_SIZE =
- key("rocksdb.block.blocksize")
- .memoryType()
- .defaultValue(MemorySize.parse("4kb"))
- .withDescription(
- "The approximate size (in bytes) of user data
packed per block. "
- + "The default blocksize is '4KB'.");
-
- public static final ConfigOption<MemorySize> METADATA_BLOCK_SIZE =
- key("rocksdb.block.metadata-blocksize")
- .memoryType()
- .defaultValue(MemorySize.parse("4kb"))
- .withDescription(
- "Approximate size of partitioned metadata packed
per block. "
- + "Currently applied to indexes block when
partitioned index/filters option is enabled. "
- + "The default blocksize is '4KB'.");
-
- public static final ConfigOption<MemorySize> BLOCK_CACHE_SIZE =
- key("rocksdb.block.cache-size")
- .memoryType()
- .defaultValue(MemorySize.parse("128mb"))
- .withDescription("The amount of the cache for data blocks
in RocksDB.");
-
- public static final ConfigOption<Boolean> USE_BLOOM_FILTER =
- key("rocksdb.use-bloom-filter")
- .booleanType()
- .defaultValue(false)
- .withDescription(
- "If true, every newly created SST file will
contain a Bloom filter. "
- + "It is disabled by default.");
-
- public static final ConfigOption<Double> BLOOM_FILTER_BITS_PER_KEY =
- key("rocksdb.bloom-filter.bits-per-key")
- .doubleType()
- .defaultValue(10.0)
- .withDescription(
- "Bits per key that bloom filter will use, this
only take effect when bloom filter is used. "
- + "The default value is 10.0.");
-
- public static final ConfigOption<Boolean> BLOOM_FILTER_BLOCK_BASED_MODE =
- key("rocksdb.bloom-filter.block-based-mode")
- .booleanType()
- .defaultValue(false)
- .withDescription(
- "If true, RocksDB will use block-based filter
instead of full filter, this only take effect when bloom filter is used. "
- + "The default value is 'false'.");
-
- public static DBOptions createDBOptions(DBOptions currentOptions, Options
options) {
-
currentOptions.setMaxBackgroundJobs(options.get(MAX_BACKGROUND_THREADS));
- currentOptions.setMaxOpenFiles(options.get(MAX_OPEN_FILES));
- currentOptions.setInfoLogLevel(options.get(LOG_LEVEL));
-
- String logDir = options.get(LOG_DIR);
- if (logDir != null && !logDir.isEmpty()) {
- currentOptions.setDbLogDir(logDir);
- }
-
-
currentOptions.setMaxLogFileSize(options.get(LOG_MAX_FILE_SIZE).getBytes());
- currentOptions.setKeepLogFileNum(options.get(LOG_FILE_NUM));
- return currentOptions;
- }
-
- public static ColumnFamilyOptions createColumnOptions(
- ColumnFamilyOptions currentOptions, Options options) {
- currentOptions.setCompressionType(options.get(COMPRESSION_TYPE));
- currentOptions.setCompactionStyle(options.get(COMPACTION_STYLE));
-
currentOptions.setLevelCompactionDynamicLevelBytes(options.get(USE_DYNAMIC_LEVEL_SIZE));
-
currentOptions.setTargetFileSizeBase(options.get(TARGET_FILE_SIZE_BASE).getBytes());
-
currentOptions.setMaxBytesForLevelBase(options.get(MAX_SIZE_LEVEL_BASE).getBytes());
-
currentOptions.setWriteBufferSize(options.get(WRITE_BUFFER_SIZE).getBytes());
-
currentOptions.setMaxWriteBufferNumber(options.get(MAX_WRITE_BUFFER_NUMBER));
- currentOptions.setMinWriteBufferNumberToMerge(
- options.get(MIN_WRITE_BUFFER_NUMBER_TO_MERGE));
-
- TableFormatConfig tableFormatConfig =
currentOptions.tableFormatConfig();
-
- BlockBasedTableConfig blockBasedTableConfig;
- if (tableFormatConfig == null) {
- blockBasedTableConfig = new BlockBasedTableConfig();
- } else {
- if (tableFormatConfig instanceof PlainTableConfig) {
- // if the table format config is PlainTableConfig, we just
return current
- // column-family options
- return currentOptions;
- } else {
- blockBasedTableConfig = (BlockBasedTableConfig)
tableFormatConfig;
- }
- }
-
- blockBasedTableConfig.setBlockSize(options.get(BLOCK_SIZE).getBytes());
-
blockBasedTableConfig.setMetadataBlockSize(options.get(METADATA_BLOCK_SIZE).getBytes());
-
blockBasedTableConfig.setBlockCacheSize(options.get(BLOCK_CACHE_SIZE).getBytes());
-
- if (options.get(USE_BLOOM_FILTER)) {
- double bitsPerKey = options.get(BLOOM_FILTER_BITS_PER_KEY);
- boolean blockBasedMode =
options.get(BLOOM_FILTER_BLOCK_BASED_MODE);
- BloomFilter bloomFilter = new BloomFilter(bitsPerKey,
blockBasedMode);
- blockBasedTableConfig.setFilterPolicy(bloomFilter);
- }
-
- return currentOptions.setTableFormatConfig(blockBasedTableConfig);
- }
-}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/lookup/rocksdb/RocksDBSetState.java
b/paimon-core/src/main/java/org/apache/paimon/lookup/rocksdb/RocksDBSetState.java
deleted file mode 100644
index 9f654e152c..0000000000
---
a/paimon-core/src/main/java/org/apache/paimon/lookup/rocksdb/RocksDBSetState.java
+++ /dev/null
@@ -1,128 +0,0 @@
-/*
- * 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.paimon.lookup.rocksdb;
-
-import org.apache.paimon.data.serializer.Serializer;
-import org.apache.paimon.lookup.ByteArray;
-import org.apache.paimon.lookup.SetState;
-
-import org.rocksdb.ColumnFamilyHandle;
-import org.rocksdb.RocksDBException;
-import org.rocksdb.RocksIterator;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-
-import static org.apache.paimon.utils.Preconditions.checkArgument;
-
-/** Rocksdb state for key -> Set values. */
-public class RocksDBSetState<K, V> extends RocksDBState<K, V, List<byte[]>>
- implements SetState<K, V> {
-
- private static final byte[] EMPTY = new byte[0];
-
- public RocksDBSetState(
- RocksDBStateFactory stateFactory,
- ColumnFamilyHandle columnFamily,
- Serializer<K> keySerializer,
- Serializer<V> valueSerializer,
- long lruCacheSize) {
- super(stateFactory, columnFamily, keySerializer, valueSerializer,
lruCacheSize);
- }
-
- @Override
- public List<V> get(K key) throws IOException {
- ByteArray keyBytes = wrap(serializeKey(key));
- List<byte[]> valueBytes = cache.getIfPresent(keyBytes);
- if (valueBytes == null) {
- valueBytes = new ArrayList<>();
- try (RocksIterator iterator = db.newIterator(columnFamily)) {
- iterator.seek(keyBytes.bytes);
-
- while (iterator.isValid() &&
startWithKeyPrefix(keyBytes.bytes, iterator.key())) {
- byte[] rawKeyBytes = iterator.key();
- byte[] value =
- Arrays.copyOfRange(
- rawKeyBytes, keyBytes.bytes.length,
rawKeyBytes.length);
- valueBytes.add(value);
- iterator.next();
- }
- }
- cache.put(keyBytes, valueBytes);
- }
-
- List<V> values = new ArrayList<>(valueBytes.size());
- for (byte[] value : valueBytes) {
- valueInputView.setBuffer(value);
- values.add(valueSerializer.deserialize(valueInputView));
- }
- return values;
- }
-
- @Override
- public void retract(K key, V value) throws IOException {
- try {
- byte[] bytes = invalidKeyAndGetKVBytes(key, value);
- if (db.get(columnFamily, bytes) != null) {
- db.delete(columnFamily, writeOptions, bytes);
- }
- } catch (RocksDBException e) {
- throw new IOException(e);
- }
- }
-
- @Override
- public void add(K key, V value) throws IOException {
- try {
- byte[] bytes = invalidKeyAndGetKVBytes(key, value);
- db.put(columnFamily, writeOptions, bytes, EMPTY);
- } catch (RocksDBException e) {
- throw new IOException(e);
- }
- }
-
- private byte[] invalidKeyAndGetKVBytes(K key, V value) throws IOException {
- checkArgument(value != null);
-
- keyOutView.clear();
- keySerializer.serialize(key, keyOutView);
-
- // it is hard to maintain cache, invalidate the key.
- cache.invalidate(wrap(keyOutView.getCopyOfBuffer()));
-
- valueSerializer.serialize(value, keyOutView);
- return keyOutView.getCopyOfBuffer();
- }
-
- private boolean startWithKeyPrefix(byte[] keyPrefixBytes, byte[]
rawKeyBytes) {
- if (rawKeyBytes.length < keyPrefixBytes.length) {
- return false;
- }
-
- for (int i = keyPrefixBytes.length; --i >= 0; ) {
- if (rawKeyBytes[i] != keyPrefixBytes[i]) {
- return false;
- }
- }
-
- return true;
- }
-}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/lookup/rocksdb/RocksDBState.java
b/paimon-core/src/main/java/org/apache/paimon/lookup/rocksdb/RocksDBState.java
deleted file mode 100644
index 6d839cdc90..0000000000
---
a/paimon-core/src/main/java/org/apache/paimon/lookup/rocksdb/RocksDBState.java
+++ /dev/null
@@ -1,140 +0,0 @@
-/*
- * 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.paimon.lookup.rocksdb;
-
-import org.apache.paimon.CoreOptions;
-import org.apache.paimon.data.serializer.Serializer;
-import org.apache.paimon.disk.IOManager;
-import org.apache.paimon.io.DataInputDeserializer;
-import org.apache.paimon.io.DataOutputSerializer;
-import org.apache.paimon.lookup.ByteArray;
-import org.apache.paimon.lookup.State;
-import org.apache.paimon.lookup.StateUtils;
-import org.apache.paimon.sort.BinaryExternalSortBuffer;
-
-import
org.apache.paimon.shade.caffeine2.com.github.benmanes.caffeine.cache.Cache;
-import
org.apache.paimon.shade.caffeine2.com.github.benmanes.caffeine.cache.Caffeine;
-
-import org.rocksdb.ColumnFamilyHandle;
-import org.rocksdb.RocksDB;
-import org.rocksdb.WriteOptions;
-
-import javax.annotation.Nullable;
-
-import java.io.IOException;
-
-/** Rocksdb state for key value. */
-public abstract class RocksDBState<K, V, CacheV> implements State<K, V> {
-
- protected final RocksDBStateFactory stateFactory;
-
- protected final RocksDB db;
-
- protected final WriteOptions writeOptions;
-
- protected final ColumnFamilyHandle columnFamily;
-
- protected final Serializer<K> keySerializer;
-
- protected final Serializer<V> valueSerializer;
-
- protected final DataOutputSerializer keyOutView;
-
- protected final DataInputDeserializer valueInputView;
-
- protected final DataOutputSerializer valueOutputView;
-
- protected final Cache<ByteArray, CacheV> cache;
-
- public RocksDBState(
- RocksDBStateFactory stateFactory,
- ColumnFamilyHandle columnFamily,
- Serializer<K> keySerializer,
- Serializer<V> valueSerializer,
- long lruCacheSize) {
- this.stateFactory = stateFactory;
- this.db = stateFactory.db();
- this.columnFamily = columnFamily;
- this.keySerializer = keySerializer;
- this.valueSerializer = valueSerializer;
- this.keyOutView = new DataOutputSerializer(32);
- this.valueInputView = new DataInputDeserializer();
- this.valueOutputView = new DataOutputSerializer(32);
- this.writeOptions = new WriteOptions().setDisableWAL(true);
- this.cache =
- Caffeine.newBuilder()
- .softValues()
- .maximumSize(lruCacheSize)
- .executor(Runnable::run)
- .build();
- }
-
- @Override
- public byte[] serializeKey(K key) throws IOException {
- keyOutView.clear();
- keySerializer.serialize(key, keyOutView);
- return keyOutView.getCopyOfBuffer();
- }
-
- @Override
- public byte[] serializeValue(V value) throws IOException {
- valueOutputView.clear();
- valueSerializer.serialize(value, valueOutputView);
- return valueOutputView.getCopyOfBuffer();
- }
-
- @Override
- public V deserializeValue(byte[] valueBytes) throws IOException {
- valueInputView.setBuffer(valueBytes);
- return valueSerializer.deserialize(valueInputView);
- }
-
- protected ByteArray wrap(byte[] bytes) {
- return new ByteArray(bytes);
- }
-
- protected Reference ref(byte[] bytes) {
- return new Reference(bytes);
- }
-
- public RocksDBBulkLoader createBulkLoader() {
- return new RocksDBBulkLoader(db, stateFactory.options(), columnFamily,
stateFactory.path());
- }
-
- /** @deprecated Use {@link StateUtils#createBulkLoadSorter(IOManager,
CoreOptions)}. */
- @Deprecated
- public static BinaryExternalSortBuffer createBulkLoadSorter(
- IOManager ioManager, CoreOptions options) {
- return StateUtils.createBulkLoadSorter(ioManager, options);
- }
-
- /** A class wraps byte[] to indicate contain or not contain. */
- protected static class Reference {
-
- @Nullable protected final byte[] bytes;
-
- protected Reference(@Nullable byte[] bytes) {
- this.bytes = bytes;
- }
-
- public boolean isPresent() {
- return bytes != null;
- }
- }
-}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/lookup/rocksdb/RocksDBStateFactory.java
b/paimon-core/src/main/java/org/apache/paimon/lookup/rocksdb/RocksDBStateFactory.java
deleted file mode 100644
index 70a6e10e5f..0000000000
---
a/paimon-core/src/main/java/org/apache/paimon/lookup/rocksdb/RocksDBStateFactory.java
+++ /dev/null
@@ -1,214 +0,0 @@
-/*
- * 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.paimon.lookup.rocksdb;
-
-import org.apache.paimon.annotation.VisibleForTesting;
-import org.apache.paimon.data.serializer.Serializer;
-import org.apache.paimon.lookup.StateFactory;
-
-import org.rocksdb.ColumnFamilyDescriptor;
-import org.rocksdb.ColumnFamilyHandle;
-import org.rocksdb.ColumnFamilyOptions;
-import org.rocksdb.DBOptions;
-import org.rocksdb.NativeLibraryLoader;
-import org.rocksdb.Options;
-import org.rocksdb.RocksDB;
-import org.rocksdb.RocksDBException;
-import org.rocksdb.TtlDB;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import javax.annotation.Nullable;
-
-import java.io.IOException;
-import java.lang.reflect.Field;
-import java.nio.charset.StandardCharsets;
-import java.time.Duration;
-
-/** Factory to create state. */
-public class RocksDBStateFactory implements StateFactory {
-
- private static final Logger LOG =
LoggerFactory.getLogger(RocksDBStateFactory.class);
-
- public static final String MERGE_OPERATOR_NAME = "stringappendtest";
- private static final int ROCKSDB_LIB_LOADING_ATTEMPTS = 3;
-
- private static boolean rocksDbInitialized = false;
-
- private final Options options;
- private final String path;
- private final ColumnFamilyOptions columnFamilyOptions;
-
- private RocksDB db;
-
- public RocksDBStateFactory(
- String path, org.apache.paimon.options.Options conf, @Nullable
Duration ttlSecs)
- throws IOException {
- try {
- ensureRocksDBIsLoaded();
- } catch (Throwable e) {
- throw new IOException("Could not load the native RocksDB library",
e);
- }
- DBOptions dbOptions =
- RocksDBOptions.createDBOptions(
- new DBOptions()
- .setUseFsync(false)
- .setStatsDumpPeriodSec(0)
- .setCreateIfMissing(true),
- conf);
- this.path = path;
- this.columnFamilyOptions =
- RocksDBOptions.createColumnOptions(new ColumnFamilyOptions(),
conf)
- .setMergeOperatorName(MERGE_OPERATOR_NAME);
-
- this.options = new Options(dbOptions, columnFamilyOptions);
- try {
- this.db =
- ttlSecs == null
- ? RocksDB.open(options, path)
- : TtlDB.open(options, path, (int)
ttlSecs.getSeconds(), false);
- } catch (RocksDBException e) {
- throw new IOException("Error while opening RocksDB instance.", e);
- }
- }
-
- public RocksDB db() {
- return db;
- }
-
- public Options options() {
- return options;
- }
-
- public String path() {
- return path;
- }
-
- @Override
- public <K, V> RocksDBValueState<K, V> valueState(
- String name,
- Serializer<K> keySerializer,
- Serializer<V> valueSerializer,
- long lruCacheSize)
- throws IOException {
- return new RocksDBValueState<>(
- this, createColumnFamily(name), keySerializer,
valueSerializer, lruCacheSize);
- }
-
- @Override
- public <K, V> RocksDBSetState<K, V> setState(
- String name,
- Serializer<K> keySerializer,
- Serializer<V> valueSerializer,
- long lruCacheSize)
- throws IOException {
- return new RocksDBSetState<>(
- this, createColumnFamily(name), keySerializer,
valueSerializer, lruCacheSize);
- }
-
- @Override
- public <K, V> RocksDBListState<K, V> listState(
- String name,
- Serializer<K> keySerializer,
- Serializer<V> valueSerializer,
- long lruCacheSize)
- throws IOException {
-
- return new RocksDBListState<>(
- this, createColumnFamily(name), keySerializer,
valueSerializer, lruCacheSize);
- }
-
- @Override
- public boolean preferBulkLoad() {
- return true;
- }
-
- private ColumnFamilyHandle createColumnFamily(String name) throws
IOException {
- try {
- return db.createColumnFamily(
- new ColumnFamilyDescriptor(
- name.getBytes(StandardCharsets.UTF_8),
columnFamilyOptions));
- } catch (RocksDBException e) {
- throw new IOException(e);
- }
- }
-
- @Override
- public void close() throws IOException {
- if (db != null) {
- db.close();
- db = null;
- }
- }
-
- // ------------------------------------------------------------------------
- // static library loading utilities
- // ------------------------------------------------------------------------
-
- @VisibleForTesting
- static void ensureRocksDBIsLoaded() throws IOException {
- synchronized (RocksDBStateFactory.class) {
- if (!rocksDbInitialized) {
- LOG.info("Attempting to load RocksDB native library");
-
- Throwable lastException = null;
- for (int attempt = 1; attempt <= ROCKSDB_LIB_LOADING_ATTEMPTS;
attempt++) {
- try {
- // keep same with RocksDB.loadLibrary
- final String tmpDir =
System.getenv("ROCKSDB_SHAREDLIB_DIR");
- // explicitly load the JNI dependency if it has not
been loaded before
- NativeLibraryLoader.getInstance().loadLibrary(tmpDir);
-
- // this initialization here should validate that the
loading succeeded
- RocksDB.loadLibrary();
-
- // seems to have worked
- LOG.info("Successfully loaded RocksDB native library");
- rocksDbInitialized = true;
- return;
- } catch (Throwable t) {
- lastException = t;
- LOG.debug("RocksDB JNI library loading attempt {}
failed", attempt, t);
- // try to force RocksDB to attempt reloading the
library
- try {
- resetRocksDBLoadedFlag();
- } catch (Throwable tt) {
- LOG.debug(
- "Failed to reset 'initialized' flag in
RocksDB native code loader",
- tt);
- }
- }
- }
- throw new IOException("Could not load the native RocksDB
library", lastException);
- }
- }
- }
-
- @VisibleForTesting
- static void resetRocksDBLoadedFlag() throws Exception {
- final Field initField =
-
org.rocksdb.NativeLibraryLoader.class.getDeclaredField("initialized");
- initField.setAccessible(true);
- initField.setBoolean(null, false);
- }
-
- public static boolean rocksDbInitialized() {
- return rocksDbInitialized;
- }
-}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/lookup/rocksdb/RocksDBValueState.java
b/paimon-core/src/main/java/org/apache/paimon/lookup/rocksdb/RocksDBValueState.java
deleted file mode 100644
index 3d8768a9c4..0000000000
---
a/paimon-core/src/main/java/org/apache/paimon/lookup/rocksdb/RocksDBValueState.java
+++ /dev/null
@@ -1,94 +0,0 @@
-/*
- * 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.paimon.lookup.rocksdb;
-
-import org.apache.paimon.data.serializer.Serializer;
-import org.apache.paimon.lookup.ByteArray;
-import org.apache.paimon.lookup.ValueState;
-
-import org.rocksdb.ColumnFamilyHandle;
-
-import javax.annotation.Nullable;
-
-import java.io.IOException;
-
-import static org.apache.paimon.utils.Preconditions.checkArgument;
-
-/** Rocksdb state for key -> a single value. */
-public class RocksDBValueState<K, V> extends RocksDBState<K, V,
RocksDBState.Reference>
- implements ValueState<K, V> {
-
- public RocksDBValueState(
- RocksDBStateFactory stateFactory,
- ColumnFamilyHandle columnFamily,
- Serializer<K> keySerializer,
- Serializer<V> valueSerializer,
- long lruCacheSize) {
- super(stateFactory, columnFamily, keySerializer, valueSerializer,
lruCacheSize);
- }
-
- @Nullable
- @Override
- public V get(K key) throws IOException {
- try {
- Reference valueRef = get(wrap(serializeKey(key)));
- return valueRef.isPresent() ? deserializeValue(valueRef.bytes) :
null;
- } catch (Exception e) {
- throw new IOException(e);
- }
- }
-
- private Reference get(ByteArray keyBytes) throws Exception {
- Reference valueRef = cache.getIfPresent(keyBytes);
- if (valueRef == null) {
- valueRef = ref(db.get(columnFamily, keyBytes.bytes));
- cache.put(keyBytes, valueRef);
- }
-
- return valueRef;
- }
-
- @Override
- public void put(K key, V value) throws IOException {
- checkArgument(value != null);
-
- try {
- byte[] keyBytes = serializeKey(key);
- byte[] valueBytes = serializeValue(value);
- db.put(columnFamily, writeOptions, keyBytes, valueBytes);
- cache.put(wrap(keyBytes), ref(valueBytes));
- } catch (Exception e) {
- throw new IOException(e);
- }
- }
-
- @Override
- public void delete(K key) throws IOException {
- try {
- byte[] keyBytes = serializeKey(key);
- ByteArray keyByteArray = wrap(keyBytes);
- if (get(keyByteArray).isPresent()) {
- db.delete(columnFamily, writeOptions, keyBytes);
- cache.put(keyByteArray, ref(null));
- }
- } catch (Exception e) {
- throw new IOException(e);
- }
- }
-}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/lookup/rocksdb/RocksDBStateFactoryTest.java
b/paimon-core/src/test/java/org/apache/paimon/lookup/rocksdb/RocksDBStateFactoryTest.java
deleted file mode 100644
index 823f77232b..0000000000
---
a/paimon-core/src/test/java/org/apache/paimon/lookup/rocksdb/RocksDBStateFactoryTest.java
+++ /dev/null
@@ -1,167 +0,0 @@
-/*
- * 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.paimon.lookup.rocksdb;
-
-import org.apache.paimon.options.Options;
-
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.io.TempDir;
-import org.mockito.MockedStatic;
-import org.mockito.Mockito;
-import org.rocksdb.NativeLibraryLoader;
-import org.rocksdb.RocksDB;
-
-import java.io.IOException;
-import java.lang.reflect.Field;
-import java.nio.file.Path;
-
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
-import static org.mockito.ArgumentMatchers.anyString;
-
-/** Test for {@link RocksDBStateFactory}. */
-public class RocksDBStateFactoryTest {
-
- @TempDir Path tempDir;
-
- @Test
- void testRocksDBLibraryLoadSuccess() throws Exception {
- // Test that RocksDBStateFactory can be created successfully when
RocksDB library loads
- // properly
- RocksDBStateFactory factory =
- new RocksDBStateFactory(tempDir.toString(), new Options(),
null);
-
- // Verify that the factory is created and can be used
- assertThat(factory).isNotNull();
- assertThat(factory.db()).isNotNull();
- assertThat(factory.path()).isEqualTo(tempDir.toString());
-
- factory.close();
- }
-
- @Test
- void testEnsureRocksDBIsLoadedSuccess() throws Exception {
- // Reset the initialization flag first
- resetRocksDBInitializedFlag();
-
- // Test successful loading
- RocksDBStateFactory.ensureRocksDBIsLoaded();
-
- // Verify that the flag is set to true after successful loading
- assertThat(RocksDBStateFactory.rocksDbInitialized()).isTrue();
- }
-
- @Test
- void testEnsureRocksDBIsLoadedFailureAfterRetries() throws Exception {
- // Reset the initialization flag first
- resetRocksDBInitializedFlag();
-
- try (MockedStatic<RocksDB> mockedRocksDB =
Mockito.mockStatic(RocksDB.class);
- MockedStatic<NativeLibraryLoader> mockedNativeLoader =
- Mockito.mockStatic(NativeLibraryLoader.class)) {
-
- // Mock NativeLibraryLoader.getInstance()
- NativeLibraryLoader mockLoader =
Mockito.mock(NativeLibraryLoader.class);
-
mockedNativeLoader.when(NativeLibraryLoader::getInstance).thenReturn(mockLoader);
-
- // Make both NativeLibraryLoader.loadLibrary and
RocksDB.loadLibrary fail
- Mockito.doThrow(new RuntimeException("Native library loading
failed"))
- .when(mockLoader)
- .loadLibrary(anyString());
- mockedRocksDB
- .when(RocksDB::loadLibrary)
- .thenThrow(new RuntimeException("RocksDB library loading
failed"));
-
- // Should throw IOException after all retry attempts fail
- assertThatThrownBy(RocksDBStateFactory::ensureRocksDBIsLoaded)
- .isInstanceOf(IOException.class)
- .hasMessage("Could not load the native RocksDB library");
-
- // Verify that the flag remains false after failure
- assertThat(RocksDBStateFactory.rocksDbInitialized()).isFalse();
-
- assertThatThrownBy(
- () -> new RocksDBStateFactory(tempDir.toString(),
new Options(), null))
- .isInstanceOf(IOException.class)
- .hasMessage("Could not load the native RocksDB library");
- }
- }
-
- @Test
- void testEnsureRocksDBIsLoadedAlreadyInitialized() throws Exception {
- // Set the initialization flag to true first
- setRocksDBInitializedFlag(true);
-
- try (MockedStatic<RocksDB> mockedRocksDB =
Mockito.mockStatic(RocksDB.class)) {
- // Should not attempt to load library if already initialized
- RocksDBStateFactory.ensureRocksDBIsLoaded();
-
- // Verify that RocksDB.loadLibrary was never called
- mockedRocksDB.verify(Mockito.never(), () -> RocksDB.loadLibrary());
- }
- }
-
- @Test
- void testEnsureRocksDBIsLoadedWithDifferentExceptionTypes() throws
Exception {
- // Reset the initialization flag first
- resetRocksDBInitializedFlag();
-
- try (MockedStatic<RocksDB> mockedRocksDB =
Mockito.mockStatic(RocksDB.class);
- MockedStatic<NativeLibraryLoader> mockedNativeLoader =
- Mockito.mockStatic(NativeLibraryLoader.class)) {
-
- // Mock NativeLibraryLoader.getInstance()
- NativeLibraryLoader mockLoader =
Mockito.mock(NativeLibraryLoader.class);
-
mockedNativeLoader.when(NativeLibraryLoader::getInstance).thenReturn(mockLoader);
-
- // Make NativeLibraryLoader.loadLibrary fail with
UnsatisfiedLinkError
- Mockito.doThrow(new UnsatisfiedLinkError("Native library not
found"))
- .when(mockLoader)
- .loadLibrary(anyString());
-
- // Make RocksDB.loadLibrary also fail
- mockedRocksDB
- .when(RocksDB::loadLibrary)
- .thenThrow(new UnsatisfiedLinkError("RocksDB library not
found"));
-
- // Should throw IOException with the correct message
- assertThatThrownBy(RocksDBStateFactory::ensureRocksDBIsLoaded)
- .isInstanceOf(IOException.class)
- .hasMessage("Could not load the native RocksDB library")
- .hasCauseInstanceOf(UnsatisfiedLinkError.class);
- }
- }
-
- @Test
- void testResetRocksDBLoadedFlag() throws Exception {
- // Test the resetRocksDBLoadedFlag method
- RocksDBStateFactory.resetRocksDBLoadedFlag();
- }
-
- // Helper methods for accessing private static fields
- private void setRocksDBInitializedFlag(boolean value) throws Exception {
- Field field =
RocksDBStateFactory.class.getDeclaredField("rocksDbInitialized");
- field.setAccessible(true);
- field.setBoolean(null, value);
- }
-
- private void resetRocksDBInitializedFlag() throws Exception {
- setRocksDBInitializedFlag(false);
- }
-}
diff --git
a/paimon-docs/src/main/java/org/apache/paimon/docs/configuration/ConfigOptionsDocGenerator.java
b/paimon-docs/src/main/java/org/apache/paimon/docs/configuration/ConfigOptionsDocGenerator.java
index 3e7db2d89a..ebb167cbac 100644
---
a/paimon-docs/src/main/java/org/apache/paimon/docs/configuration/ConfigOptionsDocGenerator.java
+++
b/paimon-docs/src/main/java/org/apache/paimon/docs/configuration/ConfigOptionsDocGenerator.java
@@ -74,7 +74,6 @@ public class ConfigOptionsDocGenerator {
new OptionsClassLocation[] {
new OptionsClassLocation("paimon-api",
"org.apache.paimon.options"),
new OptionsClassLocation("paimon-api", "org.apache.paimon"),
- new OptionsClassLocation("paimon-core",
"org.apache.paimon.lookup.rocksdb"),
new OptionsClassLocation("paimon-core",
"org.apache.paimon.jdbc"),
new OptionsClassLocation("paimon-core",
"org.apache.paimon.table"),
new OptionsClassLocation("paimon-core",
"org.apache.paimon.iceberg"),
diff --git a/paimon-flink/paimon-flink-1.17/pom.xml
b/paimon-flink/paimon-flink-1.17/pom.xml
index 2debc8e4b9..0a191141e7 100644
--- a/paimon-flink/paimon-flink-1.17/pom.xml
+++ b/paimon-flink/paimon-flink-1.17/pom.xml
@@ -36,7 +36,6 @@ under the License.
<properties>
<flink.version>1.17.2</flink.version>
<iceberg.flink.version>1.17</iceberg.flink.version>
- <frocksdbjni.version>6.20.3-ververica-2.0</frocksdbjni.version>
</properties>
<dependencies>
@@ -128,12 +127,6 @@ under the License.
<scope>test</scope>
</dependency>
- <dependency>
- <groupId>com.ververica</groupId>
- <artifactId>frocksdbjni</artifactId>
- <version>${frocksdbjni.version}</version>
- <scope>test</scope>
- </dependency>
</dependencies>
<build>
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FileStoreLookupFunction.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FileStoreLookupFunction.java
index f7c6fb2d67..2e815af0aa 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FileStoreLookupFunction.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FileStoreLookupFunction.java
@@ -75,13 +75,13 @@ import java.util.stream.Collectors;
import java.util.stream.IntStream;
import static org.apache.paimon.CoreOptions.CONTINUOUS_DISCOVERY_INTERVAL;
+import static org.apache.paimon.CoreOptions.LOOKUP_CACHE_ROWS;
+import static
org.apache.paimon.CoreOptions.LOOKUP_CONTINUOUS_DISCOVERY_INTERVAL;
import static org.apache.paimon.flink.FlinkConnectorOptions.LOOKUP_CACHE_MODE;
import static
org.apache.paimon.flink.FlinkConnectorOptions.LOOKUP_DYNAMIC_PARTITION_REFRESH_ASYNC;
import static
org.apache.paimon.flink.FlinkConnectorOptions.LOOKUP_REFRESH_FULL_LOAD_THRESHOLD;
import static
org.apache.paimon.flink.FlinkConnectorOptions.LOOKUP_REFRESH_TIME_PERIODS_BLACKLIST;
import static
org.apache.paimon.flink.query.RemoteTableQuery.isRemoteServiceAvailable;
-import static
org.apache.paimon.lookup.rocksdb.RocksDBOptions.LOOKUP_CACHE_ROWS;
-import static
org.apache.paimon.lookup.rocksdb.RocksDBOptions.LOOKUP_CONTINUOUS_DISCOVERY_INTERVAL;
import static
org.apache.paimon.predicate.PredicateBuilder.transformFieldMapping;
/** A lookup {@link TableFunction} for file store. */
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/PartitionRefresher.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/PartitionRefresher.java
index cbf6b025bb..a9caa0f6fb 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/PartitionRefresher.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/PartitionRefresher.java
@@ -42,7 +42,7 @@ import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
-import static
org.apache.paimon.lookup.rocksdb.RocksDBOptions.LOOKUP_CACHE_ROWS;
+import static org.apache.paimon.CoreOptions.LOOKUP_CACHE_ROWS;
/** Manages partition refresh logic for {@link FullCacheLookupTable}. */
public class PartitionRefresher implements Closeable {
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/DynamicPartitionLevelLoaderTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/DynamicPartitionLevelLoaderTest.java
index cc54a394de..915efe3d40 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/DynamicPartitionLevelLoaderTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/DynamicPartitionLevelLoaderTest.java
@@ -23,7 +23,6 @@ import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.flink.FlinkConnectorOptions;
-import org.apache.paimon.lookup.rocksdb.RocksDBOptions;
import org.apache.paimon.options.Options;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaManager;
@@ -221,7 +220,7 @@ public class DynamicPartitionLevelLoaderTest {
SchemaManager schemaManager = new SchemaManager(fileIO, tablePath);
Options conf = new Options(customOptions);
conf.set(CoreOptions.BUCKET, 2);
- conf.set(RocksDBOptions.LOOKUP_CONTINUOUS_DISCOVERY_INTERVAL,
Duration.ofSeconds(1));
+ conf.set(CoreOptions.LOOKUP_CONTINUOUS_DISCOVERY_INTERVAL,
Duration.ofSeconds(1));
if (primaryKeys.isEmpty()) {
conf.set(CoreOptions.BUCKET_KEY.key(), "k");
}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/FileStoreLookupFunctionTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/FileStoreLookupFunctionTest.java
index ad315237b9..eb89ca1081 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/FileStoreLookupFunctionTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/FileStoreLookupFunctionTest.java
@@ -27,7 +27,6 @@ import org.apache.paimon.flink.FlinkRowData;
import
org.apache.paimon.flink.lookup.PrimaryKeyPartialLookupTable.LocalQueryExecutor;
import
org.apache.paimon.flink.lookup.PrimaryKeyPartialLookupTable.QueryExecutor;
import
org.apache.paimon.flink.lookup.PrimaryKeyPartialLookupTable.RemoteQueryExecutor;
-import org.apache.paimon.lookup.rocksdb.RocksDBOptions;
import org.apache.paimon.options.Options;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaManager;
@@ -143,7 +142,7 @@ public class FileStoreLookupFunctionTest {
conf.set(CoreOptions.BUCKET, 2);
conf.set(CoreOptions.SNAPSHOT_NUM_RETAINED_MAX, 3);
conf.set(CoreOptions.SNAPSHOT_NUM_RETAINED_MIN, 2);
- conf.set(RocksDBOptions.LOOKUP_CONTINUOUS_DISCOVERY_INTERVAL,
Duration.ofSeconds(1));
+ conf.set(CoreOptions.LOOKUP_CONTINUOUS_DISCOVERY_INTERVAL,
Duration.ofSeconds(1));
if (dynamicPartition) {
conf.set(FlinkConnectorOptions.SCAN_PARTITIONS, "max_pt()");
}
@@ -441,7 +440,7 @@ public class FileStoreLookupFunctionTest {
SchemaManager schemaManager = new SchemaManager(fileIO, tablePath);
Options conf = new Options();
conf.set(CoreOptions.BUCKET, 2);
- conf.set(RocksDBOptions.LOOKUP_CONTINUOUS_DISCOVERY_INTERVAL,
Duration.ofSeconds(1));
+ conf.set(CoreOptions.LOOKUP_CONTINUOUS_DISCOVERY_INTERVAL,
Duration.ofSeconds(1));
RowType rowType =
RowType.of(
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/RocksDBListStateTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/RocksDBListStateTest.java
deleted file mode 100644
index d98d199623..0000000000
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/RocksDBListStateTest.java
+++ /dev/null
@@ -1,102 +0,0 @@
-/*
- * 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.paimon.flink.lookup;
-
-import org.apache.paimon.data.BinaryRow;
-import org.apache.paimon.data.BinaryRowWriter;
-import org.apache.paimon.data.BinaryString;
-import org.apache.paimon.data.GenericRow;
-import org.apache.paimon.data.InternalRow;
-import org.apache.paimon.data.serializer.InternalRowSerializer;
-import org.apache.paimon.lookup.rocksdb.RocksDBListState;
-import org.apache.paimon.lookup.rocksdb.RocksDBStateFactory;
-import org.apache.paimon.options.Options;
-import org.apache.paimon.types.DataTypes;
-import org.apache.paimon.types.RowKind;
-import org.apache.paimon.types.RowType;
-
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.io.TempDir;
-
-import java.nio.file.Path;
-import java.util.ArrayList;
-import java.util.List;
-
-import static org.assertj.core.api.Assertions.assertThat;
-
-/** Test for {@link RocksDBListState}. */
-public class RocksDBListStateTest {
-
- @TempDir Path tempDir;
-
- @Test
- void test() throws Exception {
- RocksDBStateFactory factory =
- new RocksDBStateFactory(tempDir.toString(), new Options(),
null);
-
- RowType keyType = RowType.of(DataTypes.STRING());
- RowType valueType = RowType.of(DataTypes.STRING());
- RocksDBListState listState =
- factory.listState(
- "test",
- new InternalRowSerializer(keyType),
- new InternalRowSerializer(valueType),
- 1);
-
- GenericRow key = row("aaa");
- listState.add(key, row("1"));
- List<InternalRow> result = listState.get(key);
- assertThat(getString(result)).containsExactlyInAnyOrder("1");
- listState.add(key, row("2,3"));
-
assertThat(getString(listState.get(key))).containsExactlyInAnyOrder("1", "2,3");
- listState.add(key, row("1"));
-
assertThat(getString(listState.get(key))).containsExactlyInAnyOrder("1", "2,3",
"1");
- assertThat(listState.get(row("bbb"))).isEmpty();
- factory.close();
- }
-
- public GenericRow row(String value) {
- return GenericRow.of(bs(value));
- }
-
- public GenericRow row(String value, RowKind kind) {
- GenericRow row = GenericRow.of(bs(value));
- row.setRowKind(kind);
- return row;
- }
-
- public BinaryString bs(String v) {
- return BinaryString.fromString(v);
- }
-
- public BinaryRow write(String v) {
- BinaryRow row = new BinaryRow(1);
- BinaryRowWriter write = new BinaryRowWriter(row);
- write.writeString(0, bs(v));
- return row;
- }
-
- public List<String> getString(List<InternalRow> inputs) {
- List<String> rows = new ArrayList<>();
- for (InternalRow input : inputs) {
- rows.add(input.getString(0).toString());
- }
- return rows;
- }
-}
