This is an automated email from the ASF dual-hosted git repository.
danny0405 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 c2ebc23e787a test(flink): improve table planning and lookup coverage
(#19394)
c2ebc23e787a is described below
commit c2ebc23e787ac15e85db52c8a45b19e7adf39acb
Author: Danny Chan <[email protected]>
AuthorDate: Wed Jul 29 17:35:29 2026 +0800
test(flink): improve table planning and lookup coverage (#19394)
---
.../org/apache/hudi/table/HoodieTableSink.java | 2 +-
.../apache/hudi/table/TestHoodieTablePlanning.java | 176 +++++++++++++++++++++
.../org/apache/hudi/table/TestHoodieTableSink.java | 93 +++++++++++
.../apache/hudi/table/TestHoodieTableSource.java | 39 +++++
.../lookup/TestAsyncLookupFunctionWrapper.java | 99 ++++++++++++
.../table/lookup/TestHoodieLookupFunction.java | 47 ++++++
.../hudi/table/lookup/TestRocksDBLookupCache.java | 86 ++++++++++
7 files changed, 541 insertions(+), 1 deletion(-)
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSink.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSink.java
index 45bfde998af6..f2bf465d642a 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSink.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSink.java
@@ -155,7 +155,7 @@ public class HoodieTableSink implements
@Override
public DynamicTableSink copy() {
- return new HoodieTableSink(this.conf, this.schema, this.overwrite);
+ return new HoodieTableSink(new Configuration(this.conf), this.schema,
this.overwrite);
}
@Override
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieTablePlanning.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieTablePlanning.java
new file mode 100644
index 000000000000..be05ae931f37
--- /dev/null
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieTablePlanning.java
@@ -0,0 +1,176 @@
+/*
+ * 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.hudi.table;
+
+import org.apache.hudi.common.model.WriteOperationType;
+import org.apache.hudi.configuration.FlinkOptions;
+import org.apache.hudi.util.StreamerUtil;
+import org.apache.hudi.utils.TestConfigurations;
+import org.apache.hudi.utils.TestTableEnvs;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import
org.apache.flink.streaming.api.transformations.LegacySourceTransformation;
+import org.apache.flink.streaming.api.transformations.SourceTransformation;
+import org.apache.flink.table.api.ExplainDetail;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.types.Row;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Planning tests for the Hudi table source, sink, and factory.
+ *
+ * <p>The tests translate plans but deliberately do not execute a Flink job.
+ */
+class TestHoodieTablePlanning {
+
+ @TempDir
+ File tempFile;
+
+ @Test
+ void testSourcePushDownsForBothSourceImplementations() throws Exception {
+ StreamTableEnvironment tableEnv =
+ (StreamTableEnvironment) TestTableEnvs.getBatchTableEnv();
+ createTable(tableEnv, "source_v2", new File(tempFile, "source_v2"), true);
+ createTable(tableEnv, "legacy_source", new File(tempFile,
"legacy_source"), false);
+
+ for (String tableName : new String[] {"source_v2", "legacy_source"}) {
+ String query = "SELECT name FROM " + tableName + " WHERE `partition` =
'p1'";
+ String pushedPlan = tableEnv.explainSql(
+ query,
+ ExplainDetail.CHANGELOG_MODE,
+ ExplainDetail.JSON_EXECUTION_PLAN);
+ assertTrue(pushedPlan.contains("TableSourceScan"), pushedPlan);
+ assertTrue(pushedPlan.contains("filter=[=(partition"), pushedPlan);
+ assertTrue(pushedPlan.contains("project=[name, partition]"), pushedPlan);
+ assertSourceTransformation(
+ tableEnv,
+ query,
+ tableName.equals("source_v2")
+ ? SourceTransformation.class
+ : LegacySourceTransformation.class);
+
+ String limitedPlan = tableEnv.explainSql(
+ "SELECT name FROM " + tableName + " LIMIT 2",
+ ExplainDetail.CHANGELOG_MODE,
+ ExplainDetail.JSON_EXECUTION_PLAN);
+ assertTrue(limitedPlan.contains("limit=[2]"), limitedPlan);
+ }
+ }
+
+ private static void assertSourceTransformation(
+ StreamTableEnvironment tableEnv,
+ String query,
+ Class<?> expectedSourceTransformation) {
+ DataStream<Row> plannedStream =
tableEnv.toDataStream(tableEnv.sqlQuery(query));
+ List<String> transformationNames = plannedStream.getTransformation()
+ .getTransitivePredecessors()
+ .stream()
+ .map(transformation -> transformation.getName()
+ + ":" + transformation.getClass().getName())
+ .collect(Collectors.toList());
+ assertTrue(
+ plannedStream.getTransformation()
+ .getTransitivePredecessors()
+ .stream()
+ .anyMatch(expectedSourceTransformation::isInstance),
+ "Expected " + expectedSourceTransformation.getSimpleName() + " in " +
transformationNames);
+ }
+
+ @Test
+ void testSinkPlanning() throws Exception {
+ TableEnvironment tableEnv = TestTableEnvs.getBatchTableEnv();
+ createTable(tableEnv, "sink_table", new File(tempFile, "sink_table"),
true);
+
+ String plan = tableEnv.explainSql(
+ "INSERT INTO sink_table VALUES "
+ + "('id1', 'Alice', 20, TIMESTAMP '2026-01-01 00:00:00', 'p1')",
+ ExplainDetail.CHANGELOG_MODE,
+ ExplainDetail.JSON_EXECUTION_PLAN);
+
+
assertTrue(plan.contains("Sink(table=[default_catalog.default_database.sink_table]"),
plan);
+ assertTrue(plan.contains("stream_write: default_database.sink_table"),
plan);
+
+ createTable(
+ tableEnv,
+ "append_sink",
+ new File(tempFile, "append_sink"),
+ true,
+ WriteOperationType.INSERT.value());
+ String appendPlan = tableEnv.explainSql(
+ "INSERT INTO append_sink VALUES "
+ + "('id2', 'Bob', 30, TIMESTAMP '2026-01-02 00:00:00', 'p2')",
+ ExplainDetail.CHANGELOG_MODE,
+ ExplainDetail.JSON_EXECUTION_PLAN);
+ assertTrue(
+ appendPlan.contains("hoodie_append_write:
default_database.append_sink"),
+ appendPlan);
+ }
+
+ private static void createTable(
+ TableEnvironment tableEnv,
+ String tableName,
+ File tablePath,
+ boolean sourceV2) throws Exception {
+ createTable(tableEnv, tableName, tablePath, sourceV2, null);
+ }
+
+ private static void createTable(
+ TableEnvironment tableEnv,
+ String tableName,
+ File tablePath,
+ boolean sourceV2,
+ String operation) throws Exception {
+ Configuration conf =
TestConfigurations.getDefaultConf(tablePath.getAbsolutePath());
+ if (operation != null) {
+ conf.set(FlinkOptions.OPERATION, operation);
+ }
+ StreamerUtil.initTableIfNotExists(conf);
+
+ tableEnv.executeSql(
+ "CREATE TABLE " + tableName + " ("
+ + "uuid STRING,"
+ + "name STRING,"
+ + "age INT,"
+ + "ts TIMESTAMP(3),"
+ + "`partition` STRING,"
+ + "PRIMARY KEY (uuid) NOT ENFORCED"
+ + ") PARTITIONED BY (`partition`) WITH ("
+ + "'connector' = 'hudi',"
+ + "'path' = '" + sqlLiteral(tablePath.getAbsolutePath()) + "',"
+ + "'" + FlinkOptions.ORDERING_FIELDS.key() + "' = 'ts',"
+ + "'" + FlinkOptions.READ_SOURCE_V2_ENABLED.key() + "' = '" +
sourceV2 + "'"
+ + (operation == null
+ ? ""
+ : ",'" + FlinkOptions.OPERATION.key() + "' = '" + operation +
"'")
+ + ")");
+ }
+
+ private static String sqlLiteral(String value) {
+ return value.replace("'", "''");
+ }
+}
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieTableSink.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieTableSink.java
new file mode 100644
index 000000000000..d75ca711f565
--- /dev/null
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieTableSink.java
@@ -0,0 +1,93 @@
+/*
+ * 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.hudi.table;
+
+import org.apache.hudi.common.model.WriteOperationType;
+import org.apache.hudi.configuration.FlinkOptions;
+import org.apache.hudi.util.ChangelogModes;
+import org.apache.hudi.util.DataModificationInfos;
+import org.apache.hudi.utils.TestConfigurations;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotSame;
+import static org.junit.jupiter.api.Assertions.assertSame;
+
+/**
+ * Tests for {@link HoodieTableSink}.
+ */
+class TestHoodieTableSink {
+
+ @Test
+ void testChangelogModeAndCopy() {
+ Configuration conf = new Configuration();
+ HoodieTableSink sink = new HoodieTableSink(conf,
TestConfigurations.TABLE_SCHEMA);
+
+ assertEquals(ChangelogModes.UPSERT,
sink.getChangelogMode(ChangelogMode.all()));
+ conf.set(FlinkOptions.CHANGELOG_ENABLED, true);
+ assertEquals(ChangelogModes.FULL,
sink.getChangelogMode(ChangelogMode.insertOnly()));
+ assertEquals("HoodieTableSink", sink.asSummaryString());
+
+ HoodieTableSink copied = (HoodieTableSink) sink.copy();
+ assertNotSame(sink, copied);
+ assertNotSame(conf, copied.getConf());
+ assertEquals(ChangelogModes.FULL,
copied.getChangelogMode(ChangelogMode.insertOnly()));
+
+ copied.applyRowLevelDelete(null);
+ assertEquals(
+ WriteOperationType.DELETE.value(),
+ copied.getConf().get(FlinkOptions.OPERATION));
+ assertEquals(WriteOperationType.UPSERT.value(),
conf.get(FlinkOptions.OPERATION));
+ }
+
+ @Test
+ void testOverwriteAndRowLevelOperations() {
+ Configuration conf = new Configuration();
+ HoodieTableSink sink = new HoodieTableSink(conf,
TestConfigurations.TABLE_SCHEMA);
+
+ sink.applyOverwrite(true);
+ assertEquals(
+ WriteOperationType.INSERT_OVERWRITE_TABLE.value(),
+ conf.get(FlinkOptions.OPERATION));
+ sink.applyStaticPartition(Collections.singletonMap("partition", "p1"));
+ assertEquals(
+ WriteOperationType.INSERT_OVERWRITE.value(),
+ conf.get(FlinkOptions.OPERATION));
+
+ conf.set(FlinkOptions.WRITE_PARTITION_OVERWRITE_MODE, "DYNAMIC");
+ sink.applyOverwrite(true);
+ assertEquals(
+ WriteOperationType.INSERT_OVERWRITE.value(),
+ conf.get(FlinkOptions.OPERATION));
+
+ assertSame(
+ DataModificationInfos.DEFAULT_DELETE_INFO,
+ sink.applyRowLevelDelete(null));
+ assertEquals(WriteOperationType.DELETE.value(),
conf.get(FlinkOptions.OPERATION));
+ assertSame(
+ DataModificationInfos.DEFAULT_UPDATE_INFO,
+ sink.applyRowLevelUpdate(Collections.emptyList(), null));
+ assertEquals(WriteOperationType.UPSERT.value(),
conf.get(FlinkOptions.OPERATION));
+ }
+}
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieTableSource.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieTableSource.java
index ee04e2175421..26694f2178af 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieTableSource.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieTableSource.java
@@ -24,6 +24,7 @@ import
org.apache.hudi.common.model.PartitionBucketIndexHashingConfig;
import org.apache.hudi.common.schema.HoodieSchemaField;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.configuration.FlinkOptions;
+import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.source.ExpressionPredicates;
import org.apache.hudi.source.prune.ColumnStatsProbe;
import org.apache.hudi.storage.StoragePath;
@@ -40,6 +41,9 @@ import org.apache.flink.api.common.io.InputFormat;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.connector.source.LookupTableSource;
+import
org.apache.flink.table.connector.source.lookup.AsyncLookupFunctionProvider;
+import org.apache.flink.table.connector.source.lookup.LookupFunctionProvider;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.expressions.CallExpression;
import org.apache.flink.table.expressions.FieldReferenceExpression;
@@ -78,7 +82,10 @@ import static
org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
/**
* Test cases for HoodieTableSource.
@@ -143,6 +150,16 @@ public class TestHoodieTableSource {
"Query type: 'incremental' should be supported");
}
+ @Test
+ void testStreamingInputFormatPlanning() {
+ HoodieTableSource tableSource = getEmptyStreamingSource(false);
+
+ assertThat(tableSource.getInputFormat(true),
is(instanceOf(MergeOnReadInputFormat.class)));
+
+ tableSource.getConf().set(FlinkOptions.QUERY_TYPE,
FlinkOptions.QUERY_TYPE_READ_OPTIMIZED);
+ assertThrows(HoodieException.class, () ->
tableSource.getInputFormat(true));
+ }
+
@ParameterizedTest
@ValueSource(booleans = {true, false})
void testGetTableAvroSchema(boolean isSourceV2) {
@@ -160,6 +177,28 @@ public class TestHoodieTableSource {
assertThat(schemaFields, is(expected));
}
+ @Test
+ void testLookupRuntimeProvider() {
+ HoodieTableSource tableSource = getEmptyStreamingSource(true);
+ LookupTableSource.LookupContext lookupContext =
mock(LookupTableSource.LookupContext.class);
+ when(lookupContext.getKeys()).thenReturn(new int[][] {{0}});
+
+ assertThat(
+ tableSource.getLookupRuntimeProvider(lookupContext),
+ is(instanceOf(LookupFunctionProvider.class)));
+
+ tableSource.getConf().set(FlinkOptions.LOOKUP_ASYNC, true);
+ assertThat(
+ tableSource.getLookupRuntimeProvider(lookupContext),
+ is(instanceOf(AsyncLookupFunctionProvider.class)));
+
+ LookupTableSource.LookupContext nestedLookupContext =
mock(LookupTableSource.LookupContext.class);
+ when(nestedLookupContext.getKeys()).thenReturn(new int[][] {{0, 1}});
+ assertThrows(
+ UnsupportedOperationException.class,
+ () -> tableSource.getLookupRuntimeProvider(nestedLookupContext));
+ }
+
@ParameterizedTest
@ValueSource(booleans = {true, false})
void testDataSkippingFilterShouldBeNotNullWhenTableSourceIsCopied(boolean
isSourceV2) {
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/lookup/TestAsyncLookupFunctionWrapper.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/lookup/TestAsyncLookupFunctionWrapper.java
new file mode 100644
index 000000000000..4fcdd738f04d
--- /dev/null
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/lookup/TestAsyncLookupFunctionWrapper.java
@@ -0,0 +1,99 @@
+/*
+ * 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.hudi.table.lookup;
+
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.functions.FunctionContext;
+import org.apache.flink.table.functions.LookupFunction;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.concurrent.CompletionException;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests for {@link AsyncLookupFunctionWrapper}.
+ */
+class TestAsyncLookupFunctionWrapper {
+
+ @Test
+ void testLifecycleAndAsyncLookup() throws Exception {
+ TestingLookupFunction function = new TestingLookupFunction();
+ AsyncLookupFunctionWrapper wrapper = new
AsyncLookupFunctionWrapper(function, 2);
+ RowData key = GenericRowData.of(1);
+
+ wrapper.open(null);
+ Collection<RowData> result = wrapper.asyncLookup(key).join();
+
+ assertEquals(1, result.size());
+ assertSame(key, result.iterator().next());
+ assertTrue(function.opened);
+ wrapper.close();
+ assertTrue(function.closed);
+ }
+
+ @Test
+ void testIOExceptionIsPropagatedAsUncheckedIOException() throws Exception {
+ AsyncLookupFunctionWrapper wrapper =
+ new AsyncLookupFunctionWrapper(new FailingLookupFunction(), 1);
+
+ CompletionException exception = assertThrows(
+ CompletionException.class,
+ () -> wrapper.asyncLookup(GenericRowData.of(1)).join());
+
+ assertInstanceOf(UncheckedIOException.class, exception.getCause());
+ wrapper.close();
+ }
+
+ private static class TestingLookupFunction extends LookupFunction {
+ private boolean opened;
+ private boolean closed;
+
+ @Override
+ public void open(FunctionContext context) {
+ opened = true;
+ }
+
+ @Override
+ public Collection<RowData> lookup(RowData keyRow) {
+ return Collections.singletonList(keyRow);
+ }
+
+ @Override
+ public void close() {
+ closed = true;
+ }
+ }
+
+ private static class FailingLookupFunction extends LookupFunction {
+ @Override
+ public Collection<RowData> lookup(RowData keyRow) throws IOException {
+ throw new IOException("expected");
+ }
+ }
+}
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 8d336ede565e..750f11f7956c 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
@@ -39,8 +39,14 @@ import java.util.Collections;
import java.util.List;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
/**
* Tests for {@link HoodieLookupFunction}.
@@ -98,6 +104,32 @@ class TestHoodieLookupFunction {
}
}
+ @Test
+ void testRocksDBCacheLifecycleAndLookupFailure() throws Exception {
+ Configuration conf = getConf();
+ conf.set(FlinkOptions.LOOKUP_JOIN_CACHE_TYPE, "rocksdb");
+ conf.set(FlinkOptions.LOOKUP_JOIN_ROCKSDB_PATH, new File(tempFile,
"rocksdb").getAbsolutePath());
+ StreamerUtil.initTableIfNotExists(conf);
+
+ HoodieLookupFunction function =
+ newLookupFunction(new
CountingLookupTableReader(Collections.emptyList(), conf), conf);
+ function.open(null);
+
+ assertEquals(Duration.ofDays(1), function.getReloadInterval());
+ assertInstanceOf(RocksDBLookupCache.class, getCache(function));
+ assertNull(function.lookup(lookupKey()));
+ function.close();
+
+ LookupCache failingCache = mock(LookupCache.class);
+ when(failingCache.getRows(any())).thenThrow(new IOException("expected"));
+ setCache(function, failingCache);
+ setNextLoadTime(function, Long.MAX_VALUE);
+ RuntimeException exception =
+ assertThrows(RuntimeException.class, () ->
function.lookup(lookupKey()));
+ assertInstanceOf(IOException.class, exception.getCause());
+ function.close();
+ }
+
private HoodieLookupFunction newLookupFunction(CountingLookupTableReader
reader, Configuration conf) {
return new HoodieLookupFunction(
reader,
@@ -129,6 +161,21 @@ class TestHoodieLookupFunction {
field.setLong(function, nextLoadTime);
}
+ private static void setCache(HoodieLookupFunction function, LookupCache
cache) throws Exception {
+ Field field = cacheField();
+ field.set(function, cache);
+ }
+
+ private static LookupCache getCache(HoodieLookupFunction function) throws
Exception {
+ return (LookupCache) cacheField().get(function);
+ }
+
+ private static Field cacheField() throws NoSuchFieldException {
+ Field field = HoodieLookupFunction.class.getDeclaredField("cache");
+ field.setAccessible(true);
+ return field;
+ }
+
private static class CountingLookupTableReader extends
HoodieLookupTableReader {
private final List<RowData> rows;
private int openCount;
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
new file mode 100644
index 000000000000..95b805f0809d
--- /dev/null
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/lookup/TestRocksDBLookupCache.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.hudi.table.lookup;
+
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.runtime.typeutils.InternalSerializers;
+import org.apache.flink.table.types.logical.IntType;
+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 java.io.File;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+/**
+ * Tests for {@link RocksDBLookupCache}.
+ */
+class TestRocksDBLookupCache {
+
+ @TempDir
+ File tempFile;
+
+ @Test
+ void testPutGetClearAndClose() throws Exception {
+ RowType keyType = RowType.of(
+ new VarCharType(VarCharType.MAX_LENGTH));
+ RowType rowType = RowType.of(
+ new VarCharType(VarCharType.MAX_LENGTH),
+ new IntType());
+ TypeSerializer<RowData> keySerializer =
InternalSerializers.create(keyType);
+ TypeSerializer<RowData> rowSerializer =
InternalSerializers.create(rowType);
+ RocksDBLookupCache cache =
+ new RocksDBLookupCache(keySerializer, rowSerializer,
tempFile.getAbsolutePath());
+
+ RowData key1 = key("id1");
+ cache.addRow(key1, row("id1", 10));
+ cache.addRow(key1, row("id1", 20));
+ cache.addRow(key("id2"), row("id2", 30));
+
+ List<RowData> rows = cache.getRows(key1);
+ assertEquals(2, rows.size());
+ assertEquals("id1", rows.get(0).getString(0).toString());
+ assertEquals(10, rows.get(0).getInt(1));
+ assertEquals(20, rows.get(1).getInt(1));
+ assertNull(cache.getRows(key("missing")));
+
+ cache.clear();
+ assertNull(cache.getRows(key1));
+ cache.addRow(key1, row("id1", 40));
+ assertEquals(40, cache.getRows(key1).get(0).getInt(1));
+
+ cache.close();
+ cache.close();
+ }
+
+ private static RowData key(String key) {
+ return GenericRowData.of(StringData.fromString(key));
+ }
+
+ private static RowData row(String key, int value) {
+ return GenericRowData.of(StringData.fromString(key), value);
+ }
+}
