[SPARK-54834][SQL] Add new interfaces SimpleProcedure and SimpleFunction

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

It's common that a procedure does not have any overload and the `bind` method always return the same `BoundProcedure`. This PR adds a new interface `SimpleProcedure ` for this use case, which allows people to implement a `BoundProcedure` directly without thinking about how to bind. The same applies to v2 functions.

### Why are the changes needed?

Simplify a common use case when people implement procedures and functions.

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

No, it's developer facing

### How was this patch tested?

new tests

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

cursor 2.2.43

Closes #53595 from cloud-fan/procedure.

Authored-by: Wenchen Fan <wenchen@databricks.com>
Signed-off-by: Ruifeng Zheng <ruifengz@apache.org>

Author: cloud-fan

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

@@ -0,0 +1,42 @@
+/*
+ * 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 function that does not require binding to input types.
+ * <p>
+ * This interface is designed for functions that have no overloads and do not need custom binding
+ * logic. Implementations can directly provide function parameters and execution logic without
+ * implementing the {@link UnboundFunction#bind(StructType) bind} method.
+ * <p>
+ * The default {@link #bind(StructType) bind} method simply returns {@code this}, as the function
+ * is already considered bound.
+ *
+ * @since 4.2.0
+ */
+@Evolving
+public interface SimpleFunction extends UnboundFunction, BoundFunction {
+  @Override
+  default BoundFunction bind(StructType inputType) {
+    return this;
+  }
+}
+

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

@@ -0,0 +1,42 @@
+/*
+ * 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.procedures;
+
+import org.apache.spark.annotation.Evolving;
+import org.apache.spark.sql.types.StructType;
+
+/**
+ * A procedure that does not require binding to input types.
+ * <p>
+ * This interface is designed for procedures that have no overloads and do not need custom binding
+ * logic. Implementations can directly provide procedure parameters and execution logic without
+ * implementing the {@link UnboundProcedure#bind(StructType) bind} method.
+ * <p>
+ * The default {@link #bind(StructType) bind} method simply returns {@code this}, as the procedure
+ * is already considered bound.
+ *
+ * @since 4.2.0
+ */
+@Evolving
+public interface SimpleProcedure extends UnboundProcedure, BoundProcedure {
+  @Override
+  default BoundProcedure bind(StructType inputType) {
+    return this;
+  }
+}
+

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

@@ -702,6 +702,25 @@ class DataSourceV2FunctionSuite extends DatasourceV2SQLBase {
     comparePlans(df1.queryExecution.optimizedPlan, df2.queryExecution.optimizedPlan)
     checkAnswer(df1, Row(3) :: Nil)
   }
+
+  test("simple function") {
+    catalog("testcat").asInstanceOf[SupportsNamespaces].createNamespace(Array("ns"), emptyProps)
+    addFunction(Identifier.of(Array("ns"), "simple_strlen"), SimpleStrLen)
+    checkAnswer(sql("SELECT testcat.ns.simple_strlen('abc')"), Row(3) :: Nil)
+    checkAnswer(sql("SELECT testcat.ns.simple_strlen('hello world')"), Row(11) :: Nil)
+  }
+}
+
+case object SimpleStrLen extends SimpleFunction with ScalarFunction[Int] {
+  override def inputTypes(): Array[DataType] = Array(StringType)
+  override def resultType(): DataType = IntegerType
+  override def name(): String = "simple_strlen"
+  override def description(): String = "simple string length function"
+
+  override def produceResult(input: InternalRow): Int = {
+    val s = input.getString(0)
+    s.length
+  }
 }
 
 case object StrLenDefault extends ScalarFunction[Int] {

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

@@ -26,7 +26,7 @@ import org.apache.spark.sql.{AnalysisException, QueryTest, Row}
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.util.TypeUtils.toSQLId
 import org.apache.spark.sql.connector.catalog.{BasicInMemoryTableCatalog, DefaultValue, Identifier, InMemoryCatalog}
-import org.apache.spark.sql.connector.catalog.procedures.{BoundProcedure, ProcedureParameter, UnboundProcedure}
+import org.apache.spark.sql.connector.catalog.procedures.{BoundProcedure, ProcedureParameter, SimpleProcedure, UnboundProcedure}
 import org.apache.spark.sql.connector.catalog.procedures.ProcedureParameter.Mode
 import org.apache.spark.sql.connector.catalog.procedures.ProcedureParameter.Mode.{IN, INOUT, OUT}
 import org.apache.spark.sql.connector.expressions.{Expression, GeneralScalarExpression, LiteralValue}
@@ -486,6 +486,12 @@ class ProcedureSuite extends QueryTest with SharedSparkSession with BeforeAndAft
     checkAnswer(sql("CALL cat.ns.sum(5)"), Row(9) :: Nil)
   }
 
+  test("simple procedure") {
+    catalog.createProcedure(Identifier.of(Array("ns"), "simple_sum"), SimpleSum)
+    checkAnswer(sql("CALL cat.ns.simple_sum(3, 7)"), Row(10) :: Nil)
+    checkAnswer(sql("CALL cat.ns.simple_sum(in2 => 4, in1 => 6)"), Row(10) :: Nil)
+  }
+
   test("SPARK-51780: Implement DESC PROCEDURE") {
     catalog.createProcedure(Identifier.of(Array("ns"), "foo"), UnboundSum)
     catalog.createProcedure(Identifier.of(Array("ns", "db"), "abc"), UnboundLongSum)
@@ -610,7 +616,7 @@ class ProcedureSuite extends QueryTest with SharedSparkSession with BeforeAndAft
   object UnboundNonExecutableSum extends UnboundProcedure {
     override def name: String = "sum"
     override def description: String = "sum integers"
-    override def bind(inputType: StructType): BoundProcedure = Sum
+    override def bind(inputType: StructType): BoundProcedure = NonExecutableSum
   }
 
   object NonExecutableSum extends BoundProcedure {
@@ -633,10 +639,10 @@ class ProcedureSuite extends QueryTest with SharedSparkSession with BeforeAndAft
   object UnboundSum extends UnboundProcedure {
     override def name: String = "sum"
     override def description: String = "sum integers"
-    override def bind(inputType: StructType): BoundProcedure = Sum
+    override def bind(inputType: StructType): BoundProcedure = new Sum
   }
 
-  object Sum extends BoundProcedure {
+  class Sum extends BoundProcedure {
     override def name: String = "sum"
 
     override def description: String = "sum integers"
@@ -897,4 +903,10 @@ class ProcedureSuite extends QueryTest with SharedSparkSession with BeforeAndAft
     override def defaultValue: DefaultValue = null
     override def comment: String = null
   }
+
+  object SimpleSum extends Sum with SimpleProcedure {
+    override def name: String = "simple_sum"
+
+    override def description: String = "simple sum integers"
+  }
 }