This is an automated email from the ASF dual-hosted git repository.
luoyuxia pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new b6717fed2 [server] Support altering auto partition enabled option
(#3453)
b6717fed2 is described below
commit b6717fed2ccba47d69d3e5e6818567e8c90ce7a7
Author: fhan <[email protected]>
AuthorDate: Fri Jun 12 14:12:23 2026 +0800
[server] Support altering auto partition enabled option (#3453)
---
.../org/apache/fluss/config/FlussConfigUtils.java | 1 +
.../apache/fluss/utils/AutoPartitionStrategy.java | 51 +++++++++++++--
.../fluss/flink/catalog/FlinkCatalogITCase.java | 44 ++++++++++++-
.../server/coordinator/AutoPartitionManager.java | 34 ++++++++++
.../coordinator/CoordinatorEventProcessor.java | 13 ++--
.../coordinator/AutoPartitionManagerTest.java | 72 ++++++++++++++++++++++
6 files changed, 203 insertions(+), 12 deletions(-)
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 237acf528..07334df7b 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
@@ -50,6 +50,7 @@ public class FlussConfigUtils {
ConfigOptions.TABLE_DATALAKE_FRESHNESS.key(),
ConfigOptions.TABLE_DATALAKE_AUTO_COMPACTION.key(),
ConfigOptions.TABLE_TIERED_LOG_LOCAL_SEGMENTS.key(),
+ ConfigOptions.TABLE_AUTO_PARTITION_ENABLED.key(),
ConfigOptions.TABLE_AUTO_PARTITION_NUM_RETENTION.key(),
ConfigOptions.TABLE_AUTO_PARTITION_NUM_PRECREATE.key(),
ConfigOptions.TABLE_STATISTICS_COLUMNS.key(),
diff --git
a/fluss-common/src/main/java/org/apache/fluss/utils/AutoPartitionStrategy.java
b/fluss-common/src/main/java/org/apache/fluss/utils/AutoPartitionStrategy.java
index 4da27b2b6..a3a725adc 100644
---
a/fluss-common/src/main/java/org/apache/fluss/utils/AutoPartitionStrategy.java
+++
b/fluss-common/src/main/java/org/apache/fluss/utils/AutoPartitionStrategy.java
@@ -22,12 +22,13 @@ import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
import java.util.Map;
+import java.util.Objects;
import java.util.TimeZone;
/** A class wrapping the strategy for auto partition. */
public class AutoPartitionStrategy {
- private final boolean autoPartitionEnable;
+ private final boolean autoPartitionEnabled;
private final String key;
private final AutoPartitionTimeUnit timeUnit;
private final int numPreCreate;
@@ -35,13 +36,13 @@ public class AutoPartitionStrategy {
private final TimeZone timeZone;
private AutoPartitionStrategy(
- boolean autoPartitionEnable,
+ boolean autoPartitionEnabled,
String key,
AutoPartitionTimeUnit autoPartitionTimeUnit,
int numPreCreate,
int numToRetain,
TimeZone timeZone) {
- this.autoPartitionEnable = autoPartitionEnable;
+ this.autoPartitionEnabled = autoPartitionEnabled;
this.key = key;
this.timeUnit = autoPartitionTimeUnit;
this.numPreCreate = numPreCreate;
@@ -64,7 +65,7 @@ public class AutoPartitionStrategy {
}
public boolean isAutoPartitionEnabled() {
- return autoPartitionEnable;
+ return autoPartitionEnabled;
}
public String key() {
@@ -86,4 +87,46 @@ public class AutoPartitionStrategy {
public TimeZone timeZone() {
return timeZone;
}
+
+ @Override
+ public String toString() {
+ return "AutoPartitionStrategy{"
+ + "autoPartitionEnabled="
+ + autoPartitionEnabled
+ + ", key='"
+ + key
+ + '\''
+ + ", timeUnit="
+ + timeUnit
+ + ", numPreCreate="
+ + numPreCreate
+ + ", numToRetain="
+ + numToRetain
+ + ", timeZone="
+ + timeZone
+ + '}';
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ AutoPartitionStrategy that = (AutoPartitionStrategy) o;
+ return autoPartitionEnabled == that.autoPartitionEnabled
+ && numPreCreate == that.numPreCreate
+ && numToRetain == that.numToRetain
+ && Objects.equals(key, that.key)
+ && timeUnit == that.timeUnit
+ && Objects.equals(timeZone, that.timeZone);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(
+ autoPartitionEnabled, key, timeUnit, numPreCreate,
numToRetain, timeZone);
+ }
}
diff --git
a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogITCase.java
b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogITCase.java
index c791b23b1..54942d7d8 100644
---
a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogITCase.java
+++
b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogITCase.java
@@ -251,14 +251,22 @@ abstract class FlinkCatalogITCase {
// alter table set an unsupported modification option should throw
exception
String unSupportedDml1 =
- "alter table test_alter_table_append_only set
('table.auto-partition.enabled' = 'true', 'table.kv.format' = 'indexed')";
+ "alter table test_alter_table_append_only set
('table.kv.format' = 'indexed')";
assertThatThrownBy(() -> tEnv.executeSql(unSupportedDml1))
.rootCause()
.isInstanceOf(InvalidAlterTableException.class)
.hasMessageContaining("The following options are not supported
to alter yet:")
- .hasMessageContaining("table.kv.format")
- .hasMessageContaining("table.auto-partition.enabled");
+ .hasMessageContaining("table.kv.format");
+
+ // alter table to enable auto partition for a non-partitioned table
should fail validation
+ String invalidAutoPartitionDml =
+ "alter table test_alter_table_append_only set
('table.auto-partition.enabled' = 'true')";
+ assertThatThrownBy(() -> tEnv.executeSql(invalidAutoPartitionDml))
+ .rootCause()
+ .isInstanceOf(InvalidConfigException.class)
+ .hasMessage(
+ "Currently, auto partition is only supported for
partitioned table, please set table property 'table.auto-partition.enabled' to
false.");
String unSupportedDml2 =
"alter table test_alter_table_append_only set ('bucket.num' =
'1000')";
@@ -666,6 +674,36 @@ abstract class FlinkCatalogITCase {
.doesNotContainKey(ConfigOptions.TABLE_AUTO_PARTITION_NUM_PRECREATE.key());
}
+ @Test
+ void testAlterAutoPartitionEnabled() throws Exception {
+ String tblName = "test_alter_auto_partition_enabled";
+ ObjectPath objectPath = new ObjectPath(DEFAULT_DB, tblName);
+ TablePath tablePath = new TablePath(DEFAULT_DB, tblName);
+
+ tEnv.executeSql(
+ "create table "
+ + tblName
+ + " (a int, dt string) partitioned by (dt) "
+ + "with ('table.auto-partition.enabled' = 'false',"
+ + " 'table.auto-partition.key' = 'dt',"
+ + " 'table.auto-partition.time-unit' = 'hour',"
+ + " 'table.auto-partition.num-precreate' = '2')");
+
+ tEnv.executeSql(
+ "alter table " + tblName + " set
('table.auto-partition.enabled' = 'true')");
+ FLUSS_CLUSTER_EXTENSION.waitUntilPartitionAllReady(tablePath, 2);
+
+ CatalogTable table = (CatalogTable) catalog.getTable(objectPath);
+
assertThat(table.getOptions().get(ConfigOptions.TABLE_AUTO_PARTITION_ENABLED.key()))
+ .isEqualTo("true");
+
+ tEnv.executeSql(
+ "alter table " + tblName + " set
('table.auto-partition.enabled' = 'false')");
+ table = (CatalogTable) catalog.getTable(objectPath);
+
assertThat(table.getOptions().get(ConfigOptions.TABLE_AUTO_PARTITION_ENABLED.key()))
+ .isEqualTo("false");
+ }
+
@Test
void testTableWithExpression() throws Exception {
// create a table with watermark and computed column
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/AutoPartitionManager.java
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/AutoPartitionManager.java
index 33f9513a2..37bc7c376 100644
---
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/AutoPartitionManager.java
+++
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/AutoPartitionManager.java
@@ -194,6 +194,34 @@ public class AutoPartitionManager implements AutoCloseable
{
}
}
+ /**
+ * Handles a table's auto-partition strategy change after table properties
are updated.
+ *
+ * @param newTableInfo the updated table information
+ * @param oldStrategy the old auto partition strategy
+ * @param newStrategy the updated auto partition strategy
+ */
+ public void handleAutoPartitionStrategyChange(
+ TableInfo newTableInfo,
+ AutoPartitionStrategy oldStrategy,
+ AutoPartitionStrategy newStrategy) {
+ checkNotClosed();
+ long tableId = newTableInfo.getTableId();
+ boolean oldAutoPartitionEnabled = oldStrategy.isAutoPartitionEnabled();
+ boolean newAutoPartitionEnabled = newStrategy.isAutoPartitionEnabled();
+
+ if (!oldAutoPartitionEnabled && newAutoPartitionEnabled) {
+ LOG.info("Table {} auto partition enabled from false to true.",
tableId);
+ addAutoPartitionTable(newTableInfo, true);
+ } else if (oldAutoPartitionEnabled && !newAutoPartitionEnabled) {
+ LOG.info("Table {} auto partition enabled from true to false.",
tableId);
+ removeAutoPartitionTable(tableId);
+ } else if (newAutoPartitionEnabled) {
+ LOG.info("Table {} auto partition strategy changed.", tableId);
+ updateAutoPartitionTables(newTableInfo);
+ }
+ }
+
/** Must be called while holding {@link #lock}. */
@Nullable
private TableInfo removeAutoPartitionTableLocked(long tableId) {
@@ -319,6 +347,12 @@ public class AutoPartitionManager implements AutoCloseable
{
}
TableInfo tableInfo = autoPartitionTables.get(tableId);
+ if (tableInfo == null) {
+ LOG.debug(
+ "Skipping auto partitioning for table id {} as it is
not registered.",
+ tableId);
+ continue;
+ }
TablePath tablePath = tableInfo.getTablePath();
TreeMap<String, Set<String>> currentPartitions =
partitionsByTable.computeIfAbsent(
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 6c68a5583..c2e297f19 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
@@ -853,11 +853,14 @@ public class CoordinatorEventProcessor implements
EventProcessor {
oldTableInfo.getTableConfig().getAutoPartitionStrategy();
AutoPartitionStrategy newAutoPartitionStrategy =
newTableInfo.getTableConfig().getAutoPartitionStrategy();
- if (newAutoPartitionStrategy.isAutoPartitionEnabled()
- && (newAutoPartitionStrategy.numToRetain() !=
oldAutoPartitionStrategy.numToRetain()
- || newAutoPartitionStrategy.numPreCreate()
- != oldAutoPartitionStrategy.numPreCreate())) {
- autoPartitionManager.updateAutoPartitionTables(newTableInfo);
+ if (!Objects.equals(oldAutoPartitionStrategy,
newAutoPartitionStrategy)) {
+ LOG.info(
+ "Table {} auto partition strategy changed from {} to {}.",
+ oldTableInfo.getTableId(),
+ oldAutoPartitionStrategy,
+ newAutoPartitionStrategy);
+ autoPartitionManager.handleAutoPartitionStrategyChange(
+ newTableInfo, oldAutoPartitionStrategy,
newAutoPartitionStrategy);
}
// If standby replica config changed, trigger re-election for all
online buckets
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/AutoPartitionManagerTest.java
b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/AutoPartitionManagerTest.java
index 175634129..641ae7d9e 100644
---
a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/AutoPartitionManagerTest.java
+++
b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/AutoPartitionManagerTest.java
@@ -746,6 +746,67 @@ class AutoPartitionManagerTest {
.containsExactlyInAnyOrder("2024091000", "2024091001",
"2024091002", "2024091003");
}
+ @Test
+ void testUpdateAutoPartitionEnabled() throws Exception {
+ ZonedDateTime startTime =
+
LocalDateTime.parse("2024-09-10T00:00:00").atZone(ZoneId.systemDefault());
+ long startMs = startTime.toInstant().toEpochMilli();
+ ManualClock clock = new ManualClock(startMs);
+ ManuallyTriggeredScheduledExecutorService periodicExecutor =
+ new ManuallyTriggeredScheduledExecutorService();
+
+ AutoPartitionManager autoPartitionManager =
+ new AutoPartitionManager(
+ new TestingServerMetadataCache(3),
+ metadataManager,
+ remoteDirDynamicLoader,
+ new Configuration(),
+ clock,
+ periodicExecutor);
+ autoPartitionManager.start();
+
+ TableInfo table = createPartitionedTable(-1, 4,
AutoPartitionTimeUnit.HOUR);
+ TablePath tablePath = table.getTablePath();
+ autoPartitionManager.addAutoPartitionTable(table, true);
+ periodicExecutor.triggerNonPeriodicScheduledTask();
+
+ Map<String, PartitionRegistration> partitions =
+ zookeeperClient.getPartitionRegistrations(tablePath);
+ assertThat(partitions.keySet())
+ .containsExactlyInAnyOrder("2024091000", "2024091001",
"2024091002", "2024091003");
+
+ TableInfo disabledTable =
createUpdatedAutoPartitionEnabledTableInfo(table, false);
+ autoPartitionManager.handleAutoPartitionStrategyChange(
+ disabledTable,
+ table.getTableConfig().getAutoPartitionStrategy(),
+ disabledTable.getTableConfig().getAutoPartitionStrategy());
+
+ clock.advanceTime(Duration.ofHours(4));
+ periodicExecutor.triggerPeriodicScheduledTasks();
+ partitions = zookeeperClient.getPartitionRegistrations(tablePath);
+ assertThat(partitions.keySet())
+ .containsExactlyInAnyOrder("2024091000", "2024091001",
"2024091002", "2024091003");
+
+ TableInfo reEnabledTable =
createUpdatedAutoPartitionEnabledTableInfo(disabledTable, true);
+ autoPartitionManager.handleAutoPartitionStrategyChange(
+ reEnabledTable,
+ disabledTable.getTableConfig().getAutoPartitionStrategy(),
+ reEnabledTable.getTableConfig().getAutoPartitionStrategy());
+ periodicExecutor.triggerNonPeriodicScheduledTask();
+
+ partitions = zookeeperClient.getPartitionRegistrations(tablePath);
+ assertThat(partitions.keySet())
+ .containsExactlyInAnyOrder(
+ "2024091000",
+ "2024091001",
+ "2024091002",
+ "2024091003",
+ "2024091004",
+ "2024091005",
+ "2024091006",
+ "2024091007");
+ }
+
//
---------------------------------------------------------------------------------------
// Batch / inflight tests previously housed here have moved to
TableLifecycleThrottlerTest.
// The AutoPartitionManager now drops expired partitions synchronously and
the asynchronous
@@ -1033,6 +1094,17 @@ class AutoPartitionManagerTest {
Configuration newProperties = new
Configuration(original.getProperties());
newProperties.set(ConfigOptions.TABLE_AUTO_PARTITION_NUM_RETENTION,
newNumRetention);
newProperties.set(ConfigOptions.TABLE_AUTO_PARTITION_NUM_PRECREATE,
newNumPreCreate);
+ return createUpdatedTableInfo(original, newProperties);
+ }
+
+ private TableInfo createUpdatedAutoPartitionEnabledTableInfo(
+ TableInfo original, boolean autoPartitionEnabled) {
+ Configuration newProperties = new
Configuration(original.getProperties());
+ newProperties.set(ConfigOptions.TABLE_AUTO_PARTITION_ENABLED,
autoPartitionEnabled);
+ return createUpdatedTableInfo(original, newProperties);
+ }
+
+ private TableInfo createUpdatedTableInfo(TableInfo original, Configuration
newProperties) {
return new TableInfo(
original.getTablePath(),
original.getTableId(),
