feat: add flink continuous split enumerator

[HuangZhenQiu]

Describe the issue this Pull Request addresses

Add HoodieContinuousSplitEnumerator for reading Hudi table incrementally

Summary and Changelog

Add HoodieContinuousSplitEnumerator, DefaultHoodieContinuousSplitDiscover and HoodieContinuousSplitBatch classes

Impact

none

Risk Level

none

Documentation Update

Contributor's checklist

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

[HuangZhenQiu]

@xushiyan Please review the PR at your most convenient time.

[Copilot]

Pull request overview

This PR adds support for continuous/incremental reading of Hudi tables in Flink by introducing a new HoodieContinuousSplitEnumerator that can discover new splits from the Hudi timeline as commits occur.

Key Changes:

• Introduces continuous split enumeration infrastructure for streaming incremental reads
• Adds position tracking to maintain enumeration state across checkpoints
• Provides a configurable scan context to encapsulate scanning parameters

Reviewed changes

Copilot reviewed 15 out of 15 changed files in this pull request and generated 9 comments.

Show a summary per file

File	Description
HoodieContinuousSplitEnumerator.java	Main enumerator implementation that periodically discovers new splits from Hudi commits
HoodieContinuousSplitBatch.java	Data class representing a batch of splits with instant range information
HoodieContinuousSplitDiscover.java	Interface for split discovery operations
DefaultHoodieSplitDiscover.java	Default implementation that delegates to IncrementalInputSplits
HoodieEnumeratorPosition.java	Tracks the last enumerated instant for incremental discovery
ScanContext.java	Encapsulates scan configuration including paths, instants, and skip options
HoodieSourceSplit.java	Adds static SPLIT_COUNTER for generating unique split IDs
HoodieSplitEnumeratorState.java	Extended to include last enumerated instant tracking
IncrementalInputSplits.java	Adds inputHoodieSourceSplits method to convert splits to HoodieSourceSplit type
AbstractHoodieSplitEnumerator.java	Updated to pass null values for new state parameters
TestHoodieContinuousSplitBatch.java	Comprehensive tests for HoodieContinuousSplitBatch
TestDefaultHoodieSplitDiscover.java	Tests for split discovery implementation
TestHoodieContinuousSplitEnumerator.java	Tests for continuous enumerator with mocks
TestScanContext.java	Tests for ScanContext builder and accessors
TestIncrementalInputSplits.java	Tests for the new inputHoodieSourceSplits method

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[cshuo]

Thks for the contribution, left some comments.

[hudi-bot]

CI report:

• 29df33c UNKNOWN
• 7a74e37 UNKNOWN
• 46992bc Azure: SUCCESS

Bot commands

@hudi-bot supports the following commands:

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

[xushiyan]

HoodieSourceSplit should not hold SPLIT_COUNTER for itself. better managed by the caller who creates the splits. it could be init'ed here

[xushiyan]

the current toString() is a human-readable form of info to describe the object, not suitable to be used as an id. also this should be able to uniquely identify the split reading the same set of records? it should at least contain file id, which is not present currently

[HuangZhenQiu]

The PR also resolved #14417