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 30679701b [server] Support altering auto partition num-precreate
option (#3434)
30679701b is described below
commit 30679701b302f87bb0065ac65d7befbad78b0535
Author: fhan <[email protected]>
AuthorDate: Tue Jun 9 17:57:28 2026 +0800
[server] Support altering auto partition num-precreate option (#3434)
---
.../org/apache/fluss/config/FlussConfigUtils.java | 1 +
.../fluss/flink/catalog/FlinkCatalogITCase.java | 31 +++++++++++
.../coordinator/CoordinatorEventProcessor.java | 11 ++--
.../coordinator/AutoPartitionManagerTest.java | 60 ++++++++++++++++++++++
website/docs/engine-flink/ddl.md | 2 +
website/docs/engine-flink/options.md | 4 +-
.../table-design/data-distribution/partitioning.md | 6 +--
7 files changed, 106 insertions(+), 9 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 17cfcb407..d87e11b84 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_TIERED_LOG_LOCAL_SEGMENTS.key(),
ConfigOptions.TABLE_AUTO_PARTITION_NUM_RETENTION.key(),
+ ConfigOptions.TABLE_AUTO_PARTITION_NUM_PRECREATE.key(),
ConfigOptions.TABLE_STATISTICS_COLUMNS.key(),
ConfigOptions.TABLE_KV_STANDBY_REPLICA_ENABLED.key());
}
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 8991618fd..c791b23b1 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
@@ -635,6 +635,37 @@ abstract class FlinkCatalogITCase {
.isEqualTo("1");
}
+ @Test
+ void testAlterAutoPartitionNumPrecreate() throws Exception {
+ String tblName = "test_alter_auto_partition_num_precreate";
+ 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' = 'true',"
+ + " 'table.auto-partition.key' = 'dt',"
+ + " 'table.auto-partition.time-unit' = 'hour',"
+ + " 'table.auto-partition.num-retention' = '-1',"
+ + " 'table.auto-partition.num-precreate' = '1')");
+ FLUSS_CLUSTER_EXTENSION.waitUntilPartitionAllReady(tablePath, 1);
+
+ tEnv.executeSql(
+ "alter table " + tblName + " set
('table.auto-partition.num-precreate' = '3')");
+ FLUSS_CLUSTER_EXTENSION.waitUntilPartitionAllReady(tablePath, 3);
+
+ CatalogTable table = (CatalogTable) catalog.getTable(objectPath);
+
assertThat(table.getOptions().get(ConfigOptions.TABLE_AUTO_PARTITION_NUM_PRECREATE.key()))
+ .isEqualTo("3");
+
+ tEnv.executeSql("alter table " + tblName + " reset
('table.auto-partition.num-precreate')");
+ table = (CatalogTable) catalog.getTable(objectPath);
+ assertThat(table.getOptions())
+
.doesNotContainKey(ConfigOptions.TABLE_AUTO_PARTITION_NUM_PRECREATE.key());
+ }
+
@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/CoordinatorEventProcessor.java
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java
index 456e7c587..62d63ae1f 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
@@ -818,11 +818,14 @@ public class CoordinatorEventProcessor implements
EventProcessor {
}
}
- AutoPartitionStrategy autoPartitionStrategy =
+ AutoPartitionStrategy oldAutoPartitionStrategy =
+ oldTableInfo.getTableConfig().getAutoPartitionStrategy();
+ AutoPartitionStrategy newAutoPartitionStrategy =
newTableInfo.getTableConfig().getAutoPartitionStrategy();
- if (autoPartitionStrategy.isAutoPartitionEnabled()
- && autoPartitionStrategy.numToRetain()
- !=
oldTableInfo.getTableConfig().getAutoPartitionStrategy().numToRetain()) {
+ if (newAutoPartitionStrategy.isAutoPartitionEnabled()
+ && (newAutoPartitionStrategy.numToRetain() !=
oldAutoPartitionStrategy.numToRetain()
+ || newAutoPartitionStrategy.numPreCreate()
+ != oldAutoPartitionStrategy.numPreCreate())) {
autoPartitionManager.updateAutoPartitionTables(newTableInfo);
}
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 9151ae156..9b7b33fed 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
@@ -636,6 +636,66 @@ class AutoPartitionManagerTest {
"2024091003", "2024091004", "2024091005",
"2024091006", "2024091007");
}
+ @Test
+ void testUpdateAutoPartitionNumPrecreate() 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, 1,
AutoPartitionTimeUnit.HOUR);
+ TablePath tablePath = table.getTablePath();
+ autoPartitionManager.addAutoPartitionTable(table, true);
+ periodicExecutor.triggerNonPeriodicScheduledTask();
+
+ Map<String, PartitionRegistration> partitions =
+ zookeeperClient.getPartitionRegistrations(tablePath);
+
assertThat(partitions.keySet()).containsExactlyInAnyOrder("2024091000");
+
+ TableInfo increasedPrecreateTable =
+ createUpdatedTableInfo(table, /* numRetention= */ -1, /*
numPreCreate= */ 3);
+
autoPartitionManager.updateAutoPartitionTables(increasedPrecreateTable);
+ periodicExecutor.triggerNonPeriodicScheduledTask();
+
+ partitions = zookeeperClient.getPartitionRegistrations(tablePath);
+ assertThat(partitions.keySet())
+ .containsExactlyInAnyOrder("2024091000", "2024091001",
"2024091002");
+
+ TableInfo decreasedPrecreateTable =
+ createUpdatedTableInfo(
+ increasedPrecreateTable, /* numRetention= */ -1, /*
numPreCreate= */ 1);
+
autoPartitionManager.updateAutoPartitionTables(decreasedPrecreateTable);
+ periodicExecutor.triggerNonPeriodicScheduledTask();
+
+ partitions = zookeeperClient.getPartitionRegistrations(tablePath);
+ assertThat(partitions.keySet())
+ .containsExactlyInAnyOrder("2024091000", "2024091001",
"2024091002");
+
+ clock.advanceTime(Duration.ofHours(1));
+ periodicExecutor.triggerPeriodicScheduledTasks();
+ partitions = zookeeperClient.getPartitionRegistrations(tablePath);
+ assertThat(partitions.keySet())
+ .containsExactlyInAnyOrder("2024091000", "2024091001",
"2024091002");
+
+ clock.advanceTime(Duration.ofHours(2));
+ periodicExecutor.triggerPeriodicScheduledTasks();
+ partitions = zookeeperClient.getPartitionRegistrations(tablePath);
+ assertThat(partitions.keySet())
+ .containsExactlyInAnyOrder("2024091000", "2024091001",
"2024091002", "2024091003");
+ }
+
private static class TestParams {
final AutoPartitionTimeUnit timeUnit;
final boolean multiplePartitionKeys;
diff --git a/website/docs/engine-flink/ddl.md b/website/docs/engine-flink/ddl.md
index 978bf0a41..06f586f7a 100644
--- a/website/docs/engine-flink/ddl.md
+++ b/website/docs/engine-flink/ddl.md
@@ -267,6 +267,8 @@ When using SET to modify [Storage
Options](engine-flink/options.md#storage-optio
- `table.datalake.enabled`: Enable or disable lakehouse storage for the
table.
- `table.datalake.freshness`: Set the data freshness for lakehouse storage.
- `table.log.tiered.local-segments`: Set the number of log segments to
retain locally when tiered storage is enabled.
+ - `table.auto-partition.num-retention`: Set the number of historical
partitions to retain for auto partitioning.
+ - `table.auto-partition.num-precreate`: Set the number of future partitions
to pre-create for auto partitioning.
```sql title="Flink SQL"
-- Enable lakehouse storage for the table
diff --git a/website/docs/engine-flink/options.md
b/website/docs/engine-flink/options.md
index 0b3d675dc..329ea94f1 100644
--- a/website/docs/engine-flink/options.md
+++ b/website/docs/engine-flink/options.md
@@ -70,8 +70,8 @@ See more details about [ALTER TABLE ...
SET](engine-flink/ddl.md#set-properties)
| table.auto-partition.enabled | Boolean | false
| Whether enable auto partition for the table. Disable by default.
When auto partition is enabled, the partitions of the table will be created
automatically.
[...]
| table.auto-partition.key | String | (None)
| This configuration defines the time-based partition key to be
used for auto-partitioning when a table is partitioned with multiple keys.
Auto-partitioning utilizes a time-based partition key to handle partitions
automatically, including creating new ones and removing outdated ones, by
comparing the time value of the partition with the current system time. In the
case of a table using multiple par [...]
| table.auto-partition.time-unit | ENUM | DAY
| The time granularity for auto created partitions. The default
value is `DAY`. Valid values are `HOUR`, `DAY`, `MONTH`, `QUARTER`, `YEAR`. If
the value is `HOUR`, the partition format for auto created is yyyyMMddHH. If
the value is `DAY`, the partition format for auto created is yyyyMMdd. If the
value is `MONTH`, the partition format for auto created is yyyyMM. If the value
is `QUARTER`, the parti [...]
-| table.auto-partition.num-precreate | Integer | 2
| The number of partitions to pre-create for auto created
partitions in each check for auto partition. For example, if the current check
time is 2024-11-11 and the value is configured as 3, then partitions 20241111,
20241112, 20241113 will be pre-created. If any one partition exists, it'll skip
creating the partition. The default value is 2, which means 2 partitions will
be pre-created. If the `tab [...]
-| table.auto-partition.num-retention | Integer | 7
| The number of history partitions to retain for auto created
partitions in each check for auto partition. For example, if the current check
time is 2024-11-11, time-unit is DAY, and the value is configured as 3, then
the history partitions 20241108, 20241109, 20241110 will be retained. The
partitions earlier than 20241108 will be deleted. The default value is 7, which
means that 7 partitions will [...]
+| table.auto-partition.num-precreate | Integer | 2
| The number of partitions to pre-create for auto created
partitions in each check for auto partition. For example, if the current check
time is 2024-11-11 and the value is configured as 3, then partitions 20241111,
20241112, 20241113 will be pre-created. If any one partition exists, it'll skip
creating the partition. The default value is 2, which means 2 partitions will
be pre-created. If the `tab [...]
+| table.auto-partition.num-retention | Integer | 7
| The number of history partitions to retain for auto created
partitions in each check for auto partition. For example, if the current check
time is 2024-11-11, time-unit is DAY, and the value is configured as 3, then
the history partitions 20241108, 20241109, 20241110 will be retained. The
partitions earlier than 20241108 will be deleted. The default value is 7, which
means that 7 partitions will [...]
| table.auto-partition.time-zone | String | the system time zone
| The time zone for auto partitions, which is by default the same
as the system time zone.
[...]
| table.replication.factor | Integer | (None)
| The replication factor for the log of the new table. When it's
not set, Fluss will use the cluster's default replication factor configured by
default.replication.factor. It should be a positive number and not larger than
the number of tablet servers in the Fluss cluster. A value larger than the
number of tablet servers in Fluss cluster will result in an error when the new
table is created. [...]
| table.statistics.columns | String | (None)
| Specifies which columns to collect statistics (min, max, null
count) for in log table batches. Use `*` to collect statistics for all
supported columns, or specify a comma-separated list of column names (e.g.,
`col1,col2`). Supported types: BOOLEAN, TINYINT, SMALLINT, INTEGER, BIGINT,
FLOAT, DOUBLE, STRING, CHAR, DECIMAL, DATE, TIME, TIMESTAMP, TIMESTAMP_LTZ.
Unsupported types (BYTES, BINARY, ARRA [...]
diff --git a/website/docs/table-design/data-distribution/partitioning.md
b/website/docs/table-design/data-distribution/partitioning.md
index 30460933d..2228134e7 100644
--- a/website/docs/table-design/data-distribution/partitioning.md
+++ b/website/docs/table-design/data-distribution/partitioning.md
@@ -30,7 +30,7 @@ For example, in an `Order` primary key table, the partition
key can be defined a
- The partition key must be a Fluss native data type and cannot be contained
in a map or list.
- For auto partition table, the partition keys can be one or more. If the
table has only one partition key, it supports automatic creation and automatic
expiration of partitions. Otherwise, only automatic expiration is allowed.
- If the table is a primary key table, the partition key must be a subset of
the primary key.
-- Auto-partitioning rules can only be configured at the time of creating the
partitioned table, except `table.auto-partition.num-retention`, which can be
modified by `ALTER TABLE ... SET` after table creation.
+- Auto-partitioning rules can only be configured at the time of creating the
partitioned table, except `table.auto-partition.num-retention` and
`table.auto-partition.num-precreate`, which can be modified by `ALTER TABLE ...
SET` or `ALTER TABLE ... RESET` after table creation. For a table with multiple
partition keys, `table.auto-partition.num-precreate` must remain `0`.
## Auto Partitioning
### Example
@@ -60,8 +60,8 @@ In this case, when automatic partitioning occurs (Fluss will
periodically operat
| table.auto-partition.enabled | Boolean | no | false
| Whether enable auto partition for the table. Disable by default. When auto
partition is enabled, the partitions of the table will be created
automatically.
[...]
| table.auto-partition.key | String | no | (none)
| This configuration defines the time-based partition key to be used for
auto-partitioning when a table is partitioned with multiple keys.
Auto-partitioning utilizes a time-based partition key to handle partitions
automatically, including creating new ones and removing outdated ones, by
comparing the time value of the partition with the current system time. In the
case of a table using multiple partition key [...]
| table.auto-partition.time-unit | ENUM | no | DAY
| The time granularity for auto created partitions. The default value is
'DAY'. Valid values are 'HOUR', 'DAY', 'MONTH', 'QUARTER', 'YEAR'. If the value
is 'HOUR', the partition format for auto created is yyyyMMddHH. If the value is
'DAY', the partition format for auto created is yyyyMMdd. If the value is
'MONTH', the partition format for auto created is yyyyMM. If the value is
'QUARTER', the partition forma [...]
-| table.auto-partition.num-precreate | Integer | no | 2
| The number of partitions to pre-create for auto created partitions in each
check for auto partition. For example, if the current check time is 2024-11-11
and the value is configured as 3, then partitions 20241111, 20241112, 20241113
will be pre-created. If any one partition exists, it'll skip creating the
partition. The default value is 2, which means 2 partitions will be
pre-created. If the 'table.auto-pa [...]
-| table.auto-partition.num-retention | Integer | no | 7
| The number of history partitions to retain for auto created partitions in
each check for auto partition. For example, if the current check time is
2024-11-11, time-unit is DAY, and the value is configured as 3, then the
history partitions 20241108, 20241109, 20241110 will be retained. The
partitions earlier than 20241108 will be deleted. The default value is 7. This
option can be modified after table creat [...]
+| table.auto-partition.num-precreate | Integer | no | 2
| The number of partitions to pre-create for auto created partitions in each
check for auto partition. For example, if the current check time is 2024-11-11
and the value is configured as 3, then partitions 20241111, 20241112, 20241113
will be pre-created. If any one partition exists, it'll skip creating the
partition. The default value is 2, which means 2 partitions will be
pre-created. If the 'table.auto-pa [...]
+| table.auto-partition.num-retention | Integer | no | 7
| The number of history partitions to retain for auto created partitions in
each check for auto partition. For example, if the current check time is
2024-11-11, time-unit is DAY, and the value is configured as 3, then the
history partitions 20241108, 20241109, 20241110 will be retained. The
partitions earlier than 20241108 will be deleted. The default value is 7. This
option can be modified after table creat [...]
| table.auto-partition.time-zone | String | no | the system time
zone | The time zone for auto partitions, which is by default the same as the
system time zone.
[...]
### Partition Generation Rules
