Repository navigation
Conversation
Route tiering table assignments by the configured table and service group. Keep group changes asynchronous, preserve default-group compatibility, and expose per-group tiering metrics.
|
Hi @polyzos, CI is green and this PR is ready for review. Could you take a look when you have a chance? It adds group-aware tiering table assignment with Flink-to-Paimon end-to-end coverage. Thanks! |
There was a problem hiding this comment.
Warning
Copilot couldn't run its full agentic review because it didn't start before the timeout. Make sure your repository has a runner available, or add a copilot-code-review.yml file specifying one with the runs-on attribute. See the docs for more details.
Copilot review overview
3 open issues
\"a\\nb\"passes because… · NewrequestTable(...)from O(1) queue polling to O(n) scanning on every heartbeat… · Newoption == ConfigOptions.TABLE_DATALAKE_TIERING_GROUP) makes the… · New
What changed in this PR
Adds group-aware tiering so operators can isolate tiering workloads by assigning tables and tiering services to matching “tiering groups”, with backward-compatible heartbeat/version handling.
Changes:
- Introduces table option
table.datalake.tiering-group(defaultdefault) and validates group names. - Updates coordinator scheduling and heartbeat flow to request/assign pending tables by tiering group.
- Adds/updates tests and documentation; bumps LAKE_TIERING_HEARTBEAT API max version and regenerates Rust protobuf bindings.
| File | Description |
|---|---|
| website/docs/streaming-lakehouse/tiering-service.md | Documents tiering group configuration, reassignment semantics, and upgrade/downgrade guidance |
| fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java | Adds new table config option for tiering group |
| fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java | Adds shared tiering-group validation helper |
| fluss-common/src/main/java/org/apache/fluss/config/TableConfig.java | Exposes getDataLakeTieringGroup() |
| fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java | Validates tiering-group option on descriptor parsing |
| fluss-server/src/main/java/org/apache/fluss/server/coordinator/LakeTableTieringManager.java | Stores per-table group and assigns pending tables by group |
| fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java | Routes heartbeat table requests to group-aware scheduling |
| fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java | Propagates ALTER TABLE tiering-group changes to manager |
| fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java | Bumps LAKE_TIERING_HEARTBEAT max version to 1 |
| fluss-rpc/src/main/proto/FlussApi.proto | Adds optional tiering_group field to heartbeat request |
| fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/client/ServerConnection.java | Enforces version compatibility when non-default group is used |
| fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSourceOptions.java | Adds tiering service option tiering.group |
| fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java | Includes group in all heartbeats and validates group |
| Tests (multiple) | Adds coverage for group validation, scheduling, heartbeat compatibility, and end-to-end tiering with groups |
| fluss-rust/crates/fluss/src/proto/fluss.rs | Regenerates Rust bindings for new heartbeat field |
Review effort: Lite
| /** Validates a tiering group shared by table configuration and tiering service heartbeats. */ | ||
| public static void validateTieringGroup(String group) { | ||
| if (group == null || group.isEmpty() || !group.equals(group.trim())) { | ||
| throw new IllegalConfigurationException( | ||
| "Tiering group must not be empty or have leading or trailing whitespace."); | ||
| } | ||
| } |
There was a problem hiding this comment.
Addressed in e84d7dce1.
The shared validator now reuses TablePath.validateTableName, so table configuration and tiering services follow the existing Fluss table-name rules: 1–200 ASCII letters, digits, _ or -, with the __ prefix reserved. This rejects embedded whitespace and control characters without adding a separate naming rule. Configuration descriptions and documentation are updated, with regression coverage for invalid characters and the length boundaries.
| /** Returns the next pending table in the requested group, or {@code null} if none exists. */ | ||
| @Nullable | ||
| public LakeTieringTableInfo requestTable(String group) { | ||
| return inWriteLock( | ||
| lock, | ||
| () -> { | ||
| Long tableId = pendingTieringTables.poll(); | ||
| // now no any pending table, return directly | ||
| if (tableId == null) { | ||
| return null; | ||
| } | ||
| TablePath tablePath = tablePaths.get(tableId); | ||
| // the table has been dropped, request again | ||
| if (tablePath == null) { | ||
| return requestTable(); | ||
| Iterator<Long> iterator = pendingTieringTables.iterator(); | ||
| while (iterator.hasNext()) { | ||
| long tableId = iterator.next(); | ||
| TablePath tablePath = tablePaths.get(tableId); | ||
| if (tablePath == null) { | ||
| iterator.remove(); | ||
| continue; | ||
| } | ||
| if (!group.equals(tableTieringGroups.get(tableId))) { | ||
| continue; | ||
| } | ||
| iterator.remove(); | ||
| doHandleStateChange(tableId, TieringState.Tiering); | ||
| long tieringEpoch = tableTierEpoch.get(tableId); | ||
| return new LakeTieringTableInfo(tableId, tablePath, tieringEpoch); | ||
| } | ||
| doHandleStateChange(tableId, TieringState.Tiering); | ||
| long tieringEpoch = tableTierEpoch.get(tableId); | ||
| return new LakeTieringTableInfo(tableId, tablePath, tieringEpoch); | ||
| return null; | ||
| }); | ||
| } |
There was a problem hiding this comment.
Addressed in e84d7dce1.
Pending tables are now indexed by group using Map<String, LinkedHashSet<Long>>. This preserves FIFO order within each group and avoids scanning other groups on a request. Pending group changes and table removal update the queues and pending count under the same lock; an already assigned round retains its existing assignment and epoch. Added a regression for ordering and counts after group moves and deletions. LakeTableTieringManagerTest and LakeTieringHeartbeatITCase pass.
| private static void validateOptionValue(ReadableConfig options, ConfigOption<?> option) { | ||
| try { | ||
| options.get(option); | ||
| if (option == ConfigOptions.TABLE_DATALAKE_TIERING_GROUP) { | ||
| FlussConfigUtils.validateTieringGroup( | ||
| options.get(ConfigOptions.TABLE_DATALAKE_TIERING_GROUP)); | ||
| } | ||
| } catch (Throwable t) { | ||
| throw new InvalidConfigException( | ||
| String.format( |
There was a problem hiding this comment.
Addressed in e84d7dce1.
Validation now compares the configuration option by key and reads the tiering-group value only once before passing it to the shared validator. TableDescriptorValidationTest passes.
Index pending tables by group with FIFO ordered sets, keeping reassignment, removal and pending metrics synchronized under the coordinator lock. Reuse table-name validation for tiering groups and compare configuration options by key. Document the naming contract and cover validation boundaries and queue behavior after group changes and table removal.
Purpose
Linked issue: close #3789
Allow operators to assign tables to separate tiering services so workloads with different resource or freshness requirements can be scaled independently. Services in the same group continue to share work.
Brief change log
table.datalake.tiering-groupand service option--fluss.tiering.group, both defaulting todefault, with shared group-name validation.Tests
Passed 90 unique targeted test cases after integrating
apache/mainatb25ba46d8:TableConfigTest,TableDescriptorValidationTestServerConnectionTestLakeTableTieringManagerTest,LakeTieringHeartbeatITCaseTieringSourceEnumeratorSchedulingTest,TieringSourceEnumeratorTest,TieringCommitOperatorTestFlussLakeTieringTestPaimonTieringITCase#testTieringWithGroupsand the existingtestTieringend-to-end caseThe affected modules and their full dependency reactor passed Maven
verifyon JDK 17, including Spotless, Checkstyle, and validation checks. Both Paimon end-to-end cases passed twice after integrating the upstream tiering refactor. The full repository test suite has not been run.Rust protobuf regeneration is deterministic with protoc 27.1, and
cargo check --package fluss-rs --lib --locked --offlinepassed.API and Format
Add the optional
tiering_groupfield to the internal lake-tiering heartbeat request and advertise heartbeat protocol version 1 while retaining version 0 support. An absent group meansdefault; non-default group requests are rejected before being sent to an older Coordinator. Storage formats are unchanged.Documentation
Update
website/docs/streaming-lakehouse/tiering-service.mdwith group configuration, dynamic reassignment, default behavior, legacy-table enablement, and upgrade/downgrade guidance.