ericyuan915 commented on code in PR #19520: URL: https://github.com/apache/hudi/pull/19520#discussion_r3723253167
########## hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/enumerator/TestHoodieSourceEnumeratorRouting.java: ########## @@ -0,0 +1,436 @@ +/* + * 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.source.enumerator; + +import org.apache.hudi.common.model.HoodieTableType; +import org.apache.hudi.common.schema.HoodieSchema; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.timeline.HoodieInstant; +import org.apache.hudi.common.util.Option; +import org.apache.hudi.configuration.FlinkOptions; +import org.apache.hudi.configuration.HadoopConfigurations; +import org.apache.hudi.source.HoodieScanContext; +import org.apache.hudi.source.HoodieSource; +import org.apache.hudi.source.reader.HoodieRecordEmitter; +import org.apache.hudi.source.reader.function.HoodieSplitReaderFunction; +import org.apache.hudi.source.split.DefaultHoodieSplitProvider; +import org.apache.hudi.source.split.GlobalHoodieSplitProvider; +import org.apache.hudi.source.split.HoodieCdcSourceSplit; +import org.apache.hudi.source.split.HoodieSourceSplit; +import org.apache.hudi.source.split.HoodieSourceSplitComparator; +import org.apache.hudi.source.split.HoodieSourceSplitState; +import org.apache.hudi.source.split.HoodieSourceSplitStatus; +import org.apache.hudi.source.split.HoodieSplitProvider; +import org.apache.hudi.storage.StoragePath; +import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration; +import org.apache.hudi.table.format.InternalSchemaManager; +import org.apache.hudi.util.HoodieSchemaConverter; +import org.apache.hudi.util.StreamerUtil; +import org.apache.hudi.utils.TestConfigurations; +import org.apache.hudi.utils.TestData; + +import org.apache.flink.api.connector.source.ReaderInfo; +import org.apache.flink.api.connector.source.SourceEvent; +import org.apache.flink.api.connector.source.SplitEnumerator; +import org.apache.flink.api.connector.source.SplitEnumeratorContext; +import org.apache.flink.api.connector.source.SplitsAssignment; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.metrics.groups.SplitEnumeratorMetricGroup; +import org.apache.flink.metrics.groups.UnregisteredMetricsGroup; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.types.logical.RowType; +import org.junit.jupiter.api.BeforeEach; +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.EnumSource; +import org.junit.jupiter.params.provider.ValueSource; + +import java.io.File; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.Callable; +import java.util.function.BiConsumer; +import java.util.stream.Collectors; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Tests which split provider {@link HoodieSource} wires into the enumerator. + * + * <p>Bounded reads use the shared work-stealing pool ({@link GlobalHoodieSplitProvider}); streaming + * keeps per-subtask assignment ({@link DefaultHoodieSplitProvider}) so that a file id's successive + * incremental splits stay affine to one reader. Because {@code HoodieSource.createEnumerator} + * handles fresh creation and restore in the same method, both paths are covered for every mode. + * + * <p>These tests also assert the property that makes work stealing safe for bounded reads: every + * bounded query mode emits exactly one split per file group, so there is no cross-commit + * continuation and no ordering relationship between splits that a shared pool could break. + * + * <p>Lives in the enumerator package so it can read the package-private + * {@link AbstractHoodieSplitEnumerator#splitProvider}. + */ +public class TestHoodieSourceEnumeratorRouting { + + @TempDir + File tempDir; + + private Configuration conf; + private StoragePath tablePath; + private HoodieTableMetaClient metaClient; + + /** + * The bounded query modes {@code HoodieSource.createBatchHoodieSplits()} covers. All of them are + * routed to the shared pool, so all of them are exercised here. + */ + private enum BoundedMode { + COW_SNAPSHOT(HoodieTableType.COPY_ON_WRITE, FlinkOptions.QUERY_TYPE_SNAPSHOT, false), + MOR_SNAPSHOT(HoodieTableType.MERGE_ON_READ, FlinkOptions.QUERY_TYPE_SNAPSHOT, false), + MOR_READ_OPTIMIZED(HoodieTableType.MERGE_ON_READ, FlinkOptions.QUERY_TYPE_READ_OPTIMIZED, false), + COW_INCREMENTAL(HoodieTableType.COPY_ON_WRITE, FlinkOptions.QUERY_TYPE_INCREMENTAL, false), + COW_INCREMENTAL_CDC(HoodieTableType.COPY_ON_WRITE, FlinkOptions.QUERY_TYPE_INCREMENTAL, true); + + private final HoodieTableType tableType; + private final String queryType; + private final boolean cdcEnabled; + + BoundedMode(HoodieTableType tableType, String queryType, boolean cdcEnabled) { + this.tableType = tableType; + this.queryType = queryType; + this.cdcEnabled = cdcEnabled; + } + + boolean isIncremental() { + return FlinkOptions.QUERY_TYPE_INCREMENTAL.equals(queryType); + } + } + + @BeforeEach + public void setUp() { + conf = TestConfigurations.getDefaultConf(tempDir.getAbsolutePath()); + tablePath = new StoragePath(tempDir.getAbsolutePath()); + } + + @ParameterizedTest + @EnumSource(BoundedMode.class) + public void testBoundedReadUsesSharedSplitPool(BoundedMode mode) throws Exception { + HoodieSource<RowData> source = prepareBoundedSource(mode); + MockSplitEnumeratorContext context = new MockSplitEnumeratorContext(); + + SplitEnumerator<HoodieSourceSplit, HoodieSplitEnumeratorState> enumerator = + source.createEnumerator(context); + + assertInstanceOf(HoodieStaticSplitEnumerator.class, enumerator, + "Bounded read should use the static enumerator for mode " + mode); + assertInstanceOf(GlobalHoodieSplitProvider.class, providerOf(enumerator), + "Bounded read should use the shared work-stealing pool for mode " + mode); + List<HoodieSourceSplit> splits = pendingSplits(enumerator); + assertOneSplitPerFileGroup(splits, mode); + if (mode.cdcEnabled) { + // Guards the parameterization itself: reading from earliest would fall back to a plain file + // slice scan, and this mode would silently stop covering the CDC branch. + splits.forEach(split -> assertInstanceOf(HoodieCdcSourceSplit.class, split, + "CDC mode should produce CDC splits")); + } + } + + @ParameterizedTest + @EnumSource(BoundedMode.class) + public void testBoundedRestoreKeepsSharedSplitPool(BoundedMode mode) throws Exception { + HoodieSource<RowData> source = prepareBoundedSource(mode); + // Snapshot the real splits of this mode, then restore from a subset of them. + List<HoodieSourceSplit> discovered = + pendingSplits(source.createEnumerator(new MockSplitEnumeratorContext())); + assertFalse(discovered.isEmpty(), "Expected at least one split for mode " + mode); + List<HoodieSourceSplit> checkpointed = discovered.subList(0, 1); + + MockSplitEnumeratorContext context = new MockSplitEnumeratorContext(); + SplitEnumerator<HoodieSourceSplit, HoodieSplitEnumeratorState> restored = + source.restoreEnumerator(context, enumeratorStateOf(checkpointed)); + + assertInstanceOf(HoodieStaticSplitEnumerator.class, restored, + "Restored bounded read should still use the static enumerator for mode " + mode); + assertInstanceOf(GlobalHoodieSplitProvider.class, providerOf(restored), + "Restored bounded read should still use the shared work-stealing pool for mode " + mode); + assertEquals(checkpointed.size(), providerOf(restored).pendingSplitCount(), + "Restore should replay exactly the checkpointed splits into the shared pool " + + "and must not re-run split discovery for mode " + mode); + } + + /** + * A restored pending split is not owned by any subtask: whichever reader asks for work claims it. + * Parameterized over the requesting subtask so the assertion is deterministic - under per-subtask + * pinning only the one subtask the file id hashes to could ever receive it. + */ + @ParameterizedTest + @ValueSource(ints = {0, 1, 2, 3}) + public void testRestoredSplitIsClaimedByWhicheverSubtaskAsks(int requestingSubtask) throws Exception { + HoodieSource<RowData> source = prepareBoundedSource(BoundedMode.COW_SNAPSHOT); + List<HoodieSourceSplit> discovered = + pendingSplits(source.createEnumerator(new MockSplitEnumeratorContext())); + List<HoodieSourceSplit> checkpointed = discovered.subList(0, 1); + + MockSplitEnumeratorContext context = new MockSplitEnumeratorContext(); + SplitEnumerator<HoodieSourceSplit, HoodieSplitEnumeratorState> restored = + source.restoreEnumerator(context, enumeratorStateOf(checkpointed)); + restored.start(); + for (int subtask = 0; subtask < 4; subtask++) { + context.registerReader(new ReaderInfo(subtask, "localhost")); + } + + restored.handleSplitRequest(requestingSubtask, "localhost"); + + assertEquals(checkpointed, context.getAssignedSplits().get(requestingSubtask), + "The restored split should go to whichever subtask asked for work"); + assertFalse(context.getNoMoreSplitsSignaled().contains(requestingSubtask), + "The requesting subtask received a split, so it should not be told no-more-splits"); + } + + @Test + public void testStreamingReadKeepsPerSubtaskProvider() throws Exception { + HoodieSource<RowData> source = prepareStreamingSource(); + + SplitEnumerator<HoodieSourceSplit, HoodieSplitEnumeratorState> enumerator = + source.createEnumerator(new MockSplitEnumeratorContext()); + + assertInstanceOf(HoodieContinuousSplitEnumerator.class, enumerator, + "Streaming read should use the continuous enumerator"); + assertInstanceOf(DefaultHoodieSplitProvider.class, providerOf(enumerator), + "Streaming read must keep per-subtask assignment for file id affinity"); + } + + @Test + public void testStreamingRestoreKeepsPerSubtaskProvider() throws Exception { + HoodieSource<RowData> source = prepareStreamingSource(); + List<HoodieSourceSplit> checkpointed = Collections.singletonList( + new HoodieSourceSplit(0, null, Option.empty(), tablePath.toString(), "par1", + FlinkOptions.REALTIME_PAYLOAD_COMBINE, "20260126034717000", "file-0", Option.empty())); + + SplitEnumerator<HoodieSourceSplit, HoodieSplitEnumeratorState> restored = + source.restoreEnumerator(new MockSplitEnumeratorContext(), enumeratorStateOf(checkpointed)); + + assertInstanceOf(HoodieContinuousSplitEnumerator.class, restored, + "Restored streaming read should use the continuous enumerator"); + assertInstanceOf(DefaultHoodieSplitProvider.class, providerOf(restored), + "Restored streaming read must keep per-subtask assignment"); + assertEquals(1, providerOf(restored).pendingSplitCount(), + "Restored split should be replayed into the per-subtask provider"); + } + + // Helper methods + + private static HoodieSplitProvider providerOf( + SplitEnumerator<HoodieSourceSplit, HoodieSplitEnumeratorState> enumerator) { + return ((AbstractHoodieSplitEnumerator) enumerator).splitProvider; + } + + private static List<HoodieSourceSplit> pendingSplits( + SplitEnumerator<HoodieSourceSplit, HoodieSplitEnumeratorState> enumerator) { + return providerOf(enumerator).state().stream() + .map(HoodieSourceSplitState::getSplit) + .collect(Collectors.toList()); + } + + private static HoodieSplitEnumeratorState enumeratorStateOf(List<HoodieSourceSplit> splits) { + List<HoodieSourceSplitState> states = splits.stream() + .map(split -> new HoodieSourceSplitState(split, HoodieSourceSplitStatus.UNASSIGNED)) + .collect(Collectors.toList()); + return new HoodieSplitEnumeratorState(states, Option.empty(), Option.empty()); + } + + /** + * Asserts the invariant that makes a shared pool safe for a bounded read: one split per file + * group, hence no cross-commit continuation and no ordering relationship between splits. + */ + private static void assertOneSplitPerFileGroup(List<HoodieSourceSplit> splits, BoundedMode mode) { + assertFalse(splits.isEmpty(), "Expected at least one split for mode " + mode); + Set<String> fileIds = splits.stream() + .map(HoodieSourceSplit::getFileId) + .collect(Collectors.toSet()); + assertEquals(splits.size(), fileIds.size(), + "Mode " + mode + " must emit exactly one split per file group, otherwise splits of the " + + "same file group could be read concurrently by different readers"); + } + + private HoodieSource<RowData> prepareBoundedSource(BoundedMode mode) throws Exception { + conf.set(FlinkOptions.TABLE_TYPE, mode.tableType.name()); + conf.set(FlinkOptions.READ_AS_STREAMING, false); + if (mode.tableType == HoodieTableType.MERGE_ON_READ) { + // Compact the first commit so the MOR file groups own a base file (a read-optimized read + // sees nothing otherwise); the second commit below is then written as logs only, so a + // snapshot read exercises real base + log file slices. + conf.set(FlinkOptions.COMPACTION_ASYNC_ENABLED, true); + conf.set(FlinkOptions.COMPACTION_DELTA_COMMITS, 1); + } + if (mode.cdcEnabled) { + conf.set(FlinkOptions.CDC_ENABLED, true); + conf.set(FlinkOptions.INDEX_BOOTSTRAP_ENABLED, true); // for batch update + } + + TestData.writeData(TestData.DATA_SET_INSERT, conf); + if (mode.tableType == HoodieTableType.MERGE_ON_READ) { + conf.set(FlinkOptions.COMPACTION_ASYNC_ENABLED, false); + } + TestData.writeData(TestData.DATA_SET_UPDATE_INSERT, conf); + metaClient = StreamerUtil.createMetaClient(conf); + + conf.set(FlinkOptions.QUERY_TYPE, mode.queryType); + if (mode.isIncremental()) { + // Reading from earliest triggers a full table scan, which bypasses the CDC branch, so the + // CDC mode starts from the last completed commit instead. + conf.set(FlinkOptions.READ_START_COMMIT, + mode.cdcEnabled ? lastCompletionTime() : FlinkOptions.START_COMMIT_EARLIEST); Review Comment: Good catch, you're right — `earliest` leaves the analyzer's `startInstant` empty (`IncrementalQueryAnalyzer:205`, `isConsumingFromEarliest()` is `startInstant.isEmpty()`), so `fullTableScan` was true and that parameter never reached the metadata-driven branch the safety argument cites. Fixed in 4d55875. `COW_INCREMENTAL` now starts from a real completed commit, and I added `COW_INCREMENTAL_FROM_EARLIEST` so the full-scan fallback stays covered — as you suggested, both sides of the branch. I also wanted the parameter to fail loudly rather than silently stop covering the branch again, so the modes that start from a real commit now write a last commit touching only `par5`/`par6`. The metadata-driven branch derives its read partitions from that commit's metadata alone, while a full table scan lists `par1` through `par6`, so asserting the split partitions pins which branch actually produced them: ```java if (mode.incrementalStart == IncrementalStart.LAST_COMMIT) { assertEquals(new HashSet<>(Arrays.asList("par5", "par6")), splits.stream().map(HoodieSourceSplit::getPartitionPath).collect(Collectors.toSet()), "Mode " + mode + " should read only the partitions written by the start commit, " + "which is what distinguishes the incremental branch from a full table scan"); } ``` I verified the assertion actually bites by temporarily pointing the start commit back at `earliest`: `COW_INCREMENTAL` and `COW_INCREMENTAL_CDC` both fail with `expected: <[par5, par6]> but was: <[par5, par6, par1, par2, par3, par4]>`. For the record, all three incremental shapes end at the same `getInputSplits(fileSlices, ...)`, one split per slice, and both branches source their slices from `getLatestMergedFileSlicesBeforeOrOn`; the difference is only how the partition/file set is derived. So the one-split-per-file-group claim held either way, but the test now demonstrates it on both paths instead of asserting it on one and claiming it for the other. 18 routing cases green, and 135 tests across the touched `source` suites with checkstyle clean. -- 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]
