This is an automated email from the ASF dual-hosted git repository.
wombatu-kun pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new bc8434219a50 perf(flink): reduce RocksDB lookup cache overhead (#19980)
bc8434219a50 is described below
commit bc8434219a5064d56d36b2ff43daab839163a9ce
Author: Danny Chan <[email protected]>
AuthorDate: Fri Sep 18 10:50:27 2026 +0800
perf(flink): reduce RocksDB lookup cache overhead (#19980)
---
.../hudi/common/util/collection/RocksDBDAO.java | 52 +++++++++++----
.../common/util/collection/TestRocksDBDAO.java | 52 +++++++++++++++
.../hudi/table/lookup/HoodieLookupFunction.java | 1 +
.../org/apache/hudi/table/lookup/LookupCache.java | 7 ++
.../hudi/table/lookup/RocksDBLookupCache.java | 74 ++++++++++++++--------
.../table/lookup/TestHoodieLookupFunction.java | 9 ++-
.../hudi/table/lookup/TestRocksDBLookupCache.java | 54 ++++++++++++++++
7 files changed, 209 insertions(+), 40 deletions(-)
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/util/collection/RocksDBDAO.java
b/hudi-common/src/main/java/org/apache/hudi/common/util/collection/RocksDBDAO.java
index 904f71abb64a..2ec55280b514 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/util/collection/RocksDBDAO.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/util/collection/RocksDBDAO.java
@@ -52,7 +52,6 @@ import java.io.Serializable;
import java.net.URI;
import java.util.ArrayList;
import java.util.Iterator;
-import java.util.LinkedList;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
@@ -380,24 +379,44 @@ public class RocksDBDAO {
* @param <T> Type of value stored
*/
public <T extends Serializable> Stream<Pair<String, T>> prefixSearch(String
columnFamilyName, String prefix) {
+ List<Pair<String, T>> results = new ArrayList<>();
+ this.<T, RuntimeException>prefixSearch(columnFamilyName, prefix, (key,
value) -> results.add(Pair.of(key, value)));
+ return results.stream();
+ }
+
+ /**
+ * Visits matching entries synchronously without collecting them in memory.
The iterator is
+ * closed before returning, including when the handler throws an exception.
+ *
+ * @param columnFamilyName Column family name
+ * @param prefix Prefix key
+ * @param handler Handler invoked once per matching entry, in key order
+ * @param <T> Type of value stored
+ * @param <E> Type of exception thrown by the handler
+ */
+ public <T extends Serializable, E extends Exception> void prefixSearch(
+ String columnFamilyName, String prefix, PrefixSearchHandler<T, E>
handler) throws E {
ValidationUtils.checkArgument(!closed);
- final HoodieTimer timer = HoodieTimer.start();
- long timeTakenMicro = 0;
- List<Pair<String, T>> results = new LinkedList<>();
+ final boolean debug = log.isDebugEnabled();
+ final HoodieTimer timer = debug ? HoodieTimer.start() : null;
+ long count = 0;
try (final RocksIterator it =
getRocksDB().newIterator(managedHandlesMap.get(columnFamilyName))) {
it.seek(getUTF8Bytes(prefix));
- while (it.isValid() && fromUTF8Bytes(it.key()).startsWith(prefix)) {
- long beginTs = System.nanoTime();
- T val = deserializePayload(columnFamilyName, it.value());
- timeTakenMicro += ((System.nanoTime() - beginTs) / 1000);
- results.add(Pair.of(fromUTF8Bytes(it.key()), val));
+ while (it.isValid()) {
+ String key = fromUTF8Bytes(it.key());
+ if (!key.startsWith(prefix)) {
+ break;
+ }
+ handler.accept(key, deserializePayload(columnFamilyName, it.value()));
+ count++;
it.next();
}
}
- log.info("Prefix Search for (query={}) on {}. Total Time Taken (msec)={}.
Serialization Time taken(micro)={}, num entries={}",
- prefix, columnFamilyName, timer.endTimer(), timeTakenMicro,
results.size());
- return results.stream();
+ if (debug) {
+ log.debug("Prefix Search for (query={}) on {}. Total Time Taken
(msec)={}, num entries={}",
+ prefix, columnFamilyName, timer.endTimer(), count);
+ }
}
/**
@@ -614,4 +633,13 @@ public class RocksDBDAO {
void apply(WriteBatch batch);
}
+
+ /**
+ * Handler for a prefix search that may propagate a checked exception to the
caller.
+ */
+ @FunctionalInterface
+ public interface PrefixSearchHandler<T, E extends Exception> {
+
+ void accept(String key, T value) throws E;
+ }
}
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/util/collection/TestRocksDBDAO.java
b/hudi-common/src/test/java/org/apache/hudi/common/util/collection/TestRocksDBDAO.java
index 6c4a3f7cf56c..28a1a442aadb 100644
---
a/hudi-common/src/test/java/org/apache/hudi/common/util/collection/TestRocksDBDAO.java
+++
b/hudi-common/src/test/java/org/apache/hudi/common/util/collection/TestRocksDBDAO.java
@@ -18,6 +18,7 @@
package org.apache.hudi.common.util.collection;
+import org.apache.hudi.common.serialization.CustomSerializer;
import org.apache.hudi.common.table.view.FileSystemViewStorageConfig;
import lombok.Value;
@@ -28,6 +29,7 @@ import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
import java.io.File;
+import java.io.IOException;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Arrays;
@@ -42,6 +44,7 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
@@ -49,6 +52,8 @@ import java.util.stream.IntStream;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
/**
@@ -164,6 +169,53 @@ public class TestRocksDBDAO {
assertFalse(new File(rocksDBBasePath).exists());
}
+ @Test
+ public void testPrefixSearchHandler() throws IOException {
+ String family = "prefix_handler";
+ AtomicInteger deserialized = new AtomicInteger();
+ ConcurrentHashMap<String, CustomSerializer<?>> serializers = new
ConcurrentHashMap<>();
+ serializers.put(family, new CustomSerializer<byte[]>() {
+ @Override
+ public byte[] serialize(byte[] value) {
+ return value;
+ }
+
+ @Override
+ public byte[] deserialize(byte[] bytes) {
+ deserialized.incrementAndGet();
+ return bytes;
+ }
+ });
+ RocksDBDAO dao = new RocksDBDAO("/prefix-handler",
dbManager.getRocksDBBasePath(), serializers);
+ try {
+ dao.addColumnFamily(family);
+ dao.put(family, "key_1", new byte[] {1});
+ dao.put(family, "key_2", new byte[] {2});
+ dao.put(family, "key_other", new byte[] {3});
+ dao.put(family, "other", new byte[] {4});
+ List<String> keys = new ArrayList<>();
+ dao.<byte[], IOException>prefixSearch(family, "key_", (key, value) -> {
+ keys.add(key);
+ assertEquals(keys.size(), deserialized.get(), "Values must be consumed
during the scan");
+ assertEquals(keys.size(), value[0]);
+ });
+ assertEquals(Arrays.asList("key_1", "key_2", "key_other"), keys);
+ assertEquals(keys, dao.prefixSearch(family,
"key_").map(Pair::getKey).collect(Collectors.toList()));
+ dao.prefixSearch(family, "missing", (key, value) -> {
+ throw new AssertionError("No entries should match");
+ });
+ IOException failure = new IOException("handler failed");
+ deserialized.set(0);
+ assertSame(failure, assertThrows(IOException.class, () ->
dao.prefixSearch(family, "key_", (key, value) -> {
+ throw failure;
+ })));
+ assertEquals(1, deserialized.get(), "A failed handler must stop the scan
immediately");
+ assertEquals(4, dao.prefixSearch(family, "").count());
+ } finally {
+ dao.close();
+ }
+ }
+
@Test
public void testWithSerializableKey() {
String prefix1 = "prefix1_";
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/lookup/HoodieLookupFunction.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/lookup/HoodieLookupFunction.java
index 7b02f9da826b..605cbd354937 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/lookup/HoodieLookupFunction.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/lookup/HoodieLookupFunction.java
@@ -165,6 +165,7 @@ public class HoodieLookupFunction extends LookupFunction
implements Serializable
cache.addRow(key, rowData);
}
}
+ cache.flush();
currentCommit = latestCommitInstant.get();
scheduleNextLoad();
log.info("Loaded {} row(s) into lookup join cache", count);
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/lookup/LookupCache.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/lookup/LookupCache.java
index 93e99c913e7c..6a5542f49d8e 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/lookup/LookupCache.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/lookup/LookupCache.java
@@ -44,6 +44,13 @@ public interface LookupCache extends Closeable {
*/
void addRow(RowData key, RowData row) throws IOException;
+ /**
+ * Flushes buffered writes after loading the cache. Implementations must
also make buffered
+ * rows visible to {@link #getRows(RowData)} before an explicit flush.
+ */
+ default void flush() throws IOException {
+ }
+
/**
* Returns all rows matching the given lookup key, or {@code null} / empty
list if none exist.
*
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/lookup/RocksDBLookupCache.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/lookup/RocksDBLookupCache.java
index 4bb41dfcc9a6..cbe8940ce4e6 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/lookup/RocksDBLookupCache.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/lookup/RocksDBLookupCache.java
@@ -19,6 +19,8 @@
package org.apache.hudi.table.lookup;
import org.apache.hudi.common.serialization.CustomSerializer;
+import org.apache.hudi.common.util.StringUtils;
+import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.common.util.collection.RocksDBDAO;
import lombok.extern.slf4j.Slf4j;
@@ -33,7 +35,6 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.stream.Collectors;
/**
* Off-heap {@link LookupCache} backed by RocksDB.
@@ -68,6 +69,7 @@ public class RocksDBLookupCache implements LookupCache {
private RocksDBDAO rocksDBDAO;
private long rowCounter;
+ private final WriteBuffer writeBuffer = new WriteBuffer();
public RocksDBLookupCache(
TypeSerializer<RowData> keySerializer,
@@ -77,39 +79,33 @@ public class RocksDBLookupCache implements LookupCache {
this.rowSerializer = rowSerializer;
this.rocksDbBasePath = rocksDbBasePath;
this.rocksDBDAO = createDAO();
- this.rowCounter = 0L;
}
@Override
public void addRow(RowData key, RowData row) throws IOException {
- String keyHex = serializeKeyToHex(key);
- String compoundKey = keyHex + KEY_SEPARATOR + rowCounter++;
- byte[] valueBytes = serializeRow(row);
- rocksDBDAO.put(COLUMN_FAMILY, compoundKey, valueBytes);
+ String compoundKey = serializeKeyToHex(key) + KEY_SEPARATOR + rowCounter++;
+ writeBuffer.add(rocksDBDAO, compoundKey, serializeRow(row));
+ }
+
+ @Override
+ public void flush() {
+ writeBuffer.flush(rocksDBDAO);
}
@Override
@Nullable
public List<RowData> getRows(RowData key) throws IOException {
+ flush();
String prefix = serializeKeyToHex(key) + KEY_SEPARATOR;
- List<byte[]> rawValues = rocksDBDAO.<byte[]>prefixSearch(COLUMN_FAMILY,
prefix)
- .map(pair -> pair.getValue())
- .collect(Collectors.toList());
- if (rawValues.isEmpty()) {
- return null;
- }
- List<RowData> result = new ArrayList<>(rawValues.size());
- for (byte[] bytes : rawValues) {
- result.add(deserializeRow(bytes));
- }
- return result;
+ List<RowData> result = new ArrayList<>();
+ rocksDBDAO.<byte[], IOException>prefixSearch(COLUMN_FAMILY, prefix,
+ (storedKey, bytes) -> result.add(deserializeRow(bytes)));
+ return result.isEmpty() ? null : result;
}
@Override
public void clear() {
- if (rocksDBDAO != null) {
- rocksDBDAO.close();
- }
+ close();
rocksDBDAO = createDAO();
rowCounter = 0L;
log.debug("RocksDB lookup cache cleared and reinitialized at {}",
rocksDbBasePath);
@@ -117,6 +113,7 @@ public class RocksDBLookupCache implements LookupCache {
@Override
public void close() {
+ writeBuffer.clear();
if (rocksDBDAO != null) {
rocksDBDAO.close();
rocksDBDAO = null;
@@ -138,7 +135,7 @@ public class RocksDBLookupCache implements LookupCache {
private String serializeKeyToHex(RowData key) throws IOException {
keyOutputBuffer.clear();
keySerializer.serialize(key, keyOutputBuffer);
- return bytesToHex(keyOutputBuffer.getCopyOfBuffer());
+ return StringUtils.toHexString(keyOutputBuffer.getCopyOfBuffer());
}
private byte[] serializeRow(RowData row) throws IOException {
@@ -152,12 +149,37 @@ public class RocksDBLookupCache implements LookupCache {
return rowSerializer.deserialize(rowInputBuffer);
}
- private static String bytesToHex(byte[] bytes) {
- StringBuilder sb = new StringBuilder(bytes.length * 2);
- for (byte b : bytes) {
- sb.append(String.format("%02x", b));
+ /**
+ * Pending serialized writes and their byte count, reset together after a
successful flush
+ * or when the cache is discarded.
+ */
+ private static class WriteBuffer {
+ private static final int MAX_ROWS = 1024;
+ private static final int MAX_BYTES = 1024 * 1024;
+
+ private final List<Pair<String, byte[]>> entries = new ArrayList<>();
+ private long sizeInBytes;
+
+ private void add(RocksDBDAO dao, String key, byte[] value) {
+ entries.add(Pair.of(key, value));
+ sizeInBytes += key.length() + (long) value.length;
+ if (entries.size() >= MAX_ROWS || sizeInBytes >= MAX_BYTES) {
+ flush(dao);
+ }
+ }
+
+ private void flush(RocksDBDAO dao) {
+ if (!entries.isEmpty()) {
+ dao.writeBatch(batch -> entries.forEach(
+ entry -> dao.putInBatch(batch, COLUMN_FAMILY, entry.getKey(),
entry.getValue())));
+ clear();
+ }
+ }
+
+ private void clear() {
+ entries.clear();
+ sizeInBytes = 0;
}
- return sb.toString();
}
/**
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/lookup/TestHoodieLookupFunction.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/lookup/TestHoodieLookupFunction.java
index e2af300219cd..08097d3015ea 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/lookup/TestHoodieLookupFunction.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/lookup/TestHoodieLookupFunction.java
@@ -29,6 +29,8 @@ import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.StringData;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.io.File;
import java.io.IOException;
@@ -77,9 +79,12 @@ class TestHoodieLookupFunction {
}
}
- @Test
- void testLookupCacheDoesNotReloadWhenCompletedCommitHasNotChanged() throws
Exception {
+ @ParameterizedTest
+ @ValueSource(strings = {"heap", "rocksdb"})
+ void testLookupCacheDoesNotReloadWhenCompletedCommitHasNotChanged(String
cacheType) throws Exception {
Configuration conf = getConf();
+ conf.set(FlinkOptions.LOOKUP_JOIN_CACHE_TYPE, cacheType);
+ conf.set(FlinkOptions.LOOKUP_JOIN_ROCKSDB_PATH, new File(tempFile,
"rocksdb").getAbsolutePath());
TestData.writeData(TestData.DATA_SET_SINGLE_INSERT, conf);
CountingLookupTableReader reader = new
CountingLookupTableReader(TestData.DATA_SET_SINGLE_INSERT, conf);
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/lookup/TestRocksDBLookupCache.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/lookup/TestRocksDBLookupCache.java
index 95b805f0809d..8d8a505912db 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/lookup/TestRocksDBLookupCache.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/lookup/TestRocksDBLookupCache.java
@@ -28,9 +28,14 @@ import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.table.types.logical.VarCharType;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.io.File;
+import java.util.Collections;
import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
@@ -76,6 +81,55 @@ class TestRocksDBLookupCache {
cache.close();
}
+ @ParameterizedTest
+ @ValueSource(ints = {1, 1024, 2051})
+ void testBatchedWrites(int rowCount) throws Exception {
+ RowType keyType = RowType.of(new VarCharType(VarCharType.MAX_LENGTH));
+ RowType rowType = RowType.of(new VarCharType(VarCharType.MAX_LENGTH), new
IntType());
+ try (RocksDBLookupCache cache = new RocksDBLookupCache(
+ InternalSerializers.create(keyType),
InternalSerializers.create(rowType), tempFile.getAbsolutePath())) {
+ RowData lookupKey = key("维度");
+ GenericRowData reuse = GenericRowData.of(StringData.fromString("维度"), 0);
+ for (int i = 0; i < rowCount; i++) {
+ reuse.setField(1, i);
+ cache.addRow(lookupKey, reuse);
+ }
+ reuse.setField(1, -1);
+ List<RowData> rows = cache.getRows(lookupKey);
+ assertEquals(IntStream.range(0,
rowCount).boxed().collect(Collectors.toList()),
+ rows.stream().map(row ->
row.getInt(1)).sorted().collect(Collectors.toList()));
+
+ cache.addRow(lookupKey, row("维度", rowCount));
+ cache.flush();
+ cache.flush();
+ assertEquals(rowCount + 1, cache.getRows(lookupKey).size());
+
+ // Clear must discard both persisted rows and the unflushed tail of a
load.
+ cache.addRow(key("pending"), row("pending", 1));
+ cache.clear();
+ assertNull(cache.getRows(lookupKey));
+ assertNull(cache.getRows(key("pending")));
+ cache.addRow(lookupKey, row("维度", -1));
+ assertEquals(-1, cache.getRows(lookupKey).get(0).getInt(1));
+ }
+ }
+
+ @Test
+ void testLargeRows() throws Exception {
+ RowType keyType = RowType.of(new VarCharType(VarCharType.MAX_LENGTH));
+ RowType rowType = RowType.of(new VarCharType(VarCharType.MAX_LENGTH), new
IntType());
+ String value = String.join("", Collections.nCopies(1024 * 1024, "x"));
+ try (RocksDBLookupCache cache = new RocksDBLookupCache(
+ InternalSerializers.create(keyType),
InternalSerializers.create(rowType), tempFile.getAbsolutePath())) {
+ cache.addRow(key("large"), row(value, 1));
+ cache.addRow(key("large"), row("tail", 2));
+ List<RowData> rows = cache.getRows(key("large"));
+ assertEquals(2, rows.size());
+ assertEquals(value, rows.get(0).getString(0).toString());
+ assertEquals("tail", rows.get(1).getString(0).toString());
+ }
+ }
+
private static RowData key(String key) {
return GenericRowData.of(StringData.fromString(key));
}
