[SPARK-54760][SQL] DelegatingCatalogExtension as session catalog supports both V1 and V2 functions

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

This PR fixes a bug that occurs when the user uses a custom `DelegatingCatalogExtension` as the session catalog, Spark can not load the v2 function properly provided by the catalog. A typical use case is Iceberg's `SparkSessionCatalog`
```
$ spark-sql \
  --conf spark.sql.catalog.spark_catalog=org.apache.iceberg.spark.SparkSessionCatalog \
  ...
```

```
spark-sql (default)> SELECT spark_catalog.system.iceberg_version();
[ROUTINE_NOT_FOUND] The routine `system`.`iceberg_version` cannot be found. Verify the spelling and correctness of the schema and catalog.
If you did not qualify the name with a schema and catalog, verify the current_schema() output, or qualify the name with the correct schema and catalog.
To tolerate the error on drop use DROP ... IF EXISTS. SQLSTATE: 42883; line 1 pos 7
```

### Why are the changes needed?

Fix bug.

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

Yes, it fixes a bug.

### How was this patch tested?

Add new UT. Also manually tested with Iceberg.
```
spark-sql (default)> SELECT spark_catalog.system.iceberg_version();
1.10.0
Time taken: 1.715 seconds, Fetched 1 row(s)
```

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

No.

Closes #53531 from pan3793/SPARK-54760.

Authored-by: Cheng Pan <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>

Author: pan3793

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

@@ -26,17 +26,16 @@ import org.apache.spark.sql.catalyst.expressions.aggregate._
 import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
 import org.apache.spark.sql.connector.catalog.{
   CatalogManager,
-  CatalogV2Util,
-  FunctionCatalog,
-  Identifier,
   LookupCatalog
 }
 import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
 import org.apache.spark.sql.connector.catalog.functions.{
   AggregateFunction => V2AggregateFunction,
-  ScalarFunction
+  ScalarFunction,
+  UnboundFunction
 }
 import org.apache.spark.sql.errors.{DataTypeErrorsBase, QueryCompilationErrors}
+import org.apache.spark.sql.internal.connector.V1Function
 import org.apache.spark.sql.types._
 
 class FunctionResolution(
@@ -52,10 +51,14 @@ class FunctionResolution(
       resolveBuiltinOrTempFunction(u.nameParts, u.arguments, u).getOrElse {
         val CatalogAndIdentifier(catalog, ident) =
           relationResolution.expandIdentifier(u.nameParts)
-        if (CatalogV2Util.isSessionCatalog(catalog)) {
-          resolveV1Function(ident.asFunctionIdentifier, u.arguments, u)
-        } else {
-          resolveV2Function(catalog.asFunctionCatalog, ident, u.arguments, u)
+        catalog.asFunctionCatalog.loadFunction(ident) match {
+          case V1Function(_) =>
+            // this triggers the second time v1 function resolution but should be cheap
+            // (no RPC to external catalog), since the metadata has been already cached
+            // in FunctionRegistry during the above `catalog.loadFunction` call.
+            resolveV1Function(ident.asFunctionIdentifier, u.arguments, u)
+          case unboundV2Func =>
+            resolveV2Function(unboundV2Func, u.arguments, u)
         }
       }
     }
@@ -272,11 +275,9 @@ class FunctionResolution(
   }
 
   private def resolveV2Function(
-      catalog: FunctionCatalog,
-      ident: Identifier,
+      unbound: UnboundFunction,
       arguments: Seq[Expression],
       u: UnresolvedFunction): Expression = {
-    val unbound = catalog.loadFunction(ident)
     val inputType = StructType(arguments.zipWithIndex.map {
       case (exp, pos) => StructField(s"_$pos", exp.dataType, exp.nullable)
     })

sql/core/src/test/resources/sql-tests/analyzer-results/identifier-clause-legacy.sql.out

@@ -972,11 +972,11 @@ VALUES(IDENTIFIER('a.b.c.d')())
 -- !query analysis
 org.apache.spark.sql.AnalysisException
 {
-  "errorClass" : "IDENTIFIER_TOO_MANY_NAME_PARTS",
-  "sqlState" : "42601",
+  "errorClass" : "REQUIRES_SINGLE_PART_NAMESPACE",
+  "sqlState" : "42K05",
   "messageParameters" : {
-    "identifier" : "`a`.`b`.`c`.`d`",
-    "limit" : "2"
+    "namespace" : "`a`.`b`.`c`",
+    "sessionCatalog" : "spark_catalog"
   },
   "queryContext" : [ {
     "objectType" : "",

sql/core/src/test/resources/sql-tests/analyzer-results/identifier-clause.sql.out

@@ -972,11 +972,11 @@ VALUES(IDENTIFIER('a.b.c.d')())
 -- !query analysis
 org.apache.spark.sql.AnalysisException
 {
-  "errorClass" : "IDENTIFIER_TOO_MANY_NAME_PARTS",
-  "sqlState" : "42601",
+  "errorClass" : "REQUIRES_SINGLE_PART_NAMESPACE",
+  "sqlState" : "42K05",
   "messageParameters" : {
-    "identifier" : "`a`.`b`.`c`.`d`",
-    "limit" : "2"
+    "namespace" : "`a`.`b`.`c`",
+    "sessionCatalog" : "spark_catalog"
   },
   "queryContext" : [ {
     "objectType" : "",

sql/core/src/test/resources/sql-tests/results/identifier-clause-legacy.sql.out

@@ -1112,11 +1112,11 @@ struct<>
 -- !query output
 org.apache.spark.sql.AnalysisException
 {
-  "errorClass" : "IDENTIFIER_TOO_MANY_NAME_PARTS",
-  "sqlState" : "42601",
+  "errorClass" : "REQUIRES_SINGLE_PART_NAMESPACE",
+  "sqlState" : "42K05",
   "messageParameters" : {
-    "identifier" : "`a`.`b`.`c`.`d`",
-    "limit" : "2"
+    "namespace" : "`a`.`b`.`c`",
+    "sessionCatalog" : "spark_catalog"
   },
   "queryContext" : [ {
     "objectType" : "",

sql/core/src/test/resources/sql-tests/results/identifier-clause.sql.out

@@ -1112,11 +1112,11 @@ struct<>
 -- !query output
 org.apache.spark.sql.AnalysisException
 {
-  "errorClass" : "IDENTIFIER_TOO_MANY_NAME_PARTS",
-  "sqlState" : "42601",
+  "errorClass" : "REQUIRES_SINGLE_PART_NAMESPACE",
+  "sqlState" : "42K05",
   "messageParameters" : {
-    "identifier" : "`a`.`b`.`c`.`d`",
-    "limit" : "2"
+    "namespace" : "`a`.`b`.`c`",
+    "sessionCatalog" : "spark_catalog"
   },
   "queryContext" : [ {
     "objectType" : "",

sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2DataFrameSessionCatalogSuite.scala

@@ -168,6 +168,10 @@ private [connector] trait SessionCatalogTest[T <: Table, Catalog <: TestV2Sessio
     spark.sessionState.catalogManager.catalog(name)
   }
 
+  protected def sessionCatalog: Catalog = {
+    catalog(SESSION_CATALOG_NAME).asInstanceOf[Catalog]
+  }
+
   protected val v2Format: String = classOf[FakeV2ProviderWithCustomSchema].getName
 
   protected val catalogClassName: String = classOf[InMemoryTableSessionCatalog].getName
@@ -178,7 +182,9 @@ private [connector] trait SessionCatalogTest[T <: Table, Catalog <: TestV2Sessio
 
   override def afterEach(): Unit = {
     super.afterEach()
-    catalog(SESSION_CATALOG_NAME).asInstanceOf[Catalog].clearTables()
+    sessionCatalog.checkUsage()
+    sessionCatalog.clearTables()
+    sessionCatalog.clearFunctions()
     spark.conf.unset(V2_SESSION_CATALOG_IMPLEMENTATION.key)
   }
 

sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2FunctionSuite.scala

@@ -702,127 +702,127 @@ class DataSourceV2FunctionSuite extends DatasourceV2SQLBase {
     comparePlans(df1.queryExecution.optimizedPlan, df2.queryExecution.optimizedPlan)
     checkAnswer(df1, Row(3) :: Nil)
   }
+}
 
-  private case object StrLenDefault extends ScalarFunction[Int] {
-    override def inputTypes(): Array[DataType] = Array(StringType)
-    override def resultType(): DataType = IntegerType
-    override def name(): String = "strlen_default"
+case object StrLenDefault extends ScalarFunction[Int] {
+  override def inputTypes(): Array[DataType] = Array(StringType)
+  override def resultType(): DataType = IntegerType
+  override def name(): String = "strlen_default"
 
-    override def produceResult(input: InternalRow): Int = {
-      val s = input.getString(0)
-      s.length
-    }
+  override def produceResult(input: InternalRow): Int = {
+    val s = input.getString(0)
+    s.length
   }
+}
 
-  case object StrLenMagic extends ScalarFunction[Int] {
-    override def inputTypes(): Array[DataType] = Array(StringType)
-    override def resultType(): DataType = IntegerType
-    override def name(): String = "strlen_magic"
+case object StrLenMagic extends ScalarFunction[Int] {
+  override def inputTypes(): Array[DataType] = Array(StringType)
+  override def resultType(): DataType = IntegerType
+  override def name(): String = "strlen_magic"
 
-    def invoke(input: UTF8String): Int = {
-      input.toString.length
-    }
+  def invoke(input: UTF8String): Int = {
+    input.toString.length
   }
+}
 
-  case object StrLenBadMagic extends ScalarFunction[Int] {
-    override def inputTypes(): Array[DataType] = Array(StringType)
-    override def resultType(): DataType = IntegerType
-    override def name(): String = "strlen_bad_magic"
+case object StrLenBadMagic extends ScalarFunction[Int] {
+  override def inputTypes(): Array[DataType] = Array(StringType)
+  override def resultType(): DataType = IntegerType
+  override def name(): String = "strlen_bad_magic"
 
-    def invoke(input: String): Int = {
-      input.length
-    }
+  def invoke(input: String): Int = {
+    input.length
   }
+}
 
-  case object StrLenBadMagicWithDefault extends ScalarFunction[Int] {
-    override def inputTypes(): Array[DataType] = Array(StringType)
-    override def resultType(): DataType = IntegerType
-    override def name(): String = "strlen_bad_magic"
-
-    def invoke(input: String): Int = {
-      input.length
-    }
+case object StrLenBadMagicWithDefault extends ScalarFunction[Int] {
+  override def inputTypes(): Array[DataType] = Array(StringType)
+  override def resultType(): DataType = IntegerType
+  override def name(): String = "strlen_bad_magic"
 
-    override def produceResult(input: InternalRow): Int = {
-      val s = input.getString(0)
-      s.length
-    }
+  def invoke(input: String): Int = {
+    input.length
   }
 
-  private case object StrLenNoImpl extends ScalarFunction[Int] {
-    override def inputTypes(): Array[DataType] = Array(StringType)
-    override def resultType(): DataType = IntegerType
-    override def name(): String = "strlen_noimpl"
+  override def produceResult(input: InternalRow): Int = {
+    val s = input.getString(0)
+    s.length
   }
+}
 
-  // input type doesn't match arguments accepted by `UnboundFunction.bind`
-  private case object StrLenBadInputTypes extends ScalarFunction[Int] {
-    override def inputTypes(): Array[DataType] = Array(StringType, IntegerType)
-    override def resultType(): DataType = IntegerType
-    override def name(): String = "strlen_bad_input_types"
-  }
+case object StrLenNoImpl extends ScalarFunction[Int] {
+  override def inputTypes(): Array[DataType] = Array(StringType)
+  override def resultType(): DataType = IntegerType
+  override def name(): String = "strlen_noimpl"
+}
 
-  private case object BadBoundFunction extends BoundFunction {
-    override def inputTypes(): Array[DataType] = Array(StringType)
-    override def resultType(): DataType = IntegerType
-    override def name(): String = "bad_bound_func"
-  }
+// input type doesn't match arguments accepted by `UnboundFunction.bind`
+case object StrLenBadInputTypes extends ScalarFunction[Int] {
+  override def inputTypes(): Array[DataType] = Array(StringType, IntegerType)
+  override def resultType(): DataType = IntegerType
+  override def name(): String = "strlen_bad_input_types"
+}
 
-  object UnboundDecimalAverage extends UnboundFunction {
-    override def name(): String = "decimal_avg"
+case object BadBoundFunction extends BoundFunction {
+  override def inputTypes(): Array[DataType] = Array(StringType)
+  override def resultType(): DataType = IntegerType
+  override def name(): String = "bad_bound_func"
+}
 
-    override def bind(inputType: StructType): BoundFunction = {
-      if (inputType.fields.length > 1) {
-        throw new UnsupportedOperationException("Too many arguments")
-      }
+object UnboundDecimalAverage extends UnboundFunction {
+  override def name(): String = "decimal_avg"
 
-      // put interval type here for testing purpose
-      inputType.fields(0).dataType match {
-        case _: NumericType | _: DayTimeIntervalType => DecimalAverage
-        case dataType =>
-          throw new UnsupportedOperationException(s"Unsupported input type: $dataType")
-      }
+  override def bind(inputType: StructType): BoundFunction = {
+    if (inputType.fields.length > 1) {
+      throw new UnsupportedOperationException("Too many arguments")
     }
 
-    override def description(): String =
-      "decimal_avg: produces an average using decimal division"
+    // put interval type here for testing purpose
+    inputType.fields(0).dataType match {
+      case _: NumericType | _: DayTimeIntervalType => DecimalAverage
+      case dataType =>
+        throw new UnsupportedOperationException(s"Unsupported input type: $dataType")
+    }
   }
 
-  object DecimalAverage extends AggregateFunction[(Decimal, Int), Decimal] {
-    override def name(): String = "decimal_avg"
-    override def inputTypes(): Array[DataType] = Array(DecimalType.SYSTEM_DEFAULT)
-    override def resultType(): DataType = DecimalType.SYSTEM_DEFAULT
+  override def description(): String =
+    "decimal_avg: produces an average using decimal division"
+}
 
-    override def newAggregationState(): (Decimal, Int) = (Decimal.ZERO, 0)
+object DecimalAverage extends AggregateFunction[(Decimal, Int), Decimal] {
+  override def name(): String = "decimal_avg"
+  override def inputTypes(): Array[DataType] = Array(DecimalType.SYSTEM_DEFAULT)
+  override def resultType(): DataType = DecimalType.SYSTEM_DEFAULT
 
-    override def update(state: (Decimal, Int), input: InternalRow): (Decimal, Int) = {
-      if (input.isNullAt(0)) {
-        state
-      } else {
-        val l = input.getDecimal(0, DecimalType.SYSTEM_DEFAULT.precision,
-          DecimalType.SYSTEM_DEFAULT.scale)
-        state match {
-          case (_, d) if d == 0 =>
-            (l, 1)
-          case (total, count) =>
-            (total + l, count + 1)
-        }
-      }
-    }
+  override def newAggregationState(): (Decimal, Int) = (Decimal.ZERO, 0)
 
-    override def merge(leftState: (Decimal, Int), rightState: (Decimal, Int)): (Decimal, Int) = {
-      (leftState._1 + rightState._1, leftState._2 + rightState._2)
+  override def update(state: (Decimal, Int), input: InternalRow): (Decimal, Int) = {
+    if (input.isNullAt(0)) {
+      state
+    } else {
+      val l = input.getDecimal(0, DecimalType.SYSTEM_DEFAULT.precision,
+        DecimalType.SYSTEM_DEFAULT.scale)
+      state match {
+        case (_, d) if d == 0 =>
+          (l, 1)
+        case (total, count) =>
+          (total + l, count + 1)
+      }
     }
+  }
 
-    override def produceResult(state: (Decimal, Int)): Decimal = state._1 / Decimal(state._2)
+  override def merge(leftState: (Decimal, Int), rightState: (Decimal, Int)): (Decimal, Int) = {
+    (leftState._1 + rightState._1, leftState._2 + rightState._2)
   }
 
-  object NoImplAverage extends UnboundFunction {
-    override def name(): String = "no_impl_avg"
-    override def description(): String = name()
+  override def produceResult(state: (Decimal, Int)): Decimal = state._1 / Decimal(state._2)
+}
+
+object NoImplAverage extends UnboundFunction {
+  override def name(): String = "no_impl_avg"
+  override def description(): String = name()
 
-    override def bind(inputType: StructType): BoundFunction = {
-      throw SparkUnsupportedOperationException()
-    }
+  override def bind(inputType: StructType): BoundFunction = {
+    throw SparkUnsupportedOperationException()
   }
 }

sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSessionCatalogSuite.scala

@@ -17,7 +17,7 @@
 
 package org.apache.spark.sql.connector
 
-import org.apache.spark.sql.{DataFrame, SaveMode}
+import org.apache.spark.sql.{DataFrame, Row, SaveMode}
 import org.apache.spark.sql.connector.catalog.{Identifier, InMemoryTable, Table, TableCatalog}
 
 class DataSourceV2SQLSessionCatalogSuite
@@ -79,4 +79,11 @@ class DataSourceV2SQLSessionCatalogSuite
       assert(getTableMetadata("default.t").columns().map(_.name()) === Seq("c2", "c1"))
     }
   }
+
+  test("SPARK-54760: DelegatingCatalogExtension supports both V1 and V2 functions") {
+    sessionCatalog.createFunction(Identifier.of(Array("ns"), "strlen"), StrLen(StrLenDefault))
+    checkAnswer(
+      sql("SELECT char_length('Hello') as v1, ns.strlen('Spark') as v2"),
+      Row(5, 5))
+  }
 }