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

   ### Search before asking
   
   - [x] I searched in the [issues](https://github.com/apache/paimon/issues) 
and found nothing similar.
   
   ### Paimon version
   
   master
   
   ### Compute Engine
   
   Flink and Spark dynamic-bucket writes (`HashBucketAssignerOperator` selects 
this class for overwrite; Spark uses it for new dynamic-bucket writes).
   
   ### Minimal reproduce step
   
   Pure unit test, no cluster:
   
   ```java
   SimpleHashBucketAssigner assigner = new SimpleHashBucketAssigner(1, 0, 100, 
4);
   Map<Integer, Integer> rows = new HashMap<>();
   for (int hash = 0; hash < 1000; hash++) {
       rows.merge(assigner.assign(BinaryRow.EMPTY_ROW, hash), 1, Integer::sum);
   }
   // expected: roughly even. actual: {0=700, 1=100, 2=100, 3=100}
   ```
   
   ### What doesn't meet your expectations?
   
   Once `dynamic-bucket.max-buckets` is reached, the assigner is meant to 
spread further rows over the existing buckets:
   
   ```java
   } else {
       currentBucket = ListUtils.pickRandomly(bucketList);
   }
   ```
   
   It cannot. **`bucketList` only ever contains one element**, so 
`pickRandomly` is a constant and every overflow row lands in the first bucket.
   
   `bucketList.add()` lives inside the `computeIfAbsent` mapping function:
   
   ```java
   Long num = bucketInformation.computeIfAbsent(currentBucket, bucket -> {
       bucketList.add(bucket);
       return 0L;
   });
   ...
   bucketInformation.compute(currentBucket, (i, l) -> l == null ? 1L : l + 1);
   ```
   
   When `loadNewBucket()` switches `currentBucket` to a fresh id, the 
`compute()` two lines below creates that bucket's `bucketInformation` entry 
**within the same `assign()` call**. On the next call `computeIfAbsent` 
therefore finds the key present, the mapping function never runs, and the 
bucket is never appended. Only the bucket picked in the constructor is 
registered, and nothing ever removes entries.
   
   With a cap of 4 and a target of 100 rows, 1000 rows produce **100 / 100 / 
100 / 700**.
   
   To be precise about scope: the upper bound itself is **not** violated — 
`loadNewBucket()` guards `i <= maxBucketsNum - 1` independently. This is write 
skew, not a bound violation.
   
   ### Anything else?
   
   The sibling `PartitionIndex` does it correctly, registering at the creation 
site:
   
   ```java
   totalBucketSet.add(i);
   totalBucketArray.add(i);
   ```
   
   Both classes were given this random-pick logic by the same commit — 
`cb25653f1` *"[core] Adjust 'dynamic-bucket.max-buckets' random pick logical"* 
(2025-02-19) — and this one lost the registration, so it is a regression 
against that commit's own intent rather than a design choice.
   
   `testAssignWithUpperBound` cannot catch it: it asserts `isIn(0, 2)` on the 
overflow rows, and "always 0" satisfies that.
   
   ### Are you willing to submit a PR?
   
   - [x] 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