+ * 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. + *
+ * 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; + } +} + diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/procedures/SimpleProcedure.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/procedures/SimpleProcedure.java new file mode 100644 index 0000000000000..e464a3703e094 --- /dev/null +++ b/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. + *
+ * 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. + *
+ * 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; + } +} + diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2FunctionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2FunctionSuite.scala index 366528e46ff23..23e221d6ecaf3 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2FunctionSuite.scala +++ b/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] { diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/ProcedureSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/ProcedureSuite.scala index a88e6dbbe6dee..bcf2880564719 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/connector/ProcedureSuite.scala +++ b/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" + } }