This is an automated email from the ASF dual-hosted git repository.
Zouxxyy pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new b686cd7d84 [spark] Support SQL-defined UDFs for Spark 4 (#8299)
b686cd7d84 is described below
commit b686cd7d84e9b5783f2c91779e2e72970cb000fe
Author: Zouxxyy <[email protected]>
AuthorDate: Mon Jun 22 10:58:24 2026 +0800
[spark] Support SQL-defined UDFs for Spark 4 (#8299)
---
docs/docs/spark/sql-functions.md | 33 ++++-
.../apache/spark/sql/paimon/shims/Spark4Shim.scala | 12 ++
.../paimon/spark/sql/PaimonSQLFunctionTest.scala | 21 +++
.../paimon/spark/sql/PaimonSQLFunctionTest.scala | 21 +++
.../java/org/apache/paimon/spark/SparkCatalog.java | 18 +--
.../paimon/spark/catalog/SupportV1Function.java | 14 +-
.../org/apache/paimon/spark/SparkTypeUtils.java | 7 +
...Converter.scala => FileFunctionConverter.scala} | 33 ++---
.../functions/FunctionIdentifierConverter.scala | 37 +++++
.../spark/execution/PaimonFunctionExec.scala | 35 ++++-
.../catalog/PaimonV1FunctionRegistry.scala | 90 ++++++------
.../AbstractPaimonSparkSqlExtensionsParser.scala | 2 +
.../parser/extensions/PaimonFunctionLookup.scala | 98 +++++++++++++
.../extensions/RewritePaimonFunctionCommands.scala | 124 ++++++----------
.../apache/spark/sql/paimon/shims/SparkShim.scala | 17 +++
.../spark/sql/PaimonSQLFunctionTestBase.scala | 145 ++++++++++++++++++
.../apache/spark/sql/paimon/shims/Spark3Shim.scala | 14 ++
.../catalog/functions/SQLFunctionConverter.scala | 162 +++++++++++++++++++++
.../RewritePaimonSQLFunctionCommands.scala | 79 ++++++++++
.../apache/spark/sql/paimon/shims/Spark4Shim.scala | 12 ++
20 files changed, 796 insertions(+), 178 deletions(-)
diff --git a/docs/docs/spark/sql-functions.md b/docs/docs/spark/sql-functions.md
index 410f1c99a7..7e94c46a3f 100644
--- a/docs/docs/spark/sql-functions.md
+++ b/docs/docs/spark/sql-functions.md
@@ -96,7 +96,7 @@ SELECT sys.descriptor_to_string(content) FROM t WHERE id =
'1';
## User-defined Function
-Paimon Spark supports two types of user-defined functions: lambda functions
and file-based functions.
+Paimon Spark supports three types of user-defined functions: lambda functions,
file-based functions, and SQL functions.
This feature currently only supports the REST catalog.
@@ -152,3 +152,34 @@ DESCRIBE FUNCTION [EXTENDED] <mydb>.simple_udf;
-- Drop Function
DROP [TEMPORARY] FUNCTION <mydb>.simple_udf;
```
+
+### SQL Function
+
+Define reusable scalar functions with a pure SQL body. The definition is
persisted in the Paimon catalog.
+
+This feature requires Spark 4.0 or higher. Table functions (`RETURNS
TABLE(...)`) are not supported yet.
+
+**Example**
+
+```sql
+-- Create Function (expression body)
+CREATE FUNCTION area(width DOUBLE, height DOUBLE)
+RETURNS DOUBLE
+RETURN width * height;
+
+-- Create Function (query body)
+CREATE FUNCTION dept_total(d INT) RETURNS INT
+RETURN SELECT SUM(salary) FROM emp WHERE dept_id = d;
+
+-- Parameter with DEFAULT value
+CREATE FUNCTION addd(x INT, y INT DEFAULT 10) RETURNS INT RETURN x + y;
+
+-- Create or Replace / If Not Exists
+CREATE OR REPLACE FUNCTION inc(x INT) RETURNS INT RETURN x + 100;
+CREATE FUNCTION IF NOT EXISTS inc(x INT) RETURNS INT RETURN x + 1;
+
+-- Describe / Show / Drop Function
+DESCRIBE FUNCTION [EXTENDED] area;
+SHOW USER FUNCTIONS;
+DROP FUNCTION [IF EXISTS] area;
+```
diff --git
a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
b/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
index 0f7ea24e66..9e97adc96e 100644
---
a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
+++
b/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
@@ -341,6 +341,18 @@ class Spark4Shim extends SparkShim {
dataType.isInstanceOf[VariantType]
override def SparkVariantType(): org.apache.spark.sql.types.DataType =
DataTypes.VariantType
+
+ // SQL UDFs (CREATE FUNCTION ... RETURN ...).
+ override def rewritePaimonSQLFunctionCommands(spark: SparkSession):
Rule[LogicalPlan] =
+
org.apache.spark.sql.catalyst.parser.extensions.RewritePaimonSQLFunctionCommands(spark)
+
+ override def resolvePaimonSQLFunction(
+ funcIdent: org.apache.spark.sql.catalyst.FunctionIdentifier,
+ function: org.apache.paimon.function.Function,
+ arguments: Seq[Expression],
+ parser: org.apache.spark.sql.catalyst.parser.ParserInterface):
Expression =
+ org.apache.paimon.spark.catalog.functions.SQLFunctionConverter
+ .toSQLFunctionExpression(funcIdent, function, arguments, parser)
}
object Spark4Shim {
diff --git
a/paimon-spark/paimon-spark-4.0/src/test/scala/org/apache/paimon/spark/sql/PaimonSQLFunctionTest.scala
b/paimon-spark/paimon-spark-4.0/src/test/scala/org/apache/paimon/spark/sql/PaimonSQLFunctionTest.scala
new file mode 100644
index 0000000000..967795d439
--- /dev/null
+++
b/paimon-spark/paimon-spark-4.0/src/test/scala/org/apache/paimon/spark/sql/PaimonSQLFunctionTest.scala
@@ -0,0 +1,21 @@
+/*
+ * 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.paimon.spark.sql
+
+class PaimonSQLFunctionTest extends PaimonSQLFunctionTestBase {}
diff --git
a/paimon-spark/paimon-spark-4.1/src/test/scala/org/apache/paimon/spark/sql/PaimonSQLFunctionTest.scala
b/paimon-spark/paimon-spark-4.1/src/test/scala/org/apache/paimon/spark/sql/PaimonSQLFunctionTest.scala
new file mode 100644
index 0000000000..967795d439
--- /dev/null
+++
b/paimon-spark/paimon-spark-4.1/src/test/scala/org/apache/paimon/spark/sql/PaimonSQLFunctionTest.scala
@@ -0,0 +1,21 @@
+/*
+ * 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.paimon.spark.sql
+
+class PaimonSQLFunctionTest extends PaimonSQLFunctionTestBase {}
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java
index 165be98980..7868afdd1f 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java
@@ -34,8 +34,8 @@ import org.apache.paimon.spark.catalog.FormatTableCatalog;
import org.apache.paimon.spark.catalog.SparkBaseCatalog;
import org.apache.paimon.spark.catalog.SupportV1Function;
import org.apache.paimon.spark.catalog.SupportView;
+import org.apache.paimon.spark.catalog.functions.FunctionIdentifierConverter;
import org.apache.paimon.spark.catalog.functions.PaimonFunctions;
-import org.apache.paimon.spark.catalog.functions.V1FunctionConverter;
import org.apache.paimon.spark.utils.CatalogUtils;
import org.apache.paimon.table.FormatTable;
import org.apache.paimon.table.iceberg.IcebergTable;
@@ -56,7 +56,6 @@ import
org.apache.spark.sql.catalyst.analysis.NoSuchFunctionException;
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.catalog.CatalogFunction;
import org.apache.spark.sql.catalyst.catalog.PaimonV1FunctionRegistry;
import org.apache.spark.sql.catalyst.expressions.Expression;
import
org.apache.spark.sql.catalyst.parser.extensions.UnResolvedPaimonV1Function;
@@ -749,18 +748,13 @@ public class SparkCatalog extends SparkBaseCatalog
@Override
public Function getFunction(FunctionIdentifier funcIdent) throws Exception
{
- return
paimonCatalog().getFunction(V1FunctionConverter.fromFunctionIdentifier(funcIdent));
+ return paimonCatalog()
+
.getFunction(FunctionIdentifierConverter.toPaimonIdentifier(funcIdent));
}
@Override
- public void createV1Function(CatalogFunction v1Function, boolean
ignoreIfExists)
- throws Exception {
- Function paimonFunction =
V1FunctionConverter.fromV1Function(v1Function);
- paimonCatalog()
- .createFunction(
-
V1FunctionConverter.fromFunctionIdentifier(v1Function.identifier()),
- paimonFunction,
- ignoreIfExists);
+ public void createV1Function(Function function, boolean ignoreIfExists)
throws Exception {
+ paimonCatalog().createFunction(function.identifier(), function,
ignoreIfExists);
}
@Override
@@ -778,7 +772,7 @@ public class SparkCatalog extends SparkBaseCatalog
public void dropV1Function(FunctionIdentifier funcIdent, boolean ifExists)
throws Exception {
v1FunctionRegistry().unregisterFunction(funcIdent);
paimonCatalog()
-
.dropFunction(V1FunctionConverter.fromFunctionIdentifier(funcIdent), ifExists);
+
.dropFunction(FunctionIdentifierConverter.toPaimonIdentifier(funcIdent),
ifExists);
}
// ======================= Tools methods ===============================
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/SupportV1Function.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/SupportV1Function.java
index 4c070c5949..e1bce316b4 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/SupportV1Function.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/SupportV1Function.java
@@ -21,11 +21,13 @@ package org.apache.paimon.spark.catalog;
import org.apache.paimon.function.Function;
import org.apache.spark.sql.catalyst.FunctionIdentifier;
-import org.apache.spark.sql.catalyst.catalog.CatalogFunction;
import org.apache.spark.sql.catalyst.expressions.Expression;
import
org.apache.spark.sql.catalyst.parser.extensions.UnResolvedPaimonV1Function;
-/** Catalog supports v1 function. */
+/**
+ * Catalog supports v1 function, i.e. functions surfaced through Spark's v1
(session / persistent)
+ * function mechanism: file (Hive) functions and SQL functions.
+ */
public interface SupportV1Function extends WithPaimonCatalog {
boolean v1FunctionEnabled();
@@ -33,14 +35,12 @@ public interface SupportV1Function extends
WithPaimonCatalog {
/** Look up the function in the catalog. */
Function getFunction(FunctionIdentifier funcIdent) throws Exception;
- void createV1Function(CatalogFunction v1Function, boolean ignoreIfExists)
throws Exception;
+ /** Create a v1 function (file or SQL) from an already-built Paimon {@link
Function}. */
+ void createV1Function(Function function, boolean ignoreIfExists) throws
Exception;
boolean v1FunctionRegistered(FunctionIdentifier funcIdent);
- /**
- * Register the function and resolves it to an Expression if not
registered, otherwise returns
- * the registered Expression.
- */
+ /** Resolve a v1 function reference (file or SQL) to an Expression. */
Expression registerAndResolveV1Function(UnResolvedPaimonV1Function
unresolvedV1Function)
throws Exception;
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/SparkTypeUtils.java
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/SparkTypeUtils.java
index 80a27c35ac..cd7de8f535 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/SparkTypeUtils.java
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/SparkTypeUtils.java
@@ -70,6 +70,9 @@ public class SparkTypeUtils {
*/
public static final String CURRENT_DEFAULT_COLUMN_METADATA_KEY =
"CURRENT_DEFAULT";
+ // Spark SQL functions store DEFAULT in a different metadata key than
table columns.
+ public static final String SQL_FUNCTION_DEFAULT_METADATA_KEY = "default";
+
public static RowType toPartitionType(Table table) {
int[] projections =
table.rowType().getFieldIndices(table.partitionKeys());
List<DataField> partitionTypes = new ArrayList<>();
@@ -294,6 +297,7 @@ public class SparkTypeUtils {
MetadataBuilder metadataBuilder = new MetadataBuilder();
if (field.defaultValue() != null) {
metadataBuilder.putString(CURRENT_DEFAULT_COLUMN_METADATA_KEY,
field.defaultValue());
+
metadataBuilder.putString(SQL_FUNCTION_DEFAULT_METADATA_KEY,
field.defaultValue());
}
StructField structField =
DataTypes.createStructField(
@@ -380,6 +384,9 @@ public class SparkTypeUtils {
if
(field.metadata().contains(CURRENT_DEFAULT_COLUMN_METADATA_KEY)) {
defaultValue =
field.metadata().getString(CURRENT_DEFAULT_COLUMN_METADATA_KEY);
+ } else if
(field.metadata().contains(SQL_FUNCTION_DEFAULT_METADATA_KEY)) {
+ defaultValue =
+
field.metadata().getString(SQL_FUNCTION_DEFAULT_METADATA_KEY);
}
newFields.add(
new DataField(
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalog/functions/V1FunctionConverter.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalog/functions/FileFunctionConverter.scala
similarity index 71%
rename from
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalog/functions/V1FunctionConverter.scala
rename to
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalog/functions/FileFunctionConverter.scala
index be4daa86b9..705ec09137 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalog/functions/V1FunctionConverter.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalog/functions/FileFunctionConverter.scala
@@ -18,7 +18,6 @@
package org.apache.paimon.spark.catalog.functions
-import org.apache.paimon.catalog.Identifier
import org.apache.paimon.function.{Function, FunctionDefinition, FunctionImpl}
import org.apache.paimon.spark.SparkCatalog.FUNCTION_DEFINITION_NAME
@@ -27,38 +26,32 @@ import
org.apache.spark.sql.catalyst.catalog.{CatalogFunction, FunctionResource,
import scala.collection.JavaConverters._
-object V1FunctionConverter {
-
- /** Converts spark [[FunctionIdentifier]] to paimon [[Identifier]]. */
- def fromFunctionIdentifier(ident: FunctionIdentifier): Identifier = {
- new Identifier(ident.database.get, ident.funcName)
- }
-
- /** Converts paimon [[Identifier]] to spark [[FunctionIdentifier]]. */
- def toFunctionIdentifier(ident: Identifier): FunctionIdentifier = {
- new FunctionIdentifier(ident.getObjectName, Some(ident.getDatabaseName))
- }
+/**
+ * Converts between Spark's [[CatalogFunction]] (a file/Hive className UDF)
and a Paimon
+ * [[Function]] carrying a [[FunctionDefinition.FileFunctionDefinition]].
+ */
+object FileFunctionConverter {
- /** Converts spark [[CatalogFunction]] to paimon [[Function]]. */
- def fromV1Function(v1Function: CatalogFunction): Function = {
- val functionIdentifier = v1Function.identifier
- val identifier = fromFunctionIdentifier(functionIdentifier)
- val fileResources = v1Function.resources
+ /** Converts a Spark [[CatalogFunction]] to a Paimon [[Function]]. */
+ def fromCatalogFunction(catalogFunction: CatalogFunction): Function = {
+ val functionIdentifier = catalogFunction.identifier
+ val identifier =
FunctionIdentifierConverter.toPaimonIdentifier(functionIdentifier)
+ val fileResources = catalogFunction.resources
.map(r => new
FunctionDefinition.FunctionFileResource(r.resourceType.resourceType, r.uri))
.toList
val functionDefinition: FunctionDefinition = FunctionDefinition.file(
fileResources.asJava,
"JAVA", // Apache Spark only supports JAR persistent function now.
- v1Function.className,
+ catalogFunction.className,
functionIdentifier.funcName)
val definitions = Map(FUNCTION_DEFINITION_NAME ->
functionDefinition).asJava
new FunctionImpl(identifier, definitions)
}
- /** Converts paimon [[Function]] to spark [[CatalogFunction]]. */
- def toV1Function(paimonFunction: Function): CatalogFunction = {
+ /** Converts a Paimon [[Function]] to a Spark [[CatalogFunction]]. */
+ def toCatalogFunction(paimonFunction: Function): CatalogFunction = {
paimonFunction.definition(FUNCTION_DEFINITION_NAME) match {
case functionDefinition: FunctionDefinition.FileFunctionDefinition =>
val fileResources = functionDefinition
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalog/functions/FunctionIdentifierConverter.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalog/functions/FunctionIdentifierConverter.scala
new file mode 100644
index 0000000000..137ca02859
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalog/functions/FunctionIdentifierConverter.scala
@@ -0,0 +1,37 @@
+/*
+ * 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.paimon.spark.catalog.functions
+
+import org.apache.paimon.catalog.Identifier
+
+import org.apache.spark.sql.catalyst.FunctionIdentifier
+
+/**
+ * Converts a Spark [[FunctionIdentifier]] to a Paimon [[Identifier]] (shared
by file & SQL
+ * functions).
+ */
+object FunctionIdentifierConverter {
+
+ def toPaimonIdentifier(funcIdent: FunctionIdentifier): Identifier = {
+ require(
+ funcIdent.database.isDefined,
+ s"Function identifier $funcIdent must have a database/namespace.")
+ new Identifier(funcIdent.database.get, funcIdent.funcName)
+ }
+}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonFunctionExec.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonFunctionExec.scala
index fb0a2ec014..3ca46a43de 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonFunctionExec.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonFunctionExec.scala
@@ -18,37 +18,38 @@
package org.apache.paimon.spark.execution
-import org.apache.paimon.function.FunctionDefinition
+import org.apache.paimon.function.{Function => PaimonFunction,
FunctionDefinition}
import org.apache.paimon.spark.SparkCatalog.FUNCTION_DEFINITION_NAME
import org.apache.paimon.spark.catalog.SupportV1Function
import org.apache.paimon.spark.leafnode.PaimonLeafRunnableCommand
import org.apache.spark.sql.{Row, SparkSession}
import org.apache.spark.sql.catalyst.FunctionIdentifier
-import org.apache.spark.sql.catalyst.catalog.CatalogFunction
import org.apache.spark.sql.catalyst.expressions.{Attribute,
AttributeReference}
import org.apache.spark.sql.types.StringType
import scala.collection.JavaConverters._
import scala.collection.mutable.ArrayBuffer
+/** Create a Paimon v1 function (file or SQL) from an already-built Paimon
[[PaimonFunction]]. */
case class CreatePaimonV1FunctionCommand(
catalog: SupportV1Function,
- v1Function: CatalogFunction,
+ funcIdent: FunctionIdentifier,
+ function: PaimonFunction,
ignoreIfExists: Boolean,
replace: Boolean)
extends PaimonLeafRunnableCommand {
override def run(sparkSession: SparkSession): Seq[Row] = {
- // Note: for replace just drop then create ,this operation is non-atomic.
+ // replace = drop then create (non-atomic).
if (replace) {
- catalog.dropV1Function(v1Function.identifier, true)
+ catalog.dropV1Function(funcIdent, true)
}
- catalog.createV1Function(v1Function, ignoreIfExists)
+ catalog.createV1Function(function, ignoreIfExists)
Nil
}
override def simpleString(maxFields: Int): String = {
- s"CreatePaimonV1FunctionCommand: ${v1Function.identifier}"
+ s"CreatePaimonV1FunctionCommand: $funcIdent"
}
}
@@ -86,6 +87,26 @@ case class DescribePaimonV1FunctionCommand(
rows += Row(
s"File Resources:
${functionDefinition.fileResources().asScala.map(_.uri()).mkString(", ")}")
}
+ case sqlFunctionDefinition: FunctionDefinition.SQLFunctionDefinition =>
+ rows += Row(s"Function: ${function.fullName()}")
+ rows += Row("Type: SCALAR")
+ val inputParams = function.inputParams()
+ if (inputParams.isPresent && !inputParams.get().isEmpty) {
+ val params = inputParams
+ .get()
+ .asScala
+ .map(field => s"${field.name()} ${field.`type`().asSQLString()}")
+ .mkString(", ")
+ rows += Row(s"Input: $params")
+ }
+ val returnParams = function.returnParams()
+ if (returnParams.isPresent && !returnParams.get().isEmpty) {
+ rows += Row(s"Returns:
${returnParams.get().get(0).`type`().asSQLString()}")
+ }
+ if (isExtended) {
+ Option(function.comment()).foreach(c => rows += Row(s"Comment: $c"))
+ rows += Row(s"Body: ${sqlFunctionDefinition.definition()}")
+ }
case other =>
throw new UnsupportedOperationException(s"Unsupported function
definition $other")
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/catalog/PaimonV1FunctionRegistry.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/catalog/PaimonV1FunctionRegistry.scala
index b5a70c329e..ed69abd8bb 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/catalog/PaimonV1FunctionRegistry.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/catalog/PaimonV1FunctionRegistry.scala
@@ -19,7 +19,7 @@
package org.apache.spark.sql.catalyst.catalog
import org.apache.paimon.function.{Function => PaimonFunction}
-import org.apache.paimon.spark.catalog.functions.V1FunctionConverter
+import org.apache.paimon.spark.catalog.functions.FileFunctionConverter
import org.apache.spark.sql.{PaimonUtils, SparkSession}
import org.apache.spark.sql.catalyst.{FunctionIdentifier, SQLConfHelper}
@@ -27,7 +27,7 @@ import
org.apache.spark.sql.catalyst.analysis.{FunctionAlreadyExistsException, F
import org.apache.spark.sql.catalyst.analysis.FunctionRegistry.FunctionBuilder
import org.apache.spark.sql.catalyst.expressions.{AggregateWindowFunction,
Expression, ExpressionInfo, FrameLessOffsetWindowFunction, Lag, Lead, NthValue,
WindowExpression}
import org.apache.spark.sql.catalyst.expressions.aggregate._
-import
org.apache.spark.sql.catalyst.parser.extensions.UnResolvedPaimonV1Function
+import org.apache.spark.sql.catalyst.parser.extensions.{PaimonFunctionLookup,
UnResolvedPaimonV1Function}
import org.apache.spark.sql.errors.QueryCompilationErrors
import org.apache.spark.sql.hive.HiveUDFExpressionBuilder
import org.apache.spark.sql.paimon.shims.SparkShimLoader
@@ -37,26 +37,42 @@ import java.util.Locale
case class PaimonV1FunctionRegistry(session: SparkSession) extends
SQLConfHelper {
- // ================== Start Public API ===================
-
- /**
- * Register the function and resolves it to an Expression if not registered,
otherwise returns the
- * registered Expression.
- */
+ /** Resolve a v1 function (SQL or file) to an Expression. */
def registerAndResolveFunction(u: UnResolvedPaimonV1Function): Expression = {
- val resolvedFun = resolvePersistentFunctionInternal(
- u.funcIdent,
- u.func,
- u.arguments,
- functionRegistry,
- makeFunctionBuilder)
- validateFunction(resolvedFun, u.arguments.length, u)
+ val qualifiedIdent = qualifyIdentifier(u.funcIdent)
+ val sqlFunc = u.func
+ .filter(PaimonFunctionLookup.isPaimonSQLFunction)
+ .orElse(Option(sqlFunctionCache.get(qualifiedIdent)))
+ if (sqlFunc.isDefined) {
+ // SQL UDFs resolve into a Spark `SQLFunctionExpression` (Spark 4.0+
only, via the shim), which
+ // Spark's own `ResolveSQLFunctions` rule then inlines.
+ if (u.isDistinct || u.filter.isDefined || u.ignoreNulls) {
+ throw new UnsupportedOperationException(
+ s"SQL function ${u.funcIdent} does not support DISTINCT, FILTER or
IGNORE NULLS.")
+ }
+ val resolvedFun = SparkShimLoader.shim.resolvePaimonSQLFunction(
+ u.funcIdent,
+ sqlFunc.get,
+ u.arguments,
+ session.sessionState.sqlParser)
+ sqlFunctionCache.putIfAbsent(qualifiedIdent, sqlFunc.get)
+ resolvedFun
+ } else {
+ // File (Hive) function: register into the function registry and resolve
via the Hive builder.
+ val resolvedFun = resolvePersistentFunctionInternal(
+ u.funcIdent,
+ u.func,
+ u.arguments,
+ functionRegistry,
+ makeFunctionBuilder)
+ validateFunction(resolvedFun, u.arguments.length, u)
+ }
}
- /** Check if the function is registered. */
+ /** Check if the function is registered (file function in Hive registry or
SQL function in cache). */
def isRegistered(funcIdent: FunctionIdentifier): Boolean = {
val qualifiedIdent = qualifyIdentifier(funcIdent)
- functionRegistry.functionExists(qualifiedIdent)
+ functionRegistry.functionExists(qualifiedIdent) ||
sqlFunctionCache.containsKey(qualifiedIdent)
}
/** Unregister the function. */
@@ -65,17 +81,18 @@ case class PaimonV1FunctionRegistry(session: SparkSession)
extends SQLConfHelper
if (functionRegistry.functionExists(qualifiedIdent)) {
functionRegistry.dropFunction(qualifiedIdent)
}
+ sqlFunctionCache.remove(qualifiedIdent)
}
- // ================== End Public API ===================
-
- // Most copy from spark
private val functionResourceLoader: FunctionResourceLoader =
SparkShimLoader.shim.classicApi.sessionResourceLoader(session)
private val functionRegistry: FunctionRegistry = new SimpleFunctionRegistry
private val functionExpressionBuilder: FunctionExpressionBuilder =
HiveUDFExpressionBuilder
- /** Look up a persistent scalar function by name and resolves it to an
Expression. */
+ // SQL functions bypass the Hive registry; cache them here to avoid repeated
catalog IO.
+ private val sqlFunctionCache =
+ new java.util.concurrent.ConcurrentHashMap[FunctionIdentifier,
PaimonFunction]()
+
private def resolvePersistentFunctionInternal[T](
funcIdent: FunctionIdentifier,
func: Option[PaimonFunction],
@@ -83,44 +100,27 @@ case class PaimonV1FunctionRegistry(session: SparkSession)
extends SQLConfHelper
registry: FunctionRegistryBase[T],
createFunctionBuilder: CatalogFunction =>
FunctionRegistryBase[T]#FunctionBuilder): T = {
- val name = funcIdent
- // `synchronized` is used to prevent multiple threads from concurrently
resolving the
- // same function that has not yet been loaded into the function registry.
This is needed
- // because calling `registerFunction` twice with `overrideIfExists =
false` can lead to
- // a FunctionAlreadyExistsException.
+ // Synchronized: registerFunction with overrideIfExists=false throws on
concurrent duplicate loads.
synchronized {
- val qualifiedIdent = qualifyIdentifier(name)
+ val qualifiedIdent = qualifyIdentifier(funcIdent)
if (registry.functionExists(qualifiedIdent)) {
- // This function has been already loaded into the function registry.
registry.lookupFunction(qualifiedIdent, arguments)
} else {
- // The function has not been loaded to the function registry, which
means
- // that the function is a persistent function (if it actually has been
registered
- // in the metastore). We need to first put the function in the
function registry.
require(func.isDefined, "Function must be defined")
- val catalogFunction = V1FunctionConverter.toV1Function(func.get)
+ val catalogFunction = FileFunctionConverter.toCatalogFunction(func.get)
loadFunctionResources(catalogFunction.resources)
- // Please note that qualifiedName is provided by the user. However,
- // catalogFunction.identifier.unquotedString is returned by the
underlying
- // catalog. So, it is possible that qualifiedName is not exactly the
same as
- // catalogFunction.identifier.unquotedString (difference is on
case-sensitivity).
- // At here, we preserve the input from the user.
+ // Preserve user-provided identifier (case-sensitivity may differ from
catalog).
val funcMetadata = catalogFunction.copy(identifier = qualifiedIdent)
registerFunction(
funcMetadata,
overrideIfExists = false,
registry = registry,
functionBuilder = createFunctionBuilder(funcMetadata))
- // Now, we need to create the Expression.
registry.lookupFunction(qualifiedIdent, arguments)
}
}
}
- /**
- * Loads resources such as JARs and Files for a function. Every resource is
represented by a tuple
- * (resource type, resource uri).
- */
private def loadFunctionResources(resources: Seq[FunctionResource]): Unit = {
resources.foreach(functionResourceLoader.loadResource)
}
@@ -153,7 +153,6 @@ case class PaimonV1FunctionRegistry(session: SparkSession)
extends SQLConfHelper
"hive")
}
- /** Constructs a [[FunctionBuilder]] based on the provided function
metadata. */
private def makeFunctionBuilder(func: CatalogFunction): FunctionBuilder = {
val className = func.className
if (!PaimonUtils.classIsLoadable(className)) {
@@ -164,15 +163,10 @@ case class PaimonV1FunctionRegistry(session:
SparkSession) extends SQLConfHelper
(input) => functionExpressionBuilder.makeExpression(name, clazz, input)
}
- /**
- * Qualifies the function identifier with the current database if not
specified, and normalize all
- * the names.
- */
private def qualifyIdentifier(ident: FunctionIdentifier): FunctionIdentifier
= {
FunctionIdentifier(funcName = format(ident.funcName), database =
ident.database)
}
- /** Formats object names, taking into account case sensitivity. */
protected def format(name: String): String = {
if (conf.caseSensitiveAnalysis) name else name.toLowerCase(Locale.ROOT)
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/AbstractPaimonSparkSqlExtensionsParser.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/AbstractPaimonSparkSqlExtensionsParser.scala
index 4fd533f401..7108f0715e 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/AbstractPaimonSparkSqlExtensionsParser.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/AbstractPaimonSparkSqlExtensionsParser.scala
@@ -33,6 +33,7 @@ import
org.apache.spark.sql.catalyst.parser.extensions.PaimonSqlExtensionsParser
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.internal.VariableSubstitution
+import org.apache.spark.sql.paimon.shims.SparkShimLoader
import org.apache.spark.sql.types.{DataType, StructType}
import java.util.Locale
@@ -112,6 +113,7 @@ abstract class AbstractPaimonSparkSqlExtensionsParser(val
delegate: ParserInterf
Seq(
RewritePaimonViewCommands(sparkSession),
RewritePaimonFunctionCommands(sparkSession),
+ SparkShimLoader.shim.rewritePaimonSQLFunctionCommands(sparkSession),
RewriteCreateTableLikeCommand(sparkSession),
RewriteSparkDDLCommands(sparkSession)
)
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/PaimonFunctionLookup.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/PaimonFunctionLookup.scala
new file mode 100644
index 0000000000..4a4b179ff6
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/PaimonFunctionLookup.scala
@@ -0,0 +1,98 @@
+/*
+ * 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.catalyst.parser.extensions
+
+import org.apache.paimon.catalog.Catalog.SYSTEM_DATABASE_NAME
+import org.apache.paimon.function.{Function => PaimonFunction}
+import org.apache.paimon.function.FunctionDefinition
+import org.apache.paimon.spark.SparkCatalog.FUNCTION_DEFINITION_NAME
+import org.apache.paimon.spark.catalog.SupportV1Function
+import org.apache.paimon.spark.catalog.functions.PaimonFunctions
+
+import org.apache.spark.sql.catalyst.FunctionIdentifier
+import org.apache.spark.sql.catalyst.analysis.{UnresolvedFunctionName,
UnresolvedIdentifier}
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.connector.catalog.{CatalogManager, CatalogPlugin,
LookupCatalog}
+
+/** Resolves Paimon function identifiers in parser-stage plans. */
+case class PaimonFunctionLookup(catalogManager: CatalogManager) extends
LookupCatalog {
+
+ object CatalogAndFunctionIdentifier {
+
+ def unapply(unresolved: LogicalPlan): Option[(CatalogPlugin,
FunctionIdentifier, Boolean)] =
+ unresolved match {
+ case ui: UnresolvedIdentifier =>
+ unapply(ui.nameParts)
+ case name: UnresolvedFunctionName =>
+ unapply(name.multipartIdentifier)
+ case _ =>
+ None
+ }
+
+ def unapply(nameParts: Seq[String]): Option[(CatalogPlugin,
FunctionIdentifier, Boolean)] = {
+ nameParts match {
+ // Spark's built-in or tmp functions is without database name or
catalog name.
+ case Seq(funName) if
isSparkBuiltInFunction(FunctionIdentifier(funName)) =>
+ None
+ case Seq(funName) if isSparkTmpFunc(FunctionIdentifier(funName)) =>
+ Some(null, FunctionIdentifier(funName), true)
+ case CatalogAndIdentifier(v1FunctionCatalog: SupportV1Function, ident)
+ if v1FunctionCatalog.v1FunctionEnabled() =>
+ Some(
+ v1FunctionCatalog,
+ FunctionIdentifier(
+ ident.name(),
+ Some(ident.namespace().last),
+ Some(v1FunctionCatalog.name)),
+ false)
+ case _ =>
+ None
+ }
+ }
+ }
+
+ def isPaimonBuiltInFunction(funcIdent: FunctionIdentifier): Boolean = {
+ funcIdent.database match {
+ case Some(db)
+ if db == SYSTEM_DATABASE_NAME &&
PaimonFunctions.names.contains(funcIdent.funcName) =>
+ true
+ case _ => false
+ }
+ }
+
+ def isSparkBuiltInFunction(funcIdent: FunctionIdentifier): Boolean = {
+ catalogManager.v1SessionCatalog.isBuiltinFunction(funcIdent)
+ }
+
+ def isSparkTmpFunc(funcIdent: FunctionIdentifier): Boolean = {
+ catalogManager.v1SessionCatalog.isTemporaryFunction(funcIdent)
+ }
+}
+
+object PaimonFunctionLookup {
+
+ def isPaimonFileFunction(fun: PaimonFunction): Boolean =
+
fun.definition(FUNCTION_DEFINITION_NAME).isInstanceOf[FunctionDefinition.FileFunctionDefinition]
+
+ def isPaimonSQLFunction(fun: PaimonFunction): Boolean =
+
fun.definition(FUNCTION_DEFINITION_NAME).isInstanceOf[FunctionDefinition.SQLFunctionDefinition]
+
+ def isPaimonV1Function(fun: PaimonFunction): Boolean =
+ isPaimonFileFunction(fun) || isPaimonSQLFunction(fun)
+}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/RewritePaimonFunctionCommands.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/RewritePaimonFunctionCommands.scala
index 0f9722a633..6be680cd60 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/RewritePaimonFunctionCommands.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/RewritePaimonFunctionCommands.scala
@@ -18,41 +18,42 @@
package org.apache.spark.sql.catalyst.parser.extensions
-import org.apache.paimon.catalog.Catalog.SYSTEM_DATABASE_NAME
import org.apache.paimon.function.{Function => PaimonFunction}
-import org.apache.paimon.function.FunctionDefinition
-import org.apache.paimon.spark.SparkCatalog.FUNCTION_DEFINITION_NAME
import org.apache.paimon.spark.catalog.SupportV1Function
-import org.apache.paimon.spark.catalog.functions.PaimonFunctions
+import org.apache.paimon.spark.catalog.functions.FileFunctionConverter
import org.apache.paimon.spark.execution.{CreatePaimonV1FunctionCommand,
DescribePaimonV1FunctionCommand, DropPaimonV1FunctionCommand}
import org.apache.paimon.spark.util.OptionUtils
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.FunctionIdentifier
-import org.apache.spark.sql.catalyst.analysis.{UnresolvedException,
UnresolvedFunction, UnresolvedFunctionName, UnresolvedIdentifier}
+import org.apache.spark.sql.catalyst.analysis.{UnresolvedException,
UnresolvedFunction}
import org.apache.spark.sql.catalyst.catalog.CatalogFunction
import org.apache.spark.sql.catalyst.expressions.{Expression, Unevaluable}
import org.apache.spark.sql.catalyst.plans.logical.{CreateFunction,
DescribeFunction, DropFunction, LogicalPlan, SubqueryAlias, UnresolvedWith}
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.trees.TreePattern.{TreePattern,
UNRESOLVED_FUNCTION}
-import org.apache.spark.sql.connector.catalog.{CatalogManager, CatalogPlugin,
LookupCatalog}
+import org.apache.spark.sql.connector.catalog.CatalogManager
import org.apache.spark.sql.paimon.shims.SparkShimLoader
import org.apache.spark.sql.types.DataType
-case class RewritePaimonFunctionCommands(spark: SparkSession)
- extends Rule[LogicalPlan]
- with LookupCatalog {
+case class RewritePaimonFunctionCommands(spark: SparkSession) extends
Rule[LogicalPlan] {
protected lazy val catalogManager: CatalogManager =
spark.sessionState.catalogManager
+ private lazy val lookup = PaimonFunctionLookup(catalogManager)
+
override def apply(plan: LogicalPlan): LogicalPlan = {
- // Add a global switch to enable/disable v1 function.
if (!OptionUtils.v1FunctionEnabled()) {
return plan
}
val applied = plan.resolveOperatorsUp {
- case CreateFunction(CatalogAndFunctionIdentifier(_, funcIdent, true), _,
_, _, replace) =>
+ case CreateFunction(
+ lookup.CatalogAndFunctionIdentifier(_, funcIdent, true),
+ _,
+ _,
+ _,
+ replace) =>
if (replace) {
throw new UnsupportedOperationException(
s"$funcIdent is a temporary function, you should use `CREATE OR
REPLACE TEMPORARY FUNCTION $funcIdent` or DROP TEMPORARY FUNCTION $funcIdent`.")
@@ -62,41 +63,56 @@ case class RewritePaimonFunctionCommands(spark:
SparkSession)
}
case CreateFunction(
- CatalogAndFunctionIdentifier(v1FunctionCatalog: SupportV1Function,
funcIdent, false),
+ lookup.CatalogAndFunctionIdentifier(
+ v1FunctionCatalog: SupportV1Function,
+ funcIdent,
+ false),
className,
resources,
ifExists,
replace) =>
- if (isPaimonBuildInFunction(funcIdent)) {
+ if (lookup.isPaimonBuiltInFunction(funcIdent)) {
throw new UnsupportedOperationException(s"Can't create built-in
function: $funcIdent")
}
- val v1Function = CatalogFunction(funcIdent, className, resources)
- CreatePaimonV1FunctionCommand(v1FunctionCatalog, v1Function, ifExists,
replace)
+ val paimonFunction =
+ FileFunctionConverter.fromCatalogFunction(
+ CatalogFunction(funcIdent, className, resources))
+ CreatePaimonV1FunctionCommand(
+ v1FunctionCatalog,
+ funcIdent,
+ paimonFunction,
+ ifExists,
+ replace)
case DropFunction(
- CatalogAndFunctionIdentifier(v1FunctionCatalog: SupportV1Function,
funcIdent, false),
+ lookup.CatalogAndFunctionIdentifier(
+ v1FunctionCatalog: SupportV1Function,
+ funcIdent,
+ false),
ifExists) =>
- if (isPaimonBuildInFunction(funcIdent)) {
+ if (lookup.isPaimonBuiltInFunction(funcIdent)) {
throw new UnsupportedOperationException(s"Can't drop built-in
function: $funcIdent")
}
// The function may be v1 function or not, anyway it can be safely
deleted here.
DropPaimonV1FunctionCommand(v1FunctionCatalog, funcIdent, ifExists)
case d @ DescribeFunction(
- CatalogAndFunctionIdentifier(v1FunctionCatalog: SupportV1Function,
funcIdent, false),
+ lookup.CatalogAndFunctionIdentifier(
+ v1FunctionCatalog: SupportV1Function,
+ funcIdent,
+ false),
isExtended)
// For Paimon built-in functions, Spark will resolve them by itself.
- if !isPaimonBuildInFunction(funcIdent) =>
+ if !lookup.isPaimonBuiltInFunction(funcIdent) =>
val function = v1FunctionCatalog.getFunction(funcIdent)
- if (isPaimonV1Function(function)) {
+ if (PaimonFunctionLookup.isPaimonV1Function(function)) {
DescribePaimonV1FunctionCommand(function, isExtended)
} else {
d
}
}
- // Needs to be done here and transform to `UnResolvedPaimonV1Function`, so
that spark's Analyzer can resolve
- // the 'arguments' without throwing an exception, saying that function is
not supported.
+ // Transform function references to UnResolvedPaimonV1Function so Spark
can resolve arguments.
transformPaimonV1Function(applied)
}
@@ -109,16 +125,16 @@ case class RewritePaimonFunctionCommands(spark:
SparkSession)
case l: LogicalPlan =>
l.transformExpressionsWithPruning(_.containsAnyPattern(UNRESOLVED_FUNCTION)) {
case u: UnresolvedFunction =>
- CatalogAndFunctionIdentifier.unapply(u.nameParts) match {
+ lookup.CatalogAndFunctionIdentifier.unapply(u.nameParts) match {
case Some((v1FunctionCatalog: SupportV1Function, funcIdent,
false))
// For Paimon built-in functions, Spark will resolve them by
itself.
- if !isPaimonBuildInFunction(funcIdent) =>
+ if !lookup.isPaimonBuiltInFunction(funcIdent) =>
// If the function is already registered, avoid redundant
lookup in the catalog to reduce overhead.
if (v1FunctionCatalog.v1FunctionRegistered(funcIdent)) {
UnResolvedPaimonV1Function(funcIdent, u, None)
} else {
val function = v1FunctionCatalog.getFunction(funcIdent)
- if (isPaimonV1Function(function)) {
+ if (PaimonFunctionLookup.isPaimonV1Function(function)) {
UnResolvedPaimonV1Function(funcIdent, u, Some(function))
} else {
u
@@ -129,64 +145,6 @@ case class RewritePaimonFunctionCommands(spark:
SparkSession)
}
}
}
-
- private object CatalogAndFunctionIdentifier {
-
- def unapply(unresolved: LogicalPlan): Option[(CatalogPlugin,
FunctionIdentifier, Boolean)] =
- unresolved match {
- case ui: UnresolvedIdentifier =>
- unapply(ui.nameParts)
- case name: UnresolvedFunctionName =>
- unapply(name.multipartIdentifier)
- case _ =>
- None
- }
-
- def unapply(nameParts: Seq[String]): Option[(CatalogPlugin,
FunctionIdentifier, Boolean)] = {
- nameParts match {
- // Spark's built-in or tmp functions is without database name or
catalog name.
- case Seq(funName) if
isSparkBuiltInFunction(FunctionIdentifier(funName)) =>
- None
- case Seq(funName) if isSparkTmpFunc(FunctionIdentifier(funName)) =>
- Some(null, FunctionIdentifier(funName), true)
- case CatalogAndIdentifier(v1FunctionCatalog: SupportV1Function, ident)
- if v1FunctionCatalog.v1FunctionEnabled() =>
- Some(
- v1FunctionCatalog,
- FunctionIdentifier(
- ident.name(),
- Some(ident.namespace().last),
- Some(v1FunctionCatalog.name)),
- false)
- case _ =>
- None
- }
- }
- }
-
- private def isPaimonBuildInFunction(funcIdent: FunctionIdentifier): Boolean
= {
- funcIdent.database match {
- case Some(db)
- if db == SYSTEM_DATABASE_NAME &&
PaimonFunctions.names.contains(funcIdent.funcName) =>
- true
- case _ => false
- }
- }
-
- private def isSparkBuiltInFunction(funcIdent: FunctionIdentifier): Boolean =
{
- catalogManager.v1SessionCatalog.isBuiltinFunction(funcIdent)
- }
-
- private def isSparkTmpFunc(funcIdent: FunctionIdentifier): Boolean = {
- catalogManager.v1SessionCatalog.isTemporaryFunction(funcIdent)
- }
-
- private def isPaimonV1Function(fun: PaimonFunction): Boolean = {
- fun.definition(FUNCTION_DEFINITION_NAME) match {
- case _: FunctionDefinition.FileFunctionDefinition => true
- case _ => false
- }
- }
}
/** An unresolved Paimon V1 function to let Spark resolve the necessary
variables. */
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/paimon/shims/SparkShim.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/paimon/shims/SparkShim.scala
index 7883903b30..5c8094682f 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/paimon/shims/SparkShim.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/paimon/shims/SparkShim.scala
@@ -19,12 +19,14 @@
package org.apache.spark.sql.paimon.shims
import org.apache.paimon.data.variant.Variant
+import org.apache.paimon.function.{Function => PaimonFunction}
import org.apache.paimon.spark.data.{SparkArrayData, SparkInternalRow}
import org.apache.paimon.spark.rowops.PaimonCopyOnWriteScan
import org.apache.paimon.table.{FileStoreTable, FormatTable}
import org.apache.paimon.types.{DataType, RowType}
import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.FunctionIdentifier
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression}
import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression
@@ -238,4 +240,19 @@ trait SparkShim {
def isSparkVariantType(dataType: org.apache.spark.sql.types.DataType):
Boolean
def SparkVariantType(): org.apache.spark.sql.types.DataType
+
+ // SQL UDFs (`CREATE FUNCTION ... RETURN ...`) are Spark 4.0+; the spark3
shim no-ops these.
+
+ /** Parser-stage rule rewriting a Paimon-catalog `CreateUserDefinedFunction`
into a create command. */
+ def rewritePaimonSQLFunctionCommands(spark: SparkSession): Rule[LogicalPlan]
+
+ /**
+ * Resolve a Paimon SQL function reference into a Spark
`SQLFunctionExpression` (Spark inlines
+ * it).
+ */
+ def resolvePaimonSQLFunction(
+ funcIdent: FunctionIdentifier,
+ function: PaimonFunction,
+ arguments: Seq[Expression],
+ parser: ParserInterface): Expression
}
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PaimonSQLFunctionTestBase.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PaimonSQLFunctionTestBase.scala
new file mode 100644
index 0000000000..230ec6f3dc
--- /dev/null
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PaimonSQLFunctionTestBase.scala
@@ -0,0 +1,145 @@
+/*
+ * 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.paimon.spark.sql
+
+import org.apache.paimon.spark.PaimonSparkTestWithRestCatalogBase
+
+import org.apache.spark.sql.Row
+
+abstract class PaimonSQLFunctionTestBase extends
PaimonSparkTestWithRestCatalogBase {
+
+ test("Paimon SQL Function: scalar function with expression body") {
+ withUserDefinedFunction("area" -> false) {
+ sql("""
+ |CREATE FUNCTION area(width DOUBLE, height DOUBLE)
+ |RETURNS DOUBLE
+ |RETURN width * height
+ |""".stripMargin)
+ checkAnswer(sql("SELECT area(2, 3)"), Row(6.0))
+
+ withTable("t") {
+ sql("CREATE TABLE t (a DOUBLE, b DOUBLE)")
+ sql("INSERT INTO t VALUES (1, 2), (3, 4)")
+ checkAnswer(sql("SELECT area(a, b) FROM t ORDER BY a"), Seq(Row(2.0),
Row(12.0)))
+ }
+ }
+ }
+
+ test("Paimon SQL Function: create with qualified name and call across
catalog") {
+ withUserDefinedFunction("area" -> false) {
+ sql("""
+ |CREATE FUNCTION test.area(width DOUBLE, height DOUBLE)
+ |RETURNS DOUBLE
+ |RETURN width * height
+ |""".stripMargin)
+ checkAnswer(sql("SELECT test.area(2, 3)"), Row(6.0))
+ checkAnswer(sql("SELECT paimon.test.area(2, 4)"), Row(8.0))
+ }
+ }
+
+ test("Paimon SQL Function: create or replace / if not exists") {
+ withUserDefinedFunction("inc" -> false) {
+ sql("CREATE FUNCTION inc(x INT) RETURNS INT RETURN x + 1")
+ checkAnswer(sql("SELECT inc(10)"), Row(11))
+
+ // create again should fail
+ intercept[Exception] {
+ sql("CREATE FUNCTION inc(x INT) RETURNS INT RETURN x + 1")
+ }
+
+ // if not exists: no-op, keeps the old definition
+ sql("CREATE FUNCTION IF NOT EXISTS inc(x INT) RETURNS INT RETURN x +
100")
+ checkAnswer(sql("SELECT inc(10)"), Row(11))
+
+ // or replace: new definition takes effect
+ sql("CREATE OR REPLACE FUNCTION inc(x INT) RETURNS INT RETURN x + 100")
+ checkAnswer(sql("SELECT inc(10)"), Row(110))
+ }
+ }
+
+ test("Paimon SQL Function: scalar function with query body referencing a
table") {
+ withUserDefinedFunction("dept_total" -> false) {
+ withTable("emp") {
+ sql("CREATE TABLE emp (id INT, dept_id INT, salary INT)")
+ sql("INSERT INTO emp VALUES (1, 10, 100), (2, 10, 200), (3, 20, 300)")
+ sql("""
+ |CREATE FUNCTION dept_total(d INT) RETURNS INT
+ |RETURN SELECT SUM(salary) FROM emp WHERE dept_id = d
+ |""".stripMargin)
+ checkAnswer(sql("SELECT dept_total(10)"), Row(300))
+ checkAnswer(sql("SELECT dept_total(20)"), Row(300))
+ }
+ }
+ }
+
+ test("Paimon SQL Function: parameter with DEFAULT value") {
+ withUserDefinedFunction("addd" -> false) {
+ sql("CREATE FUNCTION addd(x INT, y INT DEFAULT 10) RETURNS INT RETURN x
+ y")
+ checkAnswer(sql("SELECT addd(5)"), Row(15))
+ checkAnswer(sql("SELECT addd(5, 1)"), Row(6))
+ }
+ }
+
+ test("Paimon SQL Function: describe function") {
+ withUserDefinedFunction("area" -> false) {
+ sql("CREATE FUNCTION area(width DOUBLE, height DOUBLE) RETURNS DOUBLE
RETURN width * height")
+
+ val desc = sql("DESCRIBE FUNCTION area").collect().map(_.getString(0))
+ assert(desc.exists(_.contains("Type: SCALAR")), desc.mkString("\n"))
+ assert(desc.exists(_.contains("Input:")), desc.mkString("\n"))
+ assert(desc.exists(_.contains("width")), desc.mkString("\n"))
+ assert(desc.exists(_.contains("Returns: DOUBLE")), desc.mkString("\n"))
+
+ val descExt = sql("DESCRIBE FUNCTION EXTENDED
area").collect().map(_.getString(0))
+ assert(descExt.exists(_.contains("width * height")),
descExt.mkString("\n"))
+ }
+ }
+
+ test("Paimon SQL Function: show functions lists the created function") {
+ withUserDefinedFunction("area" -> false) {
+ sql("CREATE FUNCTION area(w DOUBLE, h DOUBLE) RETURNS DOUBLE RETURN w *
h")
+ val names = sql("SHOW USER
FUNCTIONS").collect().map(_.getString(0).toLowerCase)
+ assert(names.exists(_.contains("area")), names.mkString(", "))
+ }
+ }
+
+ test("Paimon SQL Function: drop function") {
+ withUserDefinedFunction("area" -> false) {
+ sql("CREATE FUNCTION area(w DOUBLE, h DOUBLE) RETURNS DOUBLE RETURN w *
h")
+ checkAnswer(sql("SELECT area(2, 3)"), Row(6.0))
+
+ sql("DROP FUNCTION area")
+ intercept[Exception] {
+ sql("SELECT area(2, 3)")
+ }
+ sql("DROP FUNCTION IF EXISTS area")
+ }
+ }
+
+ test("Paimon SQL Function: table function is not supported yet") {
+ val e = intercept[Exception] {
+ sql("""
+ |CREATE FUNCTION rows_of(x INT)
+ |RETURNS TABLE(a INT)
+ |RETURN SELECT x + 1 AS a
+ |""".stripMargin)
+ }
+ assert(e.getMessage.contains("does not support creating SQL table
functions"))
+ }
+}
diff --git
a/paimon-spark/paimon-spark3-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark3Shim.scala
b/paimon-spark/paimon-spark3-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark3Shim.scala
index cbefee5d04..362e2bebf0 100644
---
a/paimon-spark/paimon-spark3-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark3Shim.scala
+++
b/paimon-spark/paimon-spark3-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark3Shim.scala
@@ -324,6 +324,20 @@ class Spark3Shim extends SparkShim {
override def toPaimonVariant(array: ArrayData, pos: Int): Variant =
throw new UnsupportedOperationException()
+
+ // SQL UDFs (CREATE FUNCTION ... RETURN ...) are a Spark 4.0+ feature; no-op
rule on Spark 3.
+ override def rewritePaimonSQLFunctionCommands(spark: SparkSession):
Rule[LogicalPlan] =
+ new Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan = plan
+ }
+
+ override def resolvePaimonSQLFunction(
+ funcIdent: org.apache.spark.sql.catalyst.FunctionIdentifier,
+ function: org.apache.paimon.function.Function,
+ arguments: Seq[Expression],
+ parser: org.apache.spark.sql.catalyst.parser.ParserInterface):
Expression =
+ throw new UnsupportedOperationException(
+ "SQL user-defined functions (CREATE FUNCTION ... RETURN) require Spark
4.0 or later.")
}
object Spark3Shim {
diff --git
a/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/paimon/spark/catalog/functions/SQLFunctionConverter.scala
b/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/paimon/spark/catalog/functions/SQLFunctionConverter.scala
new file mode 100644
index 0000000000..a81baff8f3
--- /dev/null
+++
b/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/paimon/spark/catalog/functions/SQLFunctionConverter.scala
@@ -0,0 +1,162 @@
+/*
+ * 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.paimon.spark.catalog.functions
+
+import org.apache.paimon.function.{Function => PaimonFunction,
FunctionDefinition, FunctionImpl}
+import org.apache.paimon.spark.SparkCatalog.FUNCTION_DEFINITION_NAME
+import org.apache.paimon.spark.SparkTypeUtils
+import org.apache.paimon.types.{DataField, RowType}
+
+import org.apache.spark.sql.catalyst.FunctionIdentifier
+import org.apache.spark.sql.catalyst.analysis.SQLFunctionExpression
+import org.apache.spark.sql.catalyst.catalog.{SQLFunction, UserDefinedFunction}
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.catalyst.parser.ParserInterface
+import org.apache.spark.sql.types.{DataType => SparkDataType, StructType}
+
+import java.util.{Collections, HashMap => JHashMap, List => JList}
+
+/** Converts between Spark SQLFunction and Paimon Function with a
SQLFunctionDefinition body. */
+object SQLFunctionConverter {
+
+ // Spark-specific metadata stored in Paimon Function.options().
+ private val IS_QUERY = "spark.sql-function.is-query"
+ private val DETERMINISTIC = "spark.sql-function.deterministic"
+ private val CONTAINS_SQL = "spark.sql-function.contains-sql"
+
+ /** Build a Paimon function from a parsed CREATE FUNCTION ... RETURN
statement. */
+ def toPaimonFunction(
+ funcIdent: FunctionIdentifier,
+ inputParamText: Option[String],
+ returnTypeText: String,
+ exprText: Option[String],
+ queryText: Option[String],
+ comment: Option[String],
+ isDeterministic: Option[Boolean],
+ containsSQL: Option[Boolean],
+ parser: ParserInterface): PaimonFunction = {
+ require(
+ returnTypeText != null && returnTypeText.trim.nonEmpty,
+ s"SQL function $funcIdent must declare an explicit RETURNS type.")
+ val identifier = FunctionIdentifierConverter.toPaimonIdentifier(funcIdent)
+
+ val inputParams: JList[DataField] = inputParamText.filter(_.trim.nonEmpty)
match {
+ case Some(text) =>
+ SparkTypeUtils
+ .toPaimonRowType(UserDefinedFunction.parseRoutineParam(text, parser))
+ .getFields
+ case None => Collections.emptyList[DataField]()
+ }
+ val returnSparkType = parseScalarReturnType(funcIdent, returnTypeText,
parser)
+ val returnParams: JList[DataField] =
+ Collections.singletonList(
+ new DataField(0, funcIdent.funcName,
SparkTypeUtils.toPaimonType(returnSparkType)))
+
+ // Exactly one of exprText / queryText is set by the parser.
+ val isQuery = exprText.isEmpty && queryText.isDefined
+ val body = exprText
+ .orElse(queryText)
+ .getOrElse(throw new IllegalArgumentException(s"SQL function $funcIdent
has an empty body."))
+
+ val options = new JHashMap[String, String]()
+ options.put(IS_QUERY, isQuery.toString)
+ isDeterministic.foreach(d => options.put(DETERMINISTIC, d.toString))
+ containsSQL.foreach(c => options.put(CONTAINS_SQL, c.toString))
+
+ new FunctionImpl(
+ identifier,
+ inputParams,
+ returnParams,
+ isDeterministic.getOrElse(true),
+ Collections.singletonMap(FUNCTION_DEFINITION_NAME,
FunctionDefinition.sql(body)),
+ comment.orNull,
+ options
+ )
+ }
+
+ /** Resolve a Paimon-stored SQL function into a Spark SQLFunctionExpression.
*/
+ def toSQLFunctionExpression(
+ funcIdent: FunctionIdentifier,
+ function: PaimonFunction,
+ arguments: Seq[Expression],
+ parser: ParserInterface): Expression = {
+ val options = function.options()
+
+ val body = function.definition(FUNCTION_DEFINITION_NAME) match {
+ case sql: FunctionDefinition.SQLFunctionDefinition => sql.definition()
+ case other =>
+ throw new IllegalStateException(
+ s"Function $funcIdent is not a SQL function, found definition:
$other")
+ }
+
+ val inputParam: Option[StructType] = {
+ val ip = function.inputParams()
+ if (ip.isPresent && !ip.get().isEmpty) {
+ Some(SparkTypeUtils.fromPaimonType(new
RowType(ip.get())).asInstanceOf[StructType])
+ } else None
+ }
+
+ val rp = function.returnParams()
+ require(
+ rp.isPresent && !rp.get().isEmpty,
+ s"SQL function $funcIdent has no return type in returnParams.")
+ val returnType: SparkDataType =
SparkTypeUtils.fromPaimonType(rp.get().get(0).`type`())
+
+ val isQuery = Option(options.get(IS_QUERY))
+ .map(java.lang.Boolean.parseBoolean)
+ .getOrElse {
+ try { parser.parseExpression(body); false }
+ catch { case _: Exception => true }
+ }
+
+ val deterministic = Option(options.get(DETERMINISTIC))
+ .map(_.toBoolean)
+ .orElse(Some(function.isDeterministic))
+
+ val sqlFunction = SQLFunction(
+ name = funcIdent,
+ inputParam = inputParam,
+ returnType = Left(returnType),
+ exprText = if (isQuery) None else Some(body),
+ queryText = if (isQuery) Some(body) else None,
+ comment = Option(function.comment()),
+ deterministic = deterministic,
+ containsSQL = Option(options.get(CONTAINS_SQL)).map(_.toBoolean),
+ isTableFunc = false,
+ properties = Map.empty
+ )
+
+ SQLFunctionExpression(
+ sqlFunction.name.unquotedString,
+ sqlFunction,
+ arguments,
+ Some(sqlFunction.getScalarFuncReturnType))
+ }
+
+ private def parseScalarReturnType(
+ funcIdent: FunctionIdentifier,
+ returnTypeText: String,
+ parser: ParserInterface): SparkDataType =
+ SQLFunction.parseReturnTypeText(returnTypeText, isTableFunc = false,
parser) match {
+ case Some(Left(dataType)) => dataType
+ case _ =>
+ throw new UnsupportedOperationException(
+ s"Unsupported return type '$returnTypeText' for scalar SQL function
$funcIdent.")
+ }
+}
diff --git
a/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/RewritePaimonSQLFunctionCommands.scala
b/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/RewritePaimonSQLFunctionCommands.scala
new file mode 100644
index 0000000000..cc2c4df6a6
--- /dev/null
+++
b/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/RewritePaimonSQLFunctionCommands.scala
@@ -0,0 +1,79 @@
+/*
+ * 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.catalyst.parser.extensions
+
+import org.apache.paimon.spark.catalog.SupportV1Function
+import org.apache.paimon.spark.catalog.functions.SQLFunctionConverter
+import org.apache.paimon.spark.execution.CreatePaimonV1FunctionCommand
+import org.apache.paimon.spark.util.OptionUtils
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.plans.logical.{CreateUserDefinedFunction,
LogicalPlan}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.connector.catalog.CatalogManager
+
+/**
+ * Parser-stage rule that rewrites a Paimon-catalog `CREATE FUNCTION ...
RETURN ...`
+ * (`CreateUserDefinedFunction`) into [[CreatePaimonV1FunctionCommand]],
before Spark's
+ * `ResolveSessionCatalog` throws `MISSING_CATALOG_ABILITY.CREATE_FUNCTION`.
Fields are read by name
+ * (not positional unapply) since `CreateUserDefinedFunction`'s arity differs
across Spark 4.0/4.1.
+ */
+case class RewritePaimonSQLFunctionCommands(spark: SparkSession) extends
Rule[LogicalPlan] {
+
+ private lazy val catalogManager: CatalogManager =
spark.sessionState.catalogManager
+
+ private lazy val lookup = PaimonFunctionLookup(catalogManager)
+
+ override def apply(plan: LogicalPlan): LogicalPlan = {
+ if (!OptionUtils.v1FunctionEnabled()) {
+ return plan
+ }
+
+ plan.resolveOperatorsUp {
+ case c: CreateUserDefinedFunction =>
+ lookup.CatalogAndFunctionIdentifier.unapply(c.child) match {
+ case Some((catalog: SupportV1Function, funcIdent, false)) =>
+ if (lookup.isPaimonBuiltInFunction(funcIdent)) {
+ throw new UnsupportedOperationException(s"Can't create built-in
function: $funcIdent")
+ }
+ if (c.isTableFunc) {
+ throw new UnsupportedOperationException(
+ s"Paimon does not support creating SQL table functions yet:
$funcIdent")
+ }
+ val paimonFunction = SQLFunctionConverter.toPaimonFunction(
+ funcIdent,
+ c.inputParamText,
+ c.returnTypeText,
+ c.exprText,
+ c.queryText,
+ c.comment,
+ c.isDeterministic,
+ c.containsSQL,
+ spark.sessionState.sqlParser)
+ CreatePaimonV1FunctionCommand(
+ catalog,
+ funcIdent,
+ paimonFunction,
+ c.ignoreIfExists,
+ c.replace)
+ case _ => c
+ }
+ }
+ }
+}
diff --git
a/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
b/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
index 9c4a4daa6a..3a49c223b6 100644
---
a/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
+++
b/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
@@ -319,6 +319,18 @@ class Spark4Shim extends SparkShim {
dataType.isInstanceOf[VariantType]
override def SparkVariantType(): org.apache.spark.sql.types.DataType =
DataTypes.VariantType
+
+ // SQL UDFs (CREATE FUNCTION ... RETURN ...).
+ override def rewritePaimonSQLFunctionCommands(spark: SparkSession):
Rule[LogicalPlan] =
+
org.apache.spark.sql.catalyst.parser.extensions.RewritePaimonSQLFunctionCommands(spark)
+
+ override def resolvePaimonSQLFunction(
+ funcIdent: org.apache.spark.sql.catalyst.FunctionIdentifier,
+ function: org.apache.paimon.function.Function,
+ arguments: Seq[Expression],
+ parser: org.apache.spark.sql.catalyst.parser.ParserInterface):
Expression =
+ org.apache.paimon.spark.catalog.functions.SQLFunctionConverter
+ .toSQLFunctionExpression(funcIdent, function, arguments, parser)
}
object Spark4Shim {
