[spark] Extend get_cluster_configs procedure output with default value, is_default and description columns

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,35 +78,62 @@ 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 =>
         throw new RuntimeException(s"Failed to get cluster config: ${e.getMessage}", e)
     }
   }
 
+  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 = {

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)
+  }
 }

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