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 separate these responsibilities while preserving checkpoint
compatibility? The tests should independently verify that the reader includes
updates through `endInstant` and that splits from different commits retain
distinct progress/ordering timestamps.
Compatibility follow-up: this should not be implemented as a simple new
persisted split field. `StreamReadMonitoringFunction` stores pending splits
inside its private `SplitState` using `TypeInformation.of(SplitState.class)`,
which selects Kryo with the default configuration. In an isolated
old-writer/new-reader probe using local Flink 2.1.1 dependencies, adding a
`String readerEndInstant` to `MergeOnReadInputSplit` caused a `KryoException`
while restoring that state; the old-class control passed. Keeping
`serialVersionUID = 1L` does not protect this Kryo path. The separate
`StreamReadOperator` Java-serialization path did restore, but the new field was
null.
Source V2 also needs explicit format evolution:
`HoodieSourceSplitSerializer` currently writes version 1, and
`HoodieEnumeratorStateSerializer` passes its own version into the nested split
deserializer without storing a separate split version. Both formats need
coordinated backward readers if we add persisted data. Merely adding a Java
field leaves it out of the V2 checkpoint entirely.
I would avoid adding a serialized field in this fix unless we also provide
migration and old-state restore tests for these paths. Any fallback for old
splits must account for partially consumed splits: recovery skips the saved
record count, so changing the reader boundary can change the records that count
refers to.
--
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]
