Skip to content

[server][client] Support per-partition bucket count for partitioned tables - #3908

Open
Kaixuan-Duan wants to merge 2 commits into
apache:mainfrom
Kaixuan-Duan:dev-resharding-phase1
Open

[server][client] Support per-partition bucket count for partitioned tables#3908
Kaixuan-Duan wants to merge 2 commits into
apache:mainfrom
Kaixuan-Duan:dev-resharding-phase1

Conversation

@Kaixuan-Duan

@Kaixuan-Duan Kaixuan-Duan commented Aug 9, 2026

Copy link
Copy Markdown
Contributor

Purpose

Linked issue: close #3907

Support per-partition bucket count for partitioned tables. After this change, ALTER TABLE ... SET ('bucket.num' = N) only affects the default for newly created partitions; existing partitions retain their original bucket count (bucket.num.actual) in their own PartitionRegistration. This enables online bucket rescaling without disrupting data already written to existing partitions.

The PR also introduces bucketLayoutEpoch, a table-level monotonically increasing version for bucket-layout changes, to detect stale client metadata and return STALE_METADATA when a request's bucket count does not match the server's current state.

Lake table is currently only implemented for Paimon.

Brief change log

  • Migrate bucket-count ownership from PhysicalTablePath to TablePartition across client, server, and connector modules so that bucket counts and BucketAssigners are keyed by the immutable TablePartition(tableId, partitionId).
  • Add bucketLayoutEpoch to TableRegistration, propagated through PbTableMetadata and GetTableInfoResponse; withBucketCount atomically replaces the table-level count and increments the epoch.
  • Add a per-table read/write lock so ALTER bucket.num and partition creation cannot interleave; partition creation always reads the fresh bucket count from ZK.
  • Change client-side dynamic partition creation from async to synchronous — bucket assignment blocks until the partition's partitionId and actual bucket count are present in the client metadata.
  • Commit the new table-level count, bucketLayoutEpoch + 1, and legacy-partition bucketCount backfill in one version-checked, epoch-fenced ZK transaction.
  • Add bucket_count to all bucket-routed request protos and validate it server-side; mismatch returns STALE_METADATA, which triggers metadata + BucketAssigner invalidation and batch failure on the client.
  • Add epoch-aware fallback in PartitionRegistration.getBucketCountOrDefault: use the persisted count when present, fall back to table-level when epoch is 0, throw StaleMetadataException when epoch > 0 and the count is missing.
  • Read table metadata before partitions in listPartitionInfos to prevent cross-ALTER inconsistency.
  • Reject historical partition lookup on rescaled tables (bucketLayoutEpoch > 0) with a TODO for future support.
  • Propagate bucket.num changes to the Paimon lake catalog separately from schema-alter branches.

Tests

./mvnw clean verify -pl fluss-common,fluss-rpc
./mvnw clean verify -pl fluss-client
./mvnw clean verify -pl fluss-server
./mvnw clean verify -pl fluss-flink/fluss-flink-common,fluss-flink/fluss-flink-tiering
./mvnw clean verify -pl fluss-lake/fluss-lake-paimon
./mvnw clean verify -pl fluss-lake/fluss-lake-iceberg,fluss-lake/fluss-lake-hudi,fluss-lake/fluss-lake-lance
./mvnw clean verify -pl fluss-spark/fluss-spark-common,fluss-spark/fluss-spark-3.5
./mvnw clean verify -pl fluss-spark/fluss-spark-ut
cargo fmt --all -- --check
cargo check -p fluss-rs

API and Format

  • Proto: Added optional int64 bucket_layout_epoch to PbTableMetadata and GetTableInfoResponse; optional int32 bucket_count to all bucket-routed request messages. All fields are optional for backward compatibility with old servers/clients.
  • ZK JSON serialization includes the new bucketCount / bucketLayoutEpoch fields; old data without them is deserialized with null/0 defaults.

Documentation

  • Updated DDL docs (ddl.md) — ALTER TABLE bucket.num semantics
  • Updated options docs (options.md) — bucket.num option description
  • Updated bucketing docs (bucketing.md) — per-partition bucket count behavior

@luoyuxia
luoyuxia requested a lite review from Copilot August 10, 2026 02:17
Copilot AI reviewed Aug 10, 2026
View reviewed changes

Copilot AI left a comment

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.

Pull request overview

This PR implements per-partition bucket counts for partitioned tables by moving “actual bucket count” ownership to partitions, introducing a table-level bucketLayoutEpoch to detect stale client metadata, and propagating bucket-count validation across server/client/connectors/lake integrations. It updates RPC/metadata serialization paths and tightens client behavior so bucket assignment cannot proceed until the partition’s bucket metadata is known.

Changes:

  • Add bucketLayoutEpoch to table metadata and use it to detect/resist stale bucket layouts (STALE_METADATA).
  • Persist and propagate per-partition bucket counts (bucket.num.actual) and update Flink/Spark/lake tiering logic to enumerate buckets per-partition.
  • Add bucket_count to bucket-routed RPCs and validate it on the TabletServer to fail fast on stale client routing.

Reviewed changes

Copilot reviewed 110 out of 110 changed files in this pull request and generated 4 comments.

Show a summary per file
File Description
website/docs/table-design/data-distribution/bucketing.md Document ALTER bucket.num semantics for partitioned tables
website/docs/engine-flink/options.md Clarify bucket.num option semantics (new partitions only)
website/docs/engine-flink/ddl.md Document ALTER TABLE ... SET ('bucket.num' = ...) behavior/limits
fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/read/SplitPlannerLakeBucketGuardTest.scala Spark guard test for out-of-range lake buckets
fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/lake/SparkLakeTableReadTestBase.scala Spark lake read sync uses per-partition bucket counts
fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/FlussMicroBatchStream.scala Spark micro-batch offset enumeration per-partition
fluss-server/src/test/java/org/apache/fluss/server/zk/ZooKeeperClientTest.java Update ZK partition registration tests with bucketCount arg
fluss-server/src/test/java/org/apache/fluss/server/zk/data/TableRegistrationJsonSerdeTest.java Include bucket_layout_epoch in expected table JSON
fluss-server/src/test/java/org/apache/fluss/server/zk/data/PartitionRegistrationJsonSerdeTest.java Partition registration v2 JSON + bucket_count backward-compat test
fluss-server/src/test/java/org/apache/fluss/server/testutils/PartitionMetadataAssert.java Avoid failing legacy assertions when expected bucketCount is null
fluss-server/src/test/java/org/apache/fluss/server/testutils/FlussClusterExtension.java Enumerate snapshot buckets using per-partition bucket counts
fluss-server/src/test/java/org/apache/fluss/server/metadata/ZkBasedMetadataProviderTest.java Update partition registration helper calls with bucket count
fluss-server/src/test/java/org/apache/fluss/server/coordinator/TableManagerTest.java Update partition metadata registration calls with bucket count
fluss-server/src/test/java/org/apache/fluss/server/coordinator/event/watcher/TableChangeWatcherTest.java Add ZK version usage in watcher test + updated registration calls
fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java Add ZK version parameter in table-change tests + registration updates
fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperClient.java Add versioned reads + atomic CAS+epoch-fenced backfill transaction
fluss-server/src/main/java/org/apache/fluss/server/zk/data/TableRegistrationJsonSerde.java Serialize/deserialize bucket_layout_epoch
fluss-server/src/main/java/org/apache/fluss/server/zk/data/TableRegistration.java Add bucketLayoutEpoch and atomic withBucketCount()
fluss-server/src/main/java/org/apache/fluss/server/zk/data/PartitionRegistrationJsonSerde.java Add optional bucket_count and bump serde version to v2
fluss-server/src/main/java/org/apache/fluss/server/zk/data/PartitionRegistration.java Persist per-partition bucket count + epoch-aware fallback/guard
fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java Propagate bucketLayoutEpoch + partition bucketCount in RPC metadata
fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java Validate request bucket_count and throw STALE_METADATA on mismatch
fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java ListPartitionInfos reads table first + use per-partition bucketCount
fluss-server/src/main/java/org/apache/fluss/server/replica/HistoricalLakeLookupManager.java Reject historical lookup on rescaled tables (epoch > 0)
fluss-server/src/main/java/org/apache/fluss/server/metadata/ZkBasedMetadataProvider.java Populate PartitionMetadata with bucket count
fluss-server/src/main/java/org/apache/fluss/server/metadata/ServerMetadataSnapshot.java Track partition bucket counts + epochs; validate request bucket_count
fluss-server/src/main/java/org/apache/fluss/server/metadata/PartitionMetadata.java Add optional bucketCount field
fluss-server/src/main/java/org/apache/fluss/server/metadata/CoordinatorMetadataProvider.java Populate PartitionMetadata with effective bucket count
fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorRequestBatch.java Include per-partition bucketCount in UpdateMetadata
fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java Expose coordinator ZK version + refresh auto-partition tables on bucket change
fluss-server/src/main/java/org/apache/fluss/server/coordinator/AutoPartitionManager.java Guard partition creation with rescale locks + read fresh bucket count from ZK
fluss-rust/crates/fluss/src/rpc/message/put_kv.rs Add bucket_count field (legacy None for now)
fluss-rust/crates/fluss/src/rpc/message/produce_log.rs Add bucket_count field (legacy None for now)
fluss-rust/crates/fluss/src/rpc/message/prefix_lookup.rs Add bucket_count field (legacy None for now)
fluss-rust/crates/fluss/src/rpc/message/lookup.rs Add bucket_count field (legacy None for now)
fluss-rust/crates/fluss/src/rpc/message/list_offsets.rs Add bucket_count field (legacy None for now)
fluss-rust/crates/fluss/src/rpc/message/limit_scan.rs Add bucket_count field (legacy None for now)
fluss-rust/crates/fluss/src/metadata/table_stats.rs Add bucket_count field (legacy None for now)
fluss-rust/crates/fluss/src/metadata/partition.rs Add bucket_count field (legacy None for now)
fluss-rust/crates/fluss/src/client/table/scanner.rs Add bucket_count to fetch-log requests (legacy None for now)
fluss-rpc/src/main/proto/FlussApi.proto Add bucket_layout_epoch + bucket_count optional fields for compatibility
fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/Errors.java Add STALE_METADATA error mapping to StaleMetadataException
fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringTest.java Verify partition bucket-count stamping across tiering rounds
fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/PaimonLakeCatalogTest.java Ensure user-facing Paimon bucket option remains rejected; Fluss bucket.num applies
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/PaimonLakeWriter.java Override Paimon BUCKET for fixed-bucket tables using resolved actual bucket count
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/append/AppendOnlyWriter.java Refactor writer creation helper
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeCatalog.java Apply Fluss bucket.num changes directly to Paimon BUCKET option
fluss-lake/fluss-lake-lance/src/test/java/org/apache/fluss/lake/lance/tiering/LanceTieringTest.java Provide WriterInitContext.bucketCount implementation in tests
fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/tiering/IcebergTieringTest.java Provide WriterInitContext.bucketCount implementation in tests
fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/IcebergLakeCatalog.java Reject bucket.num rescale for Iceberg (unsupported)
fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/tiering/HudiTieringTest.java Provide WriterInitContext.bucketCount implementation in tests
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/FlinkTestBase.java ZK partition registration includes bucket count
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/TieringWriterInitContextTest.java Validate partition bucket count handling (fallback vs fail-loud)
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumeratorTest.java PartitionInfo now includes bucketCount in test fixtures
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/undo/RecoveryOffsetManagerTest.java Recovery enumerates buckets per-partition; adds rescale-focused tests
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/lake/LakeSplitGeneratorTest.java Flink guard test for out-of-range lake buckets
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogITCase.java Update expectations: server rejects non-partitioned bucket.num alter
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/PushdownUtils.java Count(*) enumerates buckets per-partition
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringWriterInitContext.java Add WriterInitContext.bucketCount resolution/fail-loud for partitioned
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java Snapshot per-partition bucket counts for correct lake writer stamping
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringSplitGenerator.java Generate tiering splits per-partition bucket count
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlussOnlyBatchSplitGenerator.java Generate log splits per-partition bucket count
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java Carry partition bucketCount through enumerator partition model
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/undo/RecoveryOffsetManager.java Enumerate offsets and buckets per-partition bucket count
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/LakeSplitGenerator.java Use per-partition bucket counts; add fail-loud union-read guard
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/FlinkConnectorOptions.java Allow ALTER bucket.num; update option description
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/OrphanCleanUtils.java Enumerate buckets using PartitionInfo bucketCount fallback
fluss-common/src/test/java/org/apache/fluss/lake/writer/WriterInitContextTest.java Implement new WriterInitContext.bucketCount in tests
fluss-common/src/main/java/org/apache/fluss/metadata/TableInfo.java Add bucketLayoutEpoch to core TableInfo metadata
fluss-common/src/main/java/org/apache/fluss/metadata/PartitionInfo.java Make bucketCount a first-class field; add helper bucketCountOrDefault
fluss-common/src/main/java/org/apache/fluss/lake/writer/WriterInitContext.java Add bucketCount() to writer init contract
fluss-common/src/main/java/org/apache/fluss/lake/lakestorage/LakeCatalog.java Document bucket.num SetOption contract for lake catalogs
fluss-common/src/main/java/org/apache/fluss/cluster/Cluster.java Track bucket counts by TablePartition/tableId; invalidate with metadata
fluss-client/src/test/java/org/apache/fluss/client/write/SenderTest.java Test STALE_METADATA handling and BucketAssigner invalidation
fluss-client/src/test/java/org/apache/fluss/client/utils/ClientRpcMessageUtilsTest.java Test parsing partition bucket_count with fallback
fluss-client/src/test/java/org/apache/fluss/client/table/PartitionedTableITCase.java Update expectations: dynamic partition creation is synchronous now
fluss-client/src/test/java/org/apache/fluss/client/lookup/LookupSenderTest.java Ensure lookup requests carry pinned bucketCount and legacy omit behavior
fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java End-to-end test: Kv snapshots reflect per-partition bucket counts
fluss-client/src/main/java/org/apache/fluss/client/write/WriterClient.java Key BucketAssigner cache by TablePartition/tableId; pass bucketCount into batches
fluss-client/src/main/java/org/apache/fluss/client/write/WriteBatch.java Store bucketCount used to compute bucketId for request validation
fluss-client/src/main/java/org/apache/fluss/client/write/Sender.java Fail batch on STALE_METADATA; invalidate metadata + BucketAssigner
fluss-client/src/main/java/org/apache/fluss/client/write/RecordAccumulator.java Carry bucketCount into newly created write batches
fluss-client/src/main/java/org/apache/fluss/client/write/DynamicPartitionCreator.java Make dynamic partition creation synchronous and metadata-aware
fluss-client/src/main/java/org/apache/fluss/client/utils/MetadataUtils.java Rebuild Cluster with per-partition/table bucketCount maps
fluss-client/src/main/java/org/apache/fluss/client/utils/ClientRpcMessageUtils.java Add bucket_count to routed requests; parse PartitionInfo bucket_count with fallback
fluss-client/src/main/java/org/apache/fluss/client/table/scanner/TableScan.java Enumerate batch scan buckets per partition bucketCount
fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetcher.java Include bucket_count in fetchLog requests
fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/LimitBatchScanner.java Include bucket_count in limit scan requests
fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/KvBatchScanner.java Include bucket_count in scan requests
fluss-client/src/main/java/org/apache/fluss/client/lookup/PrimaryKeyLookuper.java Re-route bucketId by per-partition bucketCount; reject historical lookup on rescaled tables
fluss-client/src/main/java/org/apache/fluss/client/lookup/PrefixLookupQuery.java Carry bucketCount through prefix lookup query
fluss-client/src/main/java/org/apache/fluss/client/lookup/PrefixLookupBatch.java Store pinned bucketCount for prefix batch requests
fluss-client/src/main/java/org/apache/fluss/client/lookup/PrefixKeyLookuper.java Compute bucketId after resolving per-partition bucketCount
fluss-client/src/main/java/org/apache/fluss/client/lookup/LookupSender.java Fail fast on STALE_METADATA; keep invalid-metadata refresh behavior
fluss-client/src/main/java/org/apache/fluss/client/lookup/LookupQuery.java Carry bucketCount through lookup query
fluss-client/src/main/java/org/apache/fluss/client/lookup/LookupClient.java Plumb bucketCount into lookup/prefixLookup
fluss-client/src/main/java/org/apache/fluss/client/lookup/LookupBatch.java Store pinned bucketCount for batch requests
fluss-client/src/main/java/org/apache/fluss/client/lookup/AbstractLookupQuery.java Add bucketCount field to lookup query base
fluss-client/src/main/java/org/apache/fluss/client/lookup/AbstractLookuper.java Add helper to resolve per-partition bucketCount from cluster metadata
fluss-client/src/main/java/org/apache/fluss/client/admin/FlussAdmin.java Parse bucketLayoutEpoch; resolve partition bucketCount fallback with epoch guard; include bucket_count in stats/offset requests
Suppressed comments (2)

fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java:1365

  • This comment says "ALTER bucket.num 4 -> 8" but the test variables change 2 -> 4 (originalBucketNum=2, newBucketNum=4). Keeping these numbers accurate is important for understanding what the test asserts.
        // ALTER bucket.num 4 -> 8.

fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java:1374

  • This comment says the new partition uses bucket.num.actual = 8, but newBucketNum is 4. The comment should reflect the actual post-ALTER bucket count used by the test.
        // New partition created AFTER the ALTER uses bucket.num.actual = 8.

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

Comment thread fluss-client/src/main/java/org/apache/fluss/client/write/DynamicPartitionCreator.java Outdated
Comment on lines +138 to +147
Throwable creationFailure = partitionCreationFailures.remove(physicalTablePath);
if (creationFailure != null) {
throw new FlussRuntimeException(
"Failed to dynamically create partition " + physicalTablePath,
creationFailure);
}
if (metadataUpdater.getPartitionId(physicalTablePath).isPresent()
|| forceCheckPartitionExist(physicalTablePath)) {
return;
}
Comment thread fluss-common/src/main/java/org/apache/fluss/metadata/TableInfo.java
Comment on lines +429 to +432
/**
* Returns the bucket layout epoch of the table. New tables start at 0; every committed
* bucket.num change increments it (see {@code TableRegistration#newBucketLayout(int)}).
*/
Comment thread fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeCatalog.java
Comment on lines +128 to +135
for (TableChange tableChange : tableChanges) {
if (tableChange instanceof TableChange.SetOption
&& BUCKET_NUM_PROPERTY.equals(((TableChange.SetOption) tableChange).getKey())) {
newBucketCount = Integer.parseInt(((TableChange.SetOption) tableChange).getValue());
} else {
remainingChanges.add(tableChange);
}
}
Comment thread fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java Outdated
admin.createTable(tablePath, partitionedPkTable, true).get();
long tableId = admin.getTableInfo(tablePath).get().getTableId();

// Old partition created BEFORE the ALTER retains bucket.num.actual = 4.
@Kaixuan-Duan

Kaixuan-Duan commented Aug 10, 2026

Copy link
Copy Markdown
Contributor Author

@luoyuxia Thanks for the review. I've addressed your comments. Could you please take another look?

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

Copilot code review Copilot
Copilot review effort
Applies to this pull request for everyone.Learn more about Copilot code review.
Copilot left review comments

Assignees

No one assigned

Labels

None yet

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

Support per-partition bucket rescale

2 participants

@Kaixuan-Duan