Improved the logs message regarding lambda function with SQL UDF

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

@@ -38,7 +38,7 @@ import org.apache.spark.sql.catalyst.analysis._
 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.{Alias, Attribute, AttributeReference, Cast, Expression, ExpressionInfo, LateralSubquery, NamedArgumentExpression, NamedExpression, NamedLambdaVariable, OuterReference, ScalarSubquery, UnresolvedNamedLambdaVariable, UpCast}
 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 +1633,24 @@ class SessionCatalog(
       throw UserDefinedFunctionErrors.notAScalarFunction(function.name.nameParts)
     }
     (input: Seq[Expression]) => {
+      // Check if any input contains a lambda variable - SQL functions cannot be used
+      // inside lambda functions in higher-order functions
+      val lambdaVar = input.flatMap(_.find {
+        case _: NamedLambdaVariable | _: UnresolvedNamedLambdaVariable => true
+        case _ => false
+      }).headOption
+      if (lambdaVar.isDefined) {
+        // Format input expressions, replacing lambda variables with "lambda <name>" format
+        val formattedInputs = input.map {
+          case v: NamedLambdaVariable => s"lambda ${v.name}"
+          case v: UnresolvedNamedLambdaVariable => s"lambda ${v.name}"
+          case e => e.sql
+        }.mkString(", ")
+        throw new AnalysisException(
+          errorClass = "UNSUPPORTED_FEATURE.LAMBDA_FUNCTION_WITH_SQL_UDF",
+          messageParameters = Map(
+            "funcName" -> s"\"${function.name.unquotedString}($formattedInputs)\""))
+      }
       val args = rearrangeArguments(function.inputParam, input, function.name.toString)
       val returnType = function.getScalarFuncReturnType
       SQLFunctionExpression(
@@ -1712,6 +1730,25 @@ class SessionCatalog(
           name, paramSize.toString, input.size)
       }
 
+      // Check if any input contains a lambda variable - SQL functions cannot be used
+      // inside lambda functions in higher-order functions
+      val lambdaVar = input.flatMap(_.find {
+        case _: NamedLambdaVariable | _: UnresolvedNamedLambdaVariable => true
+        case _ => false
+      }).headOption
+      if (lambdaVar.isDefined) {
+        // Format input expressions, replacing lambda variables with "lambda <name>" format
+        val formattedInputs = input.map {
+          case v: NamedLambdaVariable => s"lambda ${v.name}"
+          case v: UnresolvedNamedLambdaVariable => s"lambda ${v.name}"
+          case e => e.sql
+        }.mkString(", ")
+        throw new AnalysisException(
+          errorClass = "UNSUPPORTED_FEATURE.LAMBDA_FUNCTION_WITH_SQL_UDF",
+          messageParameters = Map(
+            "funcName" -> s"\"${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}")
+    }
+  }
 }