From 44bf9f07ee5c3e09010b12d146ab5d3651e4fc67 Mon Sep 17 00:00:00 2001 From: duankaixuan <1417048384@qq.com> Date: Mon, 14 Sep 2026 11:56:52 +0800 Subject: [PATCH] [server] Forbid ALTER bucket.num on aggregation merge engine tables --- .../server/coordinator/MetadataManager.java | 21 ++- .../coordinator/AlterBucketNumTest.java | 127 ++++++++++++++++++ 2 files changed, 145 insertions(+), 3 deletions(-) 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 e8dcd33e93..1ce16a9627 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 @@ -40,6 +40,7 @@ import org.apache.fluss.metadata.DatabaseInfo; import org.apache.fluss.metadata.DatabaseSummary; import org.apache.fluss.metadata.LakeTableUtil; +import org.apache.fluss.metadata.MergeEngineType; import org.apache.fluss.metadata.ResolvedPartitionSpec; import org.apache.fluss.metadata.Schema; import org.apache.fluss.metadata.SchemaInfo; @@ -569,9 +570,10 @@ private void propagateBucketCountToLake( } /** - * Validates an ALTER bucket.num request: only partitioned tables are supported and the new - * value must fall within [1, maxBucketNum]. Runs before the lake-side propagation so an invalid - * ALTER never mutates lake metadata. + * Validates an ALTER bucket.num request: only partitioned tables are supported, and only when + * neither the historical partition nor the aggregation merge engine is in use. The new value + * must fall within [1, maxBucketNum]. Runs before the lake-side propagation so an invalid ALTER + * never mutates lake metadata. */ private void validateBucketNumRescale( TablePath tablePath, TableInfo tableInfo, int newBucketNum) { @@ -595,6 +597,19 @@ private void validateBucketNumRescale( + "supported yet.", tablePath)); } + // The aggregation merge engine restores from checkpoints via undo recovery, which + // relies on the Flink sink's bucket shuffle keeping "one bucket, one writer". A + // rescaled table shards records with a stale table-level count and breaks sink + // recovery; supporting the combination is left to future work. + if (tableInfo.getTableConfig().getMergeEngineType().orElse(null) + == MergeEngineType.AGGREGATION) { + throw new InvalidAlterTableException( + String.format( + "Cannot alter 'bucket.num' on table %s with merge engine " + + "'aggregation'. Altering 'bucket.num' on such tables is " + + "not supported yet.", + tablePath)); + } if (newBucketNum < 1) { throw new InvalidAlterTableException( String.format( diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/AlterBucketNumTest.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/AlterBucketNumTest.java index 70420a862c..1e70df99cb 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/AlterBucketNumTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/AlterBucketNumTest.java @@ -31,6 +31,7 @@ import org.apache.fluss.lake.lakestorage.LakeStoragePlugin; import org.apache.fluss.metadata.DataLakeFormat; import org.apache.fluss.metadata.DatabaseDescriptor; +import org.apache.fluss.metadata.MergeEngineType; import org.apache.fluss.metadata.ResolvedPartitionSpec; import org.apache.fluss.metadata.Schema; import org.apache.fluss.metadata.TableChange; @@ -468,6 +469,95 @@ void testEnableHistoricalPartitionOnNeverRescaledTableSucceeds() throws Exceptio assertThat(mm.getTable(tablePath).getTableConfig().isHistoricalPartitionEnabled()).isTrue(); } + @Test + void testAlterBucketNumRejectedOnAggregationTable() throws Exception { + TablePath tablePath = TablePath.of(DEFAULT_DB, "test_reject_rescale_on_aggregation"); + int originalBucketCount = 4; + TableAssignment tableAssignment = + generateAssignment(originalBucketCount, 3, getTabletServers()); + metadataManager.createTable( + tablePath, + remoteDataDir, + partitionedPrimaryKeyTable(originalBucketCount, MergeEngineType.AGGREGATION.name()), + tableAssignment, + false); + TableInfo tableInfo = metadataManager.getTable(tablePath); + metadataManager.createPartition( + tablePath, + tableInfo.getTableId(), + remoteDataDir, + new PartitionAssignment( + tableInfo.getTableId(), tableAssignment.getBucketAssignments()), + fromPartitionName(tableInfo.getPartitionKeys(), "2024-01"), + false, + originalBucketCount); + + // The rejection happens during validation, before the lake propagation, so the default + // manager without a lake catalog never reaches the propagation step. + assertThatThrownBy(() -> alterBucketNum(metadataManager, tablePath, 8)) + .isInstanceOf(InvalidAlterTableException.class) + .hasMessageContaining("with merge engine 'aggregation'") + .hasMessageContaining("not supported yet"); + + // The bucket layout is untouched: neither the count nor the epoch moved. + TableInfo afterTableInfo = metadataManager.getTable(tablePath); + assertThat(afterTableInfo.getNumBuckets()).isEqualTo(originalBucketCount); + assertThat(afterTableInfo.getBucketCountEpoch()).isEqualTo(0L); + Optional partition = + zookeeperClient.getPartition(tablePath, "2024-01"); + assertThat(partition).isPresent(); + assertThat(partition.get().getBucketCount()).isEqualTo(originalBucketCount); + } + + @Test + void testAlterMergeEngineToAggregationRejectedAfterRescale() throws Exception { + // The reverse direction of the mutual exclusion: 'table.merge-engine' is not alterable, + // so a rescaled table can never switch to the aggregation merge engine afterwards. If + // altering the merge engine ever becomes supported, this test reminds that change to + // keep rejecting the switch to aggregation on rescaled tables (epoch > 0). + TablePath tablePath = TablePath.of(DEFAULT_DB, "test_reject_aggregation_after_rescale"); + int originalBucketCount = 4; + TableAssignment tableAssignment = + generateAssignment(originalBucketCount, 3, getTabletServers()); + metadataManager.createTable( + tablePath, + remoteDataDir, + partitionedPrimaryKeyTable(originalBucketCount, null), + tableAssignment, + false); + TableInfo tableInfo = metadataManager.getTable(tablePath); + metadataManager.createPartition( + tablePath, + tableInfo.getTableId(), + remoteDataDir, + new PartitionAssignment( + tableInfo.getTableId(), tableAssignment.getBucketAssignments()), + fromPartitionName(tableInfo.getPartitionKeys(), "2024-01"), + false, + originalBucketCount); + + // Rescale the default-engine table first, which advances the bucketCountEpoch. + alterBucketNum(metadataManager, tablePath, 8); + assertThat(metadataManager.getTable(tablePath).getBucketCountEpoch()).isEqualTo(1L); + + // Switching the rescaled table to the aggregation merge engine must be rejected. + assertThatThrownBy( + () -> + alterMergeEngine( + metadataManager, + tablePath, + MergeEngineType.AGGREGATION.name())) + .isInstanceOf(InvalidAlterTableException.class) + .hasMessageContaining("'table.merge-engine'") + .hasMessageContaining("not supported to alter yet"); + + // The merge engine is untouched and the bucket layout stays at the rescaled state. + assertThat(metadataManager.getTable(tablePath).getTableConfig().getMergeEngineType()) + .isEmpty(); + assertThat(metadataManager.getTable(tablePath).getNumBuckets()).isEqualTo(8); + assertThat(metadataManager.getTable(tablePath).getBucketCountEpoch()).isEqualTo(1L); + } + // ========================== Success Tests ========================== @ParameterizedTest(name = "bucketNum {0} -> {1}") @@ -958,6 +1048,43 @@ private static TableDescriptor partitionedLogTable(int bucketCount) { .withReplicationFactor(3); } + /** + * A partitioned primary key table: INT key "a", STRING partition key "b" (also part of the + * primary key); the merge engine property is set only when the given value is not null. + */ + private static TableDescriptor partitionedPrimaryKeyTable(int bucketCount, String mergeEngine) { + TableDescriptor.Builder builder = + TableDescriptor.builder() + .schema( + Schema.newBuilder() + .column("a", DataTypes.INT()) + .column("b", DataTypes.STRING()) + .primaryKey("a", "b") + .build()) + .distributedBy(bucketCount) + .partitionedBy("b"); + if (mergeEngine != null) { + builder.property(ConfigOptions.TABLE_MERGE_ENGINE.key(), mergeEngine); + } + return builder.build().withReplicationFactor(3); + } + + private static void alterMergeEngine( + MetadataManager manager, TablePath tablePath, String mergeEngine) { + String key = ConfigOptions.TABLE_MERGE_ENGINE.key(); + TablePropertyChanges.Builder builder = TablePropertyChanges.builder(); + builder.setTableProperty(key, mergeEngine); + manager.alterTableProperties( + tablePath, + Collections.singletonList(TableChange.set(key, mergeEngine)), + builder.build(), + false, + null, + (currentTable, updatedTable) -> {}, + (currentTable, updatedTable) -> {}, + ZkVersion.MATCH_ANY_VERSION.getVersion()); + } + private static void alterBucketNum( MetadataManager manager, TablePath tablePath, int newBucketCount) { manager.alterBucketCount(