ericyuan915 commented on code in PR #19376:
URL: https://github.com/apache/hudi/pull/19376#discussion_r3671601000
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/enumerator/HoodieSplitEnumeratorState.java:
##########
@@ -30,9 +31,35 @@
* State of Hoodie split enumerator. Mainly include the states of pending
splits of split provider.
*/
@Value
+@AllArgsConstructor
public class HoodieSplitEnumeratorState implements Serializable {
Collection<HoodieSourceSplitState> pendingSplitStates;
Option<String> lastEnumeratedInstant;
Option<String> lastEnumeratedInstantOffset;
+ /**
+ * The {@code read.start-commit} / {@code read.end-commit} bounds configured
when this checkpoint
+ * was taken, recorded only for bounded reads. Both hold {@code
Option.of("")} when the option was
+ * not configured, so that "recorded but unset" stays distinguishable from
"not recorded at all";
+ * both are {@link Option#empty()} for streaming reads and for checkpoints
written by serializer
+ * VERSION 1, which predates this field.
+ *
+ * <p>A bounded read's split set is frozen at enumeration time and is NOT
re-derived on restore, so
+ * {@code HoodieSource} compares these against the configured bounds and
fails fast when they
+ * differ. See {@code HoodieSource#checkBoundedCommitRangeUnchanged}.
+ */
+ Option<String> readStartCommit;
Review Comment:
For bounded reads there's nothing in those two to infer from — they're
Option.empty() on every bounded checkpoint, and were before this PR too
(HoodieStaticSplitEnumerator inherited AbstractHoodieSplitEnumerator:129;
HoodieContinuousSplitEnumerator:90 is the only writer). So recovery can't
reconstruct the configured range from state alone.
Repurposing the slots to carry the bounded bounds would avoid the version
bump, but breaks on the reverse flip. HoodieSource:141 picks the enumerator
from current config, not from what wrote the checkpoint, so a bounded
checkpoint restored with read.streaming.enabled=true feeds those values into
the resume position at :64. Both are completion times, so
IncrementalInputSplits:266 would accept read.end-commit as startCompletionTime
with OPEN_CLOSED and silently skip everything up to the old bound — this PR's
bug in the other direction, and the shared domain is exactly what makes it
silent instead of loud. Separate typed fields keep "streaming resume position"
and "bounded scope assertion" from aliasing.
On cost: it's two optional strings per checkpoint, written only by the
bounded enumerator, alongside the pending-split collection.
I also tried deriving from the splits' InstantRange — no new state at all —
but IncrementalQueryAnalyzer:443-455 returns either no range or a null start
bound when reading from earliest, and only incremental non-CDC splits carry one.
--
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]
