This is an automated email from the ASF dual-hosted git repository. danny0405 pushed a commit to branch release-1.1.2 in repository https://gitbox.apache.org/repos/asf/hudi.git
commit f2cc8768bd5541968599a9e914cc61ccab183cad Author: fhan <[email protected]> AuthorDate: Sat Sep 19 11:53:47 2026 +0800 fix(flink): use safe log-reader high watermark for pre-8 MOR incremental reads (#19949) * fix(flink): use safe log-reader high watermark for pre-8 MOR incremental reads --------- Co-authored-by: fhan <[email protected]> (cherry picked from commit f3a334eaa29c1ade90b04322cbecf9aa16ba6aef) --- .../apache/hudi/source/IncrementalInputSplits.java | 19 ++++-- .../apache/hudi/table/format/TestInputFormat.java | 72 ++++++++++++++++++++++ 2 files changed, 85 insertions(+), 6 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 c7dd6dcdf9b2..528d7b433ae3 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 @@ -24,6 +24,7 @@ import org.apache.hudi.common.model.HoodieCommitMetadata; import org.apache.hudi.common.model.HoodieFileGroupId; import org.apache.hudi.common.model.HoodieLogFile; import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.HoodieTableVersion; import org.apache.hudi.common.table.cdc.HoodieCDCExtractor; import org.apache.hudi.common.table.cdc.HoodieCDCFileSplit; import org.apache.hudi.common.table.cdc.HoodieCDCUtils; @@ -373,6 +374,8 @@ public class IncrementalInputSplits implements Serializable { InstantRange instantRange) { final AtomicInteger cnt = new AtomicInteger(0); final String mergeType = this.conf.get(FlinkOptions.MERGE_TYPE); + final boolean isPreV8 = metaClient.getTableConfig().getTableVersion() + .lesserThan(HoodieTableVersion.EIGHT); return fileSlices.stream().map(fileSlice -> { Option<List<String>> logPaths = Option.ofNullable(fileSlice.getLogFiles() .sorted(HoodieLogFile.getLogFileComparator()) @@ -380,12 +383,16 @@ public class IncrementalInputSplits implements Serializable { .filter(logPath -> !logPath.endsWith(HoodieCDCUtils.CDC_LOGFILE_SUFFIX)) .collect(Collectors.toList())); String basePath = fileSlice.getBaseFile().map(BaseFile::getPath).orElse(null); - // the latest commit is used as the limit of the log reader instant upper threshold, - // it must be at least the latest instant time of the file slice to avoid data loss. - String latestCommit = InstantComparison.minInstant(fileSlice.getLatestInstantTime(), endInstant); - return new MergeOnReadInputSplit(cnt.getAndAdd(1), - basePath, logPaths, latestCommit, - metaClient.getBasePath().toString(), maxCompactionMemoryInBytes, mergeType, instantRange, fileSlice.getFileId()); + // Pre-v8 log file names can carry the base instant while containing blocks from later + // commits, so the filename-derived file slice instant is not a safe reader upper bound. + // Keep the existing per-split upper bound for v8+ tables. The instant range remains the + // logical query boundary for both paths. + String latestCommit = isPreV8 + ? endInstant + : InstantComparison.minInstant(fileSlice.getLatestInstantTime(), endInstant); + return new MergeOnReadInputSplit(cnt.getAndAdd(1), basePath, logPaths, latestCommit, + metaClient.getBasePath().toString(), maxCompactionMemoryInBytes, mergeType, instantRange, + fileSlice.getFileId()); }).sorted(Comparator.comparing(MergeOnReadInputSplit::getLatestCommit)).collect(Collectors.toList()); } diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/format/TestInputFormat.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/format/TestInputFormat.java index fc4694bd3a6f..edfdff7b4e21 100644 --- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/format/TestInputFormat.java +++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/format/TestInputFormat.java @@ -24,6 +24,7 @@ import org.apache.hudi.client.model.PartialUpdateFlinkRecordMerger; import org.apache.hudi.common.config.RecordMergeMode; import org.apache.hudi.common.model.EventTimeAvroPayload; import org.apache.hudi.common.model.HoodieCommitMetadata; +import org.apache.hudi.common.model.HoodieLogFile; import org.apache.hudi.common.model.HoodieTableType; import org.apache.hudi.common.model.OverwriteWithLatestAvroPayload; import org.apache.hudi.common.model.PartialUpdateAvroPayload; @@ -91,6 +92,8 @@ import java.util.Map; import java.util.stream.Collectors; import java.util.stream.Stream; +import static org.apache.hudi.common.table.timeline.InstantComparison.LESSER_THAN; +import static org.apache.hudi.common.table.timeline.InstantComparison.compareTimestamps; import static org.apache.hudi.common.testutils.HoodieTestUtils.INSTANT_FILE_NAME_GENERATOR; import static org.apache.hudi.utils.TestData.insertRow; import static org.hamcrest.CoreMatchers.instanceOf; @@ -897,6 +900,75 @@ public class TestInputFormat { TestData.assertRowDataEquals(actual6, Collections.emptyList()); } + @Test + void testReadArchivedCommitsIncrementallyForPreEightMOR() throws Exception { + Map<String, String> options = new HashMap<>(); + options.put(FlinkOptions.QUERY_TYPE.key(), FlinkOptions.QUERY_TYPE_INCREMENTAL); + options.put(FlinkOptions.WRITE_TABLE_VERSION.key(), String.valueOf(HoodieTableVersion.SIX.versionCode())); + options.put(FlinkOptions.ARCHIVE_MIN_COMMITS.key(), "3"); + options.put(FlinkOptions.ARCHIVE_MAX_COMMITS.key(), "4"); + options.put(FlinkOptions.CLEAN_RETAIN_COMMITS.key(), "2"); + options.put(FlinkOptions.METADATA_ENABLED.key(), "false"); + options.put("hoodie.commits.archival.batch", "1"); + beforeEach(HoodieTableType.MERGE_ON_READ, options); + + // The second commit updates the same pre-v8 MOR file group. Its log block is appended + // to a log file whose filename still carries the first commit instant. + for (int i = 1; i <= 2; i++) { + TestData.writeData(Collections.singletonList(insertRow( + StringData.fromString("id1"), StringData.fromString("Danny"), 20 + i, + TimestampData.fromEpochMillis(i), StringData.fromString("par1"))), conf); + } + // Generate unrelated commits to archive the two commits under test without advancing + // the file slice containing id1. + for (int i = 3; i <= 8; i++) { + TestData.writeData(Collections.singletonList(insertRow( + StringData.fromString("id" + i), StringData.fromString("User" + i), 20 + i, + TimestampData.fromEpochMillis(i), StringData.fromString("par2"))), conf); + } + + HoodieTableMetaClient metaClient = StreamerUtil.createMetaClient(conf); + assertSame(HoodieTableVersion.SIX, metaClient.getTableConfig().getTableVersion()); + List<String> archivedCommits = metaClient.getArchivedTimeline().getCommitsTimeline().filterCompletedInstants() + .getInstantsAsStream().map(HoodieInstant::requestedTime).collect(Collectors.toList()); + assertTrue(archivedCommits.size() >= 2); + + String startCommit = archivedCommits.get(0); + String endCommit = archivedCommits.get(1); + conf.set(FlinkOptions.READ_START_COMMIT, startCommit); + conf.set(FlinkOptions.READ_END_COMMIT, endCommit); + this.tableSource = getTableSource(conf); + InputFormat<RowData, ?> inputFormat = this.tableSource.getInputFormat(); + assertThat(inputFormat, instanceOf(MergeOnReadInputFormat.class)); + + IncrementalInputSplits incrementalInputSplits = IncrementalInputSplits.builder() + .rowType(TestConfigurations.ROW_TYPE) + .conf(conf) + .path(FilePathUtils.toFlinkPath(metaClient.getBasePath())) + .build(); + IncrementalInputSplits.Result splits = incrementalInputSplits.inputSplits(metaClient, false); + assertFalse(splits.isEmpty()); + MergeOnReadInputSplit split = splits.getInputSplits().stream() + .filter(inputSplit -> inputSplit.getLogPaths().orElse(Collections.emptyList()).stream() + .anyMatch(logPath -> new StoragePath(logPath).getParent().getName().equals("par1"))) + .findFirst() + .orElseThrow(() -> new AssertionError("No input split found for par1")); + assertTrue(split.getLogPaths().isPresent()); + String latestLogFileInstant = split.getLogPaths().get().stream() + .map(logPath -> new HoodieLogFile(new StoragePath(logPath)).getDeltaCommitTime()) + .max(String::compareTo) + .orElseThrow(() -> new AssertionError("No log file found for par1")); + assertTrue(compareTimestamps(latestLogFileInstant, LESSER_THAN, endCommit), + "The in-range update must be newer than the filename-derived log instant"); + + List<RowData> actual = readData(inputFormat, + splits.getInputSplits().toArray(new MergeOnReadInputSplit[0])); + List<RowData> expected = Collections.singletonList(insertRow( + StringData.fromString("id1"), StringData.fromString("Danny"), 22, + TimestampData.fromEpochMillis(2), StringData.fromString("par1"))); + TestData.assertRowDataEquals(actual, expected); + } + @Test void testReadChangelogIncrementally() throws Exception { Map<String, String> options = new HashMap<>();
