[SPARK-54041][SQL] Refactor ParameterizedQuery arguments validation

### What changes were proposed in this pull request?
In this issue I propose to refactor `ParameterizedQuery` arguments validation to `ParameterizedQueryArgumentsValidator` so it can be reused between single-pass and fixed-point analyzer implementations. We also remove one redundant case from the `ParameterizedQueryArgumentsValidator.isNotAllowed` (`Alias` shouldn't call `isNotAllowed` again as it introduces unnecessary overhead to the method) to improve performance.

### Why are the changes needed?
To ease code maintenance between single-pass and fixed-point analyzer implementations.

### Does this PR introduce _any_ user-facing change?
No.

### How was this patch tested?
Existing tests (refactor).

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

Closes #52744 from mihailoale-db/refactorparamcheckargs.

Authored-by: mihailoale-db <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>

Author: mihailoale-db

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ParameterizedQueryArgumentsValidator.scala

@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.catalyst.expressions.{
+  Alias,
+  CreateArray,
+  CreateMap,
+  CreateNamedStruct,
+  Expression,
+  Literal,
+  MapFromArrays,
+  MapFromEntries,
+  VariableReference
+}
+
+/**
+ * Object used to validate arguments of [[ParameterizedQuery]] nodes.
+ */
+object ParameterizedQueryArgumentsValidator {
+
+  /**
+   * Validates the list of provided arguments. In case there is a invalid argument, throws
+   * `INVALID_SQL_ARG` exception.
+   */
+  def apply(arguments: Iterable[(String, Expression)]): Unit = {
+    arguments.find(arg => isNotAllowed(arg._2)).foreach { case (name, expr) =>
+      expr.failAnalysis(
+        errorClass = "INVALID_SQL_ARG",
+        messageParameters = Map("name" -> name))
+    }
+  }
+
+  /**
+   * Recursively checks the provided expression tree. In case there is an invalid expression type
+   * returns `false`. Otherwise, returns `true`.
+   */
+  private def isNotAllowed(expression: Expression): Boolean = expression.exists {
+    case _: Literal | _: CreateArray | _: CreateNamedStruct | _: CreateMap | _: MapFromArrays |
+        _: MapFromEntries | _: VariableReference | _: Alias =>
+      false
+    case _ => true
+  }
+}

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/parameters.scala

@@ -18,7 +18,7 @@
 package org.apache.spark.sql.catalyst.analysis
 
 import org.apache.spark.SparkException
-import org.apache.spark.sql.catalyst.expressions.{Alias, CreateArray, CreateMap, CreateNamedStruct, Expression, LeafExpression, Literal, MapFromArrays, MapFromEntries, SubqueryExpression, Unevaluable, VariableReference}
+import org.apache.spark.sql.catalyst.expressions.{Expression, LeafExpression, SubqueryExpression, Unevaluable}
 import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, SupervisingCommand}
 import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.catalyst.trees.TreePattern.{COMMAND, PARAMETER, PARAMETERIZED_QUERY, TreePattern, UNRESOLVED_WITH}
@@ -173,19 +173,6 @@ object MoveParameterizedQueriesDown extends Rule[LogicalPlan] {
  * from the user-specified arguments.
  */
 object BindParameters extends Rule[LogicalPlan] with QueryErrorsBase {
-  private def checkArgs(args: Iterable[(String, Expression)]): Unit = {
-    def isNotAllowed(expr: Expression): Boolean = expr.exists {
-      case _: Literal | _: CreateArray | _: CreateNamedStruct |
-        _: CreateMap | _: MapFromArrays |  _: MapFromEntries | _: VariableReference => false
-      case a: Alias => isNotAllowed(a.child)
-      case _ => true
-    }
-    args.find(arg => isNotAllowed(arg._2)).foreach { case (name, expr) =>
-      expr.failAnalysis(
-        errorClass = "INVALID_SQL_ARG",
-        messageParameters = Map("name" -> name))
-    }
-  }
 
   private def bind(p0: LogicalPlan)(f: PartialFunction[Expression, Expression]): LogicalPlan = {
     var stop = false
@@ -210,15 +197,15 @@ object BindParameters extends Rule[LogicalPlan] with QueryErrorsBase {
             s"must be equal to the number of argument values ${argValues.length}.")
         }
         val args = argNames.zip(argValues).toMap
-        checkArgs(args)
+        ParameterizedQueryArgumentsValidator(args)
         bind(child) { case NamedParameter(name) if args.contains(name) => args(name) }
 
       case PosParameterizedQuery(child, args)
         if !child.containsPattern(UNRESOLVED_WITH) &&
           args.forall(_.resolved) =>
 
         val indexedArgs = args.zipWithIndex
-        checkArgs(indexedArgs.map(arg => (s"_${arg._2}", arg._1)))
+        ParameterizedQueryArgumentsValidator(indexedArgs.map(arg => (s"_${arg._2}", arg._1)))
 
         val positions = scala.collection.mutable.Set.empty[Int]
         bind(child) { case p @ PosParameter(pos) => positions.add(pos); p }
@@ -238,7 +225,7 @@ object BindParameters extends Rule[LogicalPlan] with QueryErrorsBase {
           val finalName = if (name.isEmpty) s"_$index" else name
           finalName -> arg
         }
-        checkArgs(allArgs)
+        ParameterizedQueryArgumentsValidator(allArgs)
 
         // Collect parameter types used in the query to enforce invariants
         var hasNamedParam = false

sql/core/src/main/scala/org/apache/spark/sql/classic/SparkSession.scala

@@ -40,9 +40,9 @@ import org.apache.spark.sql
 import org.apache.spark.sql.{AnalysisException, Artifact, DataSourceRegistration, Encoder, Encoders, ExperimentalMethods, Row, SparkSessionBuilder, SparkSessionCompanion, SparkSessionExtensions, SparkSessionExtensionsProvider, UDTFRegistration}
 import org.apache.spark.sql.artifact.ArtifactManager
 import org.apache.spark.sql.catalyst._
-import org.apache.spark.sql.catalyst.analysis.{GeneralParameterizedQuery, NameParameterizedQuery, PosParameterizedQuery, UnresolvedRelation}
+import org.apache.spark.sql.catalyst.analysis.{GeneralParameterizedQuery, NameParameterizedQuery, ParameterizedQueryArgumentsValidator, PosParameterizedQuery, UnresolvedRelation}
 import org.apache.spark.sql.catalyst.encoders._
-import org.apache.spark.sql.catalyst.expressions.{Alias, AttributeReference, CreateArray, CreateMap, CreateNamedStruct, Expression, Literal, MapFromArrays, MapFromEntries, VariableReference}
+import org.apache.spark.sql.catalyst.expressions.{Alias, AttributeReference, Expression, Literal}
 import org.apache.spark.sql.catalyst.parser.{HybridParameterContext, NamedParameterContext, ParserInterface, PositionalParameterContext}
 import org.apache.spark.sql.catalyst.plans.logical.{CompoundBody, LocalRelation, OneRowRelation, Project, Range}
 import org.apache.spark.sql.catalyst.types.DataTypeUtils.toAttributes
@@ -453,27 +453,15 @@ class SparkSession private(
     // Analyze the plan to resolve expressions
     val analyzed = sessionState.analyzer.execute(fakePlan)
 
-    // Validate: the expression tree must only contain allowed expression types.
-    // This mirrors the validation in BindParameters.checkArgs.
-    // We check this BEFORE optimization to catch unsupported functions like str_to_map.
-    def isNotAllowed(expr: Expression): Boolean = expr.exists {
-      case _: Literal | _: CreateArray | _: CreateNamedStruct |
-        _: CreateMap | _: MapFromArrays | _: MapFromEntries | _: VariableReference => false
-      case a: Alias => isNotAllowed(a.child)
-      case _ => true
-    }
-
-    analyzed.asInstanceOf[Project].projectList.foreach { alias =>
-      val optimizedExpr = alias.asInstanceOf[Alias].child
-      if (isNotAllowed(optimizedExpr)) {
-        // Both modern and legacy modes use INVALID_SQL_ARG for sql() API argument validation.
-        // UNSUPPORTED_EXPR_FOR_PARAMETER is reserved for EXECUTE IMMEDIATE.
-        throw new AnalysisException(
-          errorClass = "INVALID_SQL_ARG",
-          messageParameters = Map("name" -> alias.name),
-          origin = optimizedExpr.origin)
-      }
+    val expressionsToValidate = analyzed.asInstanceOf[Project].projectList.map {
+      case alias: Alias =>
+        (alias.name, alias.child)
+      case other =>
+        throw SparkException.internalError(
+          s"Expected an Alias, but got ${other.getClass.getSimpleName}"
+        )
     }
+    ParameterizedQueryArgumentsValidator(expressionsToValidate)
 
     // Optimize to constant-fold expressions. After optimization, all allowed expressions
     // should be folded to Literals.