[PR] [flink] Fix postpone bucket not handled for IncrementalSplit [paimon]

["wombatu-kun (via GitHub)" <gi...@apache.org>]

wombatu-kun opened a new pull request, #8698:
URL: https://github.com/apache/paimon/pull/8698

   ### Purpose
   
   Streaming reads of postpone bucket tables with `streaming-read-overwrite` enabled silently lose data.
   
   `ContinuousFileSplitEnumerator#assignSuggestedTask(IncrementalSplit)` returns `split.bucket()` as is, which is `BucketMode.POSTPONE_BUCKET` (-2) for postpone bucket tables. `ChannelComputer#select` then computes `-2 % parallelism == -2`, and `PreAssignSplitAssigner` stores the split under map key -2 while `getNext(subtask)` is only ever called with a subtask in `[0, parallelism)`. The split is therefore never handed to any reader. Because `remainingSplits()` still includes it, the split is checkpointed and stays stranded across restore. Neither an exception nor a log is produced.
   
   The sibling overload `assignSuggestedTask(DataSplit)` already handles postpone buckets. That handling was added in #6142 ("Fix postpone bucket without changelog producer cannot read all records with multiple streaming readers"). The incremental plan path used to be covered by it, but #7093 moved incremental plans from `DataSplit` to `IncrementalSplit` and the new overload was left with a `// TODO how to deal with postpone bucket?` instead, which reintroduced the same bug.
   
   The reachable configuration is the default one. `DataTableStreamScan` only applies `onlyReadRealBuckets()` when `changelog-producer != none`, and `SchemaValidation` rejects `streaming-read-overwrite` together with `full-compaction` / `lookup`, so `none` (the default, and the affected one) and `input` (unaffected) are the only options left.
   
   This PR mirrors the postpone branch of the `DataSplit` overload. `IncrementalSplit` has no `dataFiles()`, and a diff which only removes files has no after files, so the write id is read from the after files with a fallback to the before files. It reuses the existing `PostponeBucketFileStoreWrite#getWriteId`.
   
   Since the failure mode is silent, which is precisely why the bug survived being fixed once already, `assignSuggestedTask` now also validates that the suggested task lies within `[0, parallelism)` and fails loudly otherwise. `UNAWARE_BUCKET` is 0 and `POSTPONE_BUCKET` is the only negative bucket value, so this does not affect existing paths.
   
   ### Tests
   
   Added three cases to `ContinuousFileSplitEnumeratorTest`:
   
   - `testPostponeBucketIncrementalSplitAssignedByWriteId` - a postpone bucket `IncrementalSplit` is assigned by the write id encoded in the data file name.
   - `testPostponeBucketIncrementalSplitWithoutAfterFiles` - the write id is taken from the before files when the diff only removes files.
   - `testIncrementalSplitWithRealBucketAssignedByBucket` - a regular bucket is still assigned by its bucket id.
   
   The two postpone cases fail on master with `expected: 1 but was: -2`, confirming the stranded assignment, and pass with this change. `ContinuousFileSplitEnumeratorTest` (18 tests) and `AlignedContinuousFileSplitEnumeratorTest` (3 tests) pass.
   
   🤖 Generated with [Claude Code](https://claude.com/claude-code)


-- 
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: issues-unsubscribe@paimon.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [flink] Fix postpone bucket not handled for IncrementalSplit [paimon]

["JingsongLi (via GitHub)" <gi...@apache.org>]

JingsongLi merged PR #8698:
URL: https://github.com/apache/paimon/pull/8698


-- 
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: issues-unsubscribe@paimon.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org