feat: support watermark emitter from split info #18040

HuangZhenQiu · 2026-01-29T12:55:51Z

Describe the issue this Pull Request addresses

support watermark emitter from split info

closes #14424

Summary and Changelog

add the watermark emit by looking up the base file path in HoodieRecordEmitter
add test coverage to test logic change

Impact

none

Risk Level

none

Documentation Update

none

Contributor's checklist

Read through contributor's guide
Enough context is provided in the sections above
Adequate tests were added if applicable

xushiyan

Some minor issues

xushiyan · 2026-01-30T00:36:40Z

...k-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/HoodieRecordEmitter.java

+
+  private long extractWatermark(HoodieSourceSplit split) {
+    // CDC split will be handled later
+    ValidationUtils.checkArgument(split.getBasePath().isPresent(), "Split base path can't be null.");


this check assumes no base path, while the comment implies it's CDC split. Should this check be a silent skip instead of an exception? Or should CDC splits be explicitly excluded upstream?

Consider adding explicit handling for CDC splits (even if just logging and skipping watermark emission)

If we don't include CDC read in this release, shall we keep it as it is?

for streaming read, the split base path could also be null.

xushiyan · 2026-01-30T00:37:03Z

...k-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/HoodieRecordEmitter.java

+  private long extractWatermark(HoodieSourceSplit split) {
+    // CDC split will be handled later
+    ValidationUtils.checkArgument(split.getBasePath().isPresent(), "Split base path can't be null.");
+    return Long.parseLong(FSUtils.getCommitTime(split.getBasePath().get()));


HoodieSourceSplit always has latestCommit, shouldn't it be used directly?

The latestCommit in HoodieSourceSplit is the lastest commit when scans completed commits since the last scan time. The file level commit time is in a lower granularity and introduces a more accurate of watermark.

@HuangZhenQiu i'm not sure i understand how this is scan-level latest commit. HoodieSourceSplit should be pointing to a file slice, so its latestCommit should be effectively the same as the latest commit from all files (base and log files) within the file slice

xushiyan · 2026-01-30T00:38:57Z

...k-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/HoodieRecordEmitter.java

+      long newWatermark = extractWatermark(split);
+      if (newWatermark < watermark) {
+        LOG.info(
+            "Received a new split with lower watermark. Previous watermark = {}, current watermark = {}, previous split = {}, current split = {}",


wouldn't this info level log be too noisy? looks more like debug msg

It actually shouldn't happen as we sorted split by the file commit instant time. Changed to warn.

why use the base instant time as the watermark instead of the latest commit in the split?

If so, all of the split discovered by the one time scan will emit the same watermark, right? But there could be multiple commits before the lastest commit are returned from scan. In this case, will not base instant time be more accurate?

split is discovered by timeline commits, my guess is for each split, it reports the latest instant time and the Flink would finally figure out how to allgin them then send to downsteam?

BTW, if there is no base file in the split, the base instant time is the smallest instant from the log files.

Thanks for the hint. How to find the instant from the log file? Is there a log file name pattern? If a split has both base file and log files, It should also be the the smallest instant from the log files right?

It should also be the the smallest instant from the log files right

does Flink has the contract that require split to return the min or max instant? if it is min, then it should be the smallest instant time of all files, if it is max, it should be the max instant time of all files.

We can always decode the instant time using FsUtils.

Given the split watermark is emitted after the split is processed, then it should be the max instant time of all files.

...tasource/hudi-flink/src/test/java/org/apache/hudi/source/reader/TestHoodieRecordEmitter.java

xushiyan · 2026-01-30T01:07:18Z

Pls also link issues, if this solves the issue, then use closing key words, if it's part of the issue, then just mention the it

xushiyan · 2026-01-31T02:20:23Z

...k-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/HoodieRecordEmitter.java

+    }
+
    output.collect(record.record());
    split.updatePosition(record.fileOffset(), record.recordOffset());


shouldn't the watermark be only emitted after records processed? meaning emission happens after this line?

This watermark is for split finish not for per record. Thus, it should emit after a split is processed.

xushiyan · 2026-01-31T02:39:46Z

...k-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/HoodieRecordEmitter.java

+  private long extractWatermark(HoodieSourceSplit split) {
+    // CDC split will be handled later
+    ValidationUtils.checkArgument(split.getBasePath().isPresent(), "Split base path can't be null.");
+    return Long.parseLong(FSUtils.getCommitTime(split.getBasePath().get()));


@HuangZhenQiu i'm not sure i understand how this is scan-level latest commit. HoodieSourceSplit should be pointing to a file slice, so its latestCommit should be effectively the same as the latest commit from all files (base and log files) within the file slice

xushiyan · 2026-01-31T02:41:34Z

...k-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/HoodieRecordEmitter.java

+      }
+    }
+
+    return maxInstantTime;


HoodieSourceSplit should have maintain the order of log files which have the latest log file that can be accessed directly, if no log file, then just get base file's commit time as the latest. So this extraction can be O(1). But, as i commented on the other thread, shouldn't HoodieSourceSplit#latestCommit already be serving this info?

latestCommit is not the last commit of the split. It is the latest commit when we fetch new commits from timeline. Thus, there could splits committed in different instant time have the same latestCommit time.

https://github.com/apache/hudi/blob/master/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/IncrementalInputSplits.java#L409

hudi-bot · 2026-01-31T03:43:18Z

CI report:

675b4b6 Azure: SUCCESS
152e5bf Azure: PENDING
a131124 UNKNOWN

Bot commands

@hudi-bot supports the following commands:

@hudi-bot run azure re-run the last Azure build

github-actions bot added the size:M PR with lines of changes in (100, 300] label Jan 29, 2026

xushiyan reviewed Jan 30, 2026

View reviewed changes

xushiyan requested a review from cshuo January 30, 2026 05:01

HuangZhenQiu force-pushed the watermark-record-emitter branch 2 times, most recently from e0e73f1 to 2622251 Compare January 30, 2026 07:06

github-actions bot added size:L PR with lines of changes in (300, 1000] and removed size:M PR with lines of changes in (100, 300] labels Jan 30, 2026

HuangZhenQiu force-pushed the watermark-record-emitter branch from 2622251 to 675b4b6 Compare January 30, 2026 07:28

xushiyan reviewed Jan 31, 2026

View reviewed changes

HuangZhenQiu force-pushed the watermark-record-emitter branch from 675b4b6 to 152e5bf Compare January 31, 2026 03:34

feat: support watermark emitter from split info

a131124

HuangZhenQiu force-pushed the watermark-record-emitter branch from 152e5bf to a131124 Compare January 31, 2026 03:41

feat: support watermark emitter from split info #18040

Are you sure you want to change the base?

feat: support watermark emitter from split info #18040

Conversation

HuangZhenQiu commented Jan 29, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Describe the issue this Pull Request addresses

Summary and Changelog

Impact

Risk Level

Documentation Update

Contributor's checklist

Uh oh!

xushiyan left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

HuangZhenQiu Jan 30, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

HuangZhenQiu Jan 30, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

danny0405 Jan 30, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

xushiyan commented Jan 30, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hudi-bot commented Jan 31, 2026

CI report:

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

HuangZhenQiu commented Jan 29, 2026 •

edited

Loading

HuangZhenQiu Jan 30, 2026 •

edited

Loading

HuangZhenQiu Jan 30, 2026 •

edited

Loading

danny0405 Jan 30, 2026 •

edited

Loading

xushiyan commented Jan 30, 2026 •

edited

Loading