cshuo commented on code in PR #19376:
URL: https://github.com/apache/hudi/pull/19376#discussion_r3655550875
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/HoodieSource.java:
##########
@@ -135,13 +135,36 @@ public SourceReader<T, HoodieSourceSplit>
createReader(SourceReaderContext reade
private SplitEnumerator<HoodieSourceSplit, HoodieSplitEnumeratorState>
createEnumerator(
SplitEnumeratorContext<HoodieSourceSplit> enumContext,
@Nullable HoodieSplitEnumeratorState enumeratorState) {
+ boolean streaming = scanContext.isStreaming();
+
+ // A bounded read freezes its split set at enumeration time
(createBatchHoodieSplits, below) and,
+ // on a checkpoint restore, resumes ONLY the checkpointed splits WITHOUT
re-enumerating. If the
+ // configured scope (date range / partitions / table) changed since the
checkpoint was taken,
+ // resuming would silently read the checkpoint's OLD scope. Capture the
current scope here so it is
+ // checkpointed (snapshotState) and can be compared on the next restore.
Streaming reads
+ // legitimately resume-and-continue, so they are not guarded (token stays
empty).
+ Option<String> currentScopeToken =
+ streaming ? Option.empty() :
Option.of(computeBoundedScopeToken(scanContext));
+ // Deferred bounded-scope failure. A bounded read restored from a
checkpoint whose scope changed
Review Comment:
Is deferring this failure to `start()` actually necessary? In Flink
1.18/1.20, exceptions from `restoreEnumerator()` propagate through
`resetAndStart()` to `cleanAndFailJob()`, which calls `context.failJob()` and
triggers a global failover.
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/prune/PartitionPruners.java:
##########
@@ -59,6 +59,18 @@ public interface PartitionPruner extends Serializable,
AutoCloseable {
*/
Set<String> filter(Collection<String> partitions);
+ /**
+ * Returns a stable, human-readable token describing the partition
selection this pruner
+ * enforces. Used by bounded-read scope-change detection across checkpoint
restores: two
+ * pruners that select the same partitions must return equal tokens, and
any change to the
+ * selection must change the token. The default (the pruner's class name)
is a coarse fallback
+ * for pruners whose selection is not a fixed partition list (e.g. dynamic
/ column-stats
+ * pruning); {@link StaticPartitionPruner} overrides it with the concrete
partition set.
+ */
+ default String scopeToken() {
Review Comment:
Production partition filters use `DynamicPartitionPruner`, whose
`scopeToken()` falls back to the class name. Since `candidatePartitions` is
only used in tests, changing the partition predicate produces the same token
and the restored stale splits are still accepted. Could the dynamic token
include a canonical representation of the predicate and its bound values?
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/HoodieSource.java:
##########
@@ -163,8 +186,77 @@ private SplitEnumerator<HoodieSourceSplit,
HoodieSplitEnumeratorState> createEnu
List<HoodieSourceSplit> splits = createBatchHoodieSplits();
splitProvider.onDiscoveredSplits(splits);
}
- return new HoodieStaticSplitEnumerator(tableName, enumContext,
splitProvider);
+ return new HoodieStaticSplitEnumerator(
+ tableName, enumContext, splitProvider, currentScopeToken,
boundedScopeFailure);
+ }
+ }
+
+ /**
+ * Builds a stable, human-readable token describing a bounded read's
<em>scope</em> — the inputs
+ * that determine which files the bounded read will read: table path, table
type, query type, the
+ * start/end commit-instant bounds, and the pruned partition set. The token
is checkpointed with
+ * the enumerator state and compared on restore ({@link
#computeBoundedScopeFailure}) so a changed
+ * scope is caught instead of being silently ignored.
+ *
+ * <p>Deliberately excludes {@code requiredColumns}/projection: projection
changes what is read
+ * from each file, not which files (splits) are read, so it does not
invalidate a resume.
+ */
+ @VisibleForTesting
+ static String computeBoundedScopeToken(HoodieScanContext scanContext) {
+ Configuration conf = scanContext.getConf();
+ String partitions = scanContext.getPartitionPruner() == null
+ ? "none"
+ : scanContext.getPartitionPruner().scopeToken();
+ return String.join(
Review Comment:
The scope token omits settings that affect bounded incremental split
enumeration, including CDC mode, `read.cdc.from.changelog`, and the
skip-compaction/clustering/insert-overwrite options. Restoring after changing
these settings would still accept the old splits. Could we include all
split-generation inputs in the token?
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/HoodieSource.java:
##########
@@ -135,13 +135,36 @@ public SourceReader<T, HoodieSourceSplit>
createReader(SourceReaderContext reade
private SplitEnumerator<HoodieSourceSplit, HoodieSplitEnumeratorState>
createEnumerator(
SplitEnumeratorContext<HoodieSourceSplit> enumContext,
@Nullable HoodieSplitEnumeratorState enumeratorState) {
+ boolean streaming = scanContext.isStreaming();
+
+ // A bounded read freezes its split set at enumeration time
(createBatchHoodieSplits, below) and,
+ // on a checkpoint restore, resumes ONLY the checkpointed splits WITHOUT
re-enumerating. If the
+ // configured scope (date range / partitions / table) changed since the
checkpoint was taken,
+ // resuming would silently read the checkpoint's OLD scope. Capture the
current scope here so it is
+ // checkpointed (snapshotState) and can be compared on the next restore.
Streaming reads
+ // legitimately resume-and-continue, so they are not guarded (token stays
empty).
+ Option<String> currentScopeToken =
Review Comment:
Why is scope validation disabled entirely for streaming restores? If a
partition predicate is narrowed from `partition IN ('p1', 'p2')` to `partition
= 'p1'`, the checkpointed in-flight/pending splits for `p2` are still restored,
while only newly discovered splits use the new predicate. Streaming progress
should resume from checkpoint, but split-selection settings such as partition
predicates may still require compatibility validation.
--
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]
