diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/DescribeBucketsProcedure.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/DescribeBucketsProcedure.java new file mode 100644 index 0000000000..26852d476a --- /dev/null +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/DescribeBucketsProcedure.java @@ -0,0 +1,167 @@ +/* + * 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.BucketInfo; +import org.apache.fluss.metadata.PartitionSpec; +import org.apache.fluss.metadata.ResolvedPartitionSpec; +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.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.OptionalInt; +import java.util.OptionalLong; + +/** + * Procedure to describe bucket metadata for a table. + * + *

Usage examples: + * + * 

+ * CALL sys.describe_buckets('db.table');
+ * CALL sys.describe_buckets('db.table', 'partition_key=partition_value');
+ *
+ */ +public class DescribeBucketsProcedure extends ProcedureBase { + + private static final String OUTPUT_TYPE = + "ROW, isr ARRAY>"; + + @ProcedureHint( + argument = {@ArgumentHint(name = "table_path", type = @DataTypeHint("STRING"))}, + output = @DataTypeHint(OUTPUT_TYPE)) + public Row[] call(ProcedureContext context, String tablePath) throws Exception { + TablePath parsedTablePath = parseTablePath(tablePath); + return toRows(admin.describeBuckets(parsedTablePath).get()); + } + + @ProcedureHint( + argument = { + @ArgumentHint(name = "table_path", type = @DataTypeHint("STRING")), + @ArgumentHint(name = "partition_spec", type = @DataTypeHint("STRING")) + }, + output = @DataTypeHint(OUTPUT_TYPE)) + public Row[] call(ProcedureContext context, String tablePath, String partitionSpec) + throws Exception { + TablePath parsedTablePath = parseTablePath(tablePath); + return toRows( + admin.describeBuckets(parsedTablePath, parsePartitionSpec(partitionSpec)).get()); + } + + private static Row[] toRows(List bucketInfos) { + return bucketInfos.stream().map(DescribeBucketsProcedure::toRow).toArray(Row[]::new); + } + + private static Row toRow(BucketInfo bucketInfo) { + return Row.of( + bucketInfo.getTablePath().toString(), + bucketInfo.getTableId(), + optionalLong(bucketInfo.getPartitionId()), + bucketInfo.getPartitionName(), + bucketInfo.getBucketId(), + optionalInt(bucketInfo.getLeaderId()), + optionalInt(bucketInfo.getLeaderEpoch()), + optionalInt(bucketInfo.getBucketEpoch()), + bucketInfo.getReplicas().toArray(new Integer[0]), + bucketInfo.getIsr().toArray(new Integer[0])); + } + + private static Long optionalLong(OptionalLong value) { + return value.isPresent() ? value.getAsLong() : null; + } + + private static Integer optionalInt(OptionalInt value) { + return value.isPresent() ? value.getAsInt() : null; + } + + private static TablePath parseTablePath(String tablePath) { + if (tablePath == null || tablePath.trim().isEmpty()) { + throw new IllegalArgumentException( + "table_path cannot be null or empty. Expected format is 'database.table'."); + } + + String normalizedTablePath = tablePath.trim(); + String[] parts = normalizedTablePath.split("\\.", -1); + if (parts.length != 2 || parts[0].isEmpty() || parts[1].isEmpty()) { + throw new IllegalArgumentException( + "Invalid table_path '" + tablePath + "'. Expected format is 'database.table'."); + } + TablePath parsedTablePath = TablePath.of(parts[0], parts[1]); + parsedTablePath.validate(); + return parsedTablePath; + } + + private static PartitionSpec parsePartitionSpec(String partitionSpec) { + if (partitionSpec == null || partitionSpec.trim().isEmpty()) { + throw new IllegalArgumentException( + "partition_spec cannot be null or empty. Expected format is " + + "'key=value[/key=value...]'."); + } + + String normalizedPartitionSpec = partitionSpec.trim(); + ResolvedPartitionSpec resolvedPartitionSpec; + try { + String[] keyValuePairs = normalizedPartitionSpec.split("/", -1); + for (String keyValuePair : keyValuePairs) { + if (keyValuePair.isEmpty()) { + throw new IllegalArgumentException("Empty partition key-value pair."); + } + } + resolvedPartitionSpec = + ResolvedPartitionSpec.fromPartitionQualifiedName(normalizedPartitionSpec); + } catch (IllegalArgumentException e) { + throw new IllegalArgumentException( + "Invalid partition_spec '" + + partitionSpec + + "'. Expected format is 'key=value[/key=value...]'.", + e); + } + + Map spec = new LinkedHashMap<>(); + List partitionKeys = resolvedPartitionSpec.getPartitionKeys(); + List partitionValues = resolvedPartitionSpec.getPartitionValues(); + for (int i = 0; i < partitionKeys.size(); i++) { + String partitionKey = partitionKeys.get(i); + if (partitionKey.trim().isEmpty()) { + throw new IllegalArgumentException( + "Invalid partition_spec '" + + partitionSpec + + "': partition key cannot be empty."); + } + if (spec.containsKey(partitionKey)) { + throw new IllegalArgumentException( + "Duplicate partition key '" + + partitionKey + + "' in partition_spec '" + + partitionSpec + + "'."); + } + spec.put(partitionKey, partitionValues.get(i)); + } + return new PartitionSpec(spec); + } +} 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 d7eeba7b42..311162679c 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 @@ -68,6 +68,7 @@ private static Map> initProcedureMap() { private enum ProcedureEnum { ADD_ACL("sys.add_acl", AddAclProcedure.class), + DESCRIBE_BUCKETS("sys.describe_buckets", DescribeBucketsProcedure.class), DROP_ACL("sys.drop_acl", DropAclProcedure.class), List_ACL("sys.list_acl", ListAclProcedure.class), SET_CLUSTER_CONFIGS("sys.set_cluster_configs", SetClusterConfigsProcedure.class), diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/procedure/DescribeBucketsProcedureTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/procedure/DescribeBucketsProcedureTest.java new file mode 100644 index 0000000000..42754dcc09 --- /dev/null +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/procedure/DescribeBucketsProcedureTest.java @@ -0,0 +1,183 @@ +/* + * 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.flink.sink.testutils.TestAdminAdapter; +import org.apache.fluss.metadata.BucketInfo; +import org.apache.fluss.metadata.PartitionSpec; +import org.apache.fluss.metadata.TablePath; + +import org.apache.flink.types.Row; +import org.junit.jupiter.api.Test; + +import javax.annotation.Nullable; + +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.CompletableFuture; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Test for {@link DescribeBucketsProcedure}. */ +class DescribeBucketsProcedureTest { + + private static final TablePath TABLE_PATH = TablePath.of("test_db", "test_table"); + + @Test + void testDescribeAllBucketsAndConvertNullableFields() throws Exception { + BucketInfo bucketInfo = + new BucketInfo( + TABLE_PATH, + 10L, + null, + null, + 0, + null, + null, + null, + Arrays.asList(1, 2, 3), + Collections.emptyList()); + TestingAdmin admin = new TestingAdmin(Collections.singletonList(bucketInfo)); + DescribeBucketsProcedure procedure = newProcedure(admin); + + Row[] rows = procedure.call(null, " test_db.test_table "); + + assertThat(admin.requestedTablePath).isEqualTo(TABLE_PATH); + assertThat(admin.requestedPartitionSpec).isNull(); + assertThat(rows).hasSize(1); + Row row = rows[0]; + assertThat(row.getArity()).isEqualTo(10); + assertThat(row.getField(0)).isEqualTo("test_db.test_table"); + assertThat(row.getField(1)).isEqualTo(10L); + assertThat(row.getField(2)).isNull(); + assertThat(row.getField(3)).isNull(); + assertThat(row.getField(4)).isEqualTo(0); + assertThat(row.getField(5)).isNull(); + assertThat(row.getField(6)).isNull(); + assertThat(row.getField(7)).isNull(); + assertThat((Integer[]) row.getField(8)).containsExactly(1, 2, 3); + assertThat((Integer[]) row.getField(9)).isEmpty(); + } + + @Test + void testDescribeBucketsWithPartitionSpec() throws Exception { + BucketInfo bucketInfo = + new BucketInfo( + TABLE_PATH, + 10L, + 100L, + "cn$2026-09-16", + 1, + 2, + 3, + -1, + Arrays.asList(1, 2, 3), + Arrays.asList(2, 3)); + TestingAdmin admin = new TestingAdmin(Collections.singletonList(bucketInfo)); + DescribeBucketsProcedure procedure = newProcedure(admin); + + Row[] rows = procedure.call(null, "test_db.test_table", "region=cn/dt=2026-09-16"); + + assertThat(admin.requestedTablePath).isEqualTo(TABLE_PATH); + assertThat(admin.requestedPartitionSpec).isNotNull(); + assertThat(admin.requestedPartitionSpec.getSpecMap()) + .containsEntry("region", "cn") + .containsEntry("dt", "2026-09-16"); + assertThat(rows).hasSize(1); + Row row = rows[0]; + assertThat(row.getField(2)).isEqualTo(100L); + assertThat(row.getField(3)).isEqualTo("cn$2026-09-16"); + assertThat(row.getField(4)).isEqualTo(1); + assertThat(row.getField(5)).isEqualTo(2); + assertThat(row.getField(6)).isEqualTo(3); + assertThat(row.getField(7)).isEqualTo(-1); + assertThat((Integer[]) row.getField(8)).containsExactly(1, 2, 3); + assertThat((Integer[]) row.getField(9)).containsExactly(2, 3); + } + + @Test + void testEmptyAdminResult() throws Exception { + DescribeBucketsProcedure procedure = + newProcedure(new TestingAdmin(Collections.emptyList())); + + assertThat(procedure.call(null, "test_db.test_table")).isEmpty(); + } + + @Test + void testInvalidTablePath() { + DescribeBucketsProcedure procedure = + newProcedure(new TestingAdmin(Collections.emptyList())); + + for (String tablePath : + Arrays.asList(null, "", " ", "test_db", ".test_table", "test_db.", "a.b.c")) { + assertThatThrownBy(() -> procedure.call(null, tablePath)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("table_path"); + } + } + + @Test + void testInvalidPartitionSpec() { + DescribeBucketsProcedure procedure = + newProcedure(new TestingAdmin(Collections.emptyList())); + + for (String partitionSpec : + Arrays.asList(null, "", " ", "region", "=cn", "region=cn/", "region=cn//dt=d1")) { + assertThatThrownBy(() -> procedure.call(null, "test_db.test_table", partitionSpec)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("partition_spec"); + } + + assertThatThrownBy(() -> procedure.call(null, "test_db.test_table", "region=cn/region=us")) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Duplicate partition key 'region'"); + } + + private static DescribeBucketsProcedure newProcedure(TestingAdmin admin) { + DescribeBucketsProcedure procedure = new DescribeBucketsProcedure(); + procedure.withAdmin(admin); + return procedure; + } + + private static class TestingAdmin extends TestAdminAdapter { + private final List bucketInfos; + private @Nullable TablePath requestedTablePath; + private @Nullable PartitionSpec requestedPartitionSpec; + + private TestingAdmin(List bucketInfos) { + this.bucketInfos = bucketInfos; + } + + @Override + public CompletableFuture> describeBuckets(TablePath tablePath) { + requestedTablePath = tablePath; + requestedPartitionSpec = null; + return CompletableFuture.completedFuture(bucketInfos); + } + + @Override + public CompletableFuture> describeBuckets( + TablePath tablePath, PartitionSpec partitionSpec) { + requestedTablePath = tablePath; + requestedPartitionSpec = partitionSpec; + return CompletableFuture.completedFuture(bucketInfos); + } + } +} 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 b2575e0c33..b8fa085b34 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 @@ -27,8 +27,11 @@ import org.apache.fluss.config.Configuration; import org.apache.fluss.config.MemorySize; import org.apache.fluss.exception.AuthorizationException; +import org.apache.fluss.exception.InvalidPartitionException; import org.apache.fluss.exception.NoRebalanceInProgressException; import org.apache.fluss.exception.SecurityDisabledException; +import org.apache.fluss.exception.TableNotExistException; +import org.apache.fluss.exception.TableNotPartitionedException; import org.apache.fluss.metadata.DataLakeFormat; import org.apache.fluss.metadata.PartitionInfo; import org.apache.fluss.metadata.TablePath; @@ -51,8 +54,12 @@ import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.ValueSource; +import javax.annotation.Nullable; + import java.time.Duration; +import java.util.ArrayList; import java.util.Arrays; +import java.util.Collection; import java.util.Collections; import java.util.HashMap; import java.util.List; @@ -142,6 +149,7 @@ void testShowProcedures() throws Exception { List expectedShowProceduresResult = Arrays.asList( "+I[sys.add_acl]", + "+I[sys.describe_buckets]", "+I[sys.drop_acl]", "+I[sys.get_cluster_configs]", "+I[sys.list_acl]", @@ -223,6 +231,109 @@ void testListPartitionInfos() throws Exception { } } + @Test + @MultiVersionTest + void testDescribeBuckets() throws Exception { + String tableName = "describe_buckets_table"; + tEnv.executeSql( + String.format( + "create table %s (id int, name string) " + + "with ('bucket.num' = '2')", + tableName)) + .await(); + TablePath tablePath = TablePath.of(DEFAULT_DB, tableName); + long tableId = admin.getTableInfo(tablePath).get().getTableId(); + + List tableBucketRows = waitAndDescribeBuckets(tablePath.toString(), null, 2); + assertBucketRows(tableBucketRows, tablePath.toString(), tableId, 2); + assertThat(tableBucketRows).extracting(row -> row.getField(2)).containsOnlyNulls(); + assertThat(tableBucketRows).extracting(row -> row.getField(3)).containsOnlyNulls(); + assertThat(tableBucketRows) + .extracting(row -> row.getField(4)) + .containsExactlyInAnyOrder(0, 1); + + String partitionedTableName = "describe_buckets_partitioned_table"; + tEnv.executeSql( + String.format( + "create table %s (id int, region string, dt string) " + + "partitioned by (region, dt) " + + "with ('bucket.num' = '2')", + partitionedTableName)) + .await(); + TablePath partitionedTablePath = TablePath.of(DEFAULT_DB, partitionedTableName); + long partitionedTableId = admin.getTableInfo(partitionedTablePath).get().getTableId(); + writeRows( + conn, + partitionedTablePath, + Arrays.asList( + row(1, "cn", "2026-09-15"), + row(2, "cn", "2026-09-16"), + row(3, "us", "2026-09-16")), + true); + + List allPartitionBucketRows = + waitAndDescribeBuckets(partitionedTablePath.toString(), null, 6); + assertBucketRows( + allPartitionBucketRows, partitionedTablePath.toString(), partitionedTableId, 6); + assertThat(allPartitionBucketRows).extracting(row -> row.getField(2)).doesNotContainNull(); + assertThat(allPartitionBucketRows) + .extracting(row -> row.getField(3)) + .containsExactlyInAnyOrder( + "cn$2026-09-15", + "cn$2026-09-15", + "cn$2026-09-16", + "cn$2026-09-16", + "us$2026-09-16", + "us$2026-09-16"); + + List cnBucketRows = + waitAndDescribeBuckets(partitionedTablePath.toString(), "region=cn", 4); + assertBucketRows(cnBucketRows, partitionedTablePath.toString(), partitionedTableId, 4); + assertThat(cnBucketRows) + .extracting(row -> row.getField(3)) + .containsOnly("cn$2026-09-15", "cn$2026-09-16"); + + List exactPartitionBucketRows = + waitAndDescribeBuckets( + partitionedTablePath.toString(), "region=cn/dt=2026-09-16", 2); + assertBucketRows( + exactPartitionBucketRows, partitionedTablePath.toString(), partitionedTableId, 2); + assertThat(exactPartitionBucketRows) + .extracting(row -> row.getField(3)) + .containsOnly("cn$2026-09-16"); + assertThat(exactPartitionBucketRows) + .extracting(row -> row.getField(4)) + .containsExactlyInAnyOrder(0, 1); + + assertThatThrownBy( + () -> + tEnv.executeSql( + String.format( + "call %s.sys.describe_buckets('%s.missing_table')", + CATALOG_NAME, DEFAULT_DB)) + .await()) + .rootCause() + .isInstanceOf(TableNotExistException.class); + assertThatThrownBy( + () -> + tEnv.executeSql( + String.format( + "call %s.sys.describe_buckets('%s', 'region=cn')", + CATALOG_NAME, tablePath)) + .await()) + .rootCause() + .isInstanceOf(TableNotPartitionedException.class); + assertThatThrownBy( + () -> + tEnv.executeSql( + String.format( + "call %s.sys.describe_buckets('%s', 'unknown=value')", + CATALOG_NAME, partitionedTablePath)) + .await()) + .rootCause() + .isInstanceOf(InvalidPartitionException.class); + } + @MultiVersionTest @Test void testIndexArgument() throws Exception { @@ -1343,6 +1454,79 @@ private static Configuration initConfig() { return conf; } + private List waitAndDescribeBuckets( + String tablePath, @Nullable String partitionSpec, int expectedBucketCount) + throws Exception { + retry( + Duration.ofMinutes(1), + () -> { + List rows = describeBuckets(tablePath, partitionSpec); + assertThat(rows).hasSize(expectedBucketCount); + assertThat(rows) + .allSatisfy( + row -> { + assertThat(row.getArity()).isEqualTo(10); + assertThat(row.getField(5)).isNotNull(); + assertThat(row.getField(6)).isNotNull(); + assertThat((Integer) row.getField(7)) + .isGreaterThanOrEqualTo(0); + assertThat(toIntegerList(row.getField(8))).isNotEmpty(); + assertThat(toIntegerList(row.getField(9))).isNotEmpty(); + }); + }); + return describeBuckets(tablePath, partitionSpec); + } + + private List describeBuckets(String tablePath, @Nullable String partitionSpec) + throws Exception { + String sql = + partitionSpec == null + ? String.format( + "call %s.sys.describe_buckets('%s')", CATALOG_NAME, tablePath) + : String.format( + "call %s.sys.describe_buckets('%s', '%s')", + CATALOG_NAME, tablePath, partitionSpec); + try (CloseableIterator resultIterator = tEnv.executeSql(sql).collect()) { + return CollectionUtil.iteratorToList(resultIterator); + } + } + + private static void assertBucketRows( + List rows, String tablePath, long tableId, int expectedBucketCount) { + assertThat(rows).hasSize(expectedBucketCount); + assertThat(rows).extracting(row -> row.getField(0)).containsOnly(tablePath); + assertThat(rows).extracting(row -> row.getField(1)).containsOnly(tableId); + rows.forEach( + row -> { + Integer leaderId = (Integer) row.getField(5); + List replicas = toIntegerList(row.getField(8)); + List isr = toIntegerList(row.getField(9)); + assertThat(replicas).hasSize(3).containsAll(isr); + assertThat(isr).contains(leaderId); + }); + } + + private static List toIntegerList(Object value) { + if (value instanceof int[]) { + int[] values = (int[]) value; + List result = new ArrayList<>(values.length); + for (int intValue : values) { + result.add(intValue); + } + return result; + } else if (value instanceof Object[]) { + return Arrays.stream((Object[]) value) + .map(number -> ((Number) number).intValue()) + .collect(Collectors.toList()); + } else if (value instanceof Collection) { + return ((Collection) value) + .stream() + .map(number -> ((Number) number).intValue()) + .collect(Collectors.toList()); + } + throw new AssertionError("Unsupported ARRAY value: " + value); + } + private static void assertCallResult(CloseableIterator rows, String[] expected) { List actual = CollectionUtil.iteratorToList(rows).stream() diff --git a/website/docs/engine-flink/procedures.md b/website/docs/engine-flink/procedures.md index e6a0a79678..e4094b0394 100644 --- a/website/docs/engine-flink/procedures.md +++ b/website/docs/engine-flink/procedures.md @@ -646,6 +646,74 @@ CALL sys.cancel_rebalance(); CALL sys.cancel_rebalance('rebalance-12345'); ``` +## Bucket Procedures + +Fluss provides procedures to inspect the replica placement and leader state of table buckets. + +### describe_buckets + +Describe the bucket metadata of a table. For a non-partitioned table, the procedure returns all +table buckets. For a partitioned table, it returns all partition buckets unless a partition spec is +provided. + +**Syntax:** + +```sql +-- Describe a non-partitioned table or every partition of a partitioned table +CALL [catalog_name.]sys.describe_buckets( + table_path => 'database.table' +) + +-- Describe buckets matching a complete or partial partition spec +CALL [catalog_name.]sys.describe_buckets( + table_path => 'database.table', + partition_spec => 'key1=value1/key2=value2' +) +``` + +**Parameters:** + +- `table_path` (required): The table path in `database.table` format. +- `partition_spec` (optional): A complete or partial partition spec in + `key=value[/key=value...]` format. This parameter is only valid for partitioned tables. + +**Returns:** One row per bucket, containing: + +- `table_path`: The table path. +- `table_id`: The unique table ID. +- `partition_id`: The partition ID, or `NULL` for a non-partitioned table. +- `partition_name`: The partition name, or `NULL` for a non-partitioned table. +- `bucket_id`: The bucket ID. +- `leader_id`: The leader TabletServer ID, or `NULL` if no leader is elected. +- `leader_epoch`: The leader epoch, or `NULL` if no leader is elected. +- `bucket_epoch`: The generation of the complete leader/ISR state. `NULL` indicates legacy + metadata, while `-1` indicates that no leader/ISR state exists. +- `replicas`: An `ARRAY` containing the replica TabletServer IDs. +- `isr`: An `ARRAY` containing the in-sync replica TabletServer IDs. + +**Important Notes:** + +- The caller must have `DESCRIBE` permission on the table. +- A partial partition spec matches every partition containing the specified key-value pairs. +- For a table with many partitions, provide a partition spec to avoid returning metadata for every + bucket in a single call. + +**Example:** + +```sql title="Flink SQL" +-- Use the Fluss catalog (replace 'fluss_catalog' with your catalog name if different) +USE fluss_catalog; + +-- Describe every bucket of my_db.orders +CALL sys.describe_buckets('my_db.orders'); + +-- Describe buckets in every CN partition +CALL sys.describe_buckets('my_db.orders', 'region=cn'); + +-- Describe buckets in one specific partition +CALL sys.describe_buckets('my_db.orders', 'region=cn/dt=2026-09-16'); +``` + ## Partition Procedures Fluss provides procedures to inspect partition-level metadata of partitioned tables.