[SPARK-54776][SQL] Improve error message for SQL UDF in higher-order functions

Author: Shubhambhusate

common/utils/src/main/resources/error/error-conditions.json

@@ -6747,6 +6747,11 @@
           "Lambda function with Python UDF <funcName> in a higher order function."
         ]
       },
+      "LAMBDA_FUNCTION_WITH_SQL_UDF" : {
+        "message" : [
+          "Lambda function with SQL UDF <funcName> in a higher order function."
+        ]
+      },
       "LATERAL_COLUMN_ALIAS_IN_AGGREGATE_FUNC" : {
         "message" : [
           "Referencing a lateral column alias <lca> in the aggregate function <aggFunc>."

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala

@@ -39,6 +39,8 @@ import org.apache.spark.sql.catalyst.analysis.FunctionRegistry.FunctionBuilder
 import org.apache.spark.sql.catalyst.analysis.TableFunctionRegistry.TableFunctionBuilder
 import org.apache.spark.sql.catalyst.catalog.SQLFunction.parseDefault
 import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, AttributeReference, Cast, Expression, ExpressionInfo, LateralSubquery, NamedArgumentExpression, NamedExpression, OuterReference, ScalarSubquery, UpCast}
+import org.apache.spark.sql.catalyst.expressions.NamedLambdaVariable
+import org.apache.spark.sql.catalyst.expressions.UnresolvedNamedLambdaVariable
 import org.apache.spark.sql.catalyst.parser.{CatalystSqlParser, ParserInterface}
 import org.apache.spark.sql.catalyst.plans.Inner
 import org.apache.spark.sql.catalyst.plans.logical.{FunctionSignature, InputParameter, LateralJoin, LogicalPlan, NamedParametersSupport, OneRowRelation, Project, SubqueryAlias, View}
@@ -1633,6 +1635,25 @@ class SessionCatalog(
       throw UserDefinedFunctionErrors.notAScalarFunction(function.name.nameParts)
     }
     (input: Seq[Expression]) => {
+      // Check if any input contains a lambda variable
+      val hasLambdaVar = input.exists { expr =>
+        expr.find {
+          case _: NamedLambdaVariable => true
+          case _: UnresolvedNamedLambdaVariable => true
+          case _ => false
+        }.isDefined
+      }
+      if (hasLambdaVar) {
+        val formattedInputs = input.map {
+          case v: NamedLambdaVariable => "lambda " + v.name
+          case v: UnresolvedNamedLambdaVariable => "lambda " + v.name
+          case e => e.sql
+        }.mkString(", ")
+        throw new AnalysisException(
+          errorClass = "UNSUPPORTED_FEATURE.LAMBDA_FUNCTION_WITH_SQL_UDF",
+          messageParameters = Map(
+            "funcName" -> ("\"" + function.name.unquotedString + "(" + formattedInputs + ")\"")))
+      }
       val args = rearrangeArguments(function.inputParam, input, function.name.toString)
       val returnType = function.getScalarFuncReturnType
       SQLFunctionExpression(
@@ -1712,6 +1733,26 @@ class SessionCatalog(
           name, paramSize.toString, input.size)
       }
 
+      // Check if any input contains a lambda variable
+      val hasLambdaVar = input.exists { expr =>
+        expr.find {
+          case _: NamedLambdaVariable => true
+          case _: UnresolvedNamedLambdaVariable => true
+          case _ => false
+        }.isDefined
+      }
+      if (hasLambdaVar) {
+        val formattedInputs = input.map {
+          case v: NamedLambdaVariable => "lambda " + v.name
+          case v: UnresolvedNamedLambdaVariable => "lambda " + v.name
+          case e => e.sql
+        }.mkString(", ")
+        throw new AnalysisException(
+          errorClass = "UNSUPPORTED_FEATURE.LAMBDA_FUNCTION_WITH_SQL_UDF",
+          messageParameters = Map(
+            "funcName" -> ("\"" + function.name.unquotedString + "(" + formattedInputs + ")\"")))
+      }
+
       val inputs = inputParam.map { param =>
         // Attributes referencing the input parameters inside the function can use the
         // function name as a qualifier. E.G.:

sql/core/src/test/scala/org/apache/spark/sql/execution/SQLFunctionSuite.scala

@@ -17,7 +17,7 @@
 
 package org.apache.spark.sql.execution
 
-import org.apache.spark.sql.{QueryTest, Row}
+import org.apache.spark.sql.{AnalysisException, QueryTest, Row}
 import org.apache.spark.sql.test.SharedSparkSession
 
 /**
@@ -87,4 +87,25 @@ class SQLFunctionSuite extends QueryTest with SharedSparkSession {
       checkAnswer(sql("SELECT bar(1)"), Row(2))
     }
   }
+
+  test("SQL UDF in higher-order function should fail with clear error message") {
+    withUserDefinedFunction("test_lower_udf" -> false) {
+      sql(
+        """
+          |CREATE FUNCTION test_lower_udf(s STRING)
+          |RETURNS STRING
+          |RETURN lower(s)
+          |""".stripMargin)
+      val exception = intercept[AnalysisException] {
+        sql("SELECT transform(array('A', 'B', 'C'), x -> test_lower_udf(x))").collect()
+      }
+      assert(exception.getCondition == "UNSUPPORTED_FEATURE.LAMBDA_FUNCTION_WITH_SQL_UDF",
+        s"Expected UNSUPPORTED_FEATURE.LAMBDA_FUNCTION_WITH_SQL_UDF " +
+        s"but got ${exception.getCondition}: ${exception.getMessage}")
+      assert(exception.getMessage.contains("test_lower_udf"),
+        s"Error message should contain function name: ${exception.getMessage}")
+      assert(exception.getMessage.contains("lambda x"),
+        s"Error message should contain 'lambda x': ${exception.getMessage}")
+    }
+  }
 }