This is an automated email from the ASF dual-hosted git repository. voonhous pushed a commit to branch release-1.2.1 in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 0cd87c4367b714f14fb034bb3b1d4964319da503 Author: fhan <[email protected]> AuthorDate: Mon Jun 1 17:13:07 2026 +0800 fix(flink): fix data loss in stream read from earliest (#18848) * fix(flink): fix data loss in stream read from earliest * fix(flink): optimize UTs and refine de-duplicate full-table-scan timeline comment --------- Co-authored-by: fhan <[email protected]> (cherry picked from commit 4e5034d354628bab348a55d694171632a0c7c2a0) --- .../apache/hudi/source/IncrementalInputSplits.java | 35 ++- .../apache/hudi/table/ITTestHoodieDataSource.java | 281 +++++++++++++++++++++ 2 files changed, 313 insertions(+), 3 deletions(-) diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/IncrementalInputSplits.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/IncrementalInputSplits.java index bc6ebd29da21..344b24bc10c4 100644 --- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/IncrementalInputSplits.java +++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/IncrementalInputSplits.java @@ -184,7 +184,12 @@ public class IncrementalInputSplits implements Serializable { return Result.EMPTY; } fileInfoList = fileIndex.getFilesInPartitions(); - List<FileSlice> allFileSlices = getFileSlices(metaClient, commitTimeline, readPartitions, fileInfoList, analyzingResult.getMaxCompletionTime(), false); + // Use the full commits-and-compaction timeline rather than the (possibly compaction-filtered) + // activeTimeline carried by the QueryContext. Otherwise, on a MOR table with + // 'read.streaming.skip_compaction = true', file slice boundaries would be wrongly + // computed and log files could be missed, causing data loss. + List<FileSlice> allFileSlices = getFileSlices(metaClient, getFullCommitsTimeline(metaClient), + readPartitions, fileInfoList, analyzingResult.getMaxCompletionTime(), false); fileSlices = fileIndex.filterFileSlices(allFileSlices); } else { if (cdcEnabled) { @@ -217,7 +222,11 @@ public class IncrementalInputSplits implements Serializable { return Result.EMPTY; } fileInfoList = fileIndex.getFilesInPartitions(); - List<FileSlice> allFileSlices = getFileSlices(metaClient, commitTimeline, readPartitions, fileInfoList, analyzingResult.getMaxCompletionTime(), false); + // Same reason as the full-table-scan branch above: build the FileSystemView with the + // complete commits-and-compaction timeline to avoid losing data when 'skip_compaction' + // is enabled. + List<FileSlice> allFileSlices = getFileSlices(metaClient, getFullCommitsTimeline(metaClient), + readPartitions, fileInfoList, analyzingResult.getMaxCompletionTime(), false); fileSlices = fileIndex.filterFileSlices(allFileSlices); } else { fileSlices = getFileSlices(metaClient, commitTimeline, readPartitions, files, analyzingResult.getMaxCompletionTime(), false); @@ -295,7 +304,10 @@ public class IncrementalInputSplits implements Serializable { log.warn("No files found for reading under path: {}", path); return Result.EMPTY; } - List<FileSlice> allFileSlices = getFileSlices(metaClient, commitTimeline, readPartitions, pathInfoList, offsetToIssue, false); + // Same reason as the batch full-table-scan branch: + // see getFullCommitsTimeline() for why a compaction-filtered timeline must not be used here. + List<FileSlice> allFileSlices = getFileSlices(metaClient, getFullCommitsTimeline(metaClient), + readPartitions, pathInfoList, offsetToIssue, false); List<FileSlice> fileSlices = fileIndex.filterFileSlices(allFileSlices); List<MergeOnReadInputSplit> inputSplits = getInputSplits(fileSlices, metaClient, endInstant, null); @@ -391,6 +403,23 @@ public class IncrementalInputSplits implements Serializable { return getInputSplits(fileSlices, metaClient, endInstant, instantRange); } + /** + * Returns the full commit timeline (including completed compaction instants) for building + * a {@link HoodieTableFileSystemView} during full table scan. + * + * <p>NOTE: when streaming/batch read enables {@code skip_compaction}, the {@code activeTimeline} + * carried by {@link IncrementalQueryAnalyzer.QueryContext} has already filtered out the + * compaction instants. Using such a partial timeline to construct a {@link HoodieTableFileSystemView} + * would mis-classify the file slice boundaries on a MOR table (since file slice boundaries + * are derived from compaction instants), leading to data loss when reading from the earliest + * or after start commit got archived. For full table scan we should always rely on the + * complete commits-and-compaction timeline; the {@code skip_compaction} semantics is preserved + * by the instant range filtering applied later on the generated input splits. + */ + private static HoodieTimeline getFullCommitsTimeline(HoodieTableMetaClient metaClient) { + return metaClient.getCommitsAndCompactionTimeline().filterCompletedAndCompactionInstants(); + } + private List<FileSlice> getFileSlices( HoodieTableMetaClient metaClient, HoodieTimeline commitTimeline, diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java index 1cc5beaad418..4af3598c0a97 100644 --- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java +++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java @@ -28,16 +28,21 @@ import org.apache.hudi.common.table.HoodieTableMetaClient; import org.apache.hudi.common.table.HoodieTableVersion; import org.apache.hudi.common.table.cdc.HoodieCDCSupplementalLoggingMode; import org.apache.hudi.common.table.marker.MarkerType; +import org.apache.hudi.common.table.timeline.HoodieInstant; import org.apache.hudi.common.table.timeline.HoodieTimeline; +import org.apache.hudi.common.table.timeline.TimelineUtils; +import org.apache.hudi.common.testutils.HoodieTestUtils; import org.apache.hudi.common.util.CollectionUtils; import org.apache.hudi.config.HoodieWriteConfig; import org.apache.hudi.configuration.FlinkOptions; +import org.apache.hudi.configuration.HadoopConfigurations; import org.apache.hudi.index.HoodieIndex; import org.apache.hudi.index.bucket.partition.PartitionBucketIndexUtils; import org.apache.hudi.metadata.HoodieTableMetadata; import org.apache.hudi.sink.buffer.BufferMemoryType; import org.apache.hudi.sink.buffer.BufferType; import org.apache.hudi.storage.StoragePath; +import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration; import org.apache.hudi.table.catalog.HoodieCatalogTestUtils; import org.apache.hudi.table.catalog.HoodieHiveCatalog; import org.apache.hudi.util.StreamerUtil; @@ -89,6 +94,8 @@ import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.Set; +import java.util.TreeSet; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; @@ -712,6 +719,280 @@ public class ITTestHoodieDataSource { assertRowsEquals(rows, TestData.DATA_SET_SOURCE_INSERT); } + /** + * Regression test for HUDI: data loss in stream read from earliest when + * {@code read.streaming.skip_compaction = true} on a MOR table with completed + * compaction commits. Covers the streaming earliest full table scan branch in + * {@link org.apache.hudi.source.IncrementalInputSplits#inputSplits(HoodieTableMetaClient, String, boolean)}. + * + * <p>Triggering condition: + * <ul> + * <li>{@code read.start-commit = earliest} (no instant range -> full table scan path);</li> + * <li>{@code read.streaming.skip_compaction = true} (active timeline filtered out compaction);</li> + * <li>MOR table with at least one completed compaction commit that produced a + * new base file from existing log files.</li> + * </ul> + * + * <p>Construction: + * <ol> + * <li>Offline write {@code DATA_SET_INSERT} (8 records, ids 1..8) and then + * {@code DATA_SET_UPDATE_INSERT} (8 records, where ids 1..5 update existing keys + * and ids 9..11 are new) via {@link TestData#writeDataAsBatch}, which deterministically + * triggers an inline compaction once {@code COMPACTION_DELTA_COMMITS = 1} + + * {@code COMPACTION_ASYNC_ENABLED = true} are set. After this step the table has + * both a base file (from compaction) and log files written by the UPDATE batch.</li> + * <li>Streaming read from earliest with {@code skip_compaction = true} and wait until + * the expected number of merged rows are received. Without the fix, the FS view used + * in the earliest full-table-scan branch is built from a compaction-filtered + * timeline, file slice boundaries are wrongly computed, log files are missed + * and the read will never reach the expected row count (the test would time out).</li> + * </ol> + */ + @ParameterizedTest + @ValueSource(booleans = {true, false}) + void testStreamReadMorTableWithCompactionFromEarliest(boolean useSourceV2) throws Exception { + Configuration conf = TestConfigurations.getDefaultConf(tempFile.getAbsolutePath()); + conf.set(FlinkOptions.TABLE_NAME, "t1"); + conf.set(FlinkOptions.TABLE_TYPE, MERGE_ON_READ.name()); + conf.set(FlinkOptions.INDEX_TYPE, HoodieIndex.IndexType.BUCKET.name()); + conf.set(FlinkOptions.BUCKET_INDEX_NUM_BUCKETS, 2); + // mandatory for writeDataAsBatch#inlineCompaction to actually run a compaction + conf.set(FlinkOptions.COMPACTION_ASYNC_ENABLED, true); + conf.set(FlinkOptions.COMPACTION_DELTA_COMMITS, 1); + + // Step 1: offline-write two batches with deterministic inline compaction in between. + TestData.writeDataAsBatch(TestData.DATA_SET_INSERT, conf); + TestData.writeDataAsBatch(TestData.DATA_SET_UPDATE_INSERT, conf); + + // Step 2: streaming read from earliest with skip_compaction = true. + String hoodieTableDDL = sql("t1") + .option(FlinkOptions.PATH, tempFile.getAbsolutePath()) + .options(getDefaultKeys()) + .option(FlinkOptions.TABLE_TYPE, MERGE_ON_READ) + .option(FlinkOptions.INDEX_TYPE, HoodieIndex.IndexType.BUCKET.name()) + .option(FlinkOptions.BUCKET_INDEX_NUM_BUCKETS, 2) + .option(FlinkOptions.READ_AS_STREAMING, true) + .option(FlinkOptions.READ_START_COMMIT, FlinkOptions.START_COMMIT_EARLIEST) + .option(FlinkOptions.READ_STREAMING_CHECK_INTERVAL, 2) + .option(FlinkOptions.READ_SOURCE_V2_ENABLED, useSourceV2) + // skip compaction instant -> active timeline drops compaction commit + .option(FlinkOptions.READ_STREAMING_SKIP_COMPACT, true) + .end(); + streamTableEnv.executeSql(hoodieTableDDL); + + // After the UPDATE batch, the merged result must contain all up-to-date records: + // - 5 updated records (id1..id5 from DATA_SET_UPDATE_INSERT) + // - 3 carried-over records (id6, id7, id8 from DATA_SET_INSERT, not touched by UPDATE) + // - 3 newly inserted records (id9, id10, id11 from DATA_SET_UPDATE_INSERT) + // i.e. 11 records in total. Without the fix the streaming read would never reach + // expectedNum = 11 and the test would time out via the CollectSink. + final int expectedNum = 11; + List<Row> rows = execSelectSqlWithExpectedNum(streamTableEnv, "select * from t1", expectedNum); + assertEquals(expectedNum, rows.size(), + "Expect 11 up-to-date records to be visible after earliest streaming read" + + " with skip_compaction on a MOR table that has a completed compaction commit" + + ", actual rows: " + rows); + } + + /** + * Regression test for HUDI: data loss in batch read from earliest when + * {@code read.streaming.skip_compaction = true} on a MOR table with completed + * compaction commits. Covers the batch full-table-scan branch in + * {@link org.apache.hudi.source.IncrementalInputSplits#inputSplits(HoodieTableMetaClient, boolean)}. + * + * <p>This complements {@link #testStreamReadMorTableWithCompactionFromEarliest(boolean)} + * which only exercises the streaming code path. Without the fix, building the + * {@link org.apache.hudi.common.table.view.HoodieTableFileSystemView} with a + * compaction-filtered timeline would mis-classify file slice boundaries and + * lose log files. + */ + @Test + void testBatchReadMorTableWithCompactionFromEarliest() throws Exception { + Configuration conf = TestConfigurations.getDefaultConf(tempFile.getAbsolutePath()); + conf.set(FlinkOptions.TABLE_NAME, "t1"); + conf.set(FlinkOptions.TABLE_TYPE, MERGE_ON_READ.name()); + conf.set(FlinkOptions.INDEX_TYPE, HoodieIndex.IndexType.BUCKET.name()); + conf.set(FlinkOptions.BUCKET_INDEX_NUM_BUCKETS, 2); + // mandatory for writeDataAsBatch#inlineCompaction to actually run a compaction + conf.set(FlinkOptions.COMPACTION_ASYNC_ENABLED, true); + conf.set(FlinkOptions.COMPACTION_DELTA_COMMITS, 1); + + // Offline-write two batches against overlapping record keys, the 2nd write triggers + // an inline compaction that merges existing log files into a new base file - exactly + // the scenario that exposes the buggy file-slice classification when skip_compaction + // is enabled. + TestData.writeDataAsBatch(TestData.DATA_SET_INSERT, conf); + TestData.writeDataAsBatch(TestData.DATA_SET_UPDATE_INSERT, conf); + + String hoodieTableDDL = sql("t1") + .option(FlinkOptions.PATH, tempFile.getAbsolutePath()) + .options(getDefaultKeys()) + .option(FlinkOptions.TABLE_TYPE, MERGE_ON_READ) + .option(FlinkOptions.INDEX_TYPE, HoodieIndex.IndexType.BUCKET.name()) + .option(FlinkOptions.BUCKET_INDEX_NUM_BUCKETS, 2) + .option(FlinkOptions.READ_START_COMMIT, FlinkOptions.START_COMMIT_EARLIEST) + // skip compaction instant -> active timeline drops compaction commit + .option(FlinkOptions.READ_STREAMING_SKIP_COMPACT, true) + .end(); + batchTableEnv.executeSql(hoodieTableDDL); + + List<Row> result = CollectionUtil.iteratorToList( + batchTableEnv.executeSql("select * from t1").collect()); + // After update, the merged result must contain all up-to-date records: + // - 5 updated records (id1..id5 from DATA_SET_UPDATE_INSERT) + // - 3 carried-over records (id6, id7, id8 from DATA_SET_INSERT, not touched by UPDATE) + // - 3 newly inserted records (id9, id10, id11 from DATA_SET_UPDATE_INSERT) + // i.e. 11 records in total. Without the fix, log files belonging to the file slice prior + // to the inline compaction would be silently dropped by the file system view because + // the active timeline filtered out the compaction commit, and the result size would be + // smaller than 11. + assertEquals(11, result.size(), + "Expect all up-to-date records to be visible after earliest + skip_compaction batch read" + + ", actual rows: " + result); + } + + /** + * Regression test for HUDI: data loss when the start commit has been archived + * and {@code read.streaming.skip_compaction = true} on a MOR table. + * Covers the batch "fallback to full table scan" branch in + * {@link org.apache.hudi.source.IncrementalInputSplits#inputSplits(HoodieTableMetaClient, boolean)} + * which is reached when {@code hasArchivedInstants == true}. + * + * <p>Construction: + * <ol> + * <li>Write 10 delta-commit batches of {@code (id1,id2), (id3,id4), ...} on a MOR table + * so that each batch only inserts new keys (clear, predictable per-commit semantics).</li> + * <li>Trigger one completed compaction commit by issuing an extra UPDATE batch on + * {@code id1..id4} with {@code COMPACTION_DELTA_COMMITS = 1} via + * {@link TestData#writeDataAsBatch} (which explicitly calls {@code inlineCompaction()}). + * This creates exactly the file-slice boundary that the buggy FS view would + * mis-classify.</li> + * <li>Pick the LAST archived delta-commit as {@code read.start-commit} (filtered by + * {@code action = deltacommit} to exclude any archived compaction {@code commit} + * instants). This is deterministic regardless of how many delta commits were + * archived by the cleaner+archiver and routes the reader through the + * "archived start commit -> fullTableScan" branch.</li> + * <li>Read with {@code skip_compaction = true} and assert on the SET of record-keys + * in the result (not just on count). The expected key set is derived dynamically + * from the timeline: every delta_commit whose completion time is >= the chosen + * start_commit contributes its written ids, plus id1..id4 from the UPDATE batch + * are always present because the UPDATE is the latest write. Without the fix, + * log files of the file slice that straddles the compaction commit are silently + * dropped, so some of these ids would be missing.</li> + * </ol> + */ + @Test + void testBatchReadMorTableWithCompactionStartCommitArchived() throws Exception { + Configuration conf = TestConfigurations.getDefaultConf(tempFile.getAbsolutePath()); + conf.set(FlinkOptions.TABLE_NAME, "t1"); + conf.set(FlinkOptions.RECORD_KEY_FIELD, "uuid"); + conf.set(FlinkOptions.ORDERING_FIELDS, "ts"); + conf.set(FlinkOptions.TABLE_TYPE, MERGE_ON_READ.name()); + conf.set(FlinkOptions.INDEX_TYPE, HoodieIndex.IndexType.BUCKET.name()); + conf.set(FlinkOptions.BUCKET_INDEX_NUM_BUCKETS, 2); + // aggressive archival to force older instants out of the active timeline + conf.set(FlinkOptions.ARCHIVE_MIN_COMMITS, 4); + conf.set(FlinkOptions.ARCHIVE_MAX_COMMITS, 5); + conf.set(FlinkOptions.CLEAN_RETAIN_COMMITS, 3); + conf.setString("hoodie.commits.archival.batch", "1"); + + // Step 1: write 10 batches of 2 new records each -> 10 delta_commit instants, 20 distinct keys. + for (int i = 0; i < 20; i += 2) { + List<RowData> dataset = TestData.dataSetInsert(i + 1, i + 2); + TestData.writeData(dataset, conf); + } + + // Step 2: trigger at least one completed compaction commit by issuing one more delta_commit + // that UPDATES the very first record keys (id1..id4) and enabling COMPACTION_DELTA_COMMITS=1. + // The update writes new log files for the file group that contains id1..id4, and the inline + // compaction merges them into a new base file -> a real compaction file-slice boundary. + // NOTE: use writeDataAsBatch (which explicitly calls inlineCompaction()), since the plain + // writeData helper does not run the compaction even with COMPACTION_DELTA_COMMITS=1. + conf.set(FlinkOptions.COMPACTION_ASYNC_ENABLED, true); + conf.set(FlinkOptions.COMPACTION_DELTA_COMMITS, 1); + TestData.writeDataAsBatch(TestData.dataSetInsert(1, 2, 3, 4), conf); + + // Step 3: list the full timeline in one shot to map start_commit -> expected id set. + // Delta_commit instants are strictly monotonically increasing, so the sorted list of all + // delta_commits across active + archived timelines gives a 1:1 mapping to the 10 batches + // written in Step 1: the k-th delta_commit wrote id_{2k+1} and id_{2k+2}. + HoodieTableMetaClient metaClient = HoodieTestUtils.createMetaClient( + new HadoopStorageConfiguration(HadoopConfigurations.getHadoopConf(new Configuration())), + tempFile.getAbsolutePath()); + // Use the merged (archived + active) timeline to capture all delta_commits, + // even those that may have been archived by the aggressive archival settings. + List<String> batchInstantTimes = TimelineUtils.getTimeline(metaClient, true) + .getCommitsTimeline().filterCompletedInstants() + .filter(instant -> HoodieTimeline.DELTA_COMMIT_ACTION.equals(instant.getAction())) + .getInstantsAsStream().map(HoodieInstant::requestedTime).collect(Collectors.toList()); + // Step 1 produced exactly 10 delta_commits; the 11th (if present) is from Step 2 UPDATE. + // Keep only the first 10 to build the batch-index -> id mapping. + assertTrue(batchInstantTimes.size() >= 10, + "Expected at least 10 delta_commits from Step 1, got " + batchInstantTimes.size()); + batchInstantTimes = batchInstantTimes.subList(0, 10); + + // Step 4: pick the LAST archived delta_commit that belongs to Step 1's batches as + // start commit. This avoids any drift caused by archival ordering or by compaction + // `commit` instants being interleaved with delta_commits in the archived timeline, + // and also ignores the Step 2 UPDATE batch in case it also got archived. + Set<String> step1InstantTimeSet = new TreeSet<>(batchInstantTimes); + List<HoodieInstant> archivedDeltaCommits = metaClient.getArchivedTimeline().getCommitsTimeline() + .filterCompletedInstants() + .filter(instant -> HoodieTimeline.DELTA_COMMIT_ACTION.equals(instant.getAction())) + .filter(instant -> step1InstantTimeSet.contains(instant.requestedTime())) + .getInstants(); + // make sure archival actually happened on Step 1's batches, otherwise the test premise + // (the reader hits the archived start commit + fullTableScan branch) does not hold. + assertTrue(!archivedDeltaCommits.isEmpty(), + "archival did not happen as expected on Step 1's batches, archived delta commits = " + + archivedDeltaCommits + ", Step 1 batch instant times = " + batchInstantTimes); + HoodieInstant startInstant = archivedDeltaCommits.get(archivedDeltaCommits.size() - 1); + String archivedStartInstant = startInstant.requestedTime(); + + // The expected key set: every Step 1 batch whose instant time is >= start_commit contributes + // its 2 ids; plus id1..id4 from the Step 2 UPDATE batch (always the latest write, never + // excluded since its completion time is the largest). + int firstIncludedBatchIdx = batchInstantTimes.indexOf(archivedStartInstant); + assertTrue(firstIncludedBatchIdx >= 0, + "chosen start_commit " + archivedStartInstant + " is not one of the Step 1 batch instant times " + batchInstantTimes); + Set<String> expectedIds = new TreeSet<>(); + for (int i = firstIncludedBatchIdx; i < batchInstantTimes.size(); i++) { + expectedIds.add("id" + (2 * i + 1)); + expectedIds.add("id" + (2 * i + 2)); + } + // UPDATE batch ids — always present in the merged view because the UPDATE is the latest write. + expectedIds.add("id1"); + expectedIds.add("id2"); + expectedIds.add("id3"); + expectedIds.add("id4"); + + String hoodieTableDDL = sql("t1") + .option(FlinkOptions.PATH, tempFile.getAbsolutePath()) + .options(getDefaultKeys()) + .option(FlinkOptions.TABLE_TYPE, MERGE_ON_READ) + .option(FlinkOptions.INDEX_TYPE, HoodieIndex.IndexType.BUCKET.name()) + .option(FlinkOptions.BUCKET_INDEX_NUM_BUCKETS, 2) + .option(FlinkOptions.READ_START_COMMIT, archivedStartInstant) + .option(FlinkOptions.READ_STREAMING_SKIP_COMPACT, true) + .end(); + batchTableEnv.executeSql(hoodieTableDDL); + + List<Row> result = CollectionUtil.iteratorToList( + batchTableEnv.executeSql("select uuid from t1").collect()); + Set<String> actualIds = new TreeSet<>(); + for (Row r : result) { + actualIds.add(r.getField(0).toString()); + } + // Without the fix, the FS view used to construct file slices for the fallback full-table-scan + // branch is built from a compaction-filtered timeline, so log files of the file slice that + // straddles the compaction commit are silently dropped and some ids would be missing from + // {@code actualIds}. With the fix, every expected id must be present. + assertEquals(expectedIds, actualIds, + "Expected id set " + expectedIds + " but got " + actualIds + + " when reading from archived start commit " + archivedStartInstant + + " with skip_compaction = true on a MOR table that has a completed compaction commit"); + } + @ParameterizedTest @ValueSource(booleans = {true, false}) void testStreamReadMorTableWithBucketIndex(boolean partitioned) throws Exception {
