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, I think we should separate the two responsibilities on
current master:
- Use the query's `endInstant` as the log-reader upper bound, retaining
`InstantRange` for logical filtering.
- Keep a separate per-split commit timestamp for progress metrics and split
ordering.
#9923 changed this from `endInstant` to a per-split value because assigning
the batch end to every split made `splitLatestCommit` /
`splitLatestCommitDelay` inaccurate while reading older files. It also added
ascending split ordering. These consumers still exist in `StreamReadOperator`
and `HoodieSourceSplitComparator`. With `max(...)`, all slices older than the
query end get the same timestamp; assigning `endInstant` directly to this
shared field has the same issue.
Could we pass/store the reader boundary separately and preserve the
per-split timestamp for those consumers? This would address the pre-v8
log-reading issue without undoing the metrics and ordering behavior introduced
by #9923. The tests should independently verify that the reader includes
updates through `endInstant` and that splits from different commits retain
distinct progress/ordering timestamps.
--
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]
