From d4dfa91f44f618f8a8807438ba9c99a6e0b07c15 Mon Sep 17 00:00:00 2001 From: litiliu <38579068+litiliu@users.noreply.github.com> Date: Tue, 20 Oct 2026 10:55:24 +0800 Subject: [PATCH 1/5] [server] Support tiering service groups 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. --- .../apache/fluss/config/ConfigOptions.java | 10 ++ .../apache/fluss/config/FlussConfigUtils.java | 9 ++ .../org/apache/fluss/config/TableConfig.java | 5 + .../apache/fluss/config/TableConfigTest.java | 12 ++ .../tiering/source/TieringSourceOptions.java | 8 + .../enumerator/TieringSourceEnumerator.java | 15 +- ...TieringSourceEnumeratorSchedulingTest.java | 49 +++++- .../rpc/netty/client/ServerConnection.java | 20 +++ .../apache/fluss/rpc/protocol/ApiKeys.java | 3 +- fluss-rpc/src/main/proto/FlussApi.proto | 2 + .../netty/client/ServerConnectionTest.java | 39 +++++ .../CoordinatorEventProcessor.java | 6 + .../coordinator/CoordinatorService.java | 7 +- .../coordinator/LakeTableTieringManager.java | 139 ++++++++++++++++-- .../utils/TableDescriptorValidation.java | 5 + .../LakeTableTieringManagerTest.java | 112 +++++++++++++- .../LakeTieringHeartbeatITCase.java | 68 ++++++++- .../utils/TableDescriptorValidationTest.java | 26 ++++ .../streaming-lakehouse/tiering-service.md | 13 +- 19 files changed, 523 insertions(+), 25 deletions(-) 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 f4e71c4225c..0d99a468896 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 @@ -2181,6 +2181,16 @@ public class ConfigOptions { "Specifies the table name of the datalake table. This option is currently supported only for Paimon. " + "If not set, the Fluss table name is used. The option may be configured before the Paimon table is created and cannot be changed after creation."); + public static final ConfigOption TABLE_DATALAKE_TIERING_GROUP = + key("table.datalake.tiering-group") + .stringType() + .defaultValue("default") + .withDescription( + "The tiering service group responsible for this table. " + + "Only tiering services in the same group receive the table. " + + "Group names are case-sensitive and must not be empty or have leading or trailing whitespace. " + + "Changes through ALTER TABLE apply asynchronously to subsequent tiering rounds."); + public static final ConfigOption TABLE_DATALAKE_FRESHNESS = key("table.datalake.freshness") .durationType() diff --git a/fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java b/fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java index 415103c46ea..62a640dca84 100644 --- a/fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java +++ b/fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java @@ -56,6 +56,7 @@ public class FlussConfigUtils { ConfigOptions.TABLE_DATALAKE_HISTORICAL_PARTITION_LOOKUP_MODE.key(), ConfigOptions.TABLE_DATALAKE_DATABASE_NAME.key(), ConfigOptions.TABLE_DATALAKE_TABLE_NAME.key(), + ConfigOptions.TABLE_DATALAKE_TIERING_GROUP.key(), ConfigOptions.TABLE_DATALAKE_FRESHNESS.key(), ConfigOptions.TABLE_DATALAKE_AUTO_COMPACTION.key(), ConfigOptions.TABLE_LOG_TTL.key(), @@ -76,6 +77,14 @@ public static boolean isAlterableTableOption(String key) { return ALTERABLE_TABLE_OPTIONS.contains(key); } + /** 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."); + } + } + /** * Returns the default remote data directory from the configuration. Used as a fallback for * tables or partitions that do not contain remote data directory metadata. diff --git a/fluss-common/src/main/java/org/apache/fluss/config/TableConfig.java b/fluss-common/src/main/java/org/apache/fluss/config/TableConfig.java index ed2c111c335..838b9a8c0e3 100644 --- a/fluss-common/src/main/java/org/apache/fluss/config/TableConfig.java +++ b/fluss-common/src/main/java/org/apache/fluss/config/TableConfig.java @@ -148,6 +148,11 @@ public Optional getDataLakeFormat() { return config.getOptional(ConfigOptions.TABLE_DATALAKE_FORMAT); } + /** Gets the tiering service group responsible for this table. */ + public String getDataLakeTieringGroup() { + return config.get(ConfigOptions.TABLE_DATALAKE_TIERING_GROUP); + } + /** * Gets the data lake freshness of the table. It defines the maximum amount of time that the * datalake table's content should lag behind updates to the Fluss table. diff --git a/fluss-common/src/test/java/org/apache/fluss/config/TableConfigTest.java b/fluss-common/src/test/java/org/apache/fluss/config/TableConfigTest.java index 4a732beb7c3..9467fda9d57 100644 --- a/fluss-common/src/test/java/org/apache/fluss/config/TableConfigTest.java +++ b/fluss-common/src/test/java/org/apache/fluss/config/TableConfigTest.java @@ -28,6 +28,18 @@ /** Tests for {@link TableConfig}. */ class TableConfigTest { + @Test + void testDataLakeTieringGroup() { + Configuration conf = new Configuration(); + TableConfig tableConfig = new TableConfig(conf); + + assertThat(tableConfig.getDataLakeTieringGroup()).isEqualTo("default"); + conf.set(ConfigOptions.TABLE_DATALAKE_TIERING_GROUP, "large"); + assertThat(tableConfig.getDataLakeTieringGroup()).isEqualTo("large"); + conf.removeConfig(ConfigOptions.TABLE_DATALAKE_TIERING_GROUP); + assertThat(tableConfig.getDataLakeTieringGroup()).isEqualTo("default"); + } + @Test void testDeleteBehavior() { Configuration conf = new Configuration(); diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSourceOptions.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSourceOptions.java index e281dba671e..3e1298e0fb7 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSourceOptions.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSourceOptions.java @@ -18,6 +18,7 @@ package org.apache.fluss.flink.tiering.source; import org.apache.fluss.config.ConfigOption; +import org.apache.fluss.config.ConfigOptions; import java.time.Duration; @@ -28,6 +29,13 @@ public class TieringSourceOptions { public static final String DATA_LAKE_CONFIG_PREFIX = "datalake."; + public static final ConfigOption TIERING_GROUP = + key("tiering.group") + .stringType() + .defaultValue(ConfigOptions.TABLE_DATALAKE_TIERING_GROUP.defaultValue()) + .withDescription( + "The tiering group of tables handled by this tiering service."); + public static final ConfigOption POLL_TIERING_TABLE_INTERVAL = key("tiering.poll.table.interval") .durationType() diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java index cc00e55e1a2..d72fa2facbd 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java @@ -24,10 +24,12 @@ import org.apache.fluss.client.metadata.MetadataUpdater; import org.apache.fluss.config.ConfigOptions; import org.apache.fluss.config.Configuration; +import org.apache.fluss.config.FlussConfigUtils; import org.apache.fluss.flink.metrics.FlinkMetricRegistry; import org.apache.fluss.flink.tiering.event.FailedTieringEvent; import org.apache.fluss.flink.tiering.event.FinishedTieringEvent; import org.apache.fluss.flink.tiering.event.TieringReachMaxDurationEvent; +import org.apache.fluss.flink.tiering.source.TieringSourceOptions; import org.apache.fluss.flink.tiering.source.split.TieringLogSplit; import org.apache.fluss.flink.tiering.source.split.TieringSplit; import org.apache.fluss.flink.tiering.source.split.TieringSplitGenerator; @@ -114,6 +116,7 @@ public class TieringSourceEnumerator private final ScheduledExecutorService timerService; private final SplitEnumeratorMetricGroup enumeratorMetricGroup; private final long pollTieringTableIntervalMs; + private final String tieringGroup; // Also inspected by asynchronous heartbeat requests. private final Deque pendingSplits; private final Set readersAwaitingSplit; @@ -182,6 +185,8 @@ public TieringSourceEnumerator( this.timerService = timerService; this.enumeratorMetricGroup = context.metricGroup(); this.pollTieringTableIntervalMs = pollTieringTableIntervalMs; + this.tieringGroup = flussConf.get(TieringSourceOptions.TIERING_GROUP); + FlussConfigUtils.validateTieringGroup(tieringGroup); this.pendingSplits = new ConcurrentLinkedDeque<>(); this.readersAwaitingSplit = Collections.synchronizedSet(new TreeSet<>()); this.tieringTableEpochs = new ConcurrentHashMap<>(); @@ -218,7 +223,7 @@ void start( try { LakeTieringHeartbeatResponse heartbeatResponse = waitHeartbeatResponse( - coordinatorGateway.lakeTieringHeartbeat(basicHeartBeat())); + coordinatorGateway.lakeTieringHeartbeat(basicHeartBeat(tieringGroup))); this.flussCoordinatorEpoch = heartbeatResponse.getCoordinatorEpoch(); LOG.info( "Register Tiering Service to Fluss Coordinator(epoch={}) success.", @@ -452,7 +457,7 @@ private void assignSplits() { Map currentFailedTableEpochs = new HashMap<>(this.failedTableEpochs); LakeTieringHeartbeatRequest tieringHeartbeatRequest = tieringTableHeartBeat( - basicHeartBeat(), + basicHeartBeat(tieringGroup), this.tieringTableEpochs, currentFinishedTables, currentFailedTableEpochs, @@ -624,7 +629,7 @@ public void close() throws IOException { failedTableEpochs.putAll(tieringTableEpochs); tieringTableEpochs.clear(); if (!failedTableEpochs.isEmpty()) { - reportFailedTable(basicHeartBeat(), failedTableEpochs); + reportFailedTable(basicHeartBeat(tieringGroup), failedTableEpochs); } try { LOG.info("Closing Tiering Source Enumerator of at {}.", System.currentTimeMillis()); @@ -677,8 +682,8 @@ private void reportFailedTable( @VisibleForTesting static class HeartBeatHelper { - static LakeTieringHeartbeatRequest basicHeartBeat() { - return new LakeTieringHeartbeatRequest(); + static LakeTieringHeartbeatRequest basicHeartBeat(String tieringGroup) { + return new LakeTieringHeartbeatRequest().setTieringGroup(tieringGroup); } static LakeTieringHeartbeatRequest heartBeatWithRequestNewTieringTable( diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumeratorSchedulingTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumeratorSchedulingTest.java index dd92b08fc72..0299c02b3e6 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumeratorSchedulingTest.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumeratorSchedulingTest.java @@ -20,6 +20,9 @@ import org.apache.fluss.client.admin.Admin; import org.apache.fluss.config.Configuration; import org.apache.fluss.flink.tiering.TestingLakeTieringFactory; +import org.apache.fluss.flink.tiering.event.FailedTieringEvent; +import org.apache.fluss.flink.tiering.event.FinishedTieringEvent; +import org.apache.fluss.flink.tiering.source.TieringSourceOptions; import org.apache.fluss.flink.tiering.source.split.TieringLogSplit; import org.apache.fluss.flink.tiering.source.split.TieringSplit; import org.apache.fluss.flink.tiering.source.split.TieringSplitGenerator; @@ -35,6 +38,8 @@ import org.apache.fluss.types.DataTypes; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; import java.io.IOException; import java.util.ArrayDeque; @@ -55,6 +60,44 @@ /** Tests for split assignment and table scheduling in {@link TieringSourceEnumerator}. */ class TieringSourceEnumeratorSchedulingTest { + @ParameterizedTest + @ValueSource(strings = {"default", "large"}) + void testTieringGroupIncludedInEveryHeartbeat(String group) throws Throwable { + Configuration configuration = new Configuration(); + if (!group.equals(TieringSourceOptions.TIERING_GROUP.defaultValue())) { + configuration.set(TieringSourceOptions.TIERING_GROUP, group); + } + try (Fixture fixture = new Fixture(1, configuration)) { + configuration.set(TieringSourceOptions.TIERING_GROUP, "changed-after-start"); + fixture.addTable(1, false); + fixture.requestTable(); + fixture.context.runPeriodicCallable(0); + assertThat(fixture.requests.get(2).getTieringTablesList()) + .singleElement() + .satisfies(table -> assertThat(table.getTableId()).isEqualTo(1)); + + fixture.enumerator.handleSourceEvent(0, new FinishedTieringEvent(1)); + fixture.context.runNextOneTimeCallable(); + assertThat(fixture.requests.get(3).getFinishedTablesList()) + .singleElement() + .satisfies(table -> assertThat(table.getTableId()).isEqualTo(1)); + + fixture.addTable(2, false); + fixture.requestTable(); + fixture.enumerator.handleSourceEvent(0, new FailedTieringEvent(2, "test failure")); + fixture.context.runNextOneTimeCallable(); + assertThat(fixture.requests.get(5).getFailedTablesList()) + .singleElement() + .satisfies(table -> assertThat(table.getTableId()).isEqualTo(2)); + + assertThat(fixture.requests).hasSize(6); + assertThat(fixture.claims()).hasSize(2); + assertThat(fixture.requests) + .extracting(LakeTieringHeartbeatRequest::getTieringGroup) + .containsOnly(group); + } + } + @Test void testPendingSplitsAssignedInOrder() throws Exception { try (Fixture fixture = new Fixture()) { @@ -244,13 +287,17 @@ private Fixture() { } private Fixture(int parallelism) { + this(parallelism, new Configuration()); + } + + private Fixture(int parallelism, Configuration configuration) { this.context = new TestingContext(parallelism); CoordinatorGateway gateway = mock(CoordinatorGateway.class); when(gateway.lakeTieringHeartbeat(any())) .thenAnswer(invocation -> respondToHeartbeat(invocation.getArgument(0))); enumerator = new TieringSourceEnumerator( - new Configuration(), + configuration, new Configuration(), context, new TestingLakeTieringFactory(), diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/client/ServerConnection.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/client/ServerConnection.java index 8eb9d805c2b..2b5c8ba85f7 100644 --- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/client/ServerConnection.java +++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/client/ServerConnection.java @@ -20,6 +20,7 @@ import org.apache.fluss.annotation.VisibleForTesting; import org.apache.fluss.cluster.ServerNode; import org.apache.fluss.cluster.ServerType; +import org.apache.fluss.config.ConfigOptions; import org.apache.fluss.exception.DisconnectException; import org.apache.fluss.exception.FlussRuntimeException; import org.apache.fluss.exception.InvalidServerTypeException; @@ -32,6 +33,7 @@ import org.apache.fluss.rpc.messages.ApiVersionsResponse; import org.apache.fluss.rpc.messages.AuthenticateRequest; import org.apache.fluss.rpc.messages.AuthenticateResponse; +import org.apache.fluss.rpc.messages.LakeTieringHeartbeatRequest; import org.apache.fluss.rpc.messages.ProduceLogRequest; import org.apache.fluss.rpc.messages.PutKvRequest; import org.apache.fluss.rpc.metrics.ClientMetricGroup; @@ -77,6 +79,7 @@ final class ServerConnection { private static final short HISTORICAL_PRODUCE_LOG_MIN_VERSION = 1; private static final short HISTORICAL_PUT_KV_MIN_VERSION = 3; private static final short ALTER_BUCKET_COUNT_MIN_VERSION = 1; + private static final short TIERING_GROUP_MIN_VERSION = 1; private final ServerNode node; @@ -426,6 +429,23 @@ void validateVersionCompatibility(ApiKeys apiKey, short version, ApiMessage rawR + '.'); } } + + if (apiKey == ApiKeys.LAKE_TIERING_HEARTBEAT && version < TIERING_GROUP_MIN_VERSION) { + LakeTieringHeartbeatRequest heartbeatRequest = (LakeTieringHeartbeatRequest) rawRequest; + if (heartbeatRequest.hasTieringGroup() + && !ConfigOptions.TABLE_DATALAKE_TIERING_GROUP + .defaultValue() + .equals(heartbeatRequest.getTieringGroup())) { + throw new UnsupportedVersionException( + "Non-default tiering groups require LAKE_TIERING_HEARTBEAT version " + + TIERING_GROUP_MIN_VERSION + + " or newer, but server " + + node + + " negotiated version " + + version + + '.'); + } + } } private void handleApiVersionsResponse(ApiMessage response, Throwable cause) { diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java index 8c777f65148..a7e121d41ca 100644 --- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java +++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java @@ -89,7 +89,8 @@ public enum ApiKeys { CREATE_ACLS(1039, 0, 0, PUBLIC), LIST_ACLS(1040, 0, 0, PUBLIC), DROP_ACLS(1041, 0, 0, PUBLIC), - LAKE_TIERING_HEARTBEAT(1042, 0, 0, PRIVATE), + // Version 1: supports assigning tables by tiering group. + LAKE_TIERING_HEARTBEAT(1042, 0, 1, PRIVATE), CONTROLLED_SHUTDOWN(1043, 0, 0, PRIVATE), // Version 1: supports modifying the table distribution's bucket count. ALTER_TABLE(1044, 0, 1, PUBLIC), diff --git a/fluss-rpc/src/main/proto/FlussApi.proto b/fluss-rpc/src/main/proto/FlussApi.proto index 5a5294c614e..2da6ba6058d 100644 --- a/fluss-rpc/src/main/proto/FlussApi.proto +++ b/fluss-rpc/src/main/proto/FlussApi.proto @@ -797,6 +797,8 @@ message LakeTieringHeartbeatRequest { // These table IDs must also appear in finished_tables. The coordinator uses this field to // distinguish between normally finished tables and force finished tables. repeated int64 force_finished_tables = 5; + // The tiering group of the service. If absent, the default group is used. + optional string tiering_group = 6; } message LakeTieringHeartbeatResponse { diff --git a/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/client/ServerConnectionTest.java b/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/client/ServerConnectionTest.java index c5e10f39706..166635c6352 100644 --- a/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/client/ServerConnectionTest.java +++ b/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/client/ServerConnectionTest.java @@ -20,6 +20,7 @@ import org.apache.fluss.cluster.Endpoint; import org.apache.fluss.cluster.ServerNode; import org.apache.fluss.cluster.ServerType; +import org.apache.fluss.config.ConfigOptions; import org.apache.fluss.config.Configuration; import org.apache.fluss.exception.DisconnectException; import org.apache.fluss.exception.InvalidServerTypeException; @@ -38,6 +39,7 @@ import org.apache.fluss.rpc.messages.ApiVersionsRequest; import org.apache.fluss.rpc.messages.ApiVersionsResponse; import org.apache.fluss.rpc.messages.GetTableSchemaRequest; +import org.apache.fluss.rpc.messages.LakeTieringHeartbeatRequest; import org.apache.fluss.rpc.messages.ListDatabasesRequest; import org.apache.fluss.rpc.messages.LookupRequest; import org.apache.fluss.rpc.messages.PbApiVersion; @@ -323,6 +325,43 @@ void testRejectBucketCountChangeForOldServer() throws Exception { } } + @Test + void testTieringGroupVersionCompatibility() throws Exception { + ServerConnection connection = + new ServerConnection( + bootstrap, + serverNode, + TestingClientMetricGroup.newInstance(), + clientAuthenticator, + (con, ignore) -> {}); + try { + connection.validateVersionCompatibility( + ApiKeys.LAKE_TIERING_HEARTBEAT, (short) 0, new LakeTieringHeartbeatRequest()); + connection.validateVersionCompatibility( + ApiKeys.LAKE_TIERING_HEARTBEAT, + (short) 0, + new LakeTieringHeartbeatRequest() + .setTieringGroup( + ConfigOptions.TABLE_DATALAKE_TIERING_GROUP.defaultValue())); + + LakeTieringHeartbeatRequest groupedRequest = + new LakeTieringHeartbeatRequest().setTieringGroup("large"); + connection.validateVersionCompatibility( + ApiKeys.LAKE_TIERING_HEARTBEAT, (short) 1, groupedRequest); + assertThatThrownBy( + () -> + connection.validateVersionCompatibility( + ApiKeys.LAKE_TIERING_HEARTBEAT, + (short) 0, + groupedRequest)) + .isInstanceOf(UnsupportedVersionException.class) + .hasMessageContaining("require LAKE_TIERING_HEARTBEAT version 1 or newer") + .hasMessageContaining("negotiated version 0"); + } finally { + connection.close().get(); + } + } + @Test void testRejectHistoricalWritesForOldServer() throws Exception { nettyServer.close(); diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java index 627c3af5350..abe589148a7 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java @@ -1013,6 +1013,12 @@ private void postAlterTableProperties(TableInfo oldTableInfo, TableInfo newTable lakeTableTieringManager.updateTableLakeFreshness( newTableInfo.getTableId(), newFreshness.toMillis()); } + String oldGroup = oldTableInfo.getTableConfig().getDataLakeTieringGroup(); + String newGroup = newTableInfo.getTableConfig().getDataLakeTieringGroup(); + if (!Objects.equals(oldGroup, newGroup)) { + lakeTableTieringManager.updateTableTieringGroup( + newTableInfo.getTableId(), newGroup); + } } AutoPartitionStrategy oldAutoPartitionStrategy = diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java index df8713338a8..a99a4456437 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java @@ -1301,6 +1301,11 @@ public CompletableFuture commitLakeTableSnapsho @Override public CompletableFuture lakeTieringHeartbeat( LakeTieringHeartbeatRequest request) { + String group = + request.hasTieringGroup() + ? request.getTieringGroup() + : ConfigOptions.TABLE_DATALAKE_TIERING_GROUP.defaultValue(); + FlussConfigUtils.validateTieringGroup(group); LakeTieringHeartbeatResponse heartbeatResponse = new LakeTieringHeartbeatResponse(); int currentCoordinatorEpoch = coordinatorEpochSupplier.get(); heartbeatResponse.setCoordinatorEpoch(currentCoordinatorEpoch); @@ -1362,7 +1367,7 @@ public CompletableFuture lakeTieringHeartbeat( } if (request.hasRequestTable() && request.isRequestTable()) { - LakeTieringTableInfo lakeTieringTableInfo = lakeTableTieringManager.requestTable(); + LakeTieringTableInfo lakeTieringTableInfo = lakeTableTieringManager.requestTable(group); if (lakeTieringTableInfo != null) { heartbeatResponse .setTieringTable() diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/LakeTableTieringManager.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/LakeTableTieringManager.java index 019ec9858df..a0ccb3a6e9c 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/LakeTableTieringManager.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/LakeTableTieringManager.java @@ -18,6 +18,7 @@ package org.apache.fluss.server.coordinator; import org.apache.fluss.annotation.VisibleForTesting; +import org.apache.fluss.config.ConfigOptions; import org.apache.fluss.exception.FencedTieringEpochException; import org.apache.fluss.exception.FlussRuntimeException; import org.apache.fluss.exception.TableNotExistException; @@ -48,6 +49,7 @@ import java.util.Collections; import java.util.EnumSet; import java.util.HashMap; +import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.Queue; @@ -126,6 +128,14 @@ public class LakeTableTieringManager implements AutoCloseable { // table_id -> table path private final Map tablePaths; + // table_id -> configured tiering group + private final Map tableTieringGroups; + + // table_id -> group that owns the current tiering round + private final Map runningTieringGroups; + + private final Map groupMetrics; + // table_id -> freshness (tiering interval) private final Map tableLakeFreshness; @@ -179,6 +189,9 @@ protected LakeTableTieringManager( this.tieringStates = new HashMap<>(); this.liveTieringTableIds = new HashMap<>(); this.tablePaths = new HashMap<>(); + this.tableTieringGroups = new HashMap<>(); + this.runningTieringGroups = new HashMap<>(); + this.groupMetrics = new HashMap<>(); this.tableLakeFreshness = new HashMap<>(); this.expirationReaper = new LakeTieringExpiredOperationReaper(); expirationReaper.start(); @@ -203,6 +216,37 @@ private void registerMetrics() { () -> inReadLock(lock, liveTieringTableIds::size)); } + @GuardedBy("lock") + private void registerGroupMetrics(String group) { + groupMetrics.computeIfAbsent( + group, + name -> { + MetricGroup metricGroup = tieringMetricGroup.addGroup("tieringGroup", name); + metricGroup.gauge( + MetricNames.LAKE_TIERING_PENDING_TABLES_COUNT, + () -> getPendingTablesCount(name)); + metricGroup.gauge( + MetricNames.LAKE_TIERING_RUNNING_TABLES_COUNT, + () -> getRunningTablesCount(name)); + metricGroup.gauge( + MetricNames.LAKE_TIERING_TABLE_PENDING_TIME, + () -> getOldestPendingTime(name)); + return metricGroup; + }); + } + + @GuardedBy("lock") + private void removeUnusedGroupMetrics(@Nullable String group) { + if (group != null + && !tableTieringGroups.containsValue(group) + && !runningTieringGroups.containsValue(group)) { + MetricGroup metricGroup = groupMetrics.remove(group); + if (metricGroup != null) { + metricGroup.close(); + } + } + } + public void initWithLakeTables(List> tableInfoWithTieredTime) { inWriteLock( lock, @@ -234,6 +278,9 @@ public void addNewLakeTable(TableInfo tableInfo) { private void registerLakeTable(TableInfo tableInfo, long lastTieredTime) { long tableId = tableInfo.getTableId(); tablePaths.put(tableId, tableInfo.getTablePath()); + String group = tableInfo.getTableConfig().getDataLakeTieringGroup(); + tableTieringGroups.put(tableId, group); + registerGroupMetrics(group); tableLakeFreshness.put( tableId, tableInfo.getTableConfig().getDataLakeFreshness().toMillis()); lastTieringResult.put(tableId, LastTieringResult.initial(lastTieredTime)); @@ -333,6 +380,10 @@ public void removeLakeTable(long tableId) { lock, () -> { tablePaths.remove(tableId); + String group = tableTieringGroups.remove(tableId); + String runningGroup = runningTieringGroups.remove(tableId); + removeUnusedGroupMetrics(group); + removeUnusedGroupMetrics(runningGroup); tableLakeFreshness.remove(tableId); lastTieringResult.remove(tableId); currentTieringStartTime.remove(tableId); @@ -351,6 +402,26 @@ public void removeLakeTable(long tableId) { }); } + /** Updates the group used for subsequent assignments without interrupting a running round. */ + public void updateTableTieringGroup(long tableId, String group) { + inWriteLock( + lock, + () -> { + String previousGroup = tableTieringGroups.get(tableId); + if (previousGroup == null || previousGroup.equals(group)) { + return; + } + tableTieringGroups.put(tableId, group); + registerGroupMetrics(group); + removeUnusedGroupMetrics(previousGroup); + LOG.info( + "Updated tiering group for table {} from {} to {}.", + tableId, + previousGroup, + group); + }); + } + /** * Update the lake freshness for a table. This method should be called when the table's datalake * freshness property is changed via ALTER TABLE. @@ -422,22 +493,33 @@ protected void checkTieringServiceTimeout() { @Nullable public LakeTieringTableInfo requestTable() { + return requestTable(ConfigOptions.TABLE_DATALAKE_TIERING_GROUP.defaultValue()); + } + + /** 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 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); + runningTieringGroups.put(tableId, group); + 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; }); } @@ -621,10 +703,12 @@ private void doHandleStateChange(long tableId, TieringState targetState) { pendingEnterTime.remove(tableId); break; case Tiered: + removeUnusedGroupMetrics(runningTieringGroups.remove(tableId)); liveTieringTableIds.remove(tableId); currentTieringStartTime.remove(tableId); break; case Failed: + removeUnusedGroupMetrics(runningTieringGroups.remove(tableId)); Counter counter = tableFailureCounters.get(tableId); if (counter != null) { counter.inc(); @@ -832,6 +916,37 @@ int getRunningTablesCount() { return inReadLock(lock, liveTieringTableIds::size); } + @VisibleForTesting + long getPendingTablesCount(String group) { + return inReadLock( + lock, + () -> + pendingEnterTime.keySet().stream() + .filter(tableId -> group.equals(tableTieringGroups.get(tableId))) + .count()); + } + + @VisibleForTesting + long getRunningTablesCount(String group) { + return inReadLock( + lock, () -> runningTieringGroups.values().stream().filter(group::equals).count()); + } + + @VisibleForTesting + long getOldestPendingTime(String group) { + return inReadLock( + lock, + () -> + pendingEnterTime.entrySet().stream() + .filter( + entry -> + group.equals( + tableTieringGroups.get(entry.getKey()))) + .mapToLong(entry -> clock.milliseconds() - entry.getValue()) + .max() + .orElse(0L)); + } + @VisibleForTesting Long getTableLastSuccessTime(long tableId) { return inReadLock( 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 2b5d7d82c83..a32b3799ab7 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 @@ -22,6 +22,7 @@ import org.apache.fluss.config.ConfigOption; import org.apache.fluss.config.ConfigOptions; import org.apache.fluss.config.Configuration; +import org.apache.fluss.config.FlussConfigUtils; import org.apache.fluss.config.ReadableConfig; import org.apache.fluss.config.TableConfig; import org.apache.fluss.exception.InvalidAlterTableException; @@ -839,6 +840,10 @@ private static void checkDeleteBehavior(Configuration tableConf, boolean hasPrim 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( diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/LakeTableTieringManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/LakeTableTieringManagerTest.java index 13b273c1cb8..c6afa22527e 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/LakeTableTieringManagerTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/LakeTableTieringManagerTest.java @@ -35,6 +35,8 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; import java.time.Duration; import java.util.Arrays; @@ -265,6 +267,95 @@ void testTieringFail() { assertRequestTable(tableId1, tablePath1, 2); } + @Test + void testRequestTablesByGroup() { + TablePath defaultPath = TablePath.of("db", "default_table"); + TablePath groupPath1 = TablePath.of("db", "group_table1"); + TablePath groupPath2 = TablePath.of("db", "group_table2"); + tableTieringManager.initWithLakeTables( + Arrays.asList( + Tuple2.of( + createTableInfo(1L, groupPath1, Duration.ofSeconds(10), "large"), + -10_000L), + Tuple2.of( + createTableInfo(2L, defaultPath, Duration.ofSeconds(10)), + -10_000L))); + tableTieringManager.addNewLakeTable( + createTableInfo(3L, groupPath2, Duration.ofSeconds(10), "large")); + + assertRequestTable(2L, defaultPath, 1L); + assertThat(tableTieringManager.requestTable()).isNull(); + assertThat(tableTieringManager.requestTable("unused")).isNull(); + assertRequestTable(1L, groupPath1, 1L, "large"); + assertThat(tableTieringManager.requestTable("large")).isNull(); + manualClock.advanceTime(Duration.ofSeconds(10)); + assertRequestTable(3L, groupPath2, 1L, "large"); + assertThat(tableTieringManager.requestTable("large")).isNull(); + } + + @Test + void testUpdateGroupWhileScheduledAndPending() { + TablePath tablePath = TablePath.of("db", "table"); + tableTieringManager.addNewLakeTable( + createTableInfo(1L, tablePath, Duration.ofSeconds(10), "large")); + tableTieringManager.updateTableTieringGroup(1L, "small"); + manualClock.advanceTime(Duration.ofSeconds(10)); + waitValue( + () -> + tableTieringManager.getPendingTablesCount("small") == 1 + ? Optional.of(true) + : Optional.empty(), + Duration.ofSeconds(10), + "Table should be pending in the updated group"); + + manualClock.advanceTime(Duration.ofSeconds(5)); + tableTieringManager.updateTableTieringGroup(1L, "large"); + assertThat(tableTieringManager.requestTable("small")).isNull(); + assertThat(tableTieringManager.getPendingTablesCount("small")).isZero(); + assertThat(tableTieringManager.getPendingTablesCount("large")).isEqualTo(1); + assertThat(tableTieringManager.getOldestPendingTime("large")).isEqualTo(5000L); + assertRequestTable(1L, tablePath, 1L, "large"); + } + + @ParameterizedTest + @ValueSource(strings = {"finish", "force-finish", "fail", "timeout"}) + void testUpdateGroupWhileTiering(String outcome) { + TablePath tablePath = TablePath.of("db", "table"); + tableTieringManager.addNewLakeTable( + createTableInfo(1L, tablePath, Duration.ofSeconds(10), "large")); + manualClock.advanceTime(Duration.ofSeconds(10)); + assertRequestTable(1L, tablePath, 1L, "large"); + tableTieringManager.updateTableTieringGroup(1L, "small"); + assertThat(tableTieringManager.getRunningTablesCount("large")).isEqualTo(1); + assertThat(tableTieringManager.getRunningTablesCount("small")).isZero(); + tableTieringManager.renewTieringHeartbeat(1L, 1L); + assertThat(tableTieringManager.requestTable("small")).isNull(); + + switch (outcome) { + case "finish": + tableTieringManager.finishTableTiering(1L, 1L, false, TieringStats.UNKNOWN); + manualClock.advanceTime(Duration.ofSeconds(10)); + break; + case "force-finish": + tableTieringManager.finishTableTiering(1L, 1L, true, TieringStats.UNKNOWN); + break; + case "fail": + tableTieringManager.reportTieringFail(1L, 1L); + break; + case "timeout": + manualClock.advanceTime(Duration.ofMillis(TIERING_SERVICE_TIMEOUT_MS)); + tableTieringManager.checkTieringServiceTimeout(); + break; + default: + throw new IllegalArgumentException(outcome); + } + + assertThat(tableTieringManager.getRunningTablesCount("large")).isZero(); + assertThat(tableTieringManager.requestTable("large")).isNull(); + assertRequestTable(1L, tablePath, 2L, "small"); + assertThat(tableTieringManager.getRunningTablesCount("small")).isEqualTo(1); + } + @Test void testGlobalMetrics() throws Exception { // Initially no tables - verify counts are 0 @@ -454,11 +545,21 @@ void testForceFinishTableTieringImmediatelyRePending() { } private TableInfo createTableInfo(long tableId, TablePath tablePath, Duration freshness) { + return createTableInfo( + tableId, + tablePath, + freshness, + ConfigOptions.TABLE_DATALAKE_TIERING_GROUP.defaultValue()); + } + + private TableInfo createTableInfo( + long tableId, TablePath tablePath, Duration freshness, String group) { TableDescriptor tableDescriptor = TableDescriptor.builder() .schema(Schema.newBuilder().column("c1", DataTypes.INT()).build()) .property(ConfigOptions.TABLE_DATALAKE_ENABLED, true) .property(ConfigOptions.TABLE_DATALAKE_FRESHNESS, freshness) + .property(ConfigOptions.TABLE_DATALAKE_TIERING_GROUP, group) .distributedBy(1) .build(); @@ -473,9 +574,18 @@ private TableInfo createTableInfo(long tableId, TablePath tablePath, Duration fr } private void assertRequestTable(long tableId, TablePath tablePath, long tieredEpoch) { + assertRequestTable( + tableId, + tablePath, + tieredEpoch, + ConfigOptions.TABLE_DATALAKE_TIERING_GROUP.defaultValue()); + } + + private void assertRequestTable( + long tableId, TablePath tablePath, long tieredEpoch, String group) { LakeTieringTableInfo table = waitValue( - () -> Optional.ofNullable(tableTieringManager.requestTable()), + () -> Optional.ofNullable(tableTieringManager.requestTable(group)), Duration.ofSeconds(10), "Request tiering table timout"); assertThat(table).isEqualTo(new LakeTieringTableInfo(tableId, tablePath, tieredEpoch)); diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/LakeTieringHeartbeatITCase.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/LakeTieringHeartbeatITCase.java index d4be99e1b3d..33ffad2d067 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/LakeTieringHeartbeatITCase.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/LakeTieringHeartbeatITCase.java @@ -42,6 +42,7 @@ import java.time.Duration; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collections; import java.util.HashMap; import java.util.HashSet; import java.util.List; @@ -50,6 +51,7 @@ import java.util.Set; import static org.apache.fluss.server.coordinator.CoordinatorContext.INITIAL_COORDINATOR_EPOCH; +import static org.apache.fluss.server.testutils.RpcMessageTestUtils.newAlterTableRequest; import static org.apache.fluss.server.testutils.RpcMessageTestUtils.newCreateTableRequest; import static org.apache.fluss.testutils.common.CommonTestUtils.waitValue; import static org.assertj.core.api.Assertions.assertThat; @@ -195,6 +197,60 @@ void testLakeTieringHeartBeat() throws Exception { INITIAL_COORDINATOR_EPOCH, pbHeartbeatRespForTable.getTableId())); } + + // Grouped tables are not assigned to legacy requests without a group field. + TablePath groupedTable = TablePath.of("fluss", "test_lake_table_4"); + TableDescriptor groupedDescriptor = + TableDescriptor.builder() + .schema(Schema.newBuilder().column("f1", DataTypes.INT()).build()) + .property(ConfigOptions.TABLE_DATALAKE_ENABLED, true) + .property(ConfigOptions.TABLE_DATALAKE_FRESHNESS, Duration.ofMillis(100)) + .property(ConfigOptions.TABLE_DATALAKE_TIERING_GROUP, "large") + .build(); + coordinatorGateway + .createTable(newCreateTableRequest(groupedTable, groupedDescriptor, false)) + .get(); + verifyNoAnyTableToTier(); + verifyLakeTieringTableInfo(waitGetLakeTieringTableInfo("large"), 4L, 1L); + + coordinatorGateway + .alterTable( + newAlterTableRequest( + groupedTable, + Collections.singletonMap( + ConfigOptions.TABLE_DATALAKE_TIERING_GROUP.key(), "small"), + Collections.emptyList(), + Collections.emptyList(), + false)) + .get(); + request = makeTieringHeartbeatRequest(new long[0], new long[] {4L}, new long[0]); + request.setTieringGroup("large"); + request.addForceFinishedTable(4L); + assertThat( + coordinatorGateway + .lakeTieringHeartbeat(request) + .get() + .getFinishedTableRespsList()) + .allSatisfy(response -> assertThat(response.hasError()).isFalse()); + verifyLakeTieringTableInfo(waitGetLakeTieringTableInfo("small"), 4L, 2L); + + coordinatorGateway + .alterTable( + newAlterTableRequest( + groupedTable, + Collections.emptyMap(), + Collections.singletonList( + ConfigOptions.TABLE_DATALAKE_TIERING_GROUP.key()), + Collections.emptyList(), + false)) + .get(); + request = + makeTieringHeartbeatRequest( + new long[0], new long[] {4L}, new long[0], INITIAL_COORDINATOR_EPOCH, 2); + request.setTieringGroup("small"); + request.addForceFinishedTable(4L); + coordinatorGateway.lakeTieringHeartbeat(request).get(); + verifyLakeTieringTableInfo(waitGetLakeTieringTableInfo(), 4L, 3L); } private void verifyNoAnyTableToTier() throws Exception { @@ -231,12 +287,18 @@ private void verifyFencedTieringEpochException( } private PbLakeTieringTableInfo waitGetLakeTieringTableInfo() { + return waitGetLakeTieringTableInfo(null); + } + + private PbLakeTieringTableInfo waitGetLakeTieringTableInfo(String group) { return waitValue( () -> { + LakeTieringHeartbeatRequest request = makeRequestTableTieringHeartbeat(); + if (group != null) { + request.setTieringGroup(group); + } LakeTieringHeartbeatResponse response = - coordinatorGateway - .lakeTieringHeartbeat(makeRequestTableTieringHeartbeat()) - .get(); + coordinatorGateway.lakeTieringHeartbeat(request).get(); if (response.hasTieringTable()) { return Optional.of(response.getTieringTable()); } else { diff --git a/fluss-server/src/test/java/org/apache/fluss/server/utils/TableDescriptorValidationTest.java b/fluss-server/src/test/java/org/apache/fluss/server/utils/TableDescriptorValidationTest.java index 67c70e71958..c48e39ceebe 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/utils/TableDescriptorValidationTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/utils/TableDescriptorValidationTest.java @@ -54,6 +54,32 @@ /** Tests for {@link TableDescriptorValidation}. */ class TableDescriptorValidationTest { + @Test + void testTieringGroupValidation() { + String key = ConfigOptions.TABLE_DATALAKE_TIERING_GROUP.key(); + for (String group : new String[] {"default", "large", "Large"}) { + validate(logTableWithProperty(key, group), null); + } + for (String group : new String[] {"", " ", " large", "large "}) { + assertThatThrownBy(() -> validate(logTableWithProperty(key, group), null)) + .isInstanceOf(InvalidConfigException.class) + .hasMessageContaining(key); + } + + TableInfo lakeTable = + TableInfo.of( + TestData.DATA1_TABLE_PATH_PK, + TestData.DATA1_TABLE_ID_PK, + 1, + TestData.DATA1_TABLE_DESCRIPTOR_PK.withDataLakeFormat( + DataLakeFormat.PAIMON), + "file://remote", + 1L, + 1L); + TableDescriptorValidation.validateAlterTableProperties( + lakeTable, Collections.singleton(key)); + } + @Test void testCreateLogTableWithKvTTLFails() { assertThatThrownBy( diff --git a/website/docs/streaming-lakehouse/tiering-service.md b/website/docs/streaming-lakehouse/tiering-service.md index 8292f375d06..8ec60dea88b 100644 --- a/website/docs/streaming-lakehouse/tiering-service.md +++ b/website/docs/streaming-lakehouse/tiering-service.md @@ -91,6 +91,7 @@ The following `table.datalake.*` options are configured per table when creating | Option | Type | Default | Description | |--------|------|---------|-------------| | `table.datalake.enabled` | Boolean | false | Enable lakehouse storage for this table | +| `table.datalake.tiering-group` | String | default | Tiering service group responsible for this table | | `table.datalake.freshness` | Duration | 3min | Maximum lag between Fluss and lake data | | `table.datalake.auto-compaction` | Boolean | false | Auto-trigger compaction in the data lake | | `table.datalake.auto-expire-snapshot` | Boolean | false | Auto-expire snapshots in the data lake | @@ -126,7 +127,17 @@ ${FLINK_HOME}/bin/flink run \ ... ``` -Multiple tiering service jobs can run simultaneously. They are coordinated by the Fluss cluster to ensure exactly-once semantics and automatic load balancing. +Multiple tiering service jobs can run simultaneously. Jobs in the same tiering group share the tables assigned to that group. To use a separate set of resources for large tables, start a job with `--fluss.tiering.group large` and configure those tables with: + +```sql +ALTER TABLE large_table SET ('table.datalake.tiering-group' = 'large'); +``` + +Tables and jobs without an explicit group use `default`. Group names are case-sensitive and cannot be empty or contain leading or trailing whitespace. A table waits if no service in its group is available; it is not assigned to another group. + +Group changes take effect asynchronously after the Coordinator processes the table metadata update. An already assigned tiering round can finish in its original group; subsequent rounds use the updated group. `ALTER TABLE large_table RESET ('table.datalake.tiering-group')` restores the default group. Changing the group follows the existing lake catalog property update path. For legacy log tables without a recorded lake format, primary key, or bucket key, first enable datalake and then change the group in a separate ALTER. + +Upgrade all Coordinators before using non-default groups. A non-default group service fails when connected to a Coordinator that does not support groups. Default-group services retain the existing compatibility behavior. Before downgrading Coordinators, reset the `table.datalake.tiering-group` property on all tables and stop non-default-group services. ## Monitoring From 80dc09ac3763664f0c354ad8c9c044f10088ef23 Mon Sep 17 00:00:00 2001 From: litiliu <38579068+litiliu@users.noreply.github.com> Date: Tue, 20 Oct 2026 11:18:43 +0800 Subject: [PATCH 2/5] [rust] Regenerate tiering heartbeat protobuf --- fluss-rust/crates/fluss/src/proto/fluss.rs | 3 +++ 1 file changed, 3 insertions(+) diff --git a/fluss-rust/crates/fluss/src/proto/fluss.rs b/fluss-rust/crates/fluss/src/proto/fluss.rs index 5afd9d99156..5c571c4faa1 100644 --- a/fluss-rust/crates/fluss/src/proto/fluss.rs +++ b/fluss-rust/crates/fluss/src/proto/fluss.rs @@ -1051,6 +1051,9 @@ pub struct LakeTieringHeartbeatRequest { /// distinguish between normally finished tables and force finished tables. #[prost(int64, repeated, packed = "false", tag = "5")] pub force_finished_tables: ::prost::alloc::vec::Vec, + /// The tiering group of the service. If absent, the default group is used. + #[prost(string, optional, tag = "6")] + pub tiering_group: ::core::option::Option<::prost::alloc::string::String>, } #[derive(Clone, PartialEq, ::prost::Message)] pub struct LakeTieringHeartbeatResponse { From 6403f1acd82305c513a976ab322641abd3e69e01 Mon Sep 17 00:00:00 2001 From: litiliu <38579068+litiliu@users.noreply.github.com> Date: Tue, 20 Oct 2026 14:42:53 +0800 Subject: [PATCH 3/5] [server] Defer per-group tiering metrics --- .../coordinator/LakeTableTieringManager.java | 83 +------------------ .../LakeTableTieringManagerTest.java | 16 ++-- 2 files changed, 8 insertions(+), 91 deletions(-) diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/LakeTableTieringManager.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/LakeTableTieringManager.java index a0ccb3a6e9c..2f8ec323cf7 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/LakeTableTieringManager.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/LakeTableTieringManager.java @@ -131,11 +131,6 @@ public class LakeTableTieringManager implements AutoCloseable { // table_id -> configured tiering group private final Map tableTieringGroups; - // table_id -> group that owns the current tiering round - private final Map runningTieringGroups; - - private final Map groupMetrics; - // table_id -> freshness (tiering interval) private final Map tableLakeFreshness; @@ -190,8 +185,6 @@ protected LakeTableTieringManager( this.liveTieringTableIds = new HashMap<>(); this.tablePaths = new HashMap<>(); this.tableTieringGroups = new HashMap<>(); - this.runningTieringGroups = new HashMap<>(); - this.groupMetrics = new HashMap<>(); this.tableLakeFreshness = new HashMap<>(); this.expirationReaper = new LakeTieringExpiredOperationReaper(); expirationReaper.start(); @@ -216,37 +209,6 @@ private void registerMetrics() { () -> inReadLock(lock, liveTieringTableIds::size)); } - @GuardedBy("lock") - private void registerGroupMetrics(String group) { - groupMetrics.computeIfAbsent( - group, - name -> { - MetricGroup metricGroup = tieringMetricGroup.addGroup("tieringGroup", name); - metricGroup.gauge( - MetricNames.LAKE_TIERING_PENDING_TABLES_COUNT, - () -> getPendingTablesCount(name)); - metricGroup.gauge( - MetricNames.LAKE_TIERING_RUNNING_TABLES_COUNT, - () -> getRunningTablesCount(name)); - metricGroup.gauge( - MetricNames.LAKE_TIERING_TABLE_PENDING_TIME, - () -> getOldestPendingTime(name)); - return metricGroup; - }); - } - - @GuardedBy("lock") - private void removeUnusedGroupMetrics(@Nullable String group) { - if (group != null - && !tableTieringGroups.containsValue(group) - && !runningTieringGroups.containsValue(group)) { - MetricGroup metricGroup = groupMetrics.remove(group); - if (metricGroup != null) { - metricGroup.close(); - } - } - } - public void initWithLakeTables(List> tableInfoWithTieredTime) { inWriteLock( lock, @@ -278,9 +240,7 @@ public void addNewLakeTable(TableInfo tableInfo) { private void registerLakeTable(TableInfo tableInfo, long lastTieredTime) { long tableId = tableInfo.getTableId(); tablePaths.put(tableId, tableInfo.getTablePath()); - String group = tableInfo.getTableConfig().getDataLakeTieringGroup(); - tableTieringGroups.put(tableId, group); - registerGroupMetrics(group); + tableTieringGroups.put(tableId, tableInfo.getTableConfig().getDataLakeTieringGroup()); tableLakeFreshness.put( tableId, tableInfo.getTableConfig().getDataLakeFreshness().toMillis()); lastTieringResult.put(tableId, LastTieringResult.initial(lastTieredTime)); @@ -380,10 +340,7 @@ public void removeLakeTable(long tableId) { lock, () -> { tablePaths.remove(tableId); - String group = tableTieringGroups.remove(tableId); - String runningGroup = runningTieringGroups.remove(tableId); - removeUnusedGroupMetrics(group); - removeUnusedGroupMetrics(runningGroup); + tableTieringGroups.remove(tableId); tableLakeFreshness.remove(tableId); lastTieringResult.remove(tableId); currentTieringStartTime.remove(tableId); @@ -412,8 +369,6 @@ public void updateTableTieringGroup(long tableId, String group) { return; } tableTieringGroups.put(tableId, group); - registerGroupMetrics(group); - removeUnusedGroupMetrics(previousGroup); LOG.info( "Updated tiering group for table {} from {} to {}.", tableId, @@ -515,7 +470,6 @@ public LakeTieringTableInfo requestTable(String group) { } iterator.remove(); doHandleStateChange(tableId, TieringState.Tiering); - runningTieringGroups.put(tableId, group); long tieringEpoch = tableTierEpoch.get(tableId); return new LakeTieringTableInfo(tableId, tablePath, tieringEpoch); } @@ -703,12 +657,10 @@ private void doHandleStateChange(long tableId, TieringState targetState) { pendingEnterTime.remove(tableId); break; case Tiered: - removeUnusedGroupMetrics(runningTieringGroups.remove(tableId)); liveTieringTableIds.remove(tableId); currentTieringStartTime.remove(tableId); break; case Failed: - removeUnusedGroupMetrics(runningTieringGroups.remove(tableId)); Counter counter = tableFailureCounters.get(tableId); if (counter != null) { counter.inc(); @@ -916,37 +868,6 @@ int getRunningTablesCount() { return inReadLock(lock, liveTieringTableIds::size); } - @VisibleForTesting - long getPendingTablesCount(String group) { - return inReadLock( - lock, - () -> - pendingEnterTime.keySet().stream() - .filter(tableId -> group.equals(tableTieringGroups.get(tableId))) - .count()); - } - - @VisibleForTesting - long getRunningTablesCount(String group) { - return inReadLock( - lock, () -> runningTieringGroups.values().stream().filter(group::equals).count()); - } - - @VisibleForTesting - long getOldestPendingTime(String group) { - return inReadLock( - lock, - () -> - pendingEnterTime.entrySet().stream() - .filter( - entry -> - group.equals( - tableTieringGroups.get(entry.getKey()))) - .mapToLong(entry -> clock.milliseconds() - entry.getValue()) - .max() - .orElse(0L)); - } - @VisibleForTesting Long getTableLastSuccessTime(long tableId) { return inReadLock( diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/LakeTableTieringManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/LakeTableTieringManagerTest.java index c6afa22527e..7c1f868fe7b 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/LakeTableTieringManagerTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/LakeTableTieringManagerTest.java @@ -300,21 +300,21 @@ void testUpdateGroupWhileScheduledAndPending() { createTableInfo(1L, tablePath, Duration.ofSeconds(10), "large")); tableTieringManager.updateTableTieringGroup(1L, "small"); manualClock.advanceTime(Duration.ofSeconds(10)); + assertRequestTable(1L, tablePath, 1L, "small"); + + tableTieringManager.finishTableTiering(1L, 1L, false, TieringStats.UNKNOWN); + manualClock.advanceTime(Duration.ofSeconds(10)); waitValue( () -> - tableTieringManager.getPendingTablesCount("small") == 1 + tableTieringManager.getPendingTablesCount() == 1 ? Optional.of(true) : Optional.empty(), Duration.ofSeconds(10), "Table should be pending in the updated group"); - manualClock.advanceTime(Duration.ofSeconds(5)); tableTieringManager.updateTableTieringGroup(1L, "large"); assertThat(tableTieringManager.requestTable("small")).isNull(); - assertThat(tableTieringManager.getPendingTablesCount("small")).isZero(); - assertThat(tableTieringManager.getPendingTablesCount("large")).isEqualTo(1); - assertThat(tableTieringManager.getOldestPendingTime("large")).isEqualTo(5000L); - assertRequestTable(1L, tablePath, 1L, "large"); + assertRequestTable(1L, tablePath, 2L, "large"); } @ParameterizedTest @@ -326,8 +326,6 @@ void testUpdateGroupWhileTiering(String outcome) { manualClock.advanceTime(Duration.ofSeconds(10)); assertRequestTable(1L, tablePath, 1L, "large"); tableTieringManager.updateTableTieringGroup(1L, "small"); - assertThat(tableTieringManager.getRunningTablesCount("large")).isEqualTo(1); - assertThat(tableTieringManager.getRunningTablesCount("small")).isZero(); tableTieringManager.renewTieringHeartbeat(1L, 1L); assertThat(tableTieringManager.requestTable("small")).isNull(); @@ -350,10 +348,8 @@ void testUpdateGroupWhileTiering(String outcome) { throw new IllegalArgumentException(outcome); } - assertThat(tableTieringManager.getRunningTablesCount("large")).isZero(); assertThat(tableTieringManager.requestTable("large")).isNull(); assertRequestTable(1L, tablePath, 2L, "small"); - assertThat(tableTieringManager.getRunningTablesCount("small")).isEqualTo(1); } @Test From f692c96ac412b7322631e17ba1020cb42a3cdeed Mon Sep 17 00:00:00 2001 From: litiliu <38579068+litiliu@users.noreply.github.com> Date: Tue, 20 Oct 2026 15:20:03 +0800 Subject: [PATCH 4/5] [test] Cover group-aware Flink lake tiering --- .../paimon/tiering/PaimonTieringITCase.java | 69 +++++++++++++++++++ 1 file changed, 69 insertions(+) diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringITCase.java b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringITCase.java index 5a7da996f58..d4d6df6bdaf 100644 --- a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringITCase.java +++ b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringITCase.java @@ -23,9 +23,11 @@ import org.apache.fluss.config.ConfigOptions; import org.apache.fluss.config.Configuration; import org.apache.fluss.exception.LakeTableSnapshotNotExistException; +import org.apache.fluss.flink.tiering.LakeTieringJobBuilder; import org.apache.fluss.lake.paimon.testutils.FlinkPaimonTieringTestBase; import org.apache.fluss.lake.paimon.tiering.markdone.PartitionMarkDoneState; import org.apache.fluss.lake.paimon.tiering.markdone.PartitionMarkDoneStateJsonSerde; +import org.apache.fluss.metadata.DataLakeFormat; import org.apache.fluss.metadata.PartitionInfo; import org.apache.fluss.metadata.PartitionSpec; import org.apache.fluss.metadata.Schema; @@ -79,6 +81,8 @@ import java.util.stream.Stream; import static org.apache.fluss.flink.source.testutils.FlinkRowAssertionsUtils.collectRowsUntilEndWithTimeout; +import static org.apache.fluss.flink.tiering.source.TieringSourceOptions.POLL_TIERING_TABLE_INTERVAL; +import static org.apache.fluss.flink.tiering.source.TieringSourceOptions.TIERING_GROUP; import static org.apache.fluss.lake.committer.LakeCommitter.FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY; import static org.apache.fluss.lake.paimon.testutils.PaimonTestUtils.adjustToLegacyV1Table; import static org.apache.fluss.lake.paimon.tiering.markdone.PaimonPartitionMarkDone.MARK_DONE_STATE_PROPERTY; @@ -205,6 +209,71 @@ void testTiering() throws Exception { } } + @Test + void testTieringWithGroups() throws Exception { + TablePath groupedTable = TablePath.of(DEFAULT_DB, "largeGroupTable"); + TablePath defaultTable = TablePath.of(DEFAULT_DB, "defaultGroupTable"); + try (AutoCloseable defaultTableCleanup = () -> dropTable(defaultTable); + AutoCloseable groupedTableCleanup = () -> dropTable(groupedTable)) { + long groupedTableId = + createLogTable( + groupedTable, + 1, + false, + Collections.singletonMap( + ConfigOptions.TABLE_DATALAKE_TIERING_GROUP.key(), "large"), + Collections.emptyMap()); + TableBucket groupedBucket = new TableBucket(groupedTableId, 0); + List groupedRows = Arrays.asList(row(1, "large1"), row(2, "large2")); + writeRows(groupedTable, groupedRows, true); + + TableBucket defaultBucket = new TableBucket(createLogTable(defaultTable), 0); + List defaultRows = + Arrays.asList(row(1, "default1"), row(2, "default2"), row(3, "default3")); + writeRows(defaultTable, defaultRows.subList(0, 2), true); + + JobClient defaultJob = buildTieringJob(execEnv); + try (AutoCloseable defaultJobCleanup = () -> defaultJob.cancel().get()) { + assertReplicaStatus(defaultBucket, 2); + // Complete a second round while the large-group table is still waiting. + writeRows(defaultTable, defaultRows.subList(2, 3), true); + assertReplicaStatus(defaultBucket, 3); + checkDataInPaimonAppendOnlyTable(defaultTable, defaultRows, 0); + FileStoreTable groupedLakeTable = + (FileStoreTable) + paimonCatalog.getTable( + Identifier.create( + groupedTable.getDatabaseName(), + groupedTable.getTableName())); + assertThat(groupedLakeTable.snapshotManager().latestSnapshot()) + .as("The large-group table must wait for a matching tiering service") + .isNull(); + + Configuration groupedConf = new Configuration(clientConf); + groupedConf.set(TIERING_GROUP, "large"); + groupedConf.set(POLL_TIERING_TABLE_INTERVAL, Duration.ofMillis(500)); + StreamExecutionEnvironment groupedEnv = + StreamExecutionEnvironment.getExecutionEnvironment(); + groupedEnv.setParallelism(2); + groupedEnv.enableCheckpointing(1000); + JobClient groupedJob = + LakeTieringJobBuilder.newBuilder( + groupedEnv, + groupedConf, + Configuration.fromMap(getPaimonCatalogConf()), + new Configuration(), + DataLakeFormat.PAIMON.toString()) + .build(); + try (AutoCloseable groupedJobCleanup = () -> groupedJob.cancel().get()) { + assertReplicaStatus(groupedBucket, 2); + checkDataInPaimonAppendOnlyTable(groupedTable, groupedRows, 0); + checkFlussOffsetsInSnapshot( + groupedTable, Collections.singletonMap(groupedBucket, 2L)); + } + } + } + } + @Test void testTieringForLegacyTable() throws Exception { // FIP-27: verify that tiering still works for a legacy table that carries the three From e84d7dce19505837a9d5b61ce465d8a95b151017 Mon Sep 17 00:00:00 2001 From: litiliu <38579068+litiliu@users.noreply.github.com> Date: Thu, 22 Oct 2026 11:21:41 +0800 Subject: [PATCH 5/5] [server] Address group-aware tiering review feedback 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. --- .../apache/fluss/config/ConfigOptions.java | 2 +- .../apache/fluss/config/FlussConfigUtils.java | 10 ++- .../fluss/config/FlussConfigUtilsTest.java | 40 ++++++++++ .../tiering/source/TieringSourceOptions.java | 3 +- .../coordinator/LakeTableTieringManager.java | 78 ++++++++++++++++--- .../utils/TableDescriptorValidation.java | 5 +- .../LakeTableTieringManagerTest.java | 42 ++++++++++ .../utils/TableDescriptorValidationTest.java | 5 +- .../streaming-lakehouse/tiering-service.md | 2 +- 9 files changed, 167 insertions(+), 20 deletions(-) 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 40edac0990d..75bb3dcccca 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 @@ -2229,7 +2229,7 @@ public class ConfigOptions { .withDescription( "The tiering service group responsible for this table. " + "Only tiering services in the same group receive the table. " - + "Group names are case-sensitive and must not be empty or have leading or trailing whitespace. " + + "Group names are case-sensitive and must follow Fluss table name rules. " + "Changes through ALTER TABLE apply asynchronously to subsequent tiering rounds."); public static final ConfigOption TABLE_DATALAKE_FRESHNESS = diff --git a/fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java b/fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java index 62a640dca84..4036f0fe3cb 100644 --- a/fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java +++ b/fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java @@ -20,7 +20,9 @@ import org.apache.fluss.annotation.Internal; import org.apache.fluss.annotation.VisibleForTesting; import org.apache.fluss.exception.IllegalConfigurationException; +import org.apache.fluss.exception.InvalidTableException; import org.apache.fluss.fs.FsPath; +import org.apache.fluss.metadata.TablePath; import java.lang.reflect.Field; import java.time.Duration; @@ -79,9 +81,13 @@ public static boolean isAlterableTableOption(String key) { /** 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())) { + try { + TablePath.validateTableName(group); + } catch (InvalidTableException e) { throw new IllegalConfigurationException( - "Tiering group must not be empty or have leading or trailing whitespace."); + "Invalid tiering group. Tiering group names must follow Fluss table name rules: " + + "use 1 to 200 ASCII letters, digits, '_' or '-', and do not start " + + "with '__'."); } } diff --git a/fluss-common/src/test/java/org/apache/fluss/config/FlussConfigUtilsTest.java b/fluss-common/src/test/java/org/apache/fluss/config/FlussConfigUtilsTest.java index 1a136167e77..e72d9e180a7 100644 --- a/fluss-common/src/test/java/org/apache/fluss/config/FlussConfigUtilsTest.java +++ b/fluss-common/src/test/java/org/apache/fluss/config/FlussConfigUtilsTest.java @@ -19,7 +19,11 @@ import org.apache.fluss.exception.IllegalConfigurationException; +import org.apache.commons.lang3.StringUtils; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.NullSource; +import org.junit.jupiter.params.provider.ValueSource; import java.time.Duration; import java.util.Arrays; @@ -31,12 +35,48 @@ import static org.apache.fluss.config.FlussConfigUtils.extractConfigOptions; import static org.apache.fluss.config.FlussConfigUtils.validateCoordinatorConfigs; import static org.apache.fluss.config.FlussConfigUtils.validateTabletConfigs; +import static org.apache.fluss.config.FlussConfigUtils.validateTieringGroup; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Test for {@link FlussConfigUtils}. */ class FlussConfigUtilsTest { + @Test + void testTieringGroupNameBoundaries() { + validateTieringGroup("default"); + validateTieringGroup("Large_1-small"); + validateTieringGroup(StringUtils.repeat("a", 200)); + assertThatThrownBy(() -> validateTieringGroup(StringUtils.repeat("a", 201))) + .isInstanceOf(IllegalConfigurationException.class) + .hasMessageContaining("tiering group"); + } + + @ParameterizedTest + @NullSource + @ValueSource( + strings = { + "", + " ", + " large", + "large ", + "a b", + "a\nb", + "a\rb", + "a\tb", + "a\0b", + "a\u007fb", + "a\u00a0b", + "__reserved", + "中文" + }) + void testInvalidTieringGroupNames(String group) { + assertThatThrownBy(() -> validateTieringGroup(group)) + .isInstanceOf(IllegalConfigurationException.class) + .hasMessageContaining("tiering group") + .hasMessageNotContaining("\n"); + } + @Test void testExtractOptions() { Map> tableOptions = extractConfigOptions("table."); diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSourceOptions.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSourceOptions.java index 3e1298e0fb7..0612b34e038 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSourceOptions.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSourceOptions.java @@ -34,7 +34,8 @@ public class TieringSourceOptions { .stringType() .defaultValue(ConfigOptions.TABLE_DATALAKE_TIERING_GROUP.defaultValue()) .withDescription( - "The tiering group of tables handled by this tiering service."); + "The tiering group of tables handled by this tiering service. " + + "Group names must follow Fluss table name rules."); public static final ConfigOption POLL_TIERING_TABLE_INTERVAL = key("tiering.poll.table.interval") diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/LakeTableTieringManager.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/LakeTableTieringManager.java index 2f8ec323cf7..02fff6bc736 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/LakeTableTieringManager.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/LakeTableTieringManager.java @@ -45,14 +45,13 @@ import javax.annotation.Nullable; import javax.annotation.concurrent.GuardedBy; -import java.util.ArrayDeque; import java.util.Collections; import java.util.EnumSet; import java.util.HashMap; import java.util.Iterator; +import java.util.LinkedHashSet; import java.util.List; import java.util.Map; -import java.util.Queue; import java.util.Set; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; @@ -118,7 +117,8 @@ public class LakeTableTieringManager implements AutoCloseable { private final Timer lakeTieringScheduleTimer; private final ScheduledExecutorService lakeTieringServiceTimeoutChecker; private final Clock clock; - private final Queue pendingTieringTables; + private final Map> pendingTieringTablesByGroup; + private int pendingTieringTableCount; private final LakeTieringExpiredOperationReaper expirationReaper; // the tiering state of the table to be tiered, @@ -180,7 +180,7 @@ protected LakeTableTieringManager( this.lakeTieringScheduleTimer = lakeTieringScheduleTimer; this.lakeTieringServiceTimeoutChecker = lakeTieringServiceTimeoutChecker; this.clock = clock; - this.pendingTieringTables = new ArrayDeque<>(); + this.pendingTieringTablesByGroup = new HashMap<>(); this.tieringStates = new HashMap<>(); this.liveTieringTableIds = new HashMap<>(); this.tablePaths = new HashMap<>(); @@ -203,7 +203,7 @@ protected LakeTableTieringManager( private void registerMetrics() { tieringMetricGroup.gauge( MetricNames.LAKE_TIERING_PENDING_TABLES_COUNT, - () -> inReadLock(lock, pendingTieringTables::size)); + () -> inReadLock(lock, () -> pendingTieringTableCount)); tieringMetricGroup.gauge( MetricNames.LAKE_TIERING_RUNNING_TABLES_COUNT, () -> inReadLock(lock, liveTieringTableIds::size)); @@ -339,6 +339,9 @@ public void removeLakeTable(long tableId) { inWriteLock( lock, () -> { + if (tieringStates.get(tableId) == TieringState.Pending) { + removePendingTieringTable(tableId, tableTieringGroups.get(tableId)); + } tablePaths.remove(tableId); tableTieringGroups.remove(tableId); tableLakeFreshness.remove(tableId); @@ -368,7 +371,14 @@ public void updateTableTieringGroup(long tableId, String group) { if (previousGroup == null || previousGroup.equals(group)) { return; } + boolean isPending = tieringStates.get(tableId) == TieringState.Pending; + if (isPending) { + removePendingTieringTable(tableId, previousGroup); + } tableTieringGroups.put(tableId, group); + if (isPending) { + addPendingTieringTable(tableId); + } LOG.info( "Updated tiering group for table {} from {} to {}.", tableId, @@ -457,22 +467,36 @@ public LakeTieringTableInfo requestTable(String group) { return inWriteLock( lock, () -> { - Iterator iterator = pendingTieringTables.iterator(); + LinkedHashSet pendingTables = pendingTieringTablesByGroup.get(group); + if (pendingTables == null) { + return null; + } + Iterator iterator = pendingTables.iterator(); while (iterator.hasNext()) { long tableId = iterator.next(); + iterator.remove(); + pendingTieringTableCount--; TablePath tablePath = tablePaths.get(tableId); if (tablePath == null) { - iterator.remove(); continue; } - if (!group.equals(tableTieringGroups.get(tableId))) { + if (tieringStates.get(tableId) != TieringState.Pending) { continue; } - iterator.remove(); + String tableGroup = tableTieringGroups.get(tableId); + if (tableGroup == null) { + tableGroup = ConfigOptions.TABLE_DATALAKE_TIERING_GROUP.defaultValue(); + } + if (!group.equals(tableGroup)) { + addPendingTieringTable(tableId); + continue; + } + removeEmptyPendingTieringGroup(group, pendingTables); doHandleStateChange(tableId, TieringState.Tiering); long tieringEpoch = tableTierEpoch.get(tableId); return new LakeTieringTableInfo(tableId, tablePath, tieringEpoch); } + removeEmptyPendingTieringGroup(group, pendingTables); return null; }); } @@ -648,7 +672,7 @@ private void doHandleStateChange(long tableId, TieringState targetState) { case Pending: // increase tiering epoch and initialize the heartbeat of the tiering table tableTierEpoch.computeIfPresent(tableId, (t, v) -> v + 1); - pendingTieringTables.add(tableId); + addPendingTieringTable(tableId); pendingEnterTime.put(tableId, clock.milliseconds()); break; case Tiering: @@ -860,7 +884,39 @@ static LastTieringResult initial(long tieredTime) { @VisibleForTesting int getPendingTablesCount() { - return inReadLock(lock, pendingTieringTables::size); + return inReadLock(lock, () -> pendingTieringTableCount); + } + + @GuardedBy("lock") + private void addPendingTieringTable(long tableId) { + String group = + tableTieringGroups.getOrDefault( + tableId, ConfigOptions.TABLE_DATALAKE_TIERING_GROUP.defaultValue()); + LinkedHashSet pendingTables = + pendingTieringTablesByGroup.computeIfAbsent( + group, ignored -> new LinkedHashSet<>()); + if (pendingTables.add(tableId)) { + pendingTieringTableCount++; + } + } + + @GuardedBy("lock") + private void removePendingTieringTable(long tableId, @Nullable String group) { + if (group == null) { + return; + } + LinkedHashSet pendingTables = pendingTieringTablesByGroup.get(group); + if (pendingTables != null && pendingTables.remove(tableId)) { + pendingTieringTableCount--; + removeEmptyPendingTieringGroup(group, pendingTables); + } + } + + @GuardedBy("lock") + private void removeEmptyPendingTieringGroup(String group, LinkedHashSet pendingTables) { + if (pendingTables.isEmpty()) { + pendingTieringTablesByGroup.remove(group); + } } @VisibleForTesting 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 b2785345c3b..f047b04d375 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 @@ -869,10 +869,11 @@ private static void checkDeleteBehavior(Configuration tableConf, boolean hasPrim private static void validateOptionValue(ReadableConfig options, ConfigOption option) { try { - options.get(option); - if (option == ConfigOptions.TABLE_DATALAKE_TIERING_GROUP) { + if (ConfigOptions.TABLE_DATALAKE_TIERING_GROUP.key().equals(option.key())) { FlussConfigUtils.validateTieringGroup( options.get(ConfigOptions.TABLE_DATALAKE_TIERING_GROUP)); + } else { + options.get(option); } } catch (Throwable t) { throw new InvalidConfigException( diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/LakeTableTieringManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/LakeTableTieringManagerTest.java index 7c1f868fe7b..ff1181a9a11 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/LakeTableTieringManagerTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/LakeTableTieringManagerTest.java @@ -39,6 +39,7 @@ import org.junit.jupiter.params.provider.ValueSource; import java.time.Duration; +import java.util.ArrayList; import java.util.Arrays; import java.util.List; import java.util.Optional; @@ -293,6 +294,47 @@ void testRequestTablesByGroup() { assertThat(tableTieringManager.requestTable("large")).isNull(); } + @Test + void testPendingGroupQueuesAfterMoveAndRemoval() { + List> tables = new ArrayList<>(); + for (long tableId = 1; tableId <= 5; tableId++) { + tables.add( + Tuple2.of( + createTableInfo( + tableId, + TablePath.of("db", "table" + tableId), + Duration.ofSeconds(10), + tableId <= 3 ? "large" : "small"), + -10_000L)); + } + tableTieringManager.initWithLakeTables(tables); + waitValue( + () -> + tableTieringManager.getPendingTablesCount() == 5 + ? Optional.of(true) + : Optional.empty(), + Duration.ofSeconds(10), + "All tables should be pending before moving or removing them"); + assertThat(tableTieringManager.getPendingTablesCount()).isEqualTo(5); + + tableTieringManager.updateTableTieringGroup(2L, "small"); + tableTieringManager.updateTableTieringGroup(2L, "small"); + assertThat(tableTieringManager.getPendingTablesCount()).isEqualTo(5); + tableTieringManager.removeLakeTable(1L); + assertThat(tableTieringManager.getPendingTablesCount()).isEqualTo(4); + + assertRequestTable(3L, TablePath.of("db", "table3"), 1L, "large"); + assertThat(tableTieringManager.requestTable("large")).isNull(); + assertThat(tableTieringManager.getPendingTablesCount()).isEqualTo(3); + + tableTieringManager.removeLakeTable(4L); + assertThat(tableTieringManager.getPendingTablesCount()).isEqualTo(2); + assertRequestTable(5L, TablePath.of("db", "table5"), 1L, "small"); + assertRequestTable(2L, TablePath.of("db", "table2"), 1L, "small"); + assertThat(tableTieringManager.requestTable("small")).isNull(); + assertThat(tableTieringManager.getPendingTablesCount()).isZero(); + } + @Test void testUpdateGroupWhileScheduledAndPending() { TablePath tablePath = TablePath.of("db", "table"); diff --git a/fluss-server/src/test/java/org/apache/fluss/server/utils/TableDescriptorValidationTest.java b/fluss-server/src/test/java/org/apache/fluss/server/utils/TableDescriptorValidationTest.java index e561d0fb4db..c0d88e6a70d 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/utils/TableDescriptorValidationTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/utils/TableDescriptorValidationTest.java @@ -57,10 +57,11 @@ class TableDescriptorValidationTest { @Test void testTieringGroupValidation() { String key = ConfigOptions.TABLE_DATALAKE_TIERING_GROUP.key(); - for (String group : new String[] {"default", "large", "Large"}) { + for (String group : new String[] {"default", "large", "Large", "Large_1-small"}) { validate(logTableWithProperty(key, group), null); } - for (String group : new String[] {"", " ", " large", "large "}) { + for (String group : + new String[] {"", " ", " large", "large ", "a\nb", "a\0b", "__reserved"}) { assertThatThrownBy(() -> validate(logTableWithProperty(key, group), null)) .isInstanceOf(InvalidConfigException.class) .hasMessageContaining(key); diff --git a/website/docs/streaming-lakehouse/tiering-service.md b/website/docs/streaming-lakehouse/tiering-service.md index 8ec60dea88b..5de5e2cba45 100644 --- a/website/docs/streaming-lakehouse/tiering-service.md +++ b/website/docs/streaming-lakehouse/tiering-service.md @@ -133,7 +133,7 @@ Multiple tiering service jobs can run simultaneously. Jobs in the same tiering g ALTER TABLE large_table SET ('table.datalake.tiering-group' = 'large'); ``` -Tables and jobs without an explicit group use `default`. Group names are case-sensitive and cannot be empty or contain leading or trailing whitespace. A table waits if no service in its group is available; it is not assigned to another group. +Tables and jobs without an explicit group use `default`. Group names are case-sensitive and follow Fluss table name rules: they must be 1 to 200 ASCII letters, digits, `_` or `-`, and cannot start with `__`. A table waits if no service in its group is available; it is not assigned to another group. Group changes take effect asynchronously after the Coordinator processes the table metadata update. An already assigned tiering round can finish in its original group; subsequent rounds use the updated group. `ALTER TABLE large_table RESET ('table.datalake.tiering-group')` restores the default group. Changing the group follows the existing lake catalog property update path. For legacy log tables without a recorded lake format, primary key, or bucket key, first enable datalake and then change the group in a separate ALTER.