[spark] Add v1 function support to SparkCatalog (#6075)

Author: Zouxxyy

.gitignore

@@ -12,6 +12,7 @@ target
 *.iml
 *.swp
 *.jar
+!**/resources/**/*.jar
 *.log
 *.pyc
 *.ipr

docs/content/concepts/functions.md

@@ -84,37 +84,6 @@ DROP FUNCTION mydb.parse_str;
 
 This statement deletes the existing `parse_str` function from the `mydb` database, relinquishing its functionality.
 
-## Lambda Function Usage in Spark
+## Functions in Spark
 
-### Create Function
-
-```sql
--- Spark SQL
-CALL sys.create_function(`function` => 'my_db.area_func',
-  `inputParams` => '[{"id": 0, "name":"length", "type":"INT"}, {"id": 1, "name":"width", "type":"INT"}]',
-  `returnParams` => '[{"id": 0, "name":"area", "type":"BIGINT"}]',
-  `deterministic` => true,
-  `comment` => 'comment',
-  `options` => 'k1=v1,k2=v2'
-);
-```
-
-### Alter Function
-
-```sql
--- Spark SQL
-CALL sys.alter_function(`function` => 'my_db.area_func',
-  `change` => '{"action" : "addDefinition", "name" : "spark", "definition" : {"type" : "lambda", "definition" : "(Integer length, Integer width) -> { return (long) length * width; }", "language": "JAVA" } }'
-);
-```
-```sql
--- Spark SQL
-select paimon.my_db.area_func(1, 2);
-```
-
-### Drop Function
-
-```sql
--- Spark SQL
-CALL sys.drop_function(`function` => 'my_db.area_func');
-```
+see [SQL Functions]({{< ref "spark/sql-functions#user-defined-function" >}})

docs/content/spark/sql-functions.md

@@ -29,7 +29,9 @@ under the License.
 This section introduce all available Paimon Spark functions.
 
 
-## max_pt
+## Built-in Function
+
+### max_pt
 
 `max_pt($table_name)`
 
@@ -44,12 +46,69 @@ It would throw exception when:
 
 **Example**
 
-```shell
-> SELECT max_pt('t');
- 20250101
+```sql
+SELECT max_pt('t');
+-- 20250101
 
-> SELECT * FROM t where pt = max_pt('t');
- a, 20250101
+SELECT * FROM t where pt = max_pt('t');
+-- a, 20250101
 ```
 
 **Since: 1.1.0**
+
+## User-defined Function
+
+Paimon Spark supports two types of user-defined functions: lambda functions and file-based functions.
+
+This feature currently only supports the REST catalog.
+
+### Lambda Function
+
+Empowering users to define functions using Java lambda expressions, enabling inline, concise, and functional-style operations.
+
+**Example**
+
+```sql
+-- Create Function
+CALL sys.create_function(`function` => 'my_db.area_func',
+  `inputParams` => '[{"id": 0, "name":"length", "type":"INT"}, {"id": 1, "name":"width", "type":"INT"}]',
+  `returnParams` => '[{"id": 0, "name":"area", "type":"BIGINT"}]',
+  `deterministic` => true,
+  `comment` => 'comment',
+  `options` => 'k1=v1,k2=v2'
+);
+
+-- Alter Function
+CALL sys.alter_function(`function` => 'my_db.area_func',
+  `change` => '{"action" : "addDefinition", "name" : "spark", "definition" : {"type" : "lambda", "definition" : "(Integer length, Integer width) -> { return (long) length * width; }", "language": "JAVA" } }'
+);
+
+-- Drop Function
+CALL sys.drop_function(`function` => 'my_db.area_func');
+```
+
+### File Function
+
+Users can define functions within a file, providing flexibility and modular support for function definition, only supports jar files now.
+
+This feature requires Spark 3.4 or higher.
+
+**Example**
+
+```sql
+-- Create Function
+CREATE FUNCTION mydb.simple_udf
+AS 'com.example.SimpleUdf' 
+USING JAR '/tmp/SimpleUdf.jar' [, JAR '/tmp/SimpleUdfR.jar'];
+
+-- Create or Replace Function
+CREATE OR REPLACE FUNCTION mydb.simple_udf 
+AS 'com.example.SimpleUdf'
+USING JAR '/tmp/SimpleUdf.jar';
+       
+-- Describe Function
+DESCRIBE FUNCTION [EXTENDED] mydb.simple_udf;
+
+-- Drop Function
+DROP FUNCTION mydb.simple_udf;
+```

docs/layouts/shortcodes/generated/spark_catalog_configuration.html

@@ -38,5 +38,11 @@
             <td>String</td>
             <td>The default database name.</td>
         </tr>
+        <tr>
+            <td><h5>v1Function.enabled</h5></td>
+            <td style="word-wrap: break-word;">true</td>
+            <td>Boolean</td>
+            <td>Whether to enable v1 function.</td>
+        </tr>
     </tbody>
 </table>

paimon-api/src/main/java/org/apache/paimon/function/Function.java

@@ -18,6 +18,7 @@
 
 package org.apache.paimon.function;
 
+import org.apache.paimon.catalog.Identifier;
 import org.apache.paimon.types.DataField;
 
 import java.util.List;
@@ -31,6 +32,8 @@ public interface Function {
 
     String fullName();
 
+    Identifier identifier();
+
     Optional<List<DataField>> inputParams();
 
     Optional<List<DataField>> returnParams();

paimon-api/src/main/java/org/apache/paimon/function/FunctionImpl.java

@@ -21,34 +21,38 @@
 import org.apache.paimon.catalog.Identifier;
 import org.apache.paimon.types.DataField;
 
+import javax.annotation.Nullable;
+
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
 
 /** Function implementation. */
 public class FunctionImpl implements Function {
 
     private final Identifier identifier;
 
-    private final List<DataField> inputParams;
+    @Nullable private final List<DataField> inputParams;
 
-    private final List<DataField> returnParams;
+    @Nullable private final List<DataField> returnParams;
 
     private final boolean deterministic;
 
     private final Map<String, FunctionDefinition> definitions;
 
-    private final String comment;
+    @Nullable private final String comment;
 
     private final Map<String, String> options;
 
     public FunctionImpl(
             Identifier identifier,
-            List<DataField> inputParams,
-            List<DataField> returnParams,
+            @Nullable List<DataField> inputParams,
+            @Nullable List<DataField> returnParams,
             boolean deterministic,
             Map<String, FunctionDefinition> definitions,
-            String comment,
+            @Nullable String comment,
             Map<String, String> options) {
         this.identifier = identifier;
         this.inputParams = inputParams;
@@ -66,7 +70,7 @@ public FunctionImpl(Identifier identifier, Map<String, FunctionDefinition> defin
         this.deterministic = true;
         this.definitions = definitions;
         this.comment = null;
-        this.options = null;
+        this.options = Collections.emptyMap();
     }
 
     @Override
@@ -79,6 +83,10 @@ public String fullName() {
         return identifier.getFullName();
     }
 
+    public Identifier identifier() {
+        return identifier;
+    }
+
     @Override
     public Optional<List<DataField>> inputParams() {
         return Optional.ofNullable(inputParams);
@@ -113,4 +121,31 @@ public String comment() {
     public Map<String, String> options() {
         return options;
     }
+
+    @Override
+    public boolean equals(Object o) {
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        FunctionImpl function = (FunctionImpl) o;
+        return deterministic == function.deterministic
+                && Objects.equals(identifier, function.identifier)
+                && Objects.equals(inputParams, function.inputParams)
+                && Objects.equals(returnParams, function.returnParams)
+                && Objects.equals(definitions, function.definitions)
+                && Objects.equals(comment, function.comment)
+                && Objects.equals(options, function.options);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(
+                identifier,
+                inputParams,
+                returnParams,
+                deterministic,
+                definitions,
+                comment,
+                options);
+    }
 }

paimon-spark/paimon-spark-3.2/pom.xml

@@ -61,6 +61,18 @@ under the License.
             <version>${spark.version}</version>
         </dependency>
 
+        <dependency>
+            <groupId>org.apache.spark</groupId>
+            <artifactId>spark-catalyst_${scala.binary.version}</artifactId>
+            <version>${spark.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.spark</groupId>
+            <artifactId>spark-hive_${scala.binary.version}</artifactId>
+            <version>${spark.version}</version>
+        </dependency>
+
         <dependency>
             <groupId>org.apache.paimon</groupId>
             <artifactId>paimon-bundle</artifactId>
@@ -114,6 +126,12 @@ under the License.
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>com.squareup.okhttp3</groupId>
+            <artifactId>mockwebserver</artifactId>
+            <version>${okhttp.version}</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>

paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/RewritePaimonFunctionCommands.scala

@@ -0,0 +1,29 @@
+/*
+ * 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.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.rules.Rule
+
+case class RewritePaimonFunctionCommands(spark: SparkSession) extends Rule[LogicalPlan] {
+
+  // do nothing
+  override def apply(plan: LogicalPlan): LogicalPlan = plan
+}

paimon-spark/paimon-spark-3.3/pom.xml

@@ -61,6 +61,18 @@ under the License.
             <version>${spark.version}</version>
         </dependency>
 
+        <dependency>
+            <groupId>org.apache.spark</groupId>
+            <artifactId>spark-catalyst_${scala.binary.version}</artifactId>
+            <version>${spark.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.spark</groupId>
+            <artifactId>spark-hive_${scala.binary.version}</artifactId>
+            <version>${spark.version}</version>
+        </dependency>
+
         <dependency>
             <groupId>org.apache.paimon</groupId>
             <artifactId>paimon-bundle</artifactId>
@@ -113,6 +125,13 @@ under the License.
             <version>${spark.version}</version>
             <scope>test</scope>
         </dependency>
+
+        <dependency>
+            <groupId>com.squareup.okhttp3</groupId>
+            <artifactId>mockwebserver</artifactId>
+            <version>${okhttp.version}</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>

paimon-spark/paimon-spark-3.3/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/RewritePaimonFunctionCommands.scala

@@ -0,0 +1,29 @@
+/*
+ * 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.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.rules.Rule
+
+case class RewritePaimonFunctionCommands(spark: SparkSession) extends Rule[LogicalPlan] {
+
+  // do nothing
+  override def apply(plan: LogicalPlan): LogicalPlan = plan
+}