danny0405 commented on code in PR #19520: URL: https://github.com/apache/hudi/pull/19520#discussion_r3718919011
########## 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: `COW_INCREMENTAL` sets `READ_START_COMMIT` to `earliest` here, which makes `IncrementalInputSplits.inputSplits()` take its `fullTableScan` branch. As a result, this case does not exercise the non-full-scan bounded incremental path cited in the PR one-split-per-file-group safety argument. Could this use an actual completed commit (and, if needed, a bounded end commit) so that `fullTableScan` is false? A separate `earliest` case can remain if that fallback also needs coverage. -- 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]
