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 6d60e40988..5cc3e79712 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 @@ CompletableFuture> listPartitionInfos( *
  • {@link TooManyPartitionsException} if the number of partitions is larger than the * maximum number of partitions of one table, see {@link ConfigOptions#MAX_PARTITION_NUM}. *
  • {@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}. * * * @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 18f177b642..cda9fa3345 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 @@ private List toFsPathAndFileNames(KvSnapshotHandle kvSnapshot .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 @@ public void testAddTooManyBuckets() throws Exception { 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 @@ public void testBucketLimitForNonPartitionedTable() throws Exception { .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 @@ public void testDefaultBucketLimitForNonPartitionedTable() throws Exception { .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 @@ public void testDefaultBucketLimitForNonPartitionedTable() throws Exception { .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 f3acc82795..aca5edaa83 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 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 71258346ec..0a9c5f3791 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 @@ 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 37bc7c3760..7ac0b2fcff 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 @@ private void createPartitions( } 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 65b2ee24bd..d87b85d5e6 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 void createPartition( 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 void createPartition( 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 6b60a94099..4be0a05347 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 @@ private static void checkDistribution(TableDescriptor tableDescriptor, int maxBu 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 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 641ae7d9e9..a24ab80db9 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 @@ void testDayPartitionDropShouldNotBeDelayedByJitter() throws Exception { } /** - * 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 @@ void testMaxBucketNum() throws Exception { 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 55601246b4..605bada741 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 size limit. | | 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 be discarded. The default value is 15 minutes. This setting is important to ensure that ACL changes are propagated in a timely manner and do not remain active longer than necessary. | | 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 be implemented by providing a matching plugin identifier. | diff --git a/website/docs/table-design/data-distribution/partitioning.md b/website/docs/table-design/data-distribution/partitioning.md index 2228134e7f..69255bd0fb 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 |