From 6dc686fbcdbce963779c3d4193497ba1a22400a5 Mon Sep 17 00:00:00 2001 From: forwardxu Date: Sat, 18 Jul 2026 11:48:06 +0800 Subject: [PATCH] [spark] Add input argument validation with friendly errors to CALL procedure framework --- .../fluss/spark/procedure/BaseProcedure.scala | 70 +++++++++++++++++++ .../GetClusterConfigsProcedure.scala | 7 +- .../GetClusterConfigsProcedureTest.scala | 11 +++ 3 files changed, 87 insertions(+), 1 deletion(-) diff --git a/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/procedure/BaseProcedure.scala b/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/procedure/BaseProcedure.scala index ef70f89e9d..bb26244dd9 100644 --- a/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/procedure/BaseProcedure.scala +++ b/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. * 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..7e113afaf0 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 @@ -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) 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..e4f7ff6c2e 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.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() + } + } }