You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by ab...@apache.org on 2026/07/06 17:05:44 UTC
(fluss) branch main updated: [server] Limit max bucket num per table partition (#3597)
This is an automated email from the ASF dual-hosted git repository.
fresh-borzoni pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new 20b008ce5 [server] Limit max bucket num per table partition (#3597)
20b008ce5 is described below
commit 20b008ce566740f5d92527c89a0fc731b59dc11f
Author: yunhong <33...@qq.com>
AuthorDate: Tue Jul 7 01:05:40 2026 +0800
[server] Limit max bucket num per table partition (#3597)
---
.../java/org/apache/fluss/client/admin/Admin.java | 2 +-
.../fluss/client/admin/FlussAdminITCase.java | 87 ++++++++++++++--------
.../org/apache/fluss/config/ConfigOptions.java | 9 ++-
.../fluss/exception/TooManyBucketsException.java | 3 +-
.../server/coordinator/AutoPartitionManager.java | 2 +-
.../fluss/server/coordinator/MetadataManager.java | 27 ++-----
.../server/utils/TableDescriptorValidation.java | 2 +-
.../coordinator/AutoPartitionManagerTest.java | 15 ++--
website/docs/maintenance/configuration.md | 2 +-
.../table-design/data-distribution/partitioning.md | 2 +-
10 files changed, 84 insertions(+), 67 deletions(-)
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/admin/Admin.java b/fluss-client/src/main/java/org/apache/fluss/client/admin/Admin.java
index 6d60e4098..5cc3e7971 100644
--- a/fluss-client/src/main/java/org/apache/fluss/client/admin/Admin.java
+++ b/fluss-client/src/main/java/org/apache/fluss/client/admin/Admin.java
@@ -356,7 +356,7 @@ public interface Admin extends AutoCloseable {
* <li>{@link TooManyPartitionsException} if the number of partitions is larger than the
* maximum number of partitions of one table, see {@link ConfigOptions#MAX_PARTITION_NUM}.
* <li>{@link TooManyBucketsException} if the number of buckets is larger than the maximum
- * number of buckets of one table, see {@link ConfigOptions#MAX_BUCKET_NUM}.
+ * number of buckets of one partition, see {@link ConfigOptions#MAX_BUCKET_NUM}.
* </ul>
*
* @param tablePath The table path of the table.
diff --git a/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java b/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java
index 18f177b64..cda9fa334 100644
--- a/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java
+++ b/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java
@@ -1835,12 +1835,8 @@ class FlussAdminITCase extends ClientToServerITCaseBase {
.collect(Collectors.toList());
}
- /**
- * Test that creating a partitioned table with bucket count exceeding the maximum throws
- * TooManyBucketsException.
- */
@Test
- public void testAddTooManyBuckets() throws Exception {
+ public void testBucketLimitForPartitionedTableAppliesPerPartition() throws Exception {
// Already set low maximum bucket limit to 30 for this test in ClientToServerITCaseBase
String dbName = DEFAULT_TABLE_PATH.getDatabaseName();
TableDescriptor partitionedTable =
@@ -1857,35 +1853,46 @@ class FlussAdminITCase extends ClientToServerITCaseBase {
TablePath tablePath = TablePath.of(dbName, "test_add_too_many_buckets_table");
admin.createTable(tablePath, partitionedTable, true).get();
- // Add 3 partitions (3 * 10 = 30 buckets, which is the limit)
- for (int i = 0; i < 3; i++) {
+ // Add 4 partitions. The total bucket count is 40, which is above the configured limit,
+ // but each partition only has 10 buckets.
+ for (int i = 0; i < 4; i++) {
admin.createPartition(tablePath, newPartitionSpec("age", String.valueOf(i)), false)
.get();
}
+ assertThat(admin.listPartitionInfos(tablePath).get()).hasSize(4);
+
+ TableDescriptor tooManyBucketsPartitionedTable =
+ TableDescriptor.builder()
+ .schema(
+ Schema.newBuilder()
+ .column("id", DataTypes.STRING())
+ .column("name", DataTypes.STRING())
+ .column("age", DataTypes.STRING())
+ .build())
+ .distributedBy(40, "id")
+ .partitionedBy("age")
+ .build();
+ TablePath tooManyBucketsTablePath =
+ TablePath.of(dbName, "test_too_many_buckets_partitioned");
- // Try to add one more partition, exceeding the bucket limit (4 * 10 > 30)
assertThatThrownBy(
() ->
- admin.createPartition(
- tablePath, newPartitionSpec("age", "4"), false)
+ admin.createTable(
+ tooManyBucketsTablePath,
+ tooManyBucketsPartitionedTable,
+ false)
.get())
.cause()
.isInstanceOf(TooManyBucketsException.class)
- .hasMessageContaining("exceeding the maximum of 30 buckets");
+ .hasMessageContaining(
+ "Bucket count 40 exceeds the maximum limit 30 for a non-partitioned table or partition.");
}
- /**
- * Test that creating a non-partitioned table with bucket count exceeding the maximum throws
- * TooManyBucketsException.
- */
@Test
public void testBucketLimitForNonPartitionedTable() throws Exception {
- // Set a low maximum bucket limit for this test
- // (Assuming the configuration is already set to 30 in ClientToServerITCaseBase)
String dbName = DEFAULT_TABLE_PATH.getDatabaseName();
- // Create a non-partitioned table with 40 buckets (exceeding limit of 30)
- TableDescriptor nonPartitionedTable =
+ TableDescriptor maxBucketsNonPartitionedTable =
TableDescriptor.builder()
.schema(
Schema.newBuilder()
@@ -1893,21 +1900,42 @@ class FlussAdminITCase extends ClientToServerITCaseBase {
.column("name", DataTypes.STRING())
.column("value", DataTypes.STRING())
.build())
- .distributedBy(40, "id") // 40 buckets exceeds the limit of 30
- .build(); // No partitionedBy call makes this non-partitioned
+ .distributedBy(30, "id")
+ .build();
+ TablePath maxBucketsTablePath = TablePath.of(dbName, "test_max_buckets_non_partitioned");
- TablePath tablePath = TablePath.of(dbName, "test_too_many_buckets_non_partitioned");
+ admin.createTable(maxBucketsTablePath, maxBucketsNonPartitionedTable, false).get();
- // Creating this table should throw TooManyBucketsException
- assertThatThrownBy(() -> admin.createTable(tablePath, nonPartitionedTable, false).get())
+ TableDescriptor tooManyBucketsNonPartitionedTable =
+ TableDescriptor.builder()
+ .schema(
+ Schema.newBuilder()
+ .column("id", DataTypes.STRING())
+ .column("name", DataTypes.STRING())
+ .column("value", DataTypes.STRING())
+ .build())
+ .distributedBy(31, "id")
+ .build();
+ TablePath tooManyBucketsTablePath =
+ TablePath.of(dbName, "test_too_many_buckets_non_partitioned");
+
+ assertThatThrownBy(
+ () ->
+ admin.createTable(
+ tooManyBucketsTablePath,
+ tooManyBucketsNonPartitionedTable,
+ false)
+ .get())
.cause()
.isInstanceOf(TooManyBucketsException.class)
- .hasMessageContaining("exceeds the maximum limit");
+ .hasMessageContaining(
+ "Bucket count 31 exceeds the maximum limit 30 for a non-partitioned table or partition.");
}
@Test
- public void testDefaultBucketLimitForNonPartitionedTable() throws Exception {
- assertThat(ConfigOptions.MAX_BUCKET_NUM.defaultValue()).isEqualTo(20000);
+ public void testDefaultPartitionAndBucketLimitsForNonPartitionedTable() throws Exception {
+ assertThat(ConfigOptions.MAX_PARTITION_NUM.defaultValue()).isEqualTo(1000);
+ assertThat(ConfigOptions.MAX_BUCKET_NUM.defaultValue()).isEqualTo(4096);
FlussClusterExtension defaultLimitCluster =
FlussClusterExtension.builder().setNumOfTabletServers(1).build();
@@ -1925,7 +1953,7 @@ class FlussAdminITCase extends ClientToServerITCaseBase {
.column("id", DataTypes.STRING())
.column("name", DataTypes.STRING())
.build())
- .distributedBy(30000, "id")
+ .distributedBy(4097, "id")
.build();
defaultLimitAdmin.createDatabase(dbName, DatabaseDescriptor.EMPTY, false).get();
@@ -1936,7 +1964,8 @@ class FlussAdminITCase extends ClientToServerITCaseBase {
.get())
.cause()
.isInstanceOf(TooManyBucketsException.class)
- .hasMessageContaining("Bucket count 30000 exceeds the maximum limit 20000.");
+ .hasMessageContaining(
+ "Bucket count 4097 exceeds the maximum limit 4096 for a non-partitioned table or partition.");
} finally {
defaultLimitCluster.close();
}
diff --git a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
index f3acc8279..aca5edaa8 100644
--- a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
+++ b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
@@ -290,11 +290,12 @@ public class ConfigOptions {
public static final ConfigOption<Integer> MAX_BUCKET_NUM =
key("max.bucket.num")
.intType()
- .defaultValue(20000)
+ .defaultValue(4096)
.withDescription(
- "The maximum number of buckets that can be created for a table. "
- + "The default value is 20000. "
- + "This default is capped to reduce the risk that the table assignment znode exceeds "
+ "The maximum number of buckets that can be created for a non-partitioned table "
+ + "or for each partition of a partitioned table. "
+ + "The default value is 4096. "
+ + "This default is capped to reduce the risk that an assignment znode exceeds "
+ "ZooKeeper's packet size limit.");
/**
diff --git a/fluss-common/src/main/java/org/apache/fluss/exception/TooManyBucketsException.java b/fluss-common/src/main/java/org/apache/fluss/exception/TooManyBucketsException.java
index 71258346e..0a9c5f379 100644
--- a/fluss-common/src/main/java/org/apache/fluss/exception/TooManyBucketsException.java
+++ b/fluss-common/src/main/java/org/apache/fluss/exception/TooManyBucketsException.java
@@ -20,7 +20,8 @@ package org.apache.fluss.exception;
import org.apache.fluss.annotation.PublicEvolving;
/**
- * This exception is thrown if the number of table buckets is exceed max.bucket.num.
+ * This exception is thrown if the number of buckets for a non-partitioned table or partition
+ * exceeds max.bucket.num.
*
* @since 0.7
*/
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/AutoPartitionManager.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/AutoPartitionManager.java
index 37bc7c376..7ac0b2fcf 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/AutoPartitionManager.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/AutoPartitionManager.java
@@ -441,7 +441,7 @@ public class AutoPartitionManager implements AutoCloseable {
} catch (TooManyBucketsException t) {
LOG.warn(
"Auto partitioning skip to create partition {} for table [{}], "
- + "because exceed the maximum number of buckets.",
+ + "because exceed the maximum number of buckets per partition.",
partition,
tablePath);
} catch (Exception e) {
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java
index 65b2ee24b..d87b85d5e 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java
@@ -828,9 +828,8 @@ public class MetadataManager {
partition.getPartitionQualifiedName(), tablePath));
}
- final int partitionNumber;
try {
- partitionNumber = zookeeperClient.getPartitionNumber(tablePath);
+ int partitionNumber = zookeeperClient.getPartitionNumber(tablePath);
if (partitionNumber + 1 > maxPartitionNum) {
throw new TooManyPartitionsException(
String.format(
@@ -847,24 +846,12 @@ public class MetadataManager {
e);
}
- try {
- int bucketCount = partitionAssignment.getBucketAssignments().size();
- // currently, every partition has the same bucket count
- int totalBuckets = bucketCount * (partitionNumber + 1);
- if (totalBuckets > maxBucketNum) {
- throw new TooManyBucketsException(
- String.format(
- "Adding partition '%s' would result in %d total buckets for table %s, exceeding the maximum of %d buckets.",
- partition.getPartitionName(),
- totalBuckets,
- tablePath,
- maxBucketNum));
- }
- } catch (TooManyBucketsException e) {
- throw e;
- } catch (Exception e) {
- throw new FlussRuntimeException(
- String.format("Failed to check total bucket count for table %s", tablePath), e);
+ int bucketCount = partitionAssignment.getBucketAssignments().size();
+ if (bucketCount > maxBucketNum) {
+ throw new TooManyBucketsException(
+ String.format(
+ "Partition '%s' has %d buckets for table %s, exceeding the maximum of %d buckets per partition.",
+ partition.getPartitionName(), bucketCount, tablePath, maxBucketNum));
}
try {
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java b/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java
index 6b60a9409..4be0a0534 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java
@@ -253,7 +253,7 @@ public class TableDescriptorValidation {
if (bucketCount > maxBucketNum) {
throw new TooManyBucketsException(
String.format(
- "Bucket count %s exceeds the maximum limit %s.",
+ "Bucket count %s exceeds the maximum limit %s for a non-partitioned table or partition.",
bucketCount, maxBucketNum));
}
List<String> bucketKeys = tableDescriptor.getTableDistribution().get().getBucketKeys();
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/AutoPartitionManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/AutoPartitionManagerTest.java
index 641ae7d9e..a24ab80db 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/AutoPartitionManagerTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/AutoPartitionManagerTest.java
@@ -582,14 +582,14 @@ class AutoPartitionManagerTest {
}
/**
- * Test if AutoPartionManager.createPartition adheres to maxBucketLimit while adding new
- * parition automatically, skip if it breaches limit.
+ * Test if AutoPartitionManager.createPartition applies maxBucketLimit per partition while
+ * adding new partition automatically.
*/
@Test
- void testMaxBucketNum() throws Exception {
+ void testMaxBucketNumPerPartition() throws Exception {
int bucketCountPerPartition = 10;
- int maxBucketNum = 30; // Allow only 3 partitions with 10 buckets each
+ int maxBucketNum = 30;
Configuration config = new Configuration();
config.set(ConfigOptions.MAX_BUCKET_NUM, maxBucketNum);
@@ -626,16 +626,15 @@ class AutoPartitionManagerTest {
periodicExecutor.triggerNonPeriodicScheduledTask();
int partitionsNum = zookeeperClient.getPartitionNumber(tablePath);
- // Only 3 partitions should be created (3 * 10 = 30 buckets) out of the 4 requested
- assertThat(partitionsNum).isEqualTo(3);
+ // All 4 requested partitions should be created because each partition is below the limit.
+ assertThat(partitionsNum).isEqualTo(4);
// Advance time to trigger another auto-partition cycle
clock.advanceTime(Duration.ofDays(1).plusHours(23));
periodicExecutor.triggerPeriodicScheduledTasks();
- // Check partitions again - should still have only 3 due to bucket limit
partitionsNum = zookeeperClient.getPartitionNumber(tablePath);
- assertThat(partitionsNum).isEqualTo(3);
+ assertThat(partitionsNum).isEqualTo(5);
}
@Test
diff --git a/website/docs/maintenance/configuration.md b/website/docs/maintenance/configuration.md
index 55601246b..605bada74 100644
--- a/website/docs/maintenance/configuration.md
+++ b/website/docs/maintenance/configuration.md
@@ -46,7 +46,7 @@ during the Fluss cluster working.
| allow.create.log.tables | Boolean | true | Whether to allow creation of log tables. When set to false, attempts to create log tables (tables without primary key) will be rejected. The default value is true. [...]
| allow.create.kv.tables | Boolean | true | Whether to allow creation of kv tables (primary key tables). When set to false, attempts to create kv tables (tables with primary key) will be rejected. The default value is true. [...]
| max.partition.num | Integer | 1000 | Limits the maximum number of partitions that can be created for a partitioned table to avoid creating too many partitions. [...]
-| max.bucket.num | Integer | 20000 | The maximum number of buckets that can be created for a table. The default value is 20000. This default is capped to reduce the risk that the table assignment znode exceeds ZooKeeper's packet size limit. [...]
+| max.bucket.num | Integer | 4096 | The maximum number of buckets that can be created for a non-partitioned table or for each partition of a partitioned table. The default value is 4096. This default is capped to reduce the risk that an assignment znode exceeds ZooKeeper's packet [...]
| acl.notification.expiration-time | Duration | 15min | The duration for which ACL notifications are valid before they expire. This configuration determines the time window during which an ACL notification is considered active. After this duration, the notification will no longer be valid and will b [...]
| authorizer.enabled | Boolean | false | Specifies whether to enable the authorization feature. If enabled, access control is enforced based on the authorization rules defined in the configuration. If disabled, all operations and resources are accessible to all users. [...]
| authorizer.type | String | default | Specifies the type of authorizer to be used for access control. This value corresponds to the identifier of the authorization plugin. The default value is `default`, which indicates the built-in authorizer implementation. Custom authorizers can [...]
diff --git a/website/docs/table-design/data-distribution/partitioning.md b/website/docs/table-design/data-distribution/partitioning.md
index 2228134e7..69255bd0f 100644
--- a/website/docs/table-design/data-distribution/partitioning.md
+++ b/website/docs/table-design/data-distribution/partitioning.md
@@ -86,7 +86,7 @@ Below are the configuration items related to Fluss cluster and automatic partiti
**Dynamic partitioning** is a feature that is enabled by default on client, allowing the client to automatically create partitions based on the data being written to the table. This feature is especially valuable when the set of partitions is not known in advance, eliminating the need for manual partition creation. It is also particularly useful when working with multi-field partitions, as auto-partitioning currently only supports single-field partitioning creation.
-Please note that the number of dynamically created partitions is also subject to the `max.partition.num` and `max.bucket.num` limit configured on the Fluss cluster.
+Please note that dynamically created partitions are subject to the `max.partition.num` limit, and each partition's bucket count is subject to the `max.bucket.num` limit configured on the Fluss cluster.
### Client Options
| Option | Type | Required | Default | Description |