mikedias opened a new issue, #8349:
URL: https://github.com/apache/paimon/issues/8349

   ### Search before asking
   
   - [x] I searched in the [issues](https://github.com/apache/paimon/issues) 
and found nothing similar.
   
   
   ### Motivation
   
   ## Summary
   
   Context: follow-up to PR #7865, review comment 
https://github.com/apache/paimon/pull/7865#discussion_r3458901133.
   
   Per-partition bucket counts are currently honored by the **Flink** writer 
(via `table.createRowKeyExtractor()` backed by `PartitionBucketMapping`), but 
the **Spark** fixed-bucket write path still derives the bucket from the single 
table-level bucket count. After changing the table-level `bucket` while 
existing partitions keep a different bucket count (e.g. following a 
per-partition rescale), Spark can route rows to buckets that do not belong to 
the partition, breaking the per-partition layout. 
   
   
   ## Background / Root cause
   On the Spark side, `BUCKET_COL` for fixed-bucket (`HASH_FIXED`) tables is 
computed in `PaimonSparkWriter.write` using a single table-level literal 
`coreOptions.bucket()`:
   
   - **Fast path** (`paimonExtensionEnabled` + `BucketFunction.supportsTable`): 
computes `BUCKET_COL` via the `fixed_bucket` UDF 
(`BucketExpression.FIXED_BUCKET`) with `numBuckets = coreOptions.bucket()` — a 
literal, so it cannot vary per partition.
   - **Fallback path** (`CommonBucketProcessor`): uses 
`table.createRowKeyExtractor()`, which *is* per-partition aware via 
`PartitionBucketMapping`, but reloads the mapping per task.
   - `PaimonWriteRequirement` (V2 distribution) also clusters by 
`Expressions.bucket(numBuckets, keys)` using the table-level count.
   
   The two Spark write paths are mutually exclusive per write:
   - **V2 path:** `PaimonV2Write` → `PaimonWriteRequirement` (no 
`PaimonSparkWriter`).
   - **V1 / command path** (MERGE/UPDATE/DELETE/`WriteIntoPaimonTable`): 
`PaimonSparkWriter`.
   
   ## Failure scenario (reproducible)
   1. Create a partitioned fixed-bucket table with `bucket = 2`.
   2. Rescale one partition (e.g. `p1`) to 4 buckets, leaving `p2` at 2 (`CALL 
sys.rescale(...)`).
   3. `ALTER TABLE ... SET TBLPROPERTIES ('bucket' = '4')`.
   4. Write new keys into `p2` via Spark (a `MERGE INTO` reaches the 
`PaimonSparkWriter` fast path).
   5. Spark computes the bucket with count = 4 and attempts to write `p2` into 
bucket 2/3, which don't exist in that 2-bucket partition.
   
   Today this is caught by the core write-side safety net added in commit 
`c5172e855` ("Reject bucket writes outside partition layout"), which throws:
   
   > `Trying to write bucket 3 to partition {pt=p2}, but the partition only has 
2 buckets ...`
   
   So data isn't silently corrupted, but the write simply fails — i.e. Spark 
cannot write to tables with per-partition bucket layouts.
   
   ## Goal
   Make the Spark fixed-bucket write path resolve the bucket count **per 
partition**, matching Flink, so writes succeed and rows land in the correct 
bucket for their partition.
   
   ### Solution
   
   _No response_
   
   ### Anything else?
   
   _No response_
   
   ### Are you willing to submit a PR?
   
   - [ ] I'm willing to submit a PR!


-- 
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]

Reply via email to