You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by ya...@apache.org on 2026/09/15 10:33:04 UTC
(fluss) branch main updated: [flink] Expose partition infos via sys.list_partition_infos
This is an automated email from the ASF dual-hosted git repository.
platinumhamburg 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 f35260cd2 [flink] Expose partition infos via sys.list_partition_infos
f35260cd2 is described below
commit f35260cd2e11920ef20e26e187c6446551108308
Author: Kaixuan Duan <98...@users.noreply.github.com>
AuthorDate: Tue Sep 15 18:33:01 2026 +0800
[flink] Expose partition infos via sys.list_partition_infos
Expose partition metadata through Flink SQL. After ALTER TABLE
... SET ('bucket.num' = N), existing partitions keep their
original bucket count while only newly created partitions use
the new one, but SHOW PARTITIONS returns only partition values
and SHOW CREATE TABLE shows only the table-level default, leaving
the per-partition layout invisible from Flink SQL.
Add a CALL sys.list_partition_infos('db', 'table_name') procedure
that returns (partition_id, partition_name, bucket_count) for
every partition, delegating to the existing
Admin#listPartitionInfos.
---
.../procedure/ListPartitionInfosProcedure.java | 64 ++++++++++++++++++++++
.../fluss/flink/procedure/ProcedureManager.java | 1 +
.../flink/procedure/FlinkProcedureITCase.java | 45 +++++++++++++++
website/docs/engine-flink/procedures.md | 49 +++++++++++++++++
4 files changed, 159 insertions(+)
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/ListPartitionInfosProcedure.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/ListPartitionInfosProcedure.java
new file mode 100644
index 000000000..a069958a0
--- /dev/null
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/ListPartitionInfosProcedure.java
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.flink.procedure;
+
+import org.apache.fluss.metadata.PartitionInfo;
+import org.apache.fluss.metadata.TablePath;
+
+import org.apache.flink.table.annotation.ArgumentHint;
+import org.apache.flink.table.annotation.DataTypeHint;
+import org.apache.flink.table.annotation.ProcedureHint;
+import org.apache.flink.table.procedure.ProcedureContext;
+import org.apache.flink.types.Row;
+
+import java.util.List;
+
+/**
+ * Procedure to list the partition id, partition name and bucket count of each partition of a
+ * partitioned table.
+ *
+ * <p>After an {@code ALTER TABLE ... SET ('bucket.num' = N)}, existing partitions keep their
+ * original bucket count while only newly created partitions use the new one. This procedure makes
+ * the per-partition bucket counts visible from Flink SQL; the table-level {@code bucket.num} shown
+ * by {@code SHOW CREATE TABLE} is only the default for newly created partitions.
+ *
+ * <p>Usage example:
+ *
+ * <pre>
+ * -- List the partition id, partition name and bucket count of each partition
+ * CALL sys.list_partition_infos('my_db', 'my_table');
+ * </pre>
+ */
+public class ListPartitionInfosProcedure extends ProcedureBase {
+
+ @ProcedureHint(
+ argument = {
+ @ArgumentHint(name = "db", type = @DataTypeHint("STRING")),
+ @ArgumentHint(name = "table_name", type = @DataTypeHint("STRING"))
+ },
+ output =
+ @DataTypeHint(
+ "ROW<partition_id BIGINT, partition_name STRING, bucket_count INT>"))
+ public Row[] call(ProcedureContext context, String db, String tableName) throws Exception {
+ List<PartitionInfo> partitionInfos =
+ admin.listPartitionInfos(TablePath.of(db, tableName)).get();
+ return partitionInfos.stream()
+ .map(p -> Row.of(p.getPartitionId(), p.getPartitionName(), p.getBucketCount()))
+ .toArray(Row[]::new);
+ }
+}
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/ProcedureManager.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/ProcedureManager.java
index 6a3614cb6..d7eeba7b4 100644
--- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/ProcedureManager.java
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/ProcedureManager.java
@@ -84,6 +84,7 @@ public class ProcedureManager {
REBALANCE("sys.rebalance", RebalanceProcedure.class),
CANCEL_REBALANCE("sys.cancel_rebalance", CancelRebalanceProcedure.class),
LIST_REBALANCE_PROGRESS("sys.list_rebalance", ListRebalanceProcessProcedure.class),
+ LIST_PARTITION_INFOS("sys.list_partition_infos", ListPartitionInfosProcedure.class),
DROP_KV_SNAPSHOT_LEASE("sys.drop_kv_snapshot_lease", DropKvSnapshotLeaseProcedure.class);
private final String path;
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/procedure/FlinkProcedureITCase.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/procedure/FlinkProcedureITCase.java
index 112639267..b2575e0c3 100644
--- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/procedure/FlinkProcedureITCase.java
+++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/procedure/FlinkProcedureITCase.java
@@ -30,6 +30,7 @@ import org.apache.fluss.exception.AuthorizationException;
import org.apache.fluss.exception.NoRebalanceInProgressException;
import org.apache.fluss.exception.SecurityDisabledException;
import org.apache.fluss.metadata.DataLakeFormat;
+import org.apache.fluss.metadata.PartitionInfo;
import org.apache.fluss.metadata.TablePath;
import org.apache.fluss.row.InternalRow;
import org.apache.fluss.server.testutils.FlussClusterExtension;
@@ -52,6 +53,8 @@ import org.junit.jupiter.params.provider.ValueSource;
import java.time.Duration;
import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -153,6 +156,7 @@ public abstract class FlinkProcedureITCase {
"+I[sys.rebalance]",
"+I[sys.cancel_rebalance]",
"+I[sys.list_rebalance]",
+ "+I[sys.list_partition_infos]",
"+I[sys.drop_kv_snapshot_lease]");
// make sure no more results is unread.
assertResultsIgnoreOrder(showProceduresIterator, expectedShowProceduresResult, true);
@@ -178,6 +182,47 @@ public abstract class FlinkProcedureITCase {
.hasMessageContaining("No match found for function signature generate_n");
}
+ @Test
+ void testListPartitionInfos() throws Exception {
+ String tableName = "partition_infos_table";
+ tEnv.executeSql(
+ String.format(
+ "create table %s (a int, b varchar) partitioned by (b) "
+ + "with ('connector' = 'fluss', 'bucket.num' = '2')",
+ tableName));
+ TablePath tablePath = TablePath.of(DEFAULT_DB, tableName);
+
+ // partitions created before the rescale keep the old bucket count
+ writeRows(conn, tablePath, Arrays.asList(row(1, "p1"), row(2, "p2")), true);
+
+ // rescale the table-level bucket count, only affecting newly created partitions
+ tEnv.executeSql(String.format("alter table %s set ('bucket.num' = '4')", tableName));
+
+ // partitions created after the rescale use the new bucket count
+ writeRows(conn, tablePath, Collections.singletonList(row(3, "p3")), true);
+
+ // fetch the partition ids assigned by the cluster to build the expected rows
+ Map<String, Long> partitionIds = new HashMap<>();
+ for (PartitionInfo partitionInfo : admin.listPartitionInfos(tablePath).get()) {
+ partitionIds.put(partitionInfo.getPartitionName(), partitionInfo.getPartitionId());
+ }
+
+ try (CloseableIterator<Row> rows =
+ tEnv.executeSql(
+ String.format(
+ "call %s.sys.list_partition_infos('%s', '%s')",
+ CATALOG_NAME, DEFAULT_DB, tableName))
+ .collect()) {
+ assertCallResult(
+ rows,
+ new String[] {
+ String.format("+I[%d, p1, 2]", partitionIds.get("p1")),
+ String.format("+I[%d, p2, 2]", partitionIds.get("p2")),
+ String.format("+I[%d, p3, 4]", partitionIds.get("p3"))
+ });
+ }
+ }
+
@MultiVersionTest
@Test
void testIndexArgument() throws Exception {
diff --git a/website/docs/engine-flink/procedures.md b/website/docs/engine-flink/procedures.md
index b3022508f..e6a0a7967 100644
--- a/website/docs/engine-flink/procedures.md
+++ b/website/docs/engine-flink/procedures.md
@@ -646,6 +646,55 @@ CALL sys.cancel_rebalance();
CALL sys.cancel_rebalance('rebalance-12345');
```
+## Partition Procedures
+
+Fluss provides procedures to inspect partition-level metadata of partitioned tables.
+
+### list_partition_infos
+
+List the partition id, partition name and bucket count of every partition of a partitioned table.
+
+For a partitioned table, `ALTER TABLE ... SET ('bucket.num' = N)` only changes the bucket count for newly created partitions; existing partitions keep their original bucket counts. The table-level `'bucket.num'` shown by `SHOW CREATE TABLE` is only the default for new partitions, so use this procedure to see the actual bucket count of each partition.
+
+**Syntax:**
+
+```sql
+-- List the partition id, partition name and bucket count of each partition
+CALL [catalog_name.]sys.list_partition_infos(
+ db => 'STRING',
+ table_name => 'STRING'
+)
+```
+
+**Parameters:**
+
+- `db` (required): The database name of the table.
+- `table_name` (required): The table name.
+
+**Returns:** One row per partition, containing:
+
+- `partition_id`: The unique identifier of the partition in the cluster.
+- `partition_name`: The partition name, e.g. `2024-01` for a daily partition.
+- `bucket_count`: The bucket count of the partition.
+
+**Example:**
+
+```sql title="Flink SQL"
+-- Use the Fluss catalog (replace 'fluss_catalog' with your catalog name if different)
+USE fluss_catalog;
+
+-- List the partitions of my_db.orders with their bucket counts
+CALL sys.list_partition_infos('my_db', 'orders');
+```
+
+An example result after `ALTER TABLE my_db.orders SET ('bucket.num' = '8')` on a table created with 4 buckets:
+
+```sql
++I[1001, 2024-01, 4]
++I[1002, 2024-02, 4]
++I[1003, 2024-03, 8]
+```
+
## kv snapshot lease
Fluss provides procedures to manage KV snapshot leases, allowing you to drop leased kv snapshots.