danny0405 commented on code in PR #19949:
URL: https://github.com/apache/hudi/pull/19949#discussion_r4025685717
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/IncrementalInputSplits.java:
##########
@@ -447,9 +447,10 @@ private List<MergeOnReadInputSplit> getInputSplits(
.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);
+ // The latest commit is the physical upper threshold of the log reader.
It must cover
+ // both the selected file slice and the query end to avoid data loss.
The instant range
+ // remains the logical query boundary and filters out records beyond the
query end.
+ String latestCommit =
InstantComparison.maxInstant(fileSlice.getLatestInstantTime(), endInstant);
Review Comment:
After tracing #9923 and checking checkpoint compatibility, I would scope
this fix by the actual table version. `getInputSplits()` already receives
`metaClient`, so the flag can be a local variable computed once before mapping
the file slices; no new field is needed in either `IncrementalInputSplits` or
the checkpointed split classes:
```java
final boolean isPreV8 = metaClient.getTableConfig().getTableVersion()
.lesserThan(HoodieTableVersion.EIGHT);
```
Then, inside the mapping:
```java
String latestCommit = isPreV8
? endInstant
: InstantComparison.minInstant(fileSlice.getLatestInstantTime(),
endInstant);
```
For pre-v8 tables, a log filename can contain the base instant while its
blocks contain later commits, so the filename-derived instant is not a safe
reader upper bound. For v8+ tables, keeping master's expression preserves the
per-split metrics and ordering introduced by #9923. Keep `InstantRange` for
logical filtering.
This leaves the checkpointed split schema and serializers unchanged. Adding
a persisted reader-boundary field would require migration: the legacy
monitoring function's pending splits use Kryo, and an isolated
old-state/new-class probe failed after adding a field; Source V2 also needs
coordinated split/enumerator format versioning.
The tradeoff is that pre-v8 splits use the batch end for progress metrics
and lose per-split distinction in the primary ordering key. Previously
checkpointed splits retain their stored boundary and consumed offset; this
change fixes newly generated splits.
I checked this conditional using isolated compilation and local Flink 2.1.1
dependencies: the pre-v8 archived-range result assertion passed (with the
implementation-specific `latestCommit > endCommit` assertion removed), and
master's unchanged `testInputSplitsForSplitLastCommit` passed. Please retain
that existing v8+ behavior test and add a pre-v8 output regression where an
in-range log update is newer than the filename-derived instant; that assertion
should fail with `min(...)` and pass with `endInstant`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
