[SPARK-37731][SQL] Refactor and cleanup function lookup in Analyzer

### What changes were proposed in this pull request?

Today, the function lookup code path is pretty hard to understand as it spreads over many places:
1. lookup v1 function
2. lookup v2 function
3. lookup higher-order function
4. lookup table function
5. lookup functions in different levels: builtin, temp and persistent.

This PR is a major refactor of the function lookup code path and cleans it up quite a bit. In general, it follows the idea of table lookup:
1. Analyzer looks up built-in or temp functions first.
2. Analyzer qualifies the function name with current catalog and namespace, or with view catalog/namespace if we are resolving a view.
3. Analyzer calls v1 sessin catalog if the catalog is `spark_catalog`, otherwise call the v2 catalog.

With this refactor, the analyzer is kind of the router and the v1 session catalog can just have some small functions with very specific goals.

The function DDL commands also follow similar table/view commands and can fail automatically if the command requires persistent functions but the resolved function is built-in/temp. After this change, it should be simpler to add v2 function DDL commands.

Note that, table function lookup is still in its own rule as it has a dedicated function registry and doesn't share the namespace of the scalar functions.

### Why are the changes needed?

code cleanup to improve readability.

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

Yes. Since Spark 3.3, DESCRIBE FUNCTION fails if the function does not exist. In Spark 3.2 or earlier, DESCRIBE FUNCTION can still run and print "Function: func_name not found".

### How was this patch tested?

existing tests

Closes apache#35004 from cloud-fan/func.

Authored-by: Wenchen Fan <wenchen@databricks.com>
Signed-off-by: Wenchen Fan <wenchen@databricks.com>

Author: cloud-fan

docs/sql-migration-guide.md

@@ -54,6 +54,8 @@ license: |
 
   - Since Spark 3.3, nulls are written as empty strings in CSV data source by default. In Spark 3.2 or earlier, nulls were written as empty strings as quoted empty strings, `""`. To restore the previous behavior, set `nullValue` to `""`.
 
+  - Since Spark 3.3, DESCRIBE FUNCTION fails if the function does not exist. In Spark 3.2 or earlier, DESCRIBE FUNCTION can still run and print "Function: func_name not found".
+
 ## Upgrading from Spark SQL 3.1 to 3.2
 
   - Since Spark 3.2, ADD FILE/JAR/ARCHIVE commands require each path to be enclosed by `"` or `'` if the path contains whitespaces.

sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/CatalogExtension.java

@@ -30,7 +30,7 @@
  * @since 3.0.0
  */
 @Evolving
-public interface CatalogExtension extends TableCatalog, SupportsNamespaces {
+public interface CatalogExtension extends TableCatalog, FunctionCatalog, SupportsNamespaces {
 
   /**
    * This will be called only once by Spark to pass in the Spark built-in session catalog, after

sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/DelegatingCatalogExtension.java

@@ -20,10 +20,8 @@
 import java.util.Map;
 
 import org.apache.spark.annotation.Evolving;
-import org.apache.spark.sql.catalyst.analysis.NamespaceAlreadyExistsException;
-import org.apache.spark.sql.catalyst.analysis.NoSuchNamespaceException;
-import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
-import org.apache.spark.sql.catalyst.analysis.TableAlreadyExistsException;
+import org.apache.spark.sql.catalyst.analysis.*;
+import org.apache.spark.sql.connector.catalog.functions.UnboundFunction;
 import org.apache.spark.sql.connector.expressions.Transform;
 import org.apache.spark.sql.types.StructType;
 import org.apache.spark.sql.util.CaseInsensitiveStringMap;
@@ -161,11 +159,30 @@ public boolean dropNamespace(String[] namespace) throws NoSuchNamespaceException
     return asNamespaceCatalog().dropNamespace(namespace);
   }
 
+  @Override
+  public UnboundFunction loadFunction(Identifier ident) throws NoSuchFunctionException {
+    return asFunctionCatalog().loadFunction(ident);
+  }
+
+  @Override
+  public Identifier[] listFunctions(String[] namespace) throws NoSuchNamespaceException {
+    return asFunctionCatalog().listFunctions(namespace);
+  }
+
+  @Override
+  public boolean functionExists(Identifier ident) {
+    return asFunctionCatalog().functionExists(ident);
+  }
+
   private TableCatalog asTableCatalog() {
-    return (TableCatalog)delegate;
+    return (TableCatalog) delegate;
   }
 
   private SupportsNamespaces asNamespaceCatalog() {
-    return (SupportsNamespaces)delegate;
+    return (SupportsNamespaces) delegate;
+  }
+
+  private FunctionCatalog asFunctionCatalog() {
+    return (FunctionCatalog) delegate;
   }
 }

sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/FunctionCatalog.java

@@ -50,4 +50,17 @@ public interface FunctionCatalog extends CatalogPlugin {
    */
   UnboundFunction loadFunction(Identifier ident) throws NoSuchFunctionException;
 
+  /**
+   * Returns true if the function exists, false otherwise.
+   *
+   * @since 3.3.0
+   */
+  default boolean functionExists(Identifier ident) {
+    try {
+      loadFunction(ident);
+      return true;
+    } catch (NoSuchFunctionException e) {
+      return false;
+    }
+  }
 }

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

@@ -125,6 +125,10 @@ trait CheckAnalysis extends PredicateHelper with LookupCatalog {
       case u: UnresolvedRelation =>
         u.failAnalysis(s"Table or view not found: ${u.multipartIdentifier.quoted}")
 
+      case u: UnresolvedFunc =>
+        throw QueryCompilationErrors.noSuchFunctionError(
+          u.multipartIdentifier, u, u.possibleQualifiedName)
+
       case u: UnresolvedHint =>
         u.failAnalysis(s"Hint not found: ${u.name}")
 

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

@@ -743,6 +743,37 @@ object FunctionRegistry {
 
   val functionSet: Set[FunctionIdentifier] = builtin.listFunction().toSet
 
+  private def makeExprInfoForVirtualOperator(name: String, usage: String): ExpressionInfo = {
+    new ExpressionInfo(
+      null,
+      null,
+      name,
+      usage,
+      "",
+      "",
+      "",
+      "",
+      "",
+      "",
+      "built-in")
+  }
+
+  val builtinOperators: Map[String, ExpressionInfo] = Map(
+    "<>" -> makeExprInfoForVirtualOperator("<>",
+      "expr1 <> expr2 - Returns true if `expr1` is not equal to `expr2`."),
+    "!=" -> makeExprInfoForVirtualOperator("!=",
+      "expr1 != expr2 - Returns true if `expr1` is not equal to `expr2`."),
+    "between" -> makeExprInfoForVirtualOperator("between",
+      "expr1 [NOT] BETWEEN expr2 AND expr3 - " +
+        "evaluate if `expr1` is [not] in between `expr2` and `expr3`."),
+    "case" -> makeExprInfoForVirtualOperator("case",
+      "CASE expr1 WHEN expr2 THEN expr3 [WHEN expr4 THEN expr5]* [ELSE expr6] END " +
+        "- When `expr1` = `expr2`, returns `expr3`; when `expr1` = `expr4`, return `expr5`; " +
+        "else return `expr6`."),
+    "||" -> makeExprInfoForVirtualOperator("||",
+      "expr1 || expr2 - Returns the concatenation of `expr1` and `expr2`.")
+  )
+
   /**
    * Create a SQL function builder and corresponding `ExpressionInfo`.
    * @param name The function name.

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

@@ -17,7 +17,7 @@
 
 package org.apache.spark.sql.catalyst.analysis
 
-import org.apache.spark.sql.catalyst.plans.logical.{DropTable, DropView, LogicalPlan, NoopCommand, UncacheTable}
+import org.apache.spark.sql.catalyst.plans.logical.{DropFunction, DropTable, DropView, LogicalPlan, NoopCommand, UncacheTable}
 import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.catalyst.trees.TreePattern.COMMAND
 
@@ -35,5 +35,7 @@ object ResolveCommandsWithIfExists extends Rule[LogicalPlan] {
       NoopCommand("DROP VIEW", u.multipartIdentifier)
     case UncacheTable(u: UnresolvedRelation, ifExists, _) if ifExists =>
       NoopCommand("UNCACHE TABLE", u.multipartIdentifier)
+    case DropFunction(u: UnresolvedFunc, ifExists) if ifExists =>
+      NoopCommand("DROP FUNCTION", u.multipartIdentifier)
   }
 }

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

@@ -21,50 +21,8 @@ import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
 import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.catalyst.trees.TreePattern._
-import org.apache.spark.sql.connector.catalog.{CatalogManager, LookupCatalog}
-import org.apache.spark.sql.errors.QueryCompilationErrors
 import org.apache.spark.sql.types.DataType
 
-/**
- * Resolve a higher order functions from the catalog. This is different from regular function
- * resolution because lambda functions can only be resolved after the function has been resolved;
- * so we need to resolve higher order function when all children are either resolved or a lambda
- * function.
- */
-case class ResolveHigherOrderFunctions(catalogManager: CatalogManager)
-  extends Rule[LogicalPlan] with LookupCatalog {
-
-  override def apply(plan: LogicalPlan): LogicalPlan = plan.resolveExpressionsWithPruning(
-    _.containsPattern(LAMBDA_FUNCTION), ruleId) {
-    case u @ UnresolvedFunction(AsFunctionIdentifier(ident), children, false, filter, ignoreNulls)
-        if hasLambdaAndResolvedArguments(children) =>
-      withPosition(u) {
-        catalogManager.v1SessionCatalog.lookupFunction(ident, children) match {
-          case func: HigherOrderFunction =>
-            filter.foreach(_.failAnalysis("FILTER predicate specified, " +
-              s"but ${func.prettyName} is not an aggregate function"))
-            if (ignoreNulls) {
-              throw QueryCompilationErrors.functionWithUnsupportedSyntaxError(
-                func.prettyName, "IGNORE NULLS")
-            }
-            func
-          case other => other.failAnalysis(
-            "A lambda function should only be used in a higher order function. However, " +
-              s"its class is ${other.getClass.getCanonicalName}, which is not a " +
-              s"higher order function.")
-        }
-      }
-  }
-
-  /**
-   * Check if the arguments of a function are either resolved or a lambda function.
-   */
-  private def hasLambdaAndResolvedArguments(expressions: Seq[Expression]): Boolean = {
-    val (lambdas, others) = expressions.partition(_.isInstanceOf[LambdaFunction])
-    lambdas.nonEmpty && others.forall(_.resolved)
-  }
-}
-
 /**
  * Resolve the lambda variables exposed by a higher order functions.
  *

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

@@ -23,9 +23,10 @@ import org.apache.spark.sql.catalyst.expressions.{Attribute, LeafExpression, Une
 import org.apache.spark.sql.catalyst.plans.logical.LeafNode
 import org.apache.spark.sql.catalyst.trees.TreePattern.{TreePattern, UNRESOLVED_FUNC}
 import org.apache.spark.sql.catalyst.util.CharVarcharUtils
-import org.apache.spark.sql.connector.catalog.{CatalogPlugin, Identifier, Table, TableCatalog}
+import org.apache.spark.sql.connector.catalog.{CatalogPlugin, FunctionCatalog, Identifier, Table, TableCatalog}
 import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
 import org.apache.spark.sql.connector.catalog.TableChange.ColumnPosition
+import org.apache.spark.sql.connector.catalog.functions.UnboundFunction
 import org.apache.spark.sql.types.{DataType, StructField}
 
 /**
@@ -52,7 +53,7 @@ case class UnresolvedTable(
 }
 
 /**
- * Holds the name of a view that has yet to be looked up in a catalog. It will be resolved to
+ * Holds the name of a view that has yet to be looked up. It will be resolved to
  * [[ResolvedView]] during analysis.
  */
 case class UnresolvedView(
@@ -115,10 +116,15 @@ case class UnresolvedFieldPosition(position: ColumnPosition) extends FieldPositi
 }
 
 /**
- * Holds the name of a function that has yet to be looked up in a catalog. It will be resolved to
- * [[ResolvedFunc]] during analysis.
+ * Holds the name of a function that has yet to be looked up. It will be resolved to
+ * [[ResolvedPersistentFunc]] or [[ResolvedNonPersistentFunc]] during analysis.
  */
-case class UnresolvedFunc(multipartIdentifier: Seq[String]) extends LeafNode {
+case class UnresolvedFunc(
+    multipartIdentifier: Seq[String],
+    commandName: String,
+    requirePersistent: Boolean,
+    funcTypeMismatchHint: Option[String],
+    possibleQualifiedName: Option[Seq[String]] = None) extends LeafNode {
   override lazy val resolved: Boolean = false
   override def output: Seq[Attribute] = Nil
   final override val nodePatterns: Seq[TreePattern] = Seq(UNRESOLVED_FUNC)
@@ -190,15 +196,23 @@ case class ResolvedView(identifier: Identifier, isTemp: Boolean) extends LeafNod
 }
 
 /**
- * A plan containing resolved function.
+ * A plan containing resolved persistent function.
  */
-// TODO: create a generic representation for v1, v2 function, after we add function
-//       support to v2 catalog. For now we only need the identifier to fallback to v1 command.
-case class ResolvedFunc(identifier: Identifier)
+case class ResolvedPersistentFunc(
+    catalog: FunctionCatalog,
+    identifier: Identifier,
+    func: UnboundFunction)
   extends LeafNode {
   override def output: Seq[Attribute] = Nil
 }
 
+/**
+ * A plan containing resolved non-persistent (temp or built-in) function.
+ */
+case class ResolvedNonPersistentFunc(name: String, func: UnboundFunction) extends LeafNode {
+  override def output: Seq[Attribute] = Nil
+}
+
 /**
  * A plan containing resolved database object name with catalog determined.
  */