[spark] Add input argument validation with friendly errors to CALL procedure framework

fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/procedure/BaseProcedure.scala

@@ -118,6 +118,76 @@ abstract class BaseProcedure(tableCatalog: TableCatalog) extends Procedure {
     new GenericInternalRow(values.toArray)
   }
 
+  /**
+   * Validates that a required, non-null argument has been provided for the given parameter.
+   *
+   * @param parameter
+   *   the procedure parameter that is being validated
+   * @param isProvided
+   *   whether the argument was provided by the caller
+   */
+  protected def checkRequiredArgument(parameter: ProcedureParameter, isProvided: Boolean): Unit = {
+    if (parameter.isRequired && !isProvided) {
+      throw new IllegalArgumentException(s"Required parameter '${parameter.name}' is missing.")
+    }
+  }
+
+  /**
+   * Validates that a provided argument can be safely interpreted as the declared parameter type and
+   * produces a user-friendly error message when it cannot.
+   *
+   * This is intended for procedures whose parameters are resolved as loosely typed Spark values
+   * (e.g. via the generic CALL framework). It should be called from a procedure's `call` method
+   * before the argument is used.
+   *
+   * @param parameter
+   *   the procedure parameter describing the expected type
+   * @param argument
+   *   the argument value provided by the caller (may be null)
+   * @throws IllegalArgumentException
+   *   if the argument is present but not assignable to the parameter's declared type
+   */
+  protected def checkArgumentType(parameter: ProcedureParameter, argument: Any): Unit = {
+    if (argument == null) {
+      return
+    }
+    val expectedType = parameter.dataType
+    val actualType = argument.getClass
+    val typeName = sparkTypeName(expectedType)
+    if (!isAssignable(expectedType, argument)) {
+      throw new IllegalArgumentException(
+        s"Expected $typeName for parameter '${parameter.name}', but got " +
+          s"${actualType.getSimpleName}.")
+    }
+  }
+
+  private def isAssignable(
+      expectedType: org.apache.spark.sql.types.DataType,
+      argument: Any): Boolean = {
+    expectedType match {
+      case org.apache.spark.sql.types.StringType => argument.isInstanceOf[String]
+      case org.apache.spark.sql.types.BooleanType => argument.isInstanceOf[java.lang.Boolean]
+      case org.apache.spark.sql.types.IntegerType =>
+        argument.isInstanceOf[java.lang.Integer]
+      case org.apache.spark.sql.types.LongType => argument.isInstanceOf[java.lang.Long]
+      case org.apache.spark.sql.types.DoubleType => argument.isInstanceOf[java.lang.Double]
+      case org.apache.spark.sql.types.FloatType => argument.isInstanceOf[java.lang.Float]
+      case _ => true
+    }
+  }
+
+  private def sparkTypeName(dataType: org.apache.spark.sql.types.DataType): String = {
+    dataType match {
+      case org.apache.spark.sql.types.StringType => "STRING"
+      case org.apache.spark.sql.types.BooleanType => "BOOLEAN"
+      case org.apache.spark.sql.types.IntegerType => "INT"
+      case org.apache.spark.sql.types.LongType => "BIGINT"
+      case org.apache.spark.sql.types.DoubleType => "DOUBLE"
+      case org.apache.spark.sql.types.FloatType => "FLOAT"
+      case other => other.simpleString
+    }
+  }
+
   /**
    * Converts a Spark Identifier to a Fluss TablePath.
    *

fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/procedure/GetClusterConfigsProcedure.scala

@@ -51,7 +51,12 @@ class GetClusterConfigsProcedure(tableCatalog: TableCatalog) extends BaseProcedu
   }
 
   override def call(args: InternalRow): Array[InternalRow] = {
-    val configKeys = if (args.numFields > 0 && !args.isNullAt(0)) {
+    val provided = args.numFields > 0 && !args.isNullAt(0)
+    checkRequiredArgument(GetClusterConfigsProcedure.PARAMETERS(0), provided)
+    if (provided) {
+      checkArgumentType(GetClusterConfigsProcedure.PARAMETERS(0), args.getArray(0))
+    }
+    val configKeys = if (provided) {
       val keysArray = args.getArray(0)
       (0 until keysArray.numElements())
         .map(i => keysArray.getUTF8String(i).toString)

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.AnalysisException
+
 class GetClusterConfigsProcedureTest extends FlussSparkTestBase {
 
   test("get_cluster_configs: get all configurations") {
@@ -111,4 +113,13 @@ class GetClusterConfigsProcedureTest extends FlussSparkTestBase {
 
     assert(result.length > 0)
   }
+
+  test("get_cluster_configs: rejects wrong argument type for config_keys") {
+    val testKey = ConfigOptions.KV_SNAPSHOT_INTERVAL.key()
+
+    // Passing a scalar STRING where an ARRAY is expected must fail before execution.
+    intercept[AnalysisException] {
+      sql(s"CALL $DEFAULT_CATALOG.sys.get_cluster_configs(config_keys => '$testKey')").collect()
+    }
+  }
 }