Skip to content

[server] Forbid ALTER bucket.num on aggregation merge engine tables #4315

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
wuchong merged 1 commit into from
Sep 14, 2026
+145 −3
Merged

[server] Forbid ALTER bucket.num on aggregation merge engine tables #4315

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 18 additions & 3 deletions fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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) {
Expand All @@ -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(
Expand Down
127 changes: 127 additions & 0 deletions fluss-server/src/test/java/org/apache/fluss/server/coordinator/AlterBucketNumTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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<PartitionRegistration> partition =
zookeeperClient.getPartition(tablePath, "2024-01");
assertThat(partition).isPresent();
assertThat(partition.get().getBucketCount()).isEqualTo(originalBucketCount);
}

@Test
void testAlterMergeEngineToAggregationRejectedAfterRescale() throws Exception {

@platinumhamburg platinumhamburg Sep 14, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

table.merge-engine cannot be altered regardless of whether the table has been rescaled, so this test only exercises the existing option whitelist. The rescale setup adds no coverage for the guard introduced by this PR. Could we remove this test and keep testAlterBucketNumRejectedOnAggregationTable to cover the new behavior?

// 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}")
Expand Down Expand Up @@ -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(
Expand Down
Loading