From 753a26fd228bdf8d132e30117a84c3315229f289 Mon Sep 17 00:00:00 2001 From: forwardxu Date: Sat, 18 Jul 2026 11:46:53 +0800 Subject: [PATCH] [spark] Extend get_cluster_configs procedure output with default value, is_default and description columns --- .../GetClusterConfigsProcedure.scala | 135 ++++++++++++++---- .../GetClusterConfigsProcedureTest.scala | 69 +++++++-- website/docs/engine-spark/procedures.md | 12 +- 3 files changed, 183 insertions(+), 33 deletions(-) diff --git a/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/procedure/GetClusterConfigsProcedure.scala b/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/procedure/GetClusterConfigsProcedure.scala index 15c9886ee1..6c9dd68295 100644 --- a/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/procedure/GetClusterConfigsProcedure.scala +++ b/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/procedure/GetClusterConfigsProcedure.scala @@ -17,12 +17,18 @@ package org.apache.fluss.spark.procedure +import org.apache.fluss.config.{ConfigOption, ConfigOptions, MemorySize} +import org.apache.fluss.config.cluster.ConfigEntry + import org.apache.spark.sql.catalyst.InternalRow import org.apache.spark.sql.connector.catalog.TableCatalog -import org.apache.spark.sql.types.{DataTypes, Metadata, StructField, StructType} +import org.apache.spark.sql.types.{BooleanType, DataTypes, Metadata, StructField, StructType} import org.apache.spark.unsafe.types.UTF8String +import java.time.Duration + import scala.collection.JavaConverters._ +import scala.collection.mutable /** * Procedure to get cluster configuration(s). @@ -31,6 +37,10 @@ import scala.collection.JavaConverters._ * - Specific configurations by key(s) * - All configurations (when no keys are provided) * + * Besides the resolved value and its source, each row also exposes the option's declared default + * value, whether the current value equals that default, and the option description (when the key is + * a well-known [[ConfigOptions]] entry). + * * Usage examples: * {{{ * -- Get a specific configuration @@ -68,28 +78,10 @@ class GetClusterConfigsProcedure(tableCatalog: TableCatalog) extends BaseProcedu val configs = admin.describeClusterConfigs().get().asScala if (configKeys.isEmpty) { - configs - .map( - entry => - newInternalRow( - UTF8String.fromString(entry.key()), - UTF8String.fromString(entry.value()), - UTF8String.fromString(formatConfigSource(entry.source())) - )) - .toArray + configs.map(buildRow).toArray } else { val configEntryMap = configs.map(e => e.key() -> e).toMap - configKeys.flatMap { - key => - configEntryMap.get(key).map { - entry => - newInternalRow( - UTF8String.fromString(entry.key()), - UTF8String.fromString(entry.value()), - UTF8String.fromString(formatConfigSource(entry.source())) - ) - } - } + configKeys.flatMap(key => configEntryMap.get(key).map(buildRow)) } } catch { case e: Exception => @@ -97,6 +89,51 @@ class GetClusterConfigsProcedure(tableCatalog: TableCatalog) extends BaseProcedu } } + private def buildRow(entry: ConfigEntry): InternalRow = { + val resolvedValue = if (entry.value() != null) entry.value() else null + val (defaultValue, isDefault, description) = resolveConfigMeta(entry.key(), resolvedValue) + newInternalRow( + UTF8String.fromString(entry.key()), + if (resolvedValue != null) UTF8String.fromString(resolvedValue) else null, + UTF8String.fromString(formatConfigSource(entry.source())), + if (defaultValue != null) UTF8String.fromString(defaultValue) else null, + Boolean.box(isDefault), + if (description != null) UTF8String.fromString(description) else null + ) + } + + /** + * Resolves the declared metadata for a config key from the [[ConfigOptions]] registry. + * + * @param key + * the config key + * @param resolvedValue + * the resolved value from the cluster, used to determine whether the option is at its default + * @return + * a triple of (default value as string or null, whether the option is at its default, the + * option description or null). When the key is not a well-known option, the default value and + * description are null and the option is reported as being at its default. + */ + private def resolveConfigMeta(key: String, resolvedValue: String): (String, Boolean, String) = { + val option = GetClusterConfigsProcedure.CONFIG_OPTIONS_MAP.get(key) + option match { + case Some(configOption) => + val defaultValue = GetClusterConfigsProcedure.formatDefaultValue(configOption) + val description = configOption.description() + val isDefault = + if (defaultValue == null) { + true + } else { + defaultValue == resolvedValue + } + val resolvedDescription = + if (description == null || description.isEmpty) null else description + (defaultValue, isDefault, resolvedDescription) + case None => + (null, true, null) + } + } + private def formatConfigSource( source: org.apache.fluss.config.cluster.ConfigEntry.ConfigSource): String = { if (source == null) { @@ -111,7 +148,7 @@ class GetClusterConfigsProcedure(tableCatalog: TableCatalog) extends BaseProcedu } override def description(): String = { - "Retrieve cluster configuration values." + "Retrieve cluster configuration values with default value, override status and description." } } @@ -124,11 +161,61 @@ object GetClusterConfigsProcedure { private val OUTPUT_TYPE: StructType = new StructType( Array( new StructField("config_key", DataTypes.StringType, nullable = false, Metadata.empty), - new StructField("config_value", DataTypes.StringType, nullable = false, Metadata.empty), - new StructField("config_source", DataTypes.StringType, nullable = false, Metadata.empty) + new StructField("config_value", DataTypes.StringType, nullable = true, Metadata.empty), + new StructField("config_source", DataTypes.StringType, nullable = false, Metadata.empty), + new StructField("default_value", DataTypes.StringType, nullable = true, Metadata.empty), + new StructField("is_default", BooleanType, nullable = false, Metadata.empty), + new StructField("description", DataTypes.StringType, nullable = true, Metadata.empty) ) ) + /** + * Lazily builds a map from config key to its [[ConfigOption]] by reflection over the static + * fields of [[ConfigOptions]]. This is used to enrich the procedure output with the declared + * default value and description. + */ + lazy val CONFIG_OPTIONS_MAP: Map[String, ConfigOption[_]] = { + val map = mutable.Map[String, ConfigOption[_]]() + val fields = classOf[ConfigOptions].getDeclaredFields + fields.foreach { + field => + field.setAccessible(true) + if (classOf[ConfigOption[_]].isAssignableFrom(field.getType)) { + field.get(null) match { + case option: ConfigOption[_] => map.put(option.key(), option) + case _ => + } + } + } + map.toMap + } + + /** Formats a config option's default value as a user-facing string. */ + def formatDefaultValue(option: ConfigOption[_]): String = { + val value = option.defaultValue() + if (value == null) { + return null + } + value match { + case duration: Duration => formatDuration(duration) + case memorySize: MemorySize => memorySize.toString + case other => other.toString + } + } + + private def formatDuration(duration: Duration): String = { + val seconds = duration.getSeconds + if (seconds == 0) { + "0 s" + } else if (seconds >= 3600 && seconds % 3600 == 0) { + (seconds / 3600) + " hours" + } else if (seconds >= 60 && seconds % 60 == 0) { + (seconds / 60) + " min" + } else { + seconds + " s" + } + } + def builder(): ProcedureBuilder = { new BaseProcedure.Builder[GetClusterConfigsProcedure]() { override protected def doBuild(): GetClusterConfigsProcedure = { diff --git a/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/procedure/GetClusterConfigsProcedureTest.scala b/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/procedure/GetClusterConfigsProcedureTest.scala index 1f5fa4f94e..ffc4d330df 100644 --- a/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/procedure/GetClusterConfigsProcedureTest.scala +++ b/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/procedure/GetClusterConfigsProcedureTest.scala @@ -20,6 +20,8 @@ package org.apache.fluss.spark.procedure import org.apache.fluss.config.ConfigOptions import org.apache.fluss.spark.FlussSparkTestBase +import org.apache.spark.sql.Row + class GetClusterConfigsProcedureTest extends FlussSparkTestBase { test("get_cluster_configs: get all configurations") { @@ -28,15 +30,23 @@ class GetClusterConfigsProcedureTest extends FlussSparkTestBase { assert(result.length > 0) val firstRow = result.head - assert(firstRow.length == 3) + assert(firstRow.length == 6) assert( - firstRow.schema.fieldNames.sameElements(Array("config_key", "config_value", "config_source"))) + firstRow.schema.fieldNames.sameElements( + Array( + "config_key", + "config_value", + "config_source", + "default_value", + "is_default", + "description"))) result.foreach { row => assert(row.getString(0) != null) - assert(row.getString(1) != null) + // config_value may be null for unset options, but source/default flag must be present assert(row.getString(2) != null) + assert(!row.isNullAt(4)) } val result2 = @@ -51,7 +61,13 @@ class GetClusterConfigsProcedureTest extends FlussSparkTestBase { s"CALL $DEFAULT_CATALOG.sys.get_cluster_configs(config_keys => array('$testKey'))").collect() assert(result.length == 1) - assert(result.head.toString() == "[kv.snapshot.interval,1 s,STATIC]") + val row = result.head + assert(row.getString(0) == "kv.snapshot.interval") + assert(row.getString(2) == "STATIC") + // The resolved value (1 s) differs from the declared default (10 min). + assert(row.getString(3) == "10 min") + assert(!row.getBoolean(4)) + assert(row.getString(5) != null) } test("get_cluster_configs: get multiple configurations") { @@ -64,10 +80,20 @@ class GetClusterConfigsProcedureTest extends FlussSparkTestBase { assert(result.length == 2) - // convert the result into a map of key to value for easy verification, key is the first column - val kvMap: Map[String, String] = result.map(r => r.getString(0) -> r.toString).toMap - assert(kvMap.getOrElse(key1, "") == s"[$key1,1 s,STATIC]") - assert(kvMap.getOrElse(key2, "") == s"[$key2,FLUSS://localhost:0,STATIC]") + // convert the result into a map of key to row for easy verification + val rowMap: Map[String, org.apache.spark.sql.Row] = + result.map(r => r.getString(0) -> r).toMap + val kvRow = rowMap(key1) + assert(kvRow.getString(2) == "STATIC") + assert(kvRow.getString(3) == "10 min") + assert(!kvRow.getBoolean(4)) + assert(kvRow.getString(5) != null) + + val bindRow = rowMap(key2) + // bind.listeners has no declared default value in ConfigOptions. + assert(bindRow.isNullAt(3)) + assert(bindRow.getBoolean(4)) + assert(bindRow.getString(5) != null) } test("get_cluster_configs: get non-existent configuration") { @@ -111,4 +137,31 @@ class GetClusterConfigsProcedureTest extends FlussSparkTestBase { assert(result.length > 0) } + + test("get_cluster_configs: default_value and is_default columns for an overridden key") { + val testKey = ConfigOptions.KV_SNAPSHOT_INTERVAL.key() + + val result = sql( + s"CALL $DEFAULT_CATALOG.sys.get_cluster_configs(config_keys => array('$testKey'))").collect() + + assert(result.length == 1) + val row = result.head + // KV_SNAPSHOT_INTERVAL declared default is 10 min but the test cluster overrides it to 1 s. + assert(row.getString(3) == "10 min") + assert(!row.getBoolean(4)) + assert(row.getString(5) != null) + } + + test("get_cluster_configs: key with no default reports null default and is_default=true") { + val testKey = ConfigOptions.BIND_LISTENERS.key() + + val result = sql( + s"CALL $DEFAULT_CATALOG.sys.get_cluster_configs(config_keys => array('$testKey'))").collect() + + assert(result.length == 1) + val row = result.head + assert(row.isNullAt(3)) + assert(row.getBoolean(4)) + assert(row.getString(5) != null) + } } diff --git a/website/docs/engine-spark/procedures.md b/website/docs/engine-spark/procedures.md index ace2d3a6df..812d2268f1 100644 --- a/website/docs/engine-spark/procedures.md +++ b/website/docs/engine-spark/procedures.md @@ -71,8 +71,11 @@ CALL [catalog_name.]sys.get_cluster_configs(config_keys => ARRAY('key1', 'key2') **Returns:** A table with columns: - `config_key`: The configuration key name -- `config_value`: The current value +- `config_value`: The current value (nullable) - `config_source`: The source of the configuration (e.g., `DEFAULT`, `DYNAMIC`, `STATIC`) +- `default_value`: The declared default value of the option, or `NULL` if the key is not a known `ConfigOptions` entry (nullable) +- `is_default`: Whether the current value equals the declared default +- `description`: The description of the option, or `NULL` if the key is not a known `ConfigOptions` entry (nullable) **Example:** @@ -85,6 +88,13 @@ CALL sys.get_cluster_configs(); -- Get specific configurations CALL sys.get_cluster_configs(config_keys => ARRAY('kv.rocksdb.shared-rate-limiter.bytes-per-sec', 'datalake.format')); + +-- Inspect whether a configuration has been overridden from its default +CALL sys.get_cluster_configs(config_keys => ARRAY('kv.rocksdb.shared-rate-limiter.bytes-per-sec')) +SELECT config_key, config_value, default_value, is_default, description +FROM ( + CALL sys.get_cluster_configs(config_keys => ARRAY('kv.rocksdb.shared-rate-limiter.bytes-per-sec')) +); ``` ## Error Handling