[Bug] Overflow rows all land in the first bucket once dynamic-bucket.max-buckets is reached

[PDGGK]

Search before asking

• I searched in the 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:

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:

} 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:

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:

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?

• I'm willing to submit a PR!