zhuxiangyi commented on code in PR #9370:
URL: https://github.com/apache/paimon/pull/9370#discussion_r3941344537
##########
paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreWrite.java:
##########
@@ -674,8 +674,31 @@ private RestoreFiles scanExistingFileMetas(
partInfo.get(), bucket),
e);
}
- if (restored.totalBuckets() != null && validateNumBuckets) {
- checkNumBuckets(partInfo.get(), expectedTotalBuckets,
restored.totalBuckets());
+ Integer restoredTotalBuckets = restored.totalBuckets();
+ if (restoredTotalBuckets != null
+ && validateNumBuckets
+ && expectedTotalBuckets != restoredTotalBuckets) {
+ if (partitionType.getFieldCount() > 0 &&
options.bucketPerPartitionCountEnabled()) {
+ if (bucket >= restoredTotalBuckets) {
+ throw new RuntimeException(
+ String.format(
+ "Trying to write bucket %d to %s, but the
partition only has %d "
+ + "buckets (table default: %d).
Recompute the bucket using the "
+ + "partition's bucket count, or
rescale the partition via "
+ + "INSERT OVERWRITE.",
+ bucket,
+ partInfo.get(),
+ restoredTotalBuckets,
+ expectedTotalBuckets));
+ }
+ LOG.info(
+ "{} uses {} buckets (expected: {}). Accepting
per-partition bucket count.",
Review Comment:
[P1] The relaxed bucket check silently accepts mis-routed rows on the Flink
path.
`bucket >= restoredTotalBuckets` is a range check, not a routing check, and
it cannot detect the failure this feature makes reachable.
The routing mapping is frozen at job submission: `FlinkSinkBuilder` builds
`new RowDataChannelComputer(sinkTable.createRowKeyExtractor())` on the client,
and `RowKeyExtractor` was made `Serializable` in this PR precisely so the
`PartitionBucketMapping` ships with the job graph. A long-running streaming job
therefore keeps routing with the bucket layout that existed at submission time,
and the mapping is never refreshed afterwards.
Consider a partition rescaled from 4 to 8 buckets while a streaming job is
running. The router still computes `h % 4 = b`, which is in `[0, 4)`; the true
bucket is `h % 8`, which is either `b` or `b + 4`. Since `b < 8`, the range
check passes for **every** row, and roughly half of them are written into the
wrong bucket with only a `LOG.info`. On a primary-key table the same key then
lives in two buckets — duplicates and lost updates, with no error surfaced at
any layer. (Downscaling is caught only accidentally, and only for power-of-two
counts; e.g. 6 -> 4 also mis-routes silently for `h = 8`.)
The writer cannot currently do better, because `expectedTotalBuckets` here
is the *table-level* count: the Flink fixed-bucket path goes
`TableWriteImpl.writeAndReturn(row, bucket, null)` -> `write(partition, bucket,
data)` -> `createWriterContainer(partition, bucket, numBuckets, ...)`, where
`numBuckets = options.bucket()`. The bucket count the router actually used
never reaches the writer, so there is nothing meaningful to compare
`restoredTotalBuckets` against.
Suggestion: plumb the routing bucket count
(`PartitionBucketMapping.resolveNumBuckets(partition)`) through to the writer —
the `write(partition, bucket, totalBuckets, data)` overload already exists —
and fail when it disagrees with `restoredTotalBuckets`. That is precise rather
than heuristic, and it does not break the feature: a batch job or a freshly
started streaming job loads a current mapping, so the two agree and nothing
fails. It fails only when the mapping is stale, which is exactly the corruption
case that `docs/docs/maintenance/rescale-bucket.md` currently addresses with
"Streaming jobs must be restarted after rescaling a partition" — a guideline
that cannot be enforced, as @JingsongLi already noted above.
--
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]
