fhan688 commented on code in PR #19949:
URL: https://github.com/apache/hudi/pull/19949#discussion_r4043469271
##########
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`.
Thanks for the detailed analysis. I agree that this fix should be scoped by
the table version.
I’ve updated the implementation as suggested: pre-v8 tables now use
`endInstant` as the reader boundary, while v8+ tables retain
`min(fileSlice.getLatestInstantTime(), endInstant)`.
I also updated the pre-v8 MOR archived incremental regression test to
cover a later in-range log block whose log filename carries an older instant. I
verified it against both implementations: with the original
`min(...)`, the reader returns the older record (`age=21`, `ts=1`); with
the version-aware implementation, it correctly returns the in-range update
(`age=22`, `ts=2`). The original v8+ split-boundary assertions
have been restored and pass.
No checkpointed split fields or serializers were changed.
--
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]
