This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 15a454b232 [core] Register every bucket in bucketList so overflow rows
spread (#9180)
15a454b232 is described below
commit 15a454b2328c082e7a3b0dc366547c2030e51f5a
Author: ZIHAN DAI <[email protected]>
AuthorDate: Wed Aug 12 17:25:43 2026 +1000
[core] Register every bucket in bucketList so overflow rows spread (#9180)
---
.../paimon/index/SimpleHashBucketAssigner.java | 16 +++++-
.../paimon/index/SimpleHashBucketAssignerTest.java | 60 ++++++++++++++++++++++
2 files changed, 75 insertions(+), 1 deletion(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/index/SimpleHashBucketAssigner.java
b/paimon-core/src/main/java/org/apache/paimon/index/SimpleHashBucketAssigner.java
index e5249bb0a1..a0bc5b427e 100644
---
a/paimon-core/src/main/java/org/apache/paimon/index/SimpleHashBucketAssigner.java
+++
b/paimon-core/src/main/java/org/apache/paimon/index/SimpleHashBucketAssigner.java
@@ -112,7 +112,21 @@ public class SimpleHashBucketAssigner implements
BucketAssigner {
currentBucket = ListUtils.pickRandomly(bucketList);
}
}
- bucketInformation.compute(currentBucket, (i, l) -> l == null ? 1L
: l + 1);
+ // A bucket is created by exactly one of these two calls, so both
have to register
+ // it and neither can double-register. The first bucket of a
partition is created by
+ // the computeIfAbsent above. Every later one is created here
instead: loadNewBucket()
+ // switches currentBucket to an id that is absent from
bucketInformation, and this
+ // compute() is what puts it there, so the next assign() finds it
present and that
+ // computeIfAbsent's mapping function never runs for it.
+ bucketInformation.compute(
+ currentBucket,
+ (i, l) -> {
+ if (l == null) {
+ bucketList.add(i);
+ return 1L;
+ }
+ return l + 1;
+ });
hash2Bucket.put(hash, (short) currentBucket);
return currentBucket;
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/index/SimpleHashBucketAssignerTest.java
b/paimon-core/src/test/java/org/apache/paimon/index/SimpleHashBucketAssignerTest.java
index 662a68f1d9..70061941f6 100644
---
a/paimon-core/src/test/java/org/apache/paimon/index/SimpleHashBucketAssignerTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/index/SimpleHashBucketAssignerTest.java
@@ -24,12 +24,72 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
+import java.util.HashMap;
+import java.util.Map;
+
import static org.apache.paimon.io.DataFileTestUtils.row;
import static org.assertj.core.api.Assertions.assertThat;
/** Tests for {@link SimpleHashBucketAssigner}. */
public class SimpleHashBucketAssignerTest {
+ @Test
+ public void testOverflowIsSpreadAcrossAllBuckets() {
+ // A single assigner owns every bucket id, so a cap of 4 means buckets
0..3.
+ SimpleHashBucketAssigner assigner = new SimpleHashBucketAssigner(1, 0,
100, 4);
+ BinaryRow partition = BinaryRow.EMPTY_ROW;
+
+ Map<Integer, Integer> rowsPerBucket = new HashMap<>();
+ for (int hash = 0; hash < 1000; hash++) {
+ rowsPerBucket.merge(assigner.assign(partition, hash), 1,
Integer::sum);
+ }
+
+ assertThat(rowsPerBucket.keySet()).containsExactlyInAnyOrder(0, 1, 2,
3);
+
+ // The first 400 rows fill each bucket to its target of 100. The
remaining 600 go through
+ // ListUtils.pickRandomly, so they must land across the four buckets
rather than piling
+ // into one. More than one bucket ending up past its target is what
"spread" means here.
+ // Every bucket must take a share of the overflow, the constructor's
included. Asserting
+ // merely "more than one" would still pass with the first bucket
frozen at its target.
+ assertThat(rowsPerBucket.values()).allMatch(rows -> rows > 100);
+ }
+
+ @Test
+ public void testSecondPartitionAfterTheCapIsExhausted() {
+ // maxBucketId is shared across partitions, so by the time a later
partition starts, the
+ // create-a-new-bucket branch is already closed for it and it goes
straight to
+ // pickRandomly. Its own first bucket must be in the pool or that call
has nothing to
+ // choose from.
+ SimpleHashBucketAssigner assigner = new SimpleHashBucketAssigner(1, 0,
100, 4);
+
+ for (int hash = 0; hash < 500; hash++) {
+ assigner.assign(row(1), hash);
+ }
+
+ Map<Integer, Integer> rowsPerBucket = new HashMap<>();
+ for (int hash = 0; hash < 300; hash++) {
+ rowsPerBucket.merge(assigner.assign(row(2), hash), 1,
Integer::sum);
+ }
+
+ assertThat(rowsPerBucket.keySet()).allMatch(bucket -> bucket >= 0 &&
bucket < 4);
+
assertThat(rowsPerBucket.values().stream().mapToInt(Integer::intValue).sum())
+ .isEqualTo(300);
+ }
+
+ @Test
+ public void testUnboundedAssignmentIsUnchanged() {
+ // Negative control: without an upper bound the random-pick branch is
unreachable, so the
+ // assignment sequence must stay exactly as it was.
+ SimpleHashBucketAssigner assigner = new SimpleHashBucketAssigner(1, 0,
100, -1);
+ BinaryRow partition = BinaryRow.EMPTY_ROW;
+
+ for (int bucket = 0; bucket < 10; bucket++) {
+ for (int i = 0; i < 100; i++) {
+ assertThat(assigner.assign(partition, bucket * 100 +
i)).isEqualTo(bucket);
+ }
+ }
+ }
+
@Test
public void testAssign() {
SimpleHashBucketAssigner simpleHashBucketAssigner =
