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 934f7c6166 [core] Report reachable dynamic-bucket max-buckets
misconfiguration clearly (#10104)
934f7c6166 is described below
commit 934f7c6166e8578b3b37a1706b2490f404c46727
Author: YangJie <[email protected]>
AuthorDate: Thu Sep 24 02:37:44 2026 -0400
[core] Report reachable dynamic-bucket max-buckets misconfiguration clearly
(#10104)
---
.../org/apache/paimon/index/PartitionIndex.java | 11 +++++++
.../paimon/index/SimpleHashBucketAssigner.java | 15 ++++++++++
.../paimon/index/HashBucketAssignerTest.java | 35 ++++++++++++++++++++++
.../paimon/index/SimpleHashBucketAssignerTest.java | 21 +++++++++++++
4 files changed, 82 insertions(+)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/index/PartitionIndex.java
b/paimon-core/src/main/java/org/apache/paimon/index/PartitionIndex.java
index 7a0a0802eb..aa239f4be8 100644
--- a/paimon-core/src/main/java/org/apache/paimon/index/PartitionIndex.java
+++ b/paimon-core/src/main/java/org/apache/paimon/index/PartitionIndex.java
@@ -118,6 +118,17 @@ public class PartitionIndex {
}
// 4. exceed buckets upper bound
+ if (totalBucketArray.isEmpty()) {
+ // this assigner owns no bucket at all: the bucket filter rejected
every bucket id
+ // below the upper bound, which happens when the upper bound is
smaller than the
+ // number of assigners
+ throw new RuntimeException(
+ String.format(
+ "Cannot assign a bucket: the bucket filter
rejected all buckets under "
+ + "the max buckets number %s. Check '%s'
is not smaller than "
+ + "the writer parallelism.",
+ maxBucketsNum, DYNAMIC_BUCKET_MAX_BUCKETS.key()));
+ }
int bucket = ListUtils.pickRandomly(totalBucketArray);
hash2Bucket.put(hash, toBucketShort(bucket));
return bucket;
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 e0f0b56e85..a14cefd819 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
@@ -30,7 +30,9 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
+import static org.apache.paimon.CoreOptions.DYNAMIC_BUCKET_MAX_BUCKETS;
import static org.apache.paimon.CoreOptions.MAX_DYNAMIC_BUCKETS;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
/** When we need to overwrite the table, we should use this to avoid loading
index. */
public class SimpleHashBucketAssigner implements BucketAssigner {
@@ -45,6 +47,19 @@ public class SimpleHashBucketAssigner implements
BucketAssigner {
public SimpleHashBucketAssigner(
int numAssigners, int assignId, long targetBucketRowNumber, int
maxBucketsNum) {
PartitionIndex.validateMaxBuckets(maxBucketsNum);
+ // buckets are owned by 'bucket % numAssigners == assignId %
numAssigners', so the
+ // smallest bucket this assigner owns is assignId % numAssigners. With
a cap that does
+ // not admit it, this assigner owns no bucket and would silently
default to bucket 0,
+ // which belongs to another assigner.
+ checkArgument(
+ maxBucketsNum == -1 || maxBucketsNum > assignId % numAssigners,
+ "Dynamic bucket max buckets number %s must be greater than the
assigner id %s: "
+ + "buckets are owned by 'bucket modulo assigners', so
this assigner "
+ + "would own no bucket and its records could not be
placed. "
+ + "Increase '%s' or reduce the assigner parallelism.",
+ maxBucketsNum,
+ assignId % numAssigners,
+ DYNAMIC_BUCKET_MAX_BUCKETS.key());
this.numAssigners = numAssigners;
this.assignId = assignId;
this.targetBucketRowNumber = targetBucketRowNumber;
diff --git
a/paimon-core/src/test/java/org/apache/paimon/index/HashBucketAssignerTest.java
b/paimon-core/src/test/java/org/apache/paimon/index/HashBucketAssignerTest.java
index d750dc83f3..aeff1be8bd 100644
---
a/paimon-core/src/test/java/org/apache/paimon/index/HashBucketAssignerTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/index/HashBucketAssignerTest.java
@@ -25,6 +25,7 @@ import org.apache.paimon.io.DataIncrement;
import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.table.sink.CommitMessageImpl;
import org.apache.paimon.table.sink.StreamTableCommit;
+import org.apache.paimon.utils.Int2ShortHashMap;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
@@ -85,6 +86,40 @@ public class HashBucketAssignerTest extends
PrimaryKeyTableTestBase {
maxBucketsNum);
}
+ @Test
+ public void testAssignFailsDescriptivelyWhenAssignerOwnsNoBucket() {
+ // a bucket filter rejecting every id below the cap leaves the random
pick with an
+ // empty bucket list; the error must name the cause instead of "list
is empty"
+ PartitionIndex index = new PartitionIndex(new Int2ShortHashMap(), new
HashMap<>(), 100);
+ assertThatThrownBy(() -> index.assign(42, bucket -> false, 2))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("dynamic-bucket.max-buckets");
+ }
+
+ @Test
+ public void testFreshPartitionFailsDescriptivelyWhenAssignerOwnsNoBucket()
{
+ // with a cap smaller than the assigner number, the assigner whose id
reaches the cap
+ // owns no bucket on a fresh partition: it used to die with an
unrelated "list is
+ // empty" from the random pick over its empty bucket list
+ HashBucketAssigner assigner2 = createAssigner(3, 3, 2, 1);
+ assertThatThrownBy(() -> assigner2.assign(row(9), 123))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("dynamic-bucket.max-buckets");
+ }
+
+ @Test
+ public void testAssignReusesLoadedBucketBeyondCap() {
+ // an assigner that already owns a loaded bucket beyond the cap (the
table's
+ // max-buckets was lowered after the bucket was created) must keep
serving that
+ // bucket, not throw: only the genuinely bucketless case is rejected
+ Map<Integer, Long> loaded = new HashMap<>();
+ loaded.put(5, 1L); // bucket 5 is owned by assigner 2 of 3 and already
at the row cap
+ PartitionIndex index = new PartitionIndex(new Int2ShortHashMap(),
loaded, 1);
+ // cap 2 admits only buckets 0 and 1; bucket 5 is beyond it but
already loaded, so the
+ // assign falls through to the random pick over the loaded bucket
rather than failing
+ assertThat(index.assign(42, bucket -> bucket % 3 == 2,
2)).isEqualTo(5);
+ }
+
@Test
public void testAssign() {
HashBucketAssigner assigner = createAssigner(2, 2, 0);
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 c354eda5f5..8e18a570a1 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
@@ -29,11 +29,32 @@ import java.util.Map;
import static org.apache.paimon.io.DataFileTestUtils.row;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.assertj.core.api.Assertions.entry;
/** Tests for {@link SimpleHashBucketAssigner}. */
public class SimpleHashBucketAssignerTest {
+ @Test
+ public void testRejectsMaxBucketsSmallerThanAssignerNumber() {
+ // buckets are owned by 'bucket % assigners', so with a cap of 2 and 4
assigners the
+ // assigner with id 3 owns no bucket: it used to default to bucket 0
(owned by
+ // another assigner) and write all its records there
+ assertThatThrownBy(() -> new SimpleHashBucketAssigner(4, 3, 100, 2))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("dynamic-bucket.max-buckets");
+
+ // the exact boundary also leaves this assigner bucketless: it owns
buckets 3, 7, ...
+ // but a cap of 3 only admits buckets 0-2
+ assertThatThrownBy(() -> new SimpleHashBucketAssigner(4, 3, 100, 3))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("dynamic-bucket.max-buckets");
+
+ // unlimited and the tightest legal cap (assigner id + 1) are both fine
+ new SimpleHashBucketAssigner(4, 3, 100, -1);
+ new SimpleHashBucketAssigner(4, 3, 100, 4);
+ }
+
@Test
public void testOverflowIsSpreadAcrossAllBuckets() {
// A single assigner owns every bucket id, so a cap of 4 means buckets
0..3.
