ericyuan915 commented on code in PR #19376:
URL: https://github.com/apache/hudi/pull/19376#discussion_r3671224392
##########
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:
You're right. Rather than build a canonical predicate form, I've dropped
partitions from the guard entirely and reverted PartitionPruners to master, per
danny's high-level point that these options are the user's responsibility. The
guard is now commit-range only.
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/enumerator/HoodieEnumeratorStateSerializer.java:
##########
@@ -40,7 +40,7 @@
*/
@Internal
public class HoodieEnumeratorStateSerializer implements
SimpleVersionedSerializer<HoodieSplitEnumeratorState> {
- private static final int VERSION = 1;
+ private static final int VERSION = 2;
Review Comment:
Good catch — done. v2 payloads now lead with splitSerializer.getVersion()
and that is what's passed to splitSerializer.deserialize(...); v1 payloads map
to LEGACY_SPLIT_SERIALIZER_VERSION = 1. Also added a version > VERSION guard.
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/prune/PartitionPruners.java:
##########
@@ -138,6 +150,13 @@ public Set<String> filter(Collection<String> partitions) {
return partitions.stream()
.filter(this.partitions::contains).collect(Collectors.toSet());
}
+
+ @Override
+ public String scopeToken() {
+ // Sorted so the token is order-independent: the same partition set
always yields the same
+ // token regardless of insertion order.
+ return "static(" +
this.partitions.stream().sorted().collect(Collectors.joining(",")) + ")";
Review Comment:
Correct. Resolved by removal — PartitionPruners is back to master and the
state now holds two typed fields instead of a joined string, so there's no
encoding left to collide.
##########
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:
You're right about resetAndStart — it does catch and call cleanAndFailJob.
The break is one level down, and only on the initial savepoint restore, which
is the case this guard exists for. OperatorCoordinatorHolder#resetToCheckpoint
says in its own comment that the first call happens during ExecutionGraph
construction, before lazyInitialize supplies the scheduler executor;
LazyInitializedCoordinatorContext#failJob opens with checkInitialized(), which
then throws IllegalStateException inside the closingFuture.whenComplete(...)
callback whose future is discarded. The throw also escapes before
processPendingCalls(), so hasCaughtUp stays false and the later start() only
queues — the enumerator never runs. I hit exactly this internally before moving
the failure to start(): job RUNNING, zero throughput. On a mid-run failover
your description is accurate; it's the initial restore that differs.
##########
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:
For the commit range specifically, streaming must stay exempt:
IncrementalInputSplits:266 resolves startCompletionTime as issuedOffset != null
? issuedOffset : conf.get(READ_START_COMMIT), so a checkpointed offset
deliberately supersedes read.start-commit and validating it would fail every
legitimate streaming resume. The partition case is real, but it isn't fixable
by config comparison — the honest fix is re-filtering restored pending splits
through the current pruner on restore. Can follow up in another PR.
##########
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:
After chatted with Danny, we decide to only validate the date ranges, will
skip the rest.
--
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]
