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 81c935edad [core] Add LocalKvDb-backed lookup states (#8880)
81c935edad is described below
commit 81c935edad7a2639a6d781338436dd644ed63369
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Jul 28 11:28:58 2026 +0800
[core] Add LocalKvDb-backed lookup states (#8880)
---
.../apache/paimon/lookup/sort/db/LocalKvDb.java | 11 +
.../lookup/sort/db/RecordCombiningWriter.java | 8 +-
.../paimon/lookup/sort/db/LocalKvDbTest.java | 48 ++
.../paimon/lookup/local/LocalKvBulkLoader.java | 63 +++
.../paimon/lookup/local/LocalKvCompositeKey.java | 86 ++++
.../paimon/lookup/local/LocalKvListBulkLoader.java | 87 ++++
.../lookup/local/LocalKvListMergeOperator.java | 68 +++
.../paimon/lookup/local/LocalKvListState.java | 120 +++++
.../paimon/lookup/local/LocalKvListValueCodec.java | 187 +++++++
.../paimon/lookup/local/LocalKvSetState.java | 109 ++++
.../apache/paimon/lookup/local/LocalKvState.java | 117 +++++
.../paimon/lookup/local/LocalKvStateFactory.java | 201 ++++++++
.../paimon/lookup/local/LocalKvValueCodec.java | 127 +++++
.../paimon/lookup/local/LocalKvValueState.java | 95 ++++
.../lookup/local/LocalKvStateFactoryTest.java | 562 +++++++++++++++++++++
15 files changed, 1887 insertions(+), 2 deletions(-)
diff --git
a/paimon-common/src/main/java/org/apache/paimon/lookup/sort/db/LocalKvDb.java
b/paimon-common/src/main/java/org/apache/paimon/lookup/sort/db/LocalKvDb.java
index f319ee9463..3b976a67b5 100644
---
a/paimon-common/src/main/java/org/apache/paimon/lookup/sort/db/LocalKvDb.java
+++
b/paimon-common/src/main/java/org/apache/paimon/lookup/sort/db/LocalKvDb.java
@@ -1158,6 +1158,17 @@ public class LocalKvDb implements Closeable {
boolean canMerge(MemorySlice firstKey, MemorySlice nextKey);
+ /**
+ * Return whether a tombstone can be absorbed into the pending merge
group.
+ *
+ * <p>Tombstones are merge boundaries by default. Operators which
produce tombstones for
+ * consumed physical keys can opt in so later compactions can merge
across those synthetic
+ * tombstones.
+ */
+ default boolean canMergeTombstone(MemorySlice firstKey, MemorySlice
tombstoneKey) {
+ return false;
+ }
+
byte[] merge(List<byte[]> values) throws IOException;
}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/lookup/sort/db/RecordCombiningWriter.java
b/paimon-common/src/main/java/org/apache/paimon/lookup/sort/db/RecordCombiningWriter.java
index edf5eedd0d..224ef16aa8 100644
---
a/paimon-common/src/main/java/org/apache/paimon/lookup/sort/db/RecordCombiningWriter.java
+++
b/paimon-common/src/main/java/org/apache/paimon/lookup/sort/db/RecordCombiningWriter.java
@@ -55,8 +55,12 @@ final class RecordCombiningWriter {
}
if (isTombstone(value)) {
- flushPending();
- consumer.accept(key, value);
+ if (pendingKey != null &&
mergeOperator.canMergeTombstone(pendingKey, key)) {
+ pendingKeys.add(MemorySlice.wrap(key.copyBytes()));
+ } else {
+ flushPending();
+ consumer.accept(key, value);
+ }
return;
}
diff --git
a/paimon-common/src/test/java/org/apache/paimon/lookup/sort/db/LocalKvDbTest.java
b/paimon-common/src/test/java/org/apache/paimon/lookup/sort/db/LocalKvDbTest.java
index a022cb2265..d2673e5be2 100644
---
a/paimon-common/src/test/java/org/apache/paimon/lookup/sort/db/LocalKvDbTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/lookup/sort/db/LocalKvDbTest.java
@@ -563,6 +563,54 @@ public class LocalKvDbTest {
}
}
+ @Test
+ public void testCompactionMergesAcrossAbsorbedTombstones() throws
IOException {
+ File directory = new File(tempDir.toFile(), "tombstone-merge-db");
+ LocalKvDb.MergeOperator mergeOperator =
+ new LocalKvDb.MergeOperator() {
+ @Override
+ public boolean canMerge(MemorySlice firstKey, MemorySlice
nextKey) {
+ return firstKey.readByte(0) == nextKey.readByte(0);
+ }
+
+ @Override
+ public boolean canMergeTombstone(
+ MemorySlice firstKey, MemorySlice tombstoneKey) {
+ return canMerge(firstKey, tombstoneKey);
+ }
+
+ @Override
+ public byte[] merge(List<byte[]> values) {
+ StringBuilder merged = new StringBuilder();
+ for (byte[] value : values) {
+ if (merged.length() > 0) {
+ merged.append('+');
+ }
+ merged.append(new String(value, UTF_8));
+ }
+ return merged.toString().getBytes(UTF_8);
+ }
+ };
+ try (LocalKvDb db =
+ LocalKvDb.builder(directory)
+ .level0FileNumCompactTrigger(100)
+ .compressOptions(new CompressOptions("none", 1))
+ .mergeOperator(mergeOperator)
+ .build()) {
+ putString(db, "a-0", "one");
+ putString(db, "a-1", "two");
+ db.flush();
+ putString(db, "a-2", "three");
+ db.flush();
+
+ db.compact();
+
+ Assertions.assertEquals("one+two+three", getString(db, "a-0"));
+ Assertions.assertNull(getString(db, "a-1"));
+ Assertions.assertNull(getString(db, "a-2"));
+ }
+ }
+
@Test
public void
testCompactionMergesAcrossFileGroupsBeforeFilteringExpiration() throws
IOException {
File directory = new File(tempDir.toFile(),
"cross-group-expiration-merge-db");
diff --git
a/paimon-core/src/main/java/org/apache/paimon/lookup/local/LocalKvBulkLoader.java
b/paimon-core/src/main/java/org/apache/paimon/lookup/local/LocalKvBulkLoader.java
new file mode 100644
index 0000000000..0c199f663d
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/lookup/local/LocalKvBulkLoader.java
@@ -0,0 +1,63 @@
+/*
+ * 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.local;
+
+import org.apache.paimon.lookup.ValueBulkLoader;
+import org.apache.paimon.lookup.sort.db.LocalKvDb;
+
+import java.io.IOException;
+import java.util.function.Consumer;
+
+/** State bulk loader backed by a {@link LocalKvDb.BulkLoadWriter}. */
+final class LocalKvBulkLoader implements ValueBulkLoader {
+
+ private final LocalKvDb.BulkLoadWriter writer;
+ private final LocalKvValueCodec valueCodec;
+ private final Consumer<byte[]> cacheInvalidator;
+
+ LocalKvBulkLoader(
+ LocalKvDb db, LocalKvValueCodec valueCodec, Consumer<byte[]>
cacheInvalidator) {
+ try {
+ this.writer = db.createBulkLoadWriter();
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to create LocalKvDb bulk-load
writer.", e);
+ }
+ this.valueCodec = valueCodec;
+ this.cacheInvalidator = cacheInvalidator;
+ }
+
+ @Override
+ public void write(byte[] key, byte[] value) throws WriteException {
+ try {
+ writer.put(key, valueCodec.encode(value));
+ cacheInvalidator.accept(key);
+ } catch (IOException | RuntimeException e) {
+ throw new WriteException(e);
+ }
+ }
+
+ @Override
+ public void finish() {
+ try {
+ writer.finish();
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to finish LocalKvDb bulk
load.", e);
+ }
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/lookup/local/LocalKvCompositeKey.java
b/paimon-core/src/main/java/org/apache/paimon/lookup/local/LocalKvCompositeKey.java
new file mode 100644
index 0000000000..dae7ac2097
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/lookup/local/LocalKvCompositeKey.java
@@ -0,0 +1,86 @@
+/*
+ * 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.local;
+
+import org.apache.paimon.memory.MemorySlice;
+
+import javax.annotation.Nullable;
+
+import java.util.Arrays;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/**
+ * Composite-key helpers for prefix-free serialized logical keys.
+ *
+ * <p>This matches the existing RocksDB state layout and relies on the
serialized logical key being
+ * prefix-free. Paimon's lookup states use length-delimited row keys, and
fixed-width primitive
+ * serializers such as integers are prefix-free as well.
+ */
+final class LocalKvCompositeKey {
+
+ private LocalKvCompositeKey() {}
+
+ static byte[] prefix(byte[] key) {
+ return Arrays.copyOf(key, key.length);
+ }
+
+ static byte[] append(byte[] prefix, byte[] suffix) {
+ byte[] result = Arrays.copyOf(prefix, prefix.length + suffix.length);
+ System.arraycopy(suffix, 0, result, prefix.length, suffix.length);
+ return result;
+ }
+
+ static byte[] appendLong(byte[] prefix, long suffix) {
+ checkArgument(suffix >= 0, "Composite-key sequence must be
non-negative.");
+ byte[] result = Arrays.copyOf(prefix, prefix.length + Long.BYTES);
+ for (int i = result.length - 1; i >= prefix.length; i--) {
+ result[i] = (byte) suffix;
+ suffix >>>= Byte.SIZE;
+ }
+ return result;
+ }
+
+ @Nullable
+ static byte[] upperBound(byte[] prefix) {
+ byte[] result = Arrays.copyOf(prefix, prefix.length);
+ for (int i = result.length - 1; i >= 0; i--) {
+ int value = result[i] & 0xff;
+ if (value != 0xff) {
+ result[i] = (byte) (value + 1);
+ return Arrays.copyOf(result, i + 1);
+ }
+ }
+ return null;
+ }
+
+ static byte[] suffix(byte[] compositeKey, int prefixLength) {
+ checkArgument(
+ prefixLength <= compositeKey.length,
+ "Composite key is shorter than its logical-key prefix.");
+ return Arrays.copyOfRange(compositeKey, prefixLength,
compositeKey.length);
+ }
+
+ static byte[] suffix(MemorySlice compositeKey, int prefixLength) {
+ checkArgument(
+ prefixLength <= compositeKey.length(),
+ "Composite key is shorter than its logical-key prefix.");
+ return compositeKey.slice(prefixLength, compositeKey.length() -
prefixLength).copyBytes();
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/lookup/local/LocalKvListBulkLoader.java
b/paimon-core/src/main/java/org/apache/paimon/lookup/local/LocalKvListBulkLoader.java
new file mode 100644
index 0000000000..39c765b539
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/lookup/local/LocalKvListBulkLoader.java
@@ -0,0 +1,87 @@
+/*
+ * 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.local;
+
+import org.apache.paimon.lookup.ListBulkLoader;
+import org.apache.paimon.lookup.sort.db.LocalKvDb;
+import org.apache.paimon.utils.SortUtil;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+import java.util.function.Consumer;
+import java.util.function.Function;
+
+/** List-state bulk loader which stores one packed initial list per logical
key. */
+final class LocalKvListBulkLoader implements ListBulkLoader {
+
+ private final LocalKvDb.BulkLoadWriter writer;
+ private final LocalKvValueCodec valueCodec;
+ private final LocalKvListValueCodec listValueCodec;
+ private final Function<byte[], byte[]> compositeKeyFactory;
+ private final Consumer<byte[]> cacheInvalidator;
+
+ private byte[] previousKey;
+
+ LocalKvListBulkLoader(
+ LocalKvDb db,
+ LocalKvValueCodec valueCodec,
+ LocalKvListValueCodec listValueCodec,
+ Function<byte[], byte[]> compositeKeyFactory,
+ Consumer<byte[]> cacheInvalidator) {
+ try {
+ this.writer = db.createBulkLoadWriter();
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to create LocalKvDb bulk-load
writer.", e);
+ }
+ this.valueCodec = valueCodec;
+ this.listValueCodec = listValueCodec;
+ this.compositeKeyFactory = compositeKeyFactory;
+ this.cacheInvalidator = cacheInvalidator;
+ }
+
+ @Override
+ public void write(byte[] key, List<byte[]> values) throws WriteException {
+ try {
+ if (previousKey != null && SortUtil.compareBinary(previousKey,
key) >= 0) {
+ throw new IllegalArgumentException(
+ "Bulk-load keys must be sorted in strictly increasing
order.");
+ }
+ previousKey = Arrays.copyOf(key, key.length);
+
+ byte[] prefix = LocalKvCompositeKey.prefix(key);
+ writer.put(
+ compositeKeyFactory.apply(prefix),
+ valueCodec.encode(listValueCodec.encodeList(values)));
+ cacheInvalidator.accept(key);
+ } catch (IOException | RuntimeException e) {
+ writer.close();
+ throw new WriteException(e);
+ }
+ }
+
+ @Override
+ public void finish() {
+ try {
+ writer.finish();
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to finish LocalKvDb bulk
load.", e);
+ }
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/lookup/local/LocalKvListMergeOperator.java
b/paimon-core/src/main/java/org/apache/paimon/lookup/local/LocalKvListMergeOperator.java
new file mode 100644
index 0000000000..7d9f33107f
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/lookup/local/LocalKvListMergeOperator.java
@@ -0,0 +1,68 @@
+/*
+ * 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.local;
+
+import org.apache.paimon.lookup.sort.db.LocalKvDb;
+import org.apache.paimon.memory.MemorySlice;
+
+import java.io.IOException;
+import java.util.List;
+
+/**
+ * Combines ListState fragments with the same logical key when writing SST
files.
+ *
+ * <p>When TTL is enabled, merging includes expired fragments and re-encodes
the result with a new
+ * expiration time, matching RocksDB's TTL merge behavior.
+ */
+final class LocalKvListMergeOperator implements LocalKvDb.MergeOperator {
+
+ private final LocalKvValueCodec valueCodec;
+ private final ThreadLocal<LocalKvListValueCodec> listValueCodec =
+ ThreadLocal.withInitial(LocalKvListValueCodec::new);
+
+ LocalKvListMergeOperator(LocalKvValueCodec valueCodec) {
+ this.valueCodec = valueCodec;
+ }
+
+ @Override
+ public boolean canMerge(MemorySlice firstKey, MemorySlice nextKey) {
+ if (firstKey.length() < Long.BYTES || firstKey.length() !=
nextKey.length()) {
+ return false;
+ }
+
+ int logicalKeyLength = firstKey.length() - Long.BYTES;
+ for (int i = 0; i < logicalKeyLength; i++) {
+ if (firstKey.readByte(i) != nextKey.readByte(i)) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ @Override
+ public boolean canMergeTombstone(MemorySlice firstKey, MemorySlice
tombstoneKey) {
+ return canMerge(firstKey, tombstoneKey);
+ }
+
+ @Override
+ public byte[] merge(List<byte[]> values) throws IOException {
+ // Foreground flush and background compaction can invoke the operator
concurrently.
+ return listValueCodec.get().merge(values, valueCodec);
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/lookup/local/LocalKvListState.java
b/paimon-core/src/main/java/org/apache/paimon/lookup/local/LocalKvListState.java
new file mode 100644
index 0000000000..ac293b5ba0
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/lookup/local/LocalKvListState.java
@@ -0,0 +1,120 @@
+/*
+ * 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.local;
+
+import org.apache.paimon.data.serializer.Serializer;
+import org.apache.paimon.lookup.ByteArray;
+import org.apache.paimon.lookup.ListBulkLoader;
+import org.apache.paimon.lookup.ListState;
+import org.apache.paimon.lookup.sort.db.LocalKvDb;
+import org.apache.paimon.memory.MemorySlice;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Local KV state for an insertion-ordered list per key. */
+public class LocalKvListState<K, V> extends LocalKvState<K, V, List<V>>
implements ListState<K, V> {
+
+ private final LocalKvListValueCodec listValueCodec;
+ private long nextSequence;
+
+ LocalKvListState(
+ LocalKvDb db,
+ Serializer<K> keySerializer,
+ Serializer<V> valueSerializer,
+ long lruCacheSize,
+ LocalKvValueCodec valueCodec) {
+ super(db, keySerializer, valueSerializer, lruCacheSize, valueCodec);
+ this.listValueCodec = new LocalKvListValueCodec();
+ this.nextSequence = 0;
+ }
+
+ @Override
+ public void add(K key, V value) throws IOException {
+ checkArgument(value != null, "Value must not be null.");
+ byte[] keyBytes = serializeKey(key);
+ byte[] valueBytes = serializeValue(value);
+ db.put(
+ nextCompositeKey(LocalKvCompositeKey.prefix(keyBytes)),
+ valueCodec.encode(listValueCodec.encodeSingle(valueBytes)));
+ ByteArray cacheKey = wrap(keyBytes);
+ if (cache.getIfPresent(cacheKey) != null) {
+ cache.invalidate(cacheKey);
+ }
+ }
+
+ @Override
+ public List<V> get(K key) throws IOException {
+ byte[] keyBytes = serializeKey(key);
+ ByteArray cacheKey = wrap(keyBytes);
+ List<V> values = getCached(cacheKey);
+ if (values == null) {
+ byte[] prefix = LocalKvCompositeKey.prefix(keyBytes);
+ List<V> scanned = new ArrayList<>();
+ db.forEachInRange(
+ prefix,
+ LocalKvCompositeKey.upperBound(prefix),
+ (ignored, stored) -> decodeValues(stored, scanned));
+ values =
+ scanned.isEmpty()
+ ? Collections.emptyList()
+ : Collections.unmodifiableList(scanned);
+ putCached(cacheKey, values);
+ }
+ return values;
+ }
+
+ private void decodeValues(MemorySlice storedSlice, List<V> target) throws
IOException {
+ byte[] stored = storedSlice.getHeapMemory();
+ int storedOffset = storedSlice.offset();
+ if (stored == null) {
+ stored = storedSlice.copyBytes();
+ storedOffset = 0;
+ }
+
+ int valueOffset = valueCodec.valueOffset(stored, storedOffset,
storedSlice.length());
+ listValueCodec.decode(
+ stored,
+ valueOffset,
+ storedOffset + storedSlice.length() - valueOffset,
+ valueSerializer,
+ target);
+ }
+
+ @Override
+ public ListBulkLoader createBulkLoader() {
+ return new LocalKvListBulkLoader(
+ db,
+ valueCodec,
+ listValueCodec,
+ this::nextCompositeKey,
+ key -> cache.invalidate(wrap(key)));
+ }
+
+ private byte[] nextCompositeKey(byte[] key) {
+ if (nextSequence < 0) {
+ throw new IllegalStateException("Local KV list sequence has
overflowed.");
+ }
+ return LocalKvCompositeKey.appendLong(key, nextSequence++);
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/lookup/local/LocalKvListValueCodec.java
b/paimon-core/src/main/java/org/apache/paimon/lookup/local/LocalKvListValueCodec.java
new file mode 100644
index 0000000000..3e238bb46b
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/lookup/local/LocalKvListValueCodec.java
@@ -0,0 +1,187 @@
+/*
+ * 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.local;
+
+import org.apache.paimon.data.serializer.Serializer;
+import org.apache.paimon.io.DataInputDeserializer;
+import org.apache.paimon.io.DataOutputSerializer;
+
+import java.io.IOException;
+import java.util.List;
+
+/** Codec for individual list deltas and packed initial lists. */
+final class LocalKvListValueCodec {
+
+ private static final byte SINGLE_VALUE = 1;
+ private static final byte PACKED_VALUES = 2;
+
+ private final DataInputDeserializer input = new DataInputDeserializer();
+ private final DataOutputSerializer output = new DataOutputSerializer(128);
+
+ byte[] encodeSingle(byte[] value) {
+ byte[] result = new byte[value.length + 1];
+ result[0] = SINGLE_VALUE;
+ System.arraycopy(value, 0, result, 1, value.length);
+ return result;
+ }
+
+ byte[] encodeList(List<byte[]> values) throws IOException {
+ output.clear();
+ output.writeByte(PACKED_VALUES);
+ output.writeInt(values.size());
+ for (byte[] value : values) {
+ output.writeInt(value.length);
+ output.write(value);
+ }
+ return output.getCopyOfBuffer();
+ }
+
+ byte[] merge(List<byte[]> storedValues, LocalKvValueCodec valueCodec)
throws IOException {
+ long[] stats = new long[2];
+ for (byte[] stored : storedValues) {
+ inspectStoredValue(stored, valueCodec, stats);
+ }
+ if (stats[0] > Integer.MAX_VALUE || stats[1] > Integer.MAX_VALUE - 5) {
+ throw new IOException("Merged local KV list value is too large.");
+ }
+
+ byte[] packed = new byte[5 + (int) stats[1]];
+ packed[0] = PACKED_VALUES;
+ writeInt(packed, 1, (int) stats[0]);
+ int outputOffset = 5;
+ for (byte[] stored : storedValues) {
+ int valueOffset = valueCodec.valueOffset(stored, 0, stored.length);
+ input.setBuffer(stored, valueOffset, stored.length - valueOffset);
+ int type = input.readUnsignedByte();
+ if (type == SINGLE_VALUE) {
+ int valueLength = input.available();
+ writeInt(packed, outputOffset, valueLength);
+ outputOffset += Integer.BYTES;
+ System.arraycopy(stored, input.getPosition(), packed,
outputOffset, valueLength);
+ outputOffset += valueLength;
+ } else if (type == PACKED_VALUES) {
+ input.readInt();
+ int payloadLength = input.available();
+ System.arraycopy(stored, input.getPosition(), packed,
outputOffset, payloadLength);
+ outputOffset += payloadLength;
+ } else {
+ throw new IOException("Corrupted local KV list value marker.");
+ }
+ }
+ return valueCodec.encode(packed);
+ }
+
+ <V> void decode(byte[] bytes, int offset, int length, Serializer<V>
serializer, List<V> target)
+ throws IOException {
+ if (length <= 0) {
+ throw new IOException("Corrupted empty local KV list value.");
+ }
+ input.setBuffer(bytes, offset, length);
+ int type = input.readUnsignedByte();
+ if (type == SINGLE_VALUE) {
+ target.add(deserializeElement(input.available(), serializer));
+ return;
+ }
+ if (type != PACKED_VALUES) {
+ throw new IOException("Corrupted local KV list value marker.");
+ }
+
+ int size = input.readInt();
+ if (size < 0 || size > input.available() / Integer.BYTES) {
+ throw new IOException(
+ "Corrupted local KV list size: "
+ + size
+ + ", remaining bytes: "
+ + input.available());
+ }
+
+ for (int i = 0; i < size; i++) {
+ int elementLength = input.readInt();
+ if (elementLength < 0 || elementLength > input.available()) {
+ throw new IOException(
+ "Corrupted local KV list element length: "
+ + elementLength
+ + ", remaining bytes: "
+ + input.available());
+ }
+ target.add(deserializeElement(elementLength, serializer));
+ }
+ if (input.available() != 0) {
+ throw new IOException(
+ "Corrupted local KV list with " + input.available() + "
trailing bytes.");
+ }
+ }
+
+ private <V> V deserializeElement(int length, Serializer<V> serializer)
throws IOException {
+ int start = input.getPosition();
+ V value = serializer.deserialize(input);
+ int consumed = input.getPosition() - start;
+ if (consumed != length) {
+ throw new IOException(
+ "Corrupted local KV list element length: expected "
+ + length
+ + " bytes, consumed "
+ + consumed
+ + '.');
+ }
+ return value;
+ }
+
+ private void inspectStoredValue(byte[] stored, LocalKvValueCodec
valueCodec, long[] stats)
+ throws IOException {
+ int valueOffset = valueCodec.valueOffset(stored, 0, stored.length);
+ input.setBuffer(stored, valueOffset, stored.length - valueOffset);
+ int type = input.readUnsignedByte();
+ if (type == SINGLE_VALUE) {
+ stats[0]++;
+ stats[1] += Integer.BYTES + input.available();
+ return;
+ }
+ if (type != PACKED_VALUES) {
+ throw new IOException("Corrupted local KV list value marker.");
+ }
+
+ int size = input.readInt();
+ if (size < 0 || size > input.available() / Integer.BYTES) {
+ throw new IOException("Corrupted local KV list size: " + size +
'.');
+ }
+ int payloadLength = input.available();
+ for (int i = 0; i < size; i++) {
+ int elementLength = input.readInt();
+ if (elementLength < 0 || elementLength > input.available()) {
+ throw new IOException(
+ "Corrupted local KV list element length: " +
elementLength + '.');
+ }
+ input.skipBytesToRead(elementLength);
+ }
+ if (input.available() != 0) {
+ throw new IOException(
+ "Corrupted local KV list with " + input.available() + "
trailing bytes.");
+ }
+ stats[0] += size;
+ stats[1] += payloadLength;
+ }
+
+ private static void writeInt(byte[] bytes, int offset, int value) {
+ bytes[offset] = (byte) (value >>> 24);
+ bytes[offset + 1] = (byte) (value >>> 16);
+ bytes[offset + 2] = (byte) (value >>> 8);
+ bytes[offset + 3] = (byte) value;
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/lookup/local/LocalKvSetState.java
b/paimon-core/src/main/java/org/apache/paimon/lookup/local/LocalKvSetState.java
new file mode 100644
index 0000000000..5a74d939df
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/lookup/local/LocalKvSetState.java
@@ -0,0 +1,109 @@
+/*
+ * 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.local;
+
+import org.apache.paimon.data.serializer.Serializer;
+import org.apache.paimon.lookup.ByteArray;
+import org.apache.paimon.lookup.SetState;
+import org.apache.paimon.lookup.sort.db.LocalKvDb;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Local KV state for bytewise-sorted unique values per key. */
+public class LocalKvSetState<K, V> extends LocalKvState<K, V, List<byte[]>>
+ implements SetState<K, V> {
+
+ private static final byte[] PRESENT = new byte[0];
+
+ LocalKvSetState(
+ LocalKvDb db,
+ Serializer<K> keySerializer,
+ Serializer<V> valueSerializer,
+ long lruCacheSize,
+ LocalKvValueCodec valueCodec) {
+ super(db, keySerializer, valueSerializer, lruCacheSize, valueCodec);
+ }
+
+ @Override
+ public List<V> get(K key) throws IOException {
+ List<byte[]> values = getSerializedValues(serializeKey(key));
+ List<V> result = new ArrayList<>(values.size());
+ for (byte[] value : values) {
+ result.add(deserializeValue(value));
+ }
+ return result;
+ }
+
+ @Override
+ public void retract(K key, V value) throws IOException {
+ checkArgument(value != null, "Value must not be null.");
+ byte[] keyBytes = serializeKey(key);
+ byte[] compositeKey =
+ LocalKvCompositeKey.append(
+ LocalKvCompositeKey.prefix(keyBytes),
serializeValue(value));
+ if (db.get(compositeKey) != null) {
+ db.delete(compositeKey);
+ }
+ cache.invalidate(wrap(keyBytes));
+ }
+
+ @Override
+ public void add(K key, V value) throws IOException {
+ checkArgument(value != null, "Value must not be null.");
+ byte[] keyBytes = serializeKey(key);
+ byte[] compositeKey =
+ LocalKvCompositeKey.append(
+ LocalKvCompositeKey.prefix(keyBytes),
serializeValue(value));
+ db.put(compositeKey, valueCodec.encode(PRESENT));
+ cache.invalidate(wrap(keyBytes));
+ }
+
+ private List<byte[]> getSerializedValues(byte[] keyBytes) throws
IOException {
+ ByteArray key = wrap(keyBytes);
+ List<byte[]> values = getCached(key);
+ if (values == null) {
+ byte[] prefix = LocalKvCompositeKey.prefix(keyBytes);
+ List<byte[]> scanned = new ArrayList<>();
+ db.forEachInRange(
+ prefix,
+ LocalKvCompositeKey.upperBound(prefix),
+ (compositeKey, storedSlice) -> {
+ byte[] stored = storedSlice.getHeapMemory();
+ int storedOffset = storedSlice.offset();
+ if (stored == null) {
+ stored = storedSlice.copyBytes();
+ storedOffset = 0;
+ }
+ valueCodec.valueOffset(stored, storedOffset,
storedSlice.length());
+ scanned.add(LocalKvCompositeKey.suffix(compositeKey,
prefix.length));
+ });
+ values =
+ scanned.isEmpty()
+ ? Collections.emptyList()
+ : Collections.unmodifiableList(scanned);
+ putCached(key, values);
+ }
+ return values;
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/lookup/local/LocalKvState.java
b/paimon-core/src/main/java/org/apache/paimon/lookup/local/LocalKvState.java
new file mode 100644
index 0000000000..66fd993cae
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/lookup/local/LocalKvState.java
@@ -0,0 +1,117 @@
+/*
+ * 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.local;
+
+import org.apache.paimon.data.serializer.Serializer;
+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.sort.db.LocalKvDb;
+
+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 javax.annotation.Nullable;
+
+import java.io.IOException;
+
+/** Base class for states backed by {@link LocalKvDb}. */
+abstract class LocalKvState<K, V, CacheV> implements State<K, V> {
+
+ protected final LocalKvDb db;
+ protected final Serializer<K> keySerializer;
+ protected final Serializer<V> valueSerializer;
+ protected final DataOutputSerializer keyOutput;
+ protected final DataOutputSerializer valueOutput;
+ protected final DataInputDeserializer valueInput;
+ protected final Cache<ByteArray, CacheV> cache;
+ protected final LocalKvValueCodec valueCodec;
+
+ LocalKvState(
+ LocalKvDb db,
+ Serializer<K> keySerializer,
+ Serializer<V> valueSerializer,
+ long lruCacheSize,
+ LocalKvValueCodec valueCodec) {
+ this.db = db;
+ this.keySerializer = keySerializer;
+ this.valueSerializer = valueSerializer;
+ this.keyOutput = new DataOutputSerializer(32);
+ this.valueOutput = new DataOutputSerializer(32);
+ this.valueInput = new DataInputDeserializer();
+ this.valueCodec = valueCodec;
+ this.cache =
+ Caffeine.newBuilder()
+ .softValues()
+ .maximumSize(lruCacheSize)
+ .executor(Runnable::run)
+ .build();
+ }
+
+ @Override
+ public byte[] serializeKey(K key) throws IOException {
+ keyOutput.clear();
+ keySerializer.serialize(key, keyOutput);
+ return keyOutput.getCopyOfBuffer();
+ }
+
+ @Override
+ public byte[] serializeValue(V value) throws IOException {
+ valueOutput.clear();
+ valueSerializer.serialize(value, valueOutput);
+ return valueOutput.getCopyOfBuffer();
+ }
+
+ @Override
+ public V deserializeValue(byte[] valueBytes) throws IOException {
+ valueInput.setBuffer(valueBytes);
+ return valueSerializer.deserialize(valueInput);
+ }
+
+ @Nullable
+ protected byte[] getRaw(byte[] key) throws IOException {
+ byte[] stored = db.get(key);
+ if (stored == null) {
+ return null;
+ }
+ return valueCodec.decode(stored);
+ }
+
+ protected void putRaw(byte[] key, byte[] value) throws IOException {
+ db.put(key, valueCodec.encode(value));
+ }
+
+ protected ByteArray wrap(byte[] bytes) {
+ return new ByteArray(bytes);
+ }
+
+ @Nullable
+ protected CacheV getCached(ByteArray key) {
+ return valueCodec.ttlEnabled() ? null : cache.getIfPresent(key);
+ }
+
+ protected void putCached(ByteArray key, CacheV value) {
+ if (valueCodec.ttlEnabled()) {
+ cache.invalidate(key);
+ } else {
+ cache.put(key, value);
+ }
+ }
+}
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
new file mode 100644
index 0000000000..f6f538ad0d
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/lookup/local/LocalKvStateFactory.java
@@ -0,0 +1,201 @@
+/*
+ * 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.local;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.data.serializer.Serializer;
+import org.apache.paimon.io.cache.CacheManager;
+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;
+
+import javax.annotation.Nullable;
+
+import java.io.File;
+import java.io.IOException;
+import java.time.Duration;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.function.LongSupplier;
+
+import static org.apache.paimon.CoreOptions.LOOKUP_CACHE_BLOOM_FILTER_ENABLED;
+import static org.apache.paimon.CoreOptions.LOOKUP_CACHE_BLOOM_FILTER_FPP;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/**
+ * Factory for lookup states backed by {@link LocalKvDb}.
+ *
+ * <p>Each state uses a separate database so that its serialized keys keep
their natural byte
+ * ordering and can be bulk-loaded independently. All databases share one
block cache and,
+ * optionally, one caller-owned compaction executor.
+ */
+public class LocalKvStateFactory implements StateFactory {
+
+ private final File rootDirectory;
+ private final CoreOptions coreOptions;
+ private final Options options;
+ private final CacheManager cacheManager;
+ private final LocalKvValueCodec valueCodec;
+ @Nullable private final ExecutorService compactionExecutor;
+ private final Map<String, LocalKvDb> databases;
+
+ private boolean closed;
+
+ public LocalKvStateFactory(
+ String path,
+ Options options,
+ @Nullable Duration ttl,
+ @Nullable ExecutorService compactionExecutor,
+ boolean offHeapCache) {
+ this(path, options, ttl, compactionExecutor, offHeapCache,
System::currentTimeMillis);
+ }
+
+ LocalKvStateFactory(
+ String path,
+ Options options,
+ @Nullable Duration ttl,
+ @Nullable ExecutorService compactionExecutor,
+ boolean offHeapCache,
+ LongSupplier currentTimeMillis) {
+ this.rootDirectory = new File(path);
+ if ((!rootDirectory.exists() && !rootDirectory.mkdirs()) ||
!rootDirectory.isDirectory()) {
+ throw new IllegalStateException(
+ "Failed to create LocalKvStateFactory directory: " +
rootDirectory);
+ }
+ 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);
+ this.cacheManager =
+ offHeapCache
+ ? CacheManager.createOffHeap(
+ cacheMemory,
coreOptions.lookupCacheHighPrioPoolRatio())
+ : new CacheManager(cacheMemory,
coreOptions.lookupCacheHighPrioPoolRatio());
+ this.valueCodec = new LocalKvValueCodec(ttl, currentTimeMillis);
+ this.compactionExecutor = compactionExecutor;
+ this.databases = new LinkedHashMap<>();
+ this.closed = false;
+ }
+
+ @Override
+ public <K, V> ValueState<K, V> valueState(
+ String name,
+ Serializer<K> keySerializer,
+ Serializer<V> valueSerializer,
+ long lruCacheSize)
+ throws IOException {
+ return new LocalKvValueState<>(
+ createDatabase(name, null),
+ keySerializer,
+ valueSerializer,
+ lruCacheSize,
+ valueCodec);
+ }
+
+ @Override
+ public <K, V> SetState<K, V> setState(
+ String name,
+ Serializer<K> keySerializer,
+ Serializer<V> valueSerializer,
+ long lruCacheSize)
+ throws IOException {
+ return new LocalKvSetState<>(
+ createDatabase(name, null),
+ keySerializer,
+ valueSerializer,
+ lruCacheSize,
+ valueCodec);
+ }
+
+ @Override
+ public <K, V> ListState<K, V> listState(
+ String name,
+ Serializer<K> keySerializer,
+ Serializer<V> valueSerializer,
+ long lruCacheSize)
+ throws IOException {
+ return new LocalKvListState<>(
+ createDatabase(name, new LocalKvListMergeOperator(valueCodec)),
+ keySerializer,
+ valueSerializer,
+ lruCacheSize,
+ valueCodec);
+ }
+
+ @Override
+ public boolean preferBulkLoad() {
+ return true;
+ }
+
+ private LocalKvDb createDatabase(String name, @Nullable
LocalKvDb.MergeOperator mergeOperator) {
+ checkArgument(!closed, "LocalKvStateFactory is already closed.");
+ checkArgument(!databases.containsKey(name), "State '%s' already
exists.", name);
+
+ File stateDirectory =
+ new File(rootDirectory, String.format("state-%06d",
databases.size()));
+ LocalKvDb db =
+ LocalKvDb.builder(stateDirectory)
+ .cacheManager(cacheManager)
+ .blockSize(coreOptions.localKvDbBlockSize())
+ .compressOptions(coreOptions.lookupCompressOptions())
+
.bloomFilterEnabled(options.get(LOOKUP_CACHE_BLOOM_FILTER_ENABLED))
+
.bloomFilterFpp(options.get(LOOKUP_CACHE_BLOOM_FILTER_FPP))
+ .expiredValuePredicate(
+ valueCodec.ttlEnabled() ?
valueCodec::isExpired : null)
+ .mergeOperator(mergeOperator)
+ .compactionExecutor(compactionExecutor)
+ .build();
+ databases.put(name, db);
+ return db;
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (closed) {
+ return;
+ }
+ closed = true;
+
+ IOException failure = null;
+ for (LocalKvDb db : databases.values()) {
+ try {
+ db.close();
+ } catch (IOException e) {
+ if (failure == null) {
+ failure = e;
+ } else {
+ failure.addSuppressed(e);
+ }
+ }
+ }
+ databases.clear();
+ cacheManager.close();
+ if (failure != null) {
+ throw failure;
+ }
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/lookup/local/LocalKvValueCodec.java
b/paimon-core/src/main/java/org/apache/paimon/lookup/local/LocalKvValueCodec.java
new file mode 100644
index 0000000000..e6ef3ca3e8
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/lookup/local/LocalKvValueCodec.java
@@ -0,0 +1,127 @@
+/*
+ * 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.local;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.function.LongSupplier;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/**
+ * Value envelope which distinguishes LocalKvDb tombstones and optionally
stores TTL.
+ *
+ * <p>TTL is intentionally not enforced while decoding. Like RocksDB TtlDB,
LocalKvDb removes
+ * expired values during compaction, so reads may temporarily return an
expired value.
+ */
+final class LocalKvValueCodec {
+
+ private static final byte VALUE_MARKER = 1;
+ private static final byte TTL_VALUE_MARKER = 2;
+ private static final int EXPIRATION_BYTES = Long.BYTES;
+
+ private final long ttlMillis;
+ private final LongSupplier currentTimeMillis;
+
+ LocalKvValueCodec(@Nullable Duration ttl) {
+ this(ttl, System::currentTimeMillis);
+ }
+
+ LocalKvValueCodec(@Nullable Duration ttl, LongSupplier currentTimeMillis) {
+ this.currentTimeMillis = currentTimeMillis;
+ if (ttl == null) {
+ this.ttlMillis = -1;
+ } else {
+ this.ttlMillis = ttl.toMillis();
+ checkArgument(ttlMillis > 0, "TTL must be greater than zero.");
+ }
+ }
+
+ byte[] encode(byte[] value) {
+ if (!ttlEnabled()) {
+ byte[] stored = new byte[value.length + 1];
+ stored[0] = VALUE_MARKER;
+ System.arraycopy(value, 0, stored, 1, value.length);
+ return stored;
+ }
+
+ byte[] stored = new byte[value.length + 1 + EXPIRATION_BYTES];
+ stored[0] = TTL_VALUE_MARKER;
+ long now = currentTimeMillis.getAsLong();
+ long expiration = Long.MAX_VALUE - now < ttlMillis ? Long.MAX_VALUE :
now + ttlMillis;
+ writeLong(stored, 1, expiration);
+ System.arraycopy(value, 0, stored, 1 + EXPIRATION_BYTES, value.length);
+ return stored;
+ }
+
+ byte[] decode(byte[] stored) throws IOException {
+ int valueOffset = valueOffset(stored, 0, stored.length);
+ return Arrays.copyOfRange(stored, valueOffset, stored.length);
+ }
+
+ /** Return the encoded value offset. TTL expiration is enforced only
during compaction. */
+ int valueOffset(byte[] stored, int offset, int length) throws IOException {
+ if (offset < 0
+ || length <= 0
+ || offset > stored.length
+ || length > stored.length - offset) {
+ throw new IOException("Corrupted LocalKvState value marker.");
+ }
+
+ if (!ttlEnabled()) {
+ if (stored[offset] != VALUE_MARKER) {
+ throw new IOException("Corrupted LocalKvState value marker.");
+ }
+ return offset + 1;
+ }
+
+ if (length < 1 + EXPIRATION_BYTES || stored[offset] !=
TTL_VALUE_MARKER) {
+ throw new IOException("Corrupted LocalKvState TTL value.");
+ }
+ return offset + 1 + EXPIRATION_BYTES;
+ }
+
+ boolean ttlEnabled() {
+ return ttlMillis > 0;
+ }
+
+ boolean isExpired(byte[] stored) {
+ return stored.length >= 1 + EXPIRATION_BYTES
+ && stored[0] == TTL_VALUE_MARKER
+ && currentTimeMillis.getAsLong() >= readLong(stored, 1);
+ }
+
+ private static void writeLong(byte[] bytes, int offset, long value) {
+ for (int i = Long.BYTES - 1; i >= 0; i--) {
+ bytes[offset + i] = (byte) value;
+ value >>>= Byte.SIZE;
+ }
+ }
+
+ private static long readLong(byte[] bytes, int offset) {
+ long value = 0;
+ for (int i = 0; i < Long.BYTES; i++) {
+ value = (value << Byte.SIZE) | (bytes[offset + i] & 0xffL);
+ }
+ return value;
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/lookup/local/LocalKvValueState.java
b/paimon-core/src/main/java/org/apache/paimon/lookup/local/LocalKvValueState.java
new file mode 100644
index 0000000000..68ade810e9
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/lookup/local/LocalKvValueState.java
@@ -0,0 +1,95 @@
+/*
+ * 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.local;
+
+import org.apache.paimon.data.serializer.Serializer;
+import org.apache.paimon.lookup.ByteArray;
+import org.apache.paimon.lookup.ValueBulkLoader;
+import org.apache.paimon.lookup.ValueState;
+import org.apache.paimon.lookup.sort.db.LocalKvDb;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Local KV state for one value per key. */
+public class LocalKvValueState<K, V> extends LocalKvState<K, V,
LocalKvValueState.Reference>
+ implements ValueState<K, V> {
+
+ LocalKvValueState(
+ LocalKvDb db,
+ Serializer<K> keySerializer,
+ Serializer<V> valueSerializer,
+ long lruCacheSize,
+ LocalKvValueCodec valueCodec) {
+ super(db, keySerializer, valueSerializer, lruCacheSize, valueCodec);
+ }
+
+ @Nullable
+ @Override
+ public V get(K key) throws IOException {
+ Reference reference = getReference(wrap(serializeKey(key)));
+ return reference.value == null ? null :
deserializeValue(reference.value);
+ }
+
+ private Reference getReference(ByteArray key) throws IOException {
+ Reference reference = getCached(key);
+ if (reference == null) {
+ reference = new Reference(getRaw(key.bytes));
+ putCached(key, reference);
+ }
+ return reference;
+ }
+
+ @Override
+ public void put(K key, V value) throws IOException {
+ checkArgument(value != null, "Value must not be null.");
+ byte[] keyBytes = serializeKey(key);
+ byte[] valueBytes = serializeValue(value);
+ putRaw(keyBytes, valueBytes);
+ putCached(wrap(keyBytes), new Reference(valueBytes));
+ }
+
+ @Override
+ public void delete(K key) throws IOException {
+ byte[] keyBytes = serializeKey(key);
+ ByteArray wrappedKey = wrap(keyBytes);
+ if (getReference(wrappedKey).value != null) {
+ db.delete(keyBytes);
+ putCached(wrappedKey, new Reference(null));
+ }
+ }
+
+ @Override
+ public ValueBulkLoader createBulkLoader() {
+ return new LocalKvBulkLoader(db, valueCodec, key ->
cache.invalidate(wrap(key)));
+ }
+
+ /** Nullable value wrapper used for negative cache entries. */
+ static final class Reference {
+
+ @Nullable private final byte[] value;
+
+ private Reference(@Nullable byte[] value) {
+ this.value = value;
+ }
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/lookup/local/LocalKvStateFactoryTest.java
b/paimon-core/src/test/java/org/apache/paimon/lookup/local/LocalKvStateFactoryTest.java
new file mode 100644
index 0000000000..e89fe40142
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/lookup/local/LocalKvStateFactoryTest.java
@@ -0,0 +1,562 @@
+/*
+ * 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.local;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.data.serializer.IntSerializer;
+import org.apache.paimon.data.serializer.Serializer;
+import org.apache.paimon.io.DataInputView;
+import org.apache.paimon.io.DataOutputView;
+import org.apache.paimon.lookup.BulkLoader;
+import org.apache.paimon.lookup.ListBulkLoader;
+import org.apache.paimon.lookup.ListState;
+import org.apache.paimon.lookup.SetState;
+import org.apache.paimon.lookup.ValueBulkLoader;
+import org.apache.paimon.lookup.ValueState;
+import org.apache.paimon.options.MemorySize;
+import org.apache.paimon.options.Options;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link LocalKvStateFactory}. */
+class LocalKvStateFactoryTest {
+
+ @TempDir Path tempDir;
+
+ @Test
+ void testFactoryRejectsNonDirectoryPath() throws Exception {
+ Path file = Files.write(tempDir.resolve("not-a-directory"), new byte[]
{1});
+
+ assertThatThrownBy(
+ () ->
+ new LocalKvStateFactory(
+ file.toString(), options(), null,
null, false))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("LocalKvStateFactory directory");
+ }
+
+ @Test
+ void testValueStateAndStateIsolation() throws Exception {
+ try (LocalKvStateFactory factory = createFactory()) {
+ ValueState<Integer, Integer> first =
+ factory.valueState("first", IntSerializer.INSTANCE,
IntSerializer.INSTANCE, 10);
+ ValueState<Integer, Integer> second =
+ factory.valueState(
+ "second", IntSerializer.INSTANCE,
IntSerializer.INSTANCE, 10);
+
+ assertThat(first.get(1)).isNull();
+ first.put(1, 10);
+ assertThat(first.get(1)).isEqualTo(10);
+ assertThat(second.get(1)).isNull();
+
+ first.put(1, 11);
+ assertThat(first.get(1)).isEqualTo(11);
+ first.delete(1);
+ assertThat(first.get(1)).isNull();
+
+ assertThat(factory.preferBulkLoad()).isTrue();
+ assertThatThrownBy(
+ () ->
+ factory.valueState(
+ "first",
+ IntSerializer.INSTANCE,
+ IntSerializer.INSTANCE,
+ 10))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("first");
+ }
+ }
+
+ @Test
+ void testValueAndSetStateSupportEmptySerializedValues() throws Exception {
+ EmptyIntSerializer emptySerializer = new EmptyIntSerializer();
+ try (LocalKvStateFactory factory = createFactory()) {
+ @SuppressWarnings("unchecked")
+ LocalKvValueState<Integer, Integer> valueState =
+ (LocalKvValueState<Integer, Integer>)
+ factory.valueState(
+ "empty-value", IntSerializer.INSTANCE,
emptySerializer, 10);
+ valueState.put(1, 10);
+ valueState.db.flush();
+ valueState.cache.invalidateAll();
+ assertThat(valueState.get(1)).isZero();
+
+ ValueState<Integer, Integer> bulkValueState =
+ factory.valueState(
+ "bulk-empty-value", IntSerializer.INSTANCE,
emptySerializer, 10);
+ ValueBulkLoader valueLoader = bulkValueState.createBulkLoader();
+ valueLoader.write(bulkValueState.serializeKey(2),
bulkValueState.serializeValue(20));
+ valueLoader.finish();
+ assertThat(bulkValueState.get(2)).isZero();
+
+ @SuppressWarnings("unchecked")
+ LocalKvSetState<Integer, Integer> setState =
+ (LocalKvSetState<Integer, Integer>)
+ factory.setState(
+ "empty-set", IntSerializer.INSTANCE,
emptySerializer, 10);
+ setState.add(1, 30);
+ setState.db.flush();
+ assertThat(setState.get(1)).containsExactly(0);
+ setState.retract(1, 40);
+ assertThat(setState.get(1)).isEmpty();
+ }
+ }
+
+ @Test
+ void testListState() throws Exception {
+ try (LocalKvStateFactory factory = createFactory()) {
+ ListState<Integer, Integer> state =
+ factory.listState("list", IntSerializer.INSTANCE,
IntSerializer.INSTANCE, 10);
+
+ assertThat(state.get(1)).isEmpty();
+ state.add(1, 3);
+ state.add(1, 1);
+ state.add(1, 3);
+ assertThat(state.get(1)).containsExactly(3, 1, 3);
+
+ ListState<Integer, Integer> bulkState =
+ factory.listState(
+ "bulk-list", IntSerializer.INSTANCE,
IntSerializer.INSTANCE, 10);
+ ListBulkLoader loader = bulkState.createBulkLoader();
+ loader.write(
+ bulkState.serializeKey(1),
+ Arrays.asList(
+ bulkState.serializeValue(4),
+ bulkState.serializeValue(2),
+ bulkState.serializeValue(4)));
+ loader.write(
+ bulkState.serializeKey(2),
+ Collections.singletonList(bulkState.serializeValue(5)));
+ loader.finish();
+
+ assertThat(bulkState.get(1)).containsExactly(4, 2, 4);
+ assertThat(bulkState.get(2)).containsExactly(5);
+ bulkState.add(1, 6);
+ assertThat(bulkState.get(1)).containsExactly(4, 2, 4, 6);
+
+ ListState<Integer, Integer> duplicateState =
+ factory.listState(
+ "bulk-list-duplicate",
+ IntSerializer.INSTANCE,
+ IntSerializer.INSTANCE,
+ 10);
+ ListBulkLoader duplicateLoader = duplicateState.createBulkLoader();
+ duplicateLoader.write(
+ duplicateState.serializeKey(1),
+
Collections.singletonList(duplicateState.serializeValue(1)));
+ assertThatThrownBy(
+ () ->
+ duplicateLoader.write(
+ duplicateState.serializeKey(1),
+ Collections.singletonList(
+
duplicateState.serializeValue(2))))
+ .isInstanceOf(BulkLoader.WriteException.class)
+ .hasMessageContaining("strictly increasing");
+ }
+ }
+
+ @Test
+ void testListStateCachesDeserializedValuesAndInvalidatesOnAdd() throws
Exception {
+ CountingIntSerializer serializer = new CountingIntSerializer();
+ try (LocalKvStateFactory factory = createFactory()) {
+ ListState<Integer, Integer> state =
+ factory.listState("cached-list", IntSerializer.INSTANCE,
serializer, 10);
+ ListBulkLoader loader = state.createBulkLoader();
+ loader.write(
+ state.serializeKey(1),
+ Arrays.asList(state.serializeValue(10),
state.serializeValue(20)));
+ loader.finish();
+
+ assertThat(state.get(1)).containsExactly(10, 20);
+ assertThat(serializer.deserializationCount).isEqualTo(2);
+ assertThat(state.get(1)).containsExactly(10, 20);
+ assertThat(serializer.deserializationCount).isEqualTo(2);
+
+ state.add(1, 30);
+ assertThat(state.get(1)).containsExactly(10, 20, 30);
+ assertThat(serializer.deserializationCount).isEqualTo(5);
+ assertThat(state.get(1)).containsExactly(10, 20, 30);
+ assertThat(serializer.deserializationCount).isEqualTo(5);
+ }
+ }
+
+ @Test
+ void testListStateMergesFragmentsDuringFlushAndCompaction() throws
Exception {
+ try (LocalKvStateFactory factory = createFactory()) {
+ @SuppressWarnings("unchecked")
+ LocalKvListState<Integer, Integer> state =
+ (LocalKvListState<Integer, Integer>)
+ factory.listState(
+ "merged-list",
+ IntSerializer.INSTANCE,
+ IntSerializer.INSTANCE,
+ 10);
+ ListBulkLoader loader = state.createBulkLoader();
+ loader.write(
+ state.serializeKey(1),
+ Arrays.asList(state.serializeValue(10),
state.serializeValue(20)));
+ loader.write(
+ state.serializeKey(2),
Collections.singletonList(state.serializeValue(40)));
+ loader.finish();
+
+ state.add(1, 30);
+ state.add(1, 31);
+ state.add(2, 41);
+ state.db.flush();
+ state.cache.invalidateAll();
+
+ assertThat(state.get(1)).containsExactly(10, 20, 30, 31);
+ assertThat(state.get(2)).containsExactly(40, 41);
+ assertThat(rawEntries(state, 1)).hasSize(2);
+ assertThat(rawEntries(state, 2)).hasSize(2);
+
+ state.db.compact();
+ state.cache.invalidateAll();
+ assertThat(state.get(1)).containsExactly(10, 20, 30, 31);
+ assertThat(state.get(2)).containsExactly(40, 41);
+ assertThat(rawEntries(state, 1)).hasSize(1);
+ assertThat(rawEntries(state, 2)).hasSize(1);
+ }
+ }
+
+ @Test
+ void testListStateWithTtlMergesFragmentsDuringFlush() throws Exception {
+ try (LocalKvStateFactory factory =
+ new LocalKvStateFactory(
+ tempDir.resolve("list-ttl-merge").toString(),
+ options(),
+ Duration.ofHours(1),
+ null,
+ false)) {
+ @SuppressWarnings("unchecked")
+ LocalKvListState<Integer, Integer> state =
+ (LocalKvListState<Integer, Integer>)
+ factory.listState(
+ "list", IntSerializer.INSTANCE,
IntSerializer.INSTANCE, 10);
+
+ state.add(1, 10);
+ state.add(1, 20);
+ state.db.flush();
+ state.cache.invalidateAll();
+
+ assertThat(rawEntries(state, 1)).hasSize(1);
+ assertThat(state.get(1)).containsExactly(10, 20);
+ }
+ }
+
+ @Test
+ void testListStateTtlIsRefreshedAcrossRepeatedFlushes() throws Exception {
+ AtomicLong clock = new AtomicLong(1_000);
+ try (LocalKvStateFactory factory =
+ new LocalKvStateFactory(
+ tempDir.resolve("list-ttl-repeated-flush").toString(),
+ options(),
+ Duration.ofMillis(100),
+ null,
+ false,
+ clock::get)) {
+ @SuppressWarnings("unchecked")
+ LocalKvListState<Integer, Integer> state =
+ (LocalKvListState<Integer, Integer>)
+ factory.listState(
+ "list", IntSerializer.INSTANCE,
IntSerializer.INSTANCE, 10);
+
+ state.add(1, 10);
+ state.add(1, 20);
+ state.db.flush();
+ clock.addAndGet(50);
+ state.add(1, 30);
+ state.db.flush();
+ clock.addAndGet(50);
+
+ state.db.compact();
+ state.cache.invalidateAll();
+ assertThat(state.get(1)).containsExactly(10, 20, 30);
+ }
+ }
+
+ @Test
+ void testListMergeRefreshesTtlIncludingExpiredFragments() throws Exception
{
+ AtomicLong clock = new AtomicLong(1_000);
+ LocalKvValueCodec valueCodec = new
LocalKvValueCodec(Duration.ofMillis(100), clock::get);
+ LocalKvListValueCodec listValueCodec = new LocalKvListValueCodec();
+ LocalKvListMergeOperator mergeOperator = new
LocalKvListMergeOperator(valueCodec);
+
+ byte[] first = valueCodec.encode(listValueCodec.encodeSingle(new
byte[] {0, 0, 0, 10}));
+ clock.set(1_050);
+ byte[] second = valueCodec.encode(listValueCodec.encodeSingle(new
byte[] {0, 0, 0, 20}));
+ clock.set(1_200);
+ assertThat(valueCodec.isExpired(first)).isTrue();
+ assertThat(valueCodec.isExpired(second)).isTrue();
+ assertThat(valueCodec.valueOffset(first, 0,
first.length)).isPositive();
+ assertThat(valueCodec.valueOffset(second, 0,
second.length)).isPositive();
+
+ byte[] merged = mergeOperator.merge(Arrays.asList(first, second));
+ int valueOffset = valueCodec.valueOffset(merged, 0, merged.length);
+ assertThat(valueOffset).isPositive();
+ assertThat(valueCodec.isExpired(merged)).isFalse();
+ List<Integer> values = new ArrayList<>();
+ listValueCodec.decode(
+ merged, valueOffset, merged.length - valueOffset,
IntSerializer.INSTANCE, values);
+ assertThat(values).containsExactly(10, 20);
+
+ clock.set(1_299);
+ assertThat(valueCodec.isExpired(merged)).isFalse();
+ clock.set(1_300);
+ assertThat(valueCodec.isExpired(merged)).isTrue();
+ assertThat(valueCodec.valueOffset(merged, 0,
merged.length)).isPositive();
+ }
+
+ @Test
+ void testSetState() throws Exception {
+ try (LocalKvStateFactory factory = createFactory()) {
+ SetState<Integer, Integer> state =
+ factory.setState("set", IntSerializer.INSTANCE,
IntSerializer.INSTANCE, 10);
+
+ assertThat(state.get(1)).isEmpty();
+ state.retract(1, 9);
+ state.add(1, 3);
+ state.add(1, 1);
+ state.add(1, 3);
+ state.add(1, 2);
+ assertThat(state.get(1)).containsExactly(1, 2, 3);
+
+ state.retract(1, 2);
+ assertThat(state.get(1)).containsExactly(1, 3);
+ for (int value = 4; value < 260; value++) {
+ state.add(1, value);
+ }
+ assertThat(state.get(1)).hasSize(258);
+ state.retract(1, 1);
+ state.retract(1, 3);
+ for (int value = 4; value < 260; value++) {
+ state.retract(1, value);
+ }
+ assertThat(state.get(1)).isEmpty();
+ }
+ }
+
+ @Test
+ void testValueBulkLoadAndStrictOrdering() throws Exception {
+ try (LocalKvStateFactory factory = createFactory()) {
+ ValueState<Integer, Integer> state =
+ factory.valueState("bulk", IntSerializer.INSTANCE,
IntSerializer.INSTANCE, 10);
+ assertThat(state.get(1)).isNull();
+ ValueBulkLoader loader = state.createBulkLoader();
+ loader.write(state.serializeKey(1), state.serializeValue(10));
+ loader.write(state.serializeKey(2), state.serializeValue(20));
+ loader.finish();
+
+ assertThat(state.get(1)).isEqualTo(10);
+ assertThat(state.get(2)).isEqualTo(20);
+
+ ValueState<Integer, Integer> duplicateState =
+ factory.valueState(
+ "bulk-duplicate", IntSerializer.INSTANCE,
IntSerializer.INSTANCE, 10);
+ ValueBulkLoader duplicateLoader =
duplicateState.createBulkLoader();
+ duplicateLoader.write(
+ duplicateState.serializeKey(1),
duplicateState.serializeValue(10));
+ assertThatThrownBy(
+ () ->
+ duplicateLoader.write(
+ duplicateState.serializeKey(1),
+ duplicateState.serializeValue(11)))
+ .isInstanceOf(BulkLoader.WriteException.class)
+ .hasMessageContaining("strictly increasing");
+ }
+ }
+
+ @Test
+ void testValueStateTtlIsRemovedOnlyByCompaction() throws Exception {
+ Options options = options();
+ AtomicLong clock = new AtomicLong(1_000);
+ try (LocalKvStateFactory factory =
+ new LocalKvStateFactory(
+ tempDir.resolve("ttl").toString(),
+ options,
+ Duration.ofMillis(100),
+ null,
+ false,
+ clock::get)) {
+ @SuppressWarnings("unchecked")
+ LocalKvValueState<Integer, Integer> state =
+ (LocalKvValueState<Integer, Integer>)
+ factory.valueState(
+ "ttl", IntSerializer.INSTANCE,
IntSerializer.INSTANCE, 10);
+ state.put(1, 10);
+ state.db.flush();
+ assertThat(state.get(1)).isEqualTo(10);
+
+ @SuppressWarnings("unchecked")
+ LocalKvValueState<Integer, Integer> bulkState =
+ (LocalKvValueState<Integer, Integer>)
+ factory.valueState(
+ "bulk-ttl", IntSerializer.INSTANCE,
IntSerializer.INSTANCE, 10);
+ ValueBulkLoader loader = bulkState.createBulkLoader();
+ loader.write(bulkState.serializeKey(2),
bulkState.serializeValue(20));
+ loader.finish();
+ assertThat(bulkState.get(2)).isEqualTo(20);
+
+ clock.addAndGet(100);
+ assertThat(state.get(1)).isEqualTo(10);
+ assertThat(bulkState.get(2)).isEqualTo(20);
+
+ state.db.compact();
+ bulkState.db.compact();
+ assertThat(state.get(1)).isNull();
+ assertThat(bulkState.get(2)).isNull();
+ }
+ }
+
+ @Test
+ void testUnmergedListAndSetStateTtlRemainVisibleBeforeCompaction() throws
Exception {
+ AtomicLong clock = new AtomicLong(1_000);
+ try (LocalKvStateFactory factory =
+ new LocalKvStateFactory(
+ tempDir.resolve("collection-ttl").toString(),
+ options(),
+ Duration.ofMillis(500),
+ null,
+ false,
+ clock::get)) {
+ ListState<Integer, Integer> list =
+ factory.listState("list", IntSerializer.INSTANCE,
IntSerializer.INSTANCE, 10);
+ SetState<Integer, Integer> set =
+ factory.setState("set", IntSerializer.INSTANCE,
IntSerializer.INSTANCE, 10);
+
+ list.add(1, 10);
+ set.add(1, 10);
+ clock.addAndGet(350);
+ list.add(1, 20);
+ set.add(1, 20);
+ clock.addAndGet(250);
+
+ assertThat(list.get(1)).containsExactly(10, 20);
+ assertThat(set.get(1)).containsExactly(10, 20);
+ }
+ }
+
+ @Test
+ void testSetStateTtlCompactionDropsOnlyExpiredValues() throws Exception {
+ AtomicLong clock = new AtomicLong(1_000);
+ try (LocalKvStateFactory factory =
+ new LocalKvStateFactory(
+ tempDir.resolve("set-ttl-compaction").toString(),
+ options(),
+ Duration.ofMillis(100),
+ null,
+ false,
+ clock::get)) {
+ @SuppressWarnings("unchecked")
+ LocalKvSetState<Integer, Integer> state =
+ (LocalKvSetState<Integer, Integer>)
+ factory.setState(
+ "set", IntSerializer.INSTANCE,
IntSerializer.INSTANCE, 10);
+
+ state.add(1, 10);
+ state.db.flush();
+ clock.addAndGet(50);
+ state.add(1, 20);
+ state.db.flush();
+ clock.addAndGet(50);
+
+ assertThat(state.get(1)).containsExactly(10, 20);
+ state.db.compact();
+ assertThat(state.get(1)).containsExactly(20);
+ }
+ }
+
+ private LocalKvStateFactory createFactory() {
+ return new LocalKvStateFactory(tempDir.toString(), options(), null,
null, false);
+ }
+
+ private Options options() {
+ Options options = new Options();
+ options.set(CoreOptions.LOOKUP_CACHE_MAX_MEMORY_SIZE,
MemorySize.ofMebiBytes(8));
+ return options;
+ }
+
+ private static List<Map.Entry<byte[], byte[]>> rawEntries(
+ LocalKvListState<Integer, Integer> state, int key) throws
IOException {
+ byte[] prefix = LocalKvCompositeKey.prefix(state.serializeKey(key));
+ return state.db.rangeScan(prefix,
LocalKvCompositeKey.upperBound(prefix));
+ }
+
+ private static class CountingIntSerializer implements Serializer<Integer> {
+
+ private int deserializationCount;
+
+ @Override
+ public Serializer<Integer> duplicate() {
+ return this;
+ }
+
+ @Override
+ public Integer copy(Integer from) {
+ return from;
+ }
+
+ @Override
+ public void serialize(Integer record, DataOutputView target) throws
IOException {
+ target.writeInt(record);
+ }
+
+ @Override
+ public Integer deserialize(DataInputView source) throws IOException {
+ deserializationCount++;
+ return source.readInt();
+ }
+ }
+
+ private static class EmptyIntSerializer implements Serializer<Integer> {
+
+ @Override
+ public Serializer<Integer> duplicate() {
+ return this;
+ }
+
+ @Override
+ public Integer copy(Integer from) {
+ return from;
+ }
+
+ @Override
+ public void serialize(Integer record, DataOutputView target) {}
+
+ @Override
+ public Integer deserialize(DataInputView source) {
+ return 0;
+ }
+ }
+}
