[SPARK-24951][SQL] Table valued functions should throw AnalysisException

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

@@ -19,6 +19,7 @@ package org.apache.spark.sql.catalyst.analysis
 
 import java.util.Locale
 
+import org.apache.spark.sql.AnalysisException
 import org.apache.spark.sql.catalyst.expressions.{Alias, Expression}
 import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Project, Range}
 import org.apache.spark.sql.catalyst.rules._
@@ -68,9 +69,11 @@ object ResolveTableValuedFunctions extends Rule[LogicalPlan] {
       : (ArgumentList, Seq[Any] => LogicalPlan) = {
     (ArgumentList(args: _*),
      pf orElse {
-       case args =>
-         throw new IllegalArgumentException(
-           "Invalid arguments for resolved function: " + args.mkString(", "))
+       case arguments =>
+         // This is caught again by the apply function and rethrow with richer information about
+         // position, etc, for a better error message.
+         throw new AnalysisException(
+           "Invalid arguments for resolved function: " + arguments.mkString(", "))
      })
   }
 
@@ -105,22 +108,35 @@ object ResolveTableValuedFunctions extends Rule[LogicalPlan] {
 
   override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
     case u: UnresolvedTableValuedFunction if u.functionArgs.forall(_.resolved) =>
+      // The whole resolution is somewhat difficult to understand here due to too much abstractions.
+      // We should probably rewrite the following at some point. Reynold was just here to improve
+      // error messages and didn't have time to do a proper rewrite.
       val resolvedFunc = builtinFunctions.get(u.functionName.toLowerCase(Locale.ROOT)) match {
         case Some(tvf) =>
+
+          def failAnalysis(): Nothing = {
+            val argTypes = u.functionArgs.map(_.dataType.typeName).mkString(", ")
+            u.failAnalysis(
+              s"""error: table-valued function ${u.functionName} with alternatives:
+                 |${tvf.keys.map(_.toString).toSeq.sorted.map(x => s" ($x)").mkString("\n")}
+                 |cannot be applied to: ($argTypes)""".stripMargin)
+          }
+
           val resolved = tvf.flatMap { case (argList, resolver) =>
             argList.implicitCast(u.functionArgs) match {
               case Some(casted) =>
-                Some(resolver(casted.map(_.eval())))
+                try {
+                  Some(resolver(casted.map(_.eval())))
+                } catch {
+                  case e: AnalysisException =>
+                    failAnalysis()
+                }
               case _ =>
                 None
             }
           }
           resolved.headOption.getOrElse {
-            val argTypes = u.functionArgs.map(_.dataType.typeName).mkString(", ")
-            u.failAnalysis(
-              s"""error: table-valued function ${u.functionName} with alternatives:
-                |${tvf.keys.map(_.toString).toSeq.sorted.map(x => s" ($x)").mkString("\n")}
-                |cannot be applied to: (${argTypes})""".stripMargin)
+            failAnalysis()
           }
         case _ =>
           u.failAnalysis(s"could not resolve `${u.functionName}` to a table-valued function")

sql/core/src/test/resources/sql-tests/results/table-valued-functions.sql.out

@@ -83,8 +83,13 @@ select * from range(1, null)
 -- !query 6 schema
 struct<>
 -- !query 6 output
-java.lang.IllegalArgumentException
-Invalid arguments for resolved function: 1, null
+org.apache.spark.sql.AnalysisException
+error: table-valued function range with alternatives:
+ (end: long)
+ (start: long, end: long)
+ (start: long, end: long, step: long)
+ (start: long, end: long, step: long, numPartitions: integer)
+cannot be applied to: (integer, null); line 1 pos 14
 
 
 -- !query 7