cshuo commented on code in PR #19394: URL: https://github.com/apache/hudi/pull/19394#discussion_r3670594841
########## hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieTableSink.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.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.apache.flink.table.connector.sink.DynamicTableSink; +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()); + + DynamicTableSink copied = sink.copy(); + assertNotSame(sink, copied); + assertSame(conf, ((HoodieTableSink) copied).getConf()); Review Comment: `DynamicTableSink.copy()` requires a deep copy of all mutable members, but this assertion explicitly requires the copied sink to share the same mutable `Configuration`. `applyOverwrite`, `applyStaticPartition`, `applyRowLevelDelete`, and `applyRowLevelUpdate` mutate that object, so planner mutations on one copy can leak into another and change the selected write operation. Please copy the configuration in `HoodieTableSink.copy()` (for example, `new Configuration(conf)`) and make this test assert both a distinct config instance and mutation isolation. ########## hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/lookup/TestHoodieLookupFunction.java: ########## @@ -98,6 +103,31 @@ void testLookupCacheDoesNotReloadWhenCompletedCommitHasNotChanged() throws Excep } } + @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()); + assertNull(function.lookup(lookupKey())); Review Comment: This assertion does not verify that the `rocksdb` option selected `RocksDBLookupCache`: both an empty heap cache and an empty RocksDB cache return `null` and close normally. A regression that ignores `LOOKUP_JOIN_CACHE_TYPE` would therefore leave this test green. Since this test is intended to cover cache selection, please assert the selected cache type immediately after `open()` using a test-visible accessor or the existing reflection approach. ########## hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieTablePlanning.java: ########## @@ -0,0 +1,111 @@ +/* + * 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.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.table.api.ExplainDetail; +import org.apache.flink.table.api.TableEnvironment; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.File; + +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 { + TableEnvironment tableEnv = 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 pushedPlan = tableEnv.explainSql( + "SELECT name FROM " + tableName + " WHERE `partition` = 'p1'", + ExplainDetail.CHANGELOG_MODE, + ExplainDetail.JSON_EXECUTION_PLAN); + assertTrue(pushedPlan.contains("TableSourceScan"), pushedPlan); Review Comment: All assertions here describe logical pushdowns shared by both tables; none proves that `source_v2` used the FLIP-27 source while `legacy_source` used the legacy runtime path. If `READ_SOURCE_V2_ENABLED` were ignored and both tables always selected the same implementation, this test would still pass. Please assert branch-specific runtime-plan evidence (for example, the V2 versus legacy source operator) so the stated “both source implementations” coverage detects selection regressions. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
