hudi-agent commented on code in PR #19376:
URL: https://github.com/apache/hudi/pull/19376#discussion_r3653991022
##########
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:
🤖 Only `StaticPartitionPruner` overrides `scopeToken()` with the concrete
partition set — but in the real `HoodieTableSource` path partition filters
always become `partitionEvaluators` → `DynamicPartitionPruner` (nothing in main
ever calls `.candidatePartitions(...)`, so `StaticPartitionPruner` is
effectively test-only). That means the `partitions=` component of the token is
a constant class name in production, so changing the partition `WHERE` clause
won't change the token and the guard won't catch it — even though "partition
selection" is one of the listed target scenarios. The tests exercise only the
static path, so this gap isn't caught. Could you confirm whether dynamic
pruners were meant to contribute a meaningful token (e.g. from the
evaluators/probe)? @danny0405
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
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:
🤖 nit: this 7-line block restates almost word-for-word the Javadoc already
written on `computeBoundedScopeFailure`. Could you trim it to a one-liner like
`// Deferred to avoid zombie job — see computeBoundedScopeFailure.` and let the
Javadoc be the single source of truth?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
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]
