This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new f3a334eaa29c fix(flink): use safe log-reader high watermark for pre-8
MOR incremental reads (#19949)
f3a334eaa29c is described below
commit f3a334eaa29c1ade90b04322cbecf9aa16ba6aef
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]>
---
.../apache/hudi/source/IncrementalInputSplits.java | 13 +++-
.../apache/hudi/table/format/TestInputFormat.java | 71 ++++++++++++++++++++++
2 files changed, 81 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 68e81271d6c8..6609fb5a50ab 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;
@@ -440,6 +441,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())
@@ -447,9 +450,13 @@ 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);
+ // 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(), fileSlice.getPartitionPath());
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 68dd9f8052f7..713e160a98da 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
@@ -118,6 +118,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.common.util.StringUtils.fromUTF8Bytes;
import static org.apache.hudi.utils.TestData.insertRow;
@@ -984,6 +986,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(HoodieTableConfig.TABLE_STORAGE_LAYOUT.key(),
HoodieTableConfig.TableStorageLayout.DEFAULT.configValue());
+ 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.getPartitionPath().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<>();
