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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
