[SQL] Support scalar-argument table-valued functions in the DataSource V2 catalog API

Add a `TableFunctionCatalog` capability so a custom (non-session) V2 catalog can
contribute a table-valued function usable in the FROM clause:

    SELECT * FROM my_catalog.my_ns.my_tvf(42)

New connector API (connector/catalog + connector/catalog/functions):
- `TableFunctionCatalog` (loadTableFunction / listTableFunctions)
- `TableFunction` / `UnboundTableFunction` / `BoundTableFunction`
- `SupportsScalarInvocation` mixin -- produces a relation via a `ScanBuilder`
- `TableFunctionParameter` (+ Scala impl) for positional and named arguments

The invocation surface is split into a mixin so that this scalar path is
forward-stable when TABLE-argument support is added later.

Resolution opens the previously-unconditional
`MISSING_CATALOG_ABILITY.TABLE_VALUED_FUNCTIONS` guard in
`FunctionResolution.resolveTableFunctionCandidate` for a `TableFunctionCatalog`,
binds arguments (rearranging named/positional args via `NamedParametersSupport`),
and emits a `TableValuedFunctionRelation`. `InvokeTableFunctions` wraps the bound
function as a read-only V2 table so the rows flow through the ordinary DSv2 read
path (`DataSourceV2Relation` -> `V2ScanRelationPushDown` -> `BatchScanExec`),
giving distributed execution and column pruning.

Non-foldable and current_date/current_timestamp scalar arguments are rejected in
this PR; TABLE arguments are rejected until the `SupportsTableArgument` follow-up.

Author: LuciferYang

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

@@ -7591,6 +7591,12 @@
     ],
     "sqlState" : "22023"
   },
+  "TABLE_VALUED_FUNCTION_REQUIRES_FOLDABLE_ARGUMENTS" : {
+    "message" : [
+      "The table-valued function <functionName> requires all of its arguments to be constant (foldable) expressions."
+    ],
+    "sqlState" : "42K09"
+  },
   "TABLE_VALUED_FUNCTION_TOO_MANY_TABLE_ARGUMENTS" : {
     "message" : [
       "There are too many table arguments for table-valued function.",
@@ -7599,6 +7605,13 @@
     ],
     "sqlState" : "54023"
   },
+  "TABLE_VALUED_FUNCTION_WITH_TABLE_ARGUMENT_UNSUPPORTED" : {
+    "message" : [
+      "The table-valued function <functionName> is provided by a catalog that does not yet support TABLE arguments.",
+      "Only scalar arguments are supported."
+    ],
+    "sqlState" : "0A000"
+  },
   "TASK_WRITE_FAILED" : {
     "message" : [
       "Task failed while writing rows to <path>."

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

@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connector.catalog;
+
+import org.apache.spark.annotation.Evolving;
+import org.apache.spark.sql.catalyst.analysis.NoSuchFunctionException;
+import org.apache.spark.sql.catalyst.analysis.NoSuchNamespaceException;
+import org.apache.spark.sql.connector.catalog.functions.UnboundTableFunction;
+
+/**
+ * A catalog API for working with table-valued functions (functions that appear in the FROM clause
+ * and produce a relation).
+ * <p>
+ * This is a marker capability, parallel to {@link FunctionCatalog} (scalar/aggregate functions)
+ * and {@link ProcedureCatalog} (stored procedures). A catalog that does not implement this
+ * interface reports the {@code MISSING_CATALOG_ABILITY.TABLE_VALUED_FUNCTIONS} error when a
+ * table-valued function is referenced against it.
+ *
+ * @since 4.3.0
+ */
+@Evolving
+public interface TableFunctionCatalog extends CatalogPlugin {
+
+  /**
+   * List the table-valued functions in a namespace from the catalog.
+   *
+   * @param namespace a multi-part namespace
+   * @return an array of table-valued function identifiers in the namespace
+   * @throws NoSuchNamespaceException if the namespace does not exist (optional)
+   */
+  Identifier[] listTableFunctions(String[] namespace) throws NoSuchNamespaceException;
+
+  /**
+   * Load a table-valued function by identifier from the catalog.
+   *
+   * @param ident a table-valued function identifier
+   * @return an unbound table-valued function
+   * @throws NoSuchFunctionException if there is no such function
+   */
+  UnboundTableFunction loadTableFunction(Identifier ident) throws NoSuchFunctionException;
+
+  /**
+   * Returns true if the table-valued function exists, false otherwise.
+   */
+  default boolean tableFunctionExists(Identifier ident) {
+    try {
+      loadTableFunction(ident);
+      return true;
+    } catch (NoSuchFunctionException e) {
+      return false;
+    }
+  }
+}

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

@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connector.catalog.functions;
+
+import org.apache.spark.annotation.Evolving;
+import org.apache.spark.sql.types.StructType;
+
+/**
+ * A table-valued function that is bound to argument types.
+ * <p>
+ * A bound function declares its parameters (for argument validation and by-name rearrangement),
+ * its determinism, and its (possibly polymorphic) output schema. It carries NO execution method
+ * itself; a concrete function additionally implements exactly one of the invocation mixins:
+ * <ul>
+ *   <li>{@link SupportsScalarInvocation} -- a scalar-argument function that produces a relation
+ *       through a {@code ScanBuilder} (a source; reuses the DSv2 read path, column pruning,
+ *       distributed execution);</li>
+ *   <li>{@code SupportsTableArgument} -- a function that consumes a TABLE argument and transforms
+ *       its rows (added in a later phase).</li>
+ * </ul>
+ * Splitting execution into mixins keeps this interface stable as capabilities are added, and lets
+ * a connector implement only the invocation styles it supports.
+ *
+ * @since 4.3.0
+ */
+@Evolving
+public interface BoundTableFunction extends TableFunction {
+
+  /**
+   * Returns the parameters of this table-valued function, used by Spark to validate and rearrange
+   * call-site arguments (including by-name arguments). An empty array means the function takes no
+   * arguments.
+   */
+  TableFunctionParameter[] parameters();
+
+  /**
+   * Indicates whether this table-valued function is deterministic.
+   */
+  boolean isDeterministic();
+
+  /**
+   * Returns the output schema (the relation's columns) produced by this bound function. May depend
+   * on the bound arguments (polymorphic output).
+   */
+  StructType resultSchema();
+}

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

@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connector.catalog.functions;
+
+import org.apache.spark.annotation.Evolving;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.connector.read.ScanBuilder;
+
+/**
+ * A mixin for a {@link BoundTableFunction} that is invoked with scalar arguments only and produces
+ * its relation as a source.
+ * <p>
+ * The function returns a {@link ScanBuilder}, mirroring
+ * {@link org.apache.spark.sql.connector.catalog.SupportsRead#newScanBuilder}: this lets it
+ * participate in column pruning and filter pushdown (by implementing
+ * {@link org.apache.spark.sql.connector.read.SupportsPushDownRequiredColumns} etc. on the returned
+ * builder) and get distributed execution for free -- exactly like reading a table. For a small or
+ * metadata-only result, the builder's {@code build()} may return a
+ * {@link org.apache.spark.sql.connector.read.LocalScan}.
+ *
+ * @since 4.3.0
+ */
+@Evolving
+public interface SupportsScalarInvocation extends BoundTableFunction {
+
+  /**
+   * Returns a scan builder for this table-valued function given the scalar arguments.
+   * <p>
+   * Spark validates and rearranges the call-site arguments so that the order and data types of the
+   * fields in {@code scalarArgs} match those reported by {@link #parameters()}. The {@code Scan}
+   * produced by the returned builder must report a {@code readSchema()} consistent with
+   * {@link #resultSchema()} (after any column pruning the builder performs).
+   *
+   * @param scalarArgs the scalar argument values
+   * @return a scan builder producing the function's rows
+   */
+  ScanBuilder newScanBuilder(InternalRow scalarArgs);
+}

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

@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connector.catalog.functions;
+
+import org.apache.spark.annotation.Evolving;
+
+/**
+ * A base interface for all table-valued functions.
+ * <p>
+ * A table-valued function is loaded from a
+ * {@link org.apache.spark.sql.connector.catalog.TableFunctionCatalog} and, once bound to its
+ * argument types, produces a relation. This base interface carries only identity metadata; the
+ * {@link UnboundTableFunction} / {@link BoundTableFunction} pair carries binding, and the
+ * {@code SupportsScalarInvocation} / {@code SupportsTableArgument} mixins on
+ * {@link BoundTableFunction} carry the two possible execution contracts. This mirrors the scalar
+ * function ({@code UnboundFunction} / {@code BoundFunction}) and procedure
+ * ({@code UnboundProcedure} / {@code BoundProcedure}) families.
+ *
+ * @since 4.3.0
+ */
+@Evolving
+public interface TableFunction {
+
+  /**
+   * Returns the name of this table-valued function.
+   */
+  String name();
+
+  /**
+   * Returns the description of this table-valued function.
+   */
+  default String description() {
+    return getClass().toString();
+  }
+}