[SPARK-53523][SQL] Named parameters respect spark.sql.caseSensitive

[pan3793]

What changes were proposed in this pull request?

As the title.

Why are the changes needed?

The issue was originally found during

• Spark 4.0: Migrate Iceberg Stored Procedures to Spark built-in implementations iceberg#13106

I don't see any special reason that named parameters should always be case sensitive. (correct me if I'm wrong)

I tested PostgreSQL, and the named parameters are case-insensitive by default.

psql (17.6 (Debian 17.6-1.pgdg13+1))
Type "help" for help.

postgres=# CREATE FUNCTION concat_lower_or_upper(a text, b text, uppercase boolean DEFAULT false)
RETURNS text
AS
$$
 SELECT CASE
        WHEN $3 THEN UPPER($1 || ' ' || $2)
        ELSE LOWER($1 || ' ' || $2)
        END;
$$
LANGUAGE SQL IMMUTABLE STRICT;
CREATE FUNCTION
postgres=# SELECT concat_lower_or_upper('Hello', 'World', true);
 concat_lower_or_upper
-----------------------
 HELLO WORLD
(1 row)

postgres=# SELECT concat_lower_or_upper(a => 'Hello', b => 'World');
 concat_lower_or_upper
-----------------------
 hello world
(1 row)

postgres=# SELECT concat_lower_or_upper(A => 'Hello', b => 'World');
 concat_lower_or_upper
-----------------------
 hello world
(1 row)

postgres=#

Does this PR introduce any user-facing change?

Yes, named parameters used by functions, procedures now respect spark.sql.caseSensitive, instead of always performing case sensitive.

How was this patch tested?

Added UT.

Was this patch authored or co-authored using generative AI tooling?

No.

[pan3793]

cc @cloud-fan @aokolnychyi @szehon-ho

[pan3793]

cc @dtenedor

[pan3793]

cc @srielau, is this by design that named parameters are always case sensitive?

[dongjoon-hyun]

+1 from my side. Thank you, @pan3793 .

cc @yaooqinn , @LuciferYang , too.

[dongjoon-hyun]

Thank you, @pan3793 and @LuciferYang . Merged to master for Apache Spark 4.1.0.

[cloud-fan]

LGTM. Shall we add a validation in function definition to make sure the parameter names are unique case insensitiely? like how we validate column names in CREATE TABLE.

[pan3793]

Shall we add a validation in function definition to make sure the parameter names are unique case insensitiely? like how we validate column names in CREATE TABLE.

@cloud-fan I think it's already there. (correct me if I'm wrong)

scala> spark.sql("CREATE FUNCTION area(x DOUBLE, X DOUBLE) RETURNS DOUBLE RETURN x * X")
org.apache.spark.sql.AnalysisException: [DUPLICATE_ROUTINE_PARAMETER_NAMES] Found duplicate name(s) in the parameter list of the user-defined routine area: `x`. SQLSTATE: 42734
  at org.apache.spark.sql.catalyst.catalog.UserDefinedFunctionErrors$.duplicateParameterNames(UserDefinedFunctionErrors.scala:40)
  at org.apache.spark.sql.execution.command.CreateUserDefinedFunctionCommand$.checkParameterNameDuplication(CreateUserDefinedFunctionCommand.scala:108)
  at org.apache.spark.sql.execution.command.CreateSQLFunctionCommand.run(CreateSQLFunctionCommand.scala:103)
  at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult$lzycompute(commands.scala:79)
  at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult(commands.scala:77)
  at org.apache.spark.sql.execution.command.ExecutedCommandExec.executeCollect(commands.scala:88)
  at org.apache.spark.sql.execution.QueryExecution.$anonfun$eagerlyExecuteCommands$2(QueryExecution.scala:157)
  at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId0$8(SQLExecution.scala:176)
  at org.apache.spark.sql.execution.SQLExecution$.withSessionTagsApplied(SQLExecution.scala:284)
  at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId0$7(SQLExecution.scala:138)
  at org.apache.spark.JobArtifactSet$.withActiveJobArtifactState(JobArtifactSet.scala:94)
  at org.apache.spark.sql.artifact.ArtifactManager.$anonfun$withResources$1(ArtifactManager.scala:113)
  at org.apache.spark.sql.artifact.ArtifactManager.withClassLoaderIfNeeded(ArtifactManager.scala:107)
  at org.apache.spark.sql.artifact.ArtifactManager.withResources(ArtifactManager.scala:112)
  at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId0$6(SQLExecution.scala:138)
  at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:307)
  at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId0$1(SQLExecution.scala:137)
  at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:804)
  at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId0(SQLExecution.scala:91)
  at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:249)
  at org.apache.spark.sql.execution.QueryExecution.$anonfun$eagerlyExecuteCommands$1(QueryExecution.scala:157)
  at org.apache.spark.sql.execution.QueryExecution$.withInternalError(QueryExecution.scala:660)
  at org.apache.spark.sql.execution.QueryExecution.org$apache$spark$sql$execution$QueryExecution$$eagerlyExecute$1(QueryExecution.scala:156)
  at org.apache.spark.sql.execution.QueryExecution$$anonfun$eagerlyExecuteCommands$3.applyOrElse(QueryExecution.scala:173)
  at org.apache.spark.sql.execution.QueryExecution$$anonfun$eagerlyExecuteCommands$3.applyOrElse(QueryExecution.scala:166)
  at org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$transformDownWithPruning$1(TreeNode.scala:491)
  at org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(origin.scala:86)
  at org.apache.spark.sql.catalyst.trees.TreeNode.transformDownWithPruning(TreeNode.scala:491)
  at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.org$apache$spark$sql$catalyst$plans$logical$AnalysisHelper$$super$transformDownWithPruning(LogicalPlan.scala:37)
  at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDownWithPruning(AnalysisHelper.scala:360)
  at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDownWithPruning$(AnalysisHelper.scala:356)
  at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDownWithPruning(LogicalPlan.scala:37)
  at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDownWithPruning(LogicalPlan.scala:37)
  at org.apache.spark.sql.catalyst.trees.TreeNode.transformDown(TreeNode.scala:467)
  at org.apache.spark.sql.execution.QueryExecution.eagerlyExecuteCommands(QueryExecution.scala:166)
  at org.apache.spark.sql.execution.QueryExecution.$anonfun$lazyCommandExecuted$1(QueryExecution.scala:127)
  at scala.util.Try$.apply(Try.scala:217)
  at org.apache.spark.util.Utils$.doTryWithCallerStacktrace(Utils.scala:1375)
  at org.apache.spark.util.Utils$.getTryWithCallerStacktrace(Utils.scala:1436)
  at org.apache.spark.util.LazyTry.get(LazyTry.scala:58)
  at org.apache.spark.sql.execution.QueryExecution.commandExecuted(QueryExecution.scala:132)
  ... 52 elided

[szehon-ho]

Some late comment, be nice to add end to end test to ProcdureSuite.

+1 in any case, thanks for looking at it.

[szehon-ho]

nit: fix the javadoc to add the new args (in this and 'defaultRearrange')