This is an automated email from the ASF dual-hosted git repository.
nicholasjiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 7a6198474 [spark] Supports parser of Spark call procedure command
(#1785)
7a6198474 is described below
commit 7a6198474b714a68bcf82ff2bbaa394f0b0cdc1c
Author: Nicholas Jiang <[email protected]>
AuthorDate: Fri Aug 25 13:09:47 2023 +0800
[spark] Supports parser of Spark call procedure command (#1785)
---
LICENSE | 9 +-
paimon-spark/paimon-spark-common/pom.xml | 18 ++
.../PaimonSqlExtensions.g4 | 229 ++++++++++++++++++
.../java/org/apache/paimon/spark/SparkCatalog.java | 17 +-
.../org/apache/paimon/spark/SparkProcedures.java | 47 ++++
.../spark/analysis/NoSuchProcedureException.java} | 23 +-
.../paimon/spark/catalog/ProcedureCatalog.java | 41 ++++
.../apache/paimon/spark/procedure/Procedure.java | 45 ++++
.../paimon/spark/procedure/ProcedureBuilder.java | 41 ++++
.../paimon/spark/procedure/ProcedureParameter.java | 56 +++++
.../spark/procedure/ProcedureParameterImpl.java | 80 +++++++
.../PaimonSparkSessionExtensions.scala} | 17 +-
.../sql/catalyst/analysis/CoerceArguments.scala | 61 +++++
.../sql/catalyst/analysis}/PaimonAnalysis.scala | 10 +-
.../sql/catalyst/analysis/ResolveProcedures.scala | 213 +++++++++++++++++
.../PaimonSparkSqlExtensionsParser.scala | 263 +++++++++++++++++++++
.../extensions/PaimonSqlExtensionsAstBuilder.scala | 132 +++++++++++
.../sql/catalyst/plans/logical/CallCommand.scala} | 20 +-
.../catalyst/plans/logical/CallStatement.scala} | 24 +-
.../PaimonDynamicPartitionOverwriteCommand.scala | 3 +-
.../logical}/PaimonTableValuedFunctions.scala | 4 +-
.../spark/extensions/CallStatementParserTest.java | 164 +++++++++++++
.../apache/paimon/spark/PaimonSparkTestBase.scala | 3 +-
.../spark/sql/TableValuedFunctionsTest.scala | 5 +-
pom.xml | 1 +
25 files changed, 1472 insertions(+), 54 deletions(-)
diff --git a/LICENSE b/LICENSE
index ef08963ea..74613fb9d 100644
--- a/LICENSE
+++ b/LICENSE
@@ -232,10 +232,15 @@
paimon-common/src/main/java/org/apache/paimon/types/DataType.java
paimon-common/src/main/java/org/apache/paimon/options/ConfigOption.java
from http://flink.apache.org/ version 1.17.0
-paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkGenericCatalog.java
-paimon-hive/paimon-hive-common/src/test/java/org/apache/paimon/hive/TestHiveMetastore.java
paimon-core/src/main/java/org/apache/paimon/utils/ZOrderByteUtils.java
paimon-core/src/test/java/org/apache/paimon/utils/TestZOrderByteUtil.java
+paimon-hive/paimon-hive-common/src/test/java/org/apache/paimon/hive/TestHiveMetastore.java
+paimon-spark/paimon-spark-common/src/main/antlr4/org.apache.spark.sql.catalyst.parser.extensions/PaimonSqlExtensions.g4
+paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkGenericCatalog.java
+paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/analysis/CoerceArguments.scala
+paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveProcedures.scala
+paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/PaimonSparkSqlExtensionsParser.scala
+paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/PaimonSqlExtensionsAstBuilder.scala
from http://iceberg.apache.org/ version 1.3.0
paimon-hive/paimon-hive-common/src/test/resources/hive-schema-3.1.0.derby.sql
diff --git a/paimon-spark/paimon-spark-common/pom.xml
b/paimon-spark/paimon-spark-common/pom.xml
index 9a065808f..2f34e06d8 100644
--- a/paimon-spark/paimon-spark-common/pom.xml
+++ b/paimon-spark/paimon-spark-common/pom.xml
@@ -190,6 +190,7 @@ under the License.
</dependencies>
<build>
+
<pluginManagement>
<plugins>
<plugin>
@@ -276,6 +277,23 @@ under the License.
</scala>
</configuration>
</plugin>
+ <plugin>
+ <groupId>org.antlr</groupId>
+ <artifactId>antlr4-maven-plugin</artifactId>
+ <version>${antlr4-maven-plugin.version}</version>
+ <executions>
+ <execution>
+ <goals>
+ <goal>antlr4</goal>
+ </goals>
+ </execution>
+ </executions>
+ <configuration>
+ <visitor>true</visitor>
+ <listener>true</listener>
+ <sourceDirectory>src/main/antlr4</sourceDirectory>
+ </configuration>
+ </plugin>
</plugins>
</build>
</project>
diff --git
a/paimon-spark/paimon-spark-common/src/main/antlr4/org.apache.spark.sql.catalyst.parser.extensions/PaimonSqlExtensions.g4
b/paimon-spark/paimon-spark-common/src/main/antlr4/org.apache.spark.sql.catalyst.parser.extensions/PaimonSqlExtensions.g4
new file mode 100644
index 000000000..54e71b362
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/main/antlr4/org.apache.spark.sql.catalyst.parser.extensions/PaimonSqlExtensions.g4
@@ -0,0 +1,229 @@
+/*
+ * 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.
+ *
+ * This file is an adaptation of Spark's grammar files.
+*/
+
+/* This file is based on source code from the Iceberg Project
(http://iceberg.apache.org/), licensed by the Apache
+ * Software Foundation (ASF) under the Apache License, Version 2.0. See the
NOTICE file distributed with this work for
+ * additional information regarding copyright ownership. */
+
+grammar PaimonSqlExtensions;
+
+@lexer::members {
+ /**
+ * Verify whether current token is a valid decimal token (which contains
dot).
+ * Returns true if the character that follows the token is not a digit or
letter or underscore.
+ *
+ * For example:
+ * For char stream "2.3", "2." is not a valid decimal token, because it is
followed by digit '3'.
+ * For char stream "2.3_", "2.3" is not a valid decimal token, because it is
followed by '_'.
+ * For char stream "2.3W", "2.3" is not a valid decimal token, because it is
followed by 'W'.
+ * For char stream "12.0D 34.E2+0.12 " 12.0D is a valid decimal token
because it is followed
+ * by a space. 34.E2 is a valid decimal token because it is followed by
symbol '+'
+ * which is not a digit or letter or underscore.
+ */
+ public boolean isValidDecimal() {
+ int nextChar = _input.LA(1);
+ if (nextChar >= 'A' && nextChar <= 'Z' || nextChar >= '0' && nextChar <=
'9' ||
+ nextChar == '_') {
+ return false;
+ } else {
+ return true;
+ }
+ }
+
+ /**
+ * This method will be called when we see '/*' and try to match it as a
bracketed comment.
+ * If the next character is '+', it should be parsed as hint later, and we
cannot match
+ * it as a bracketed comment.
+ *
+ * Returns true if the next character is '+'.
+ */
+ public boolean isHint() {
+ int nextChar = _input.LA(1);
+ if (nextChar == '+') {
+ return true;
+ } else {
+ return false;
+ }
+ }
+}
+
+singleStatement
+ : statement ';'* EOF
+ ;
+
+statement
+ : CALL multipartIdentifier '(' (callArgument (',' callArgument)*)? ')'
#call
+ ;
+
+callArgument
+ : expression #positionalArgument
+ | identifier '=>' expression #namedArgument
+ ;
+
+expression
+ : constant
+ | stringMap
+ ;
+
+constant
+ : number #numericLiteral
+ | booleanValue #booleanLiteral
+ | STRING+ #stringLiteral
+ | identifier STRING #typeConstructor
+ ;
+
+stringMap
+ : MAP '(' constant (',' constant)* ')'
+ ;
+
+booleanValue
+ : TRUE | FALSE
+ ;
+
+number
+ : MINUS? EXPONENT_VALUE #exponentLiteral
+ | MINUS? DECIMAL_VALUE #decimalLiteral
+ | MINUS? INTEGER_VALUE #integerLiteral
+ | MINUS? BIGINT_LITERAL #bigIntLiteral
+ | MINUS? SMALLINT_LITERAL #smallIntLiteral
+ | MINUS? TINYINT_LITERAL #tinyIntLiteral
+ | MINUS? DOUBLE_LITERAL #doubleLiteral
+ | MINUS? FLOAT_LITERAL #floatLiteral
+ | MINUS? BIGDECIMAL_LITERAL #bigDecimalLiteral
+ ;
+
+multipartIdentifier
+ : parts+=identifier ('.' parts+=identifier)*
+ ;
+
+identifier
+ : IDENTIFIER #unquotedIdentifier
+ | quotedIdentifier #quotedIdentifierAlternative
+ | nonReserved #unquotedIdentifier
+ ;
+
+quotedIdentifier
+ : BACKQUOTED_IDENTIFIER
+ ;
+
+nonReserved
+ : CALL
+ | TRUE | FALSE
+ | MAP
+ ;
+
+CALL: 'CALL';
+
+TRUE: 'TRUE';
+FALSE: 'FALSE';
+
+MAP: 'MAP';
+
+PLUS: '+';
+MINUS: '-';
+
+STRING
+ : '\'' ( ~('\''|'\\') | ('\\' .) )* '\''
+ | '"' ( ~('"'|'\\') | ('\\' .) )* '"'
+ ;
+
+BIGINT_LITERAL
+ : DIGIT+ 'L'
+ ;
+
+SMALLINT_LITERAL
+ : DIGIT+ 'S'
+ ;
+
+TINYINT_LITERAL
+ : DIGIT+ 'Y'
+ ;
+
+INTEGER_VALUE
+ : DIGIT+
+ ;
+
+EXPONENT_VALUE
+ : DIGIT+ EXPONENT
+ | DECIMAL_DIGITS EXPONENT {isValidDecimal()}?
+ ;
+
+DECIMAL_VALUE
+ : DECIMAL_DIGITS {isValidDecimal()}?
+ ;
+
+FLOAT_LITERAL
+ : DIGIT+ EXPONENT? 'F'
+ | DECIMAL_DIGITS EXPONENT? 'F' {isValidDecimal()}?
+ ;
+
+DOUBLE_LITERAL
+ : DIGIT+ EXPONENT? 'D'
+ | DECIMAL_DIGITS EXPONENT? 'D' {isValidDecimal()}?
+ ;
+
+BIGDECIMAL_LITERAL
+ : DIGIT+ EXPONENT? 'BD'
+ | DECIMAL_DIGITS EXPONENT? 'BD' {isValidDecimal()}?
+ ;
+
+IDENTIFIER
+ : (LETTER | DIGIT | '_')+
+ ;
+
+BACKQUOTED_IDENTIFIER
+ : '`' ( ~'`' | '``' )* '`'
+ ;
+
+fragment DECIMAL_DIGITS
+ : DIGIT+ '.' DIGIT*
+ | '.' DIGIT+
+ ;
+
+fragment EXPONENT
+ : 'E' [+-]? DIGIT+
+ ;
+
+fragment DIGIT
+ : [0-9]
+ ;
+
+fragment LETTER
+ : [A-Z]
+ ;
+
+SIMPLE_COMMENT
+ : '--' ('\\\n' | ~[\r\n])* '\r'? '\n'? -> channel(HIDDEN)
+ ;
+
+BRACKETED_COMMENT
+ : '/*' {!isHint()}? (BRACKETED_COMMENT|.)*? '*/' -> channel(HIDDEN)
+ ;
+
+WS
+ : [ \r\n\t]+ -> channel(HIDDEN)
+ ;
+
+// Catch-all for anything we can't recognize.
+// We use this to be able to ignore and recover all the text
+// when splitting statements with DelimiterLexer
+UNRECOGNIZED
+ : .
+ ;
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 879d14c57..81dc8c27e 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
@@ -25,6 +25,10 @@ import org.apache.paimon.catalog.CatalogFactory;
import org.apache.paimon.options.Options;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaChange;
+import org.apache.paimon.spark.analysis.NoSuchProcedureException;
+import org.apache.paimon.spark.catalog.ProcedureCatalog;
+import org.apache.paimon.spark.procedure.Procedure;
+import org.apache.paimon.spark.procedure.ProcedureBuilder;
import org.apache.paimon.table.DataTable;
import org.apache.paimon.table.Table;
import org.apache.paimon.utils.Preconditions;
@@ -58,7 +62,7 @@ import java.util.stream.Collectors;
import static org.apache.paimon.spark.SparkTypeUtils.toPaimonType;
/** Spark {@link TableCatalog} for paimon. */
-public class SparkCatalog implements TableCatalog, SupportsNamespaces {
+public class SparkCatalog implements TableCatalog, ProcedureCatalog,
SupportsNamespaces {
private static final Logger LOG =
LoggerFactory.getLogger(SparkCatalog.class);
@@ -315,6 +319,17 @@ public class SparkCatalog implements TableCatalog,
SupportsNamespaces {
}
}
+ @Override
+ public Procedure loadProcedure(Identifier identifier) throws
NoSuchProcedureException {
+ if (isValidateNamespace(identifier.namespace())) {
+ ProcedureBuilder builder = SparkProcedures.newBuilder(name);
+ if (builder != null) {
+ return builder.withTableCatalog(this).build();
+ }
+ }
+ throw new NoSuchProcedureException(identifier);
+ }
+
private SchemaChange toSchemaChange(TableChange change) {
if (change instanceof TableChange.SetProperty) {
TableChange.SetProperty set = (TableChange.SetProperty) change;
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java
new file mode 100644
index 000000000..8eb08b659
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java
@@ -0,0 +1,47 @@
+/*
+ * 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;
+
+import org.apache.paimon.spark.procedure.Procedure;
+import org.apache.paimon.spark.procedure.ProcedureBuilder;
+
+import org.apache.hadoop.shaded.com.google.common.collect.ImmutableMap;
+
+import java.util.Locale;
+import java.util.Map;
+import java.util.function.Supplier;
+
+/** The {@link Procedure}s including all the stored procedures. */
+public class SparkProcedures {
+
+ private static final Map<String, Supplier<ProcedureBuilder>> BUILDERS =
initProcedureBuilders();
+
+ private SparkProcedures() {}
+
+ public static ProcedureBuilder newBuilder(String name) {
+ Supplier<ProcedureBuilder> builderSupplier =
BUILDERS.get(name.toLowerCase(Locale.ROOT));
+ return builderSupplier != null ? builderSupplier.get() : null;
+ }
+
+ private static Map<String, Supplier<ProcedureBuilder>>
initProcedureBuilders() {
+ ImmutableMap.Builder<String, Supplier<ProcedureBuilder>>
procedureBuilders =
+ ImmutableMap.builder();
+ return procedureBuilders.build();
+ }
+}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonSparkSessionExtension.scala
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/analysis/NoSuchProcedureException.java
similarity index 57%
copy from
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonSparkSessionExtension.scala
copy to
paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/analysis/NoSuchProcedureException.java
index 9d6c8bfd5..f83c5f16c 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonSparkSessionExtension.scala
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/analysis/NoSuchProcedureException.java
@@ -7,7 +7,7 @@
* "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
+ * 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,
@@ -15,21 +15,20 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.paimon.spark
-import org.apache.spark.sql.SparkSessionExtensions
+package org.apache.paimon.spark.analysis;
-class PaimonSparkSessionExtension extends (SparkSessionExtensions => Unit) {
+import org.apache.spark.SparkThrowableHelper;
+import org.apache.spark.sql.connector.catalog.Identifier;
- override def apply(extensions: SparkSessionExtensions): Unit = {
+import java.util.Collections;
- extensions.injectResolutionRule(sparkSession => new
PaimonAnalysis(sparkSession))
+/** Thrown by a catalog when a stored procedure cannot be found. */
+public class NoSuchProcedureException extends RuntimeException {
- PaimonTableValuedFunctions.supportedFnNames.foreach {
- fnName =>
- extensions.injectTableFunction(
- PaimonTableValuedFunctions.getTableValueFunctionInjection(fnName))
+ public NoSuchProcedureException(Identifier identifier) {
+ super(
+ SparkThrowableHelper.getMessage(
+ "Procedure " + identifier + " is not found",
Collections.emptyMap()));
}
- }
-
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/ProcedureCatalog.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/ProcedureCatalog.java
new file mode 100644
index 000000000..b982c1896
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/ProcedureCatalog.java
@@ -0,0 +1,41 @@
+/*
+ * 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;
+
+import org.apache.paimon.spark.analysis.NoSuchProcedureException;
+import org.apache.paimon.spark.procedure.Procedure;
+
+import org.apache.spark.sql.connector.catalog.CatalogPlugin;
+import org.apache.spark.sql.connector.catalog.Identifier;
+
+/**
+ * A {@link CatalogPlugin catalog} interface that loads stored procedures
called via CALL
+ * statements.
+ */
+public interface ProcedureCatalog extends CatalogPlugin {
+
+ /**
+ * Loads a {@link Procedure stored procedure} by {@link Identifier
identifier}.
+ *
+ * @param identifier A stored procedure identifier.
+ * @return The procedure's metadata of given identifier.
+ * @throws NoSuchProcedureException Thrown, if there is no matching
procedure stored.
+ */
+ Procedure loadProcedure(Identifier identifier) throws
NoSuchProcedureException;
+}
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/Procedure.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/Procedure.java
new file mode 100644
index 000000000..127a46647
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/Procedure.java
@@ -0,0 +1,45 @@
+/*
+ * 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.procedure;
+
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.types.StructType;
+
+/** An interface that defines an executable stored procedure. */
+public interface Procedure {
+
+ /** Returns the input parameters of stored procedure. */
+ ProcedureParameter[] parameters();
+
+ /** Returns the type of rows produced by stored procedure. */
+ StructType outputType();
+
+ /**
+ * Executes the given stored procedure.
+ *
+ * @param args Input arguments.
+ * @return The result of executing stored procedure with the given
arguments.
+ */
+ InternalRow[] call(InternalRow args);
+
+ /** Returns the description of stored procedure. */
+ default String description() {
+ return this.getClass().toString();
+ }
+}
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/ProcedureBuilder.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/ProcedureBuilder.java
new file mode 100644
index 000000000..8b81112af
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/ProcedureBuilder.java
@@ -0,0 +1,41 @@
+/*
+ * 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.procedure;
+
+import org.apache.spark.sql.connector.catalog.Identifier;
+import org.apache.spark.sql.connector.catalog.TableCatalog;
+
+/** A builder of {@link Procedure} that builds a stored procedure. */
+public interface ProcedureBuilder {
+
+ /**
+ * Returns a {@link ProcedureBuilder procedure builder} via given table
catalog.
+ *
+ * @param tableCatalog The table catalog.
+ * @return The procedure builder with given catalog.
+ */
+ ProcedureBuilder withTableCatalog(TableCatalog tableCatalog);
+
+ /**
+ * Builds a {@link Procedure stored procedure} via given {@link Identifier
identifier}.
+ *
+ * @return The stored procedure of given identifier.
+ */
+ Procedure build();
+}
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/ProcedureParameter.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/ProcedureParameter.java
new file mode 100644
index 000000000..94306424a
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/ProcedureParameter.java
@@ -0,0 +1,56 @@
+/*
+ * 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.procedure;
+
+import org.apache.spark.sql.types.DataType;
+
+/** An input parameter of a {@link Procedure stored procedure}. */
+public interface ProcedureParameter {
+
+ /**
+ * Creates a required input parameter.
+ *
+ * @param name The name of input parameter.
+ * @param dataType The data type of input parameter.
+ * @return The constructed stored procedure parameter.
+ */
+ static ProcedureParameter required(String name, DataType dataType) {
+ return new ProcedureParameterImpl(name, dataType, true);
+ }
+
+ /**
+ * Creates an optional input parameter.
+ *
+ * @param name The name of input parameter.
+ * @param dataType The data type of input parameter.
+ * @return The constructed optional stored procedure parameter.
+ */
+ static ProcedureParameter optional(String name, DataType dataType) {
+ return new ProcedureParameterImpl(name, dataType, false);
+ }
+
+ /** Returns the name of input parameter. */
+ String name();
+
+ /** Returns the type of input parameter. */
+ DataType dataType();
+
+ /** Returns whether input parameter is required. */
+ boolean required();
+}
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/ProcedureParameterImpl.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/ProcedureParameterImpl.java
new file mode 100644
index 000000000..4fe239f46
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/ProcedureParameterImpl.java
@@ -0,0 +1,80 @@
+/*
+ * 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.procedure;
+
+import org.apache.spark.sql.types.DataType;
+
+import java.util.Objects;
+
+/** An implementation of {@link ProcedureParameter}. */
+public class ProcedureParameterImpl implements ProcedureParameter {
+
+ /** The name of input parameter. */
+ private final String name;
+ /** The data type of input parameter. */
+ private final DataType dataType;
+ /** Whether input parameter is required. */
+ private final boolean required;
+
+ public ProcedureParameterImpl(String name, DataType dataType, boolean
required) {
+ this.name = name;
+ this.dataType = dataType;
+ this.required = required;
+ }
+
+ @Override
+ public String name() {
+ return name;
+ }
+
+ @Override
+ public DataType dataType() {
+ return dataType;
+ }
+
+ @Override
+ public boolean required() {
+ return required;
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ if (this == other) {
+ return true;
+ } else if (other == null || getClass() != other.getClass()) {
+ return false;
+ }
+
+ ProcedureParameterImpl that = (ProcedureParameterImpl) other;
+ return required == that.required
+ && Objects.equals(name, that.name)
+ && Objects.equals(dataType, that.dataType);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(name, dataType, required);
+ }
+
+ @Override
+ public String toString() {
+ return String.format(
+ "ProcedureParameter(name='%s', type=%s, required=%b)", name,
dataType, required);
+ }
+}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonSparkSessionExtension.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/extensions/PaimonSparkSessionExtensions.scala
similarity index 60%
copy from
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonSparkSessionExtension.scala
copy to
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/extensions/PaimonSparkSessionExtensions.scala
index 9d6c8bfd5..4e1117b26 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonSparkSessionExtension.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/extensions/PaimonSparkSessionExtensions.scala
@@ -15,21 +15,30 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.paimon.spark
+package org.apache.paimon.spark.extensions
import org.apache.spark.sql.SparkSessionExtensions
+import org.apache.spark.sql.catalyst.analysis.{CoerceArguments,
PaimonAnalysis, ResolveProcedures}
+import
org.apache.spark.sql.catalyst.parser.extensions.PaimonSparkSqlExtensionsParser
+import org.apache.spark.sql.catalyst.plans.logical.PaimonTableValuedFunctions
-class PaimonSparkSessionExtension extends (SparkSessionExtensions => Unit) {
+/** Spark session extension to extends the syntax and adds the rules. */
+class PaimonSparkSessionExtensions extends (SparkSessionExtensions => Unit) {
override def apply(extensions: SparkSessionExtensions): Unit = {
+ // parser extensions
+ extensions.injectParser { case (_, parser) => new
PaimonSparkSqlExtensionsParser(parser) }
+ // analyzer extensions
+ // resolution rule extensions
extensions.injectResolutionRule(sparkSession => new
PaimonAnalysis(sparkSession))
-
+ extensions.injectResolutionRule(spark => ResolveProcedures(spark))
+ extensions.injectResolutionRule(_ => CoerceArguments)
+ // table function extensions
PaimonTableValuedFunctions.supportedFnNames.foreach {
fnName =>
extensions.injectTableFunction(
PaimonTableValuedFunctions.getTableValueFunctionInjection(fnName))
}
}
-
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/analysis/CoerceArguments.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/analysis/CoerceArguments.scala
new file mode 100644
index 000000000..389c79bbf
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/analysis/CoerceArguments.scala
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.expressions.Cast
+import org.apache.spark.sql.catalyst.plans.logical.{CallCommand, LogicalPlan}
+import org.apache.spark.sql.catalyst.rules.Rule
+
+/* This file is based on source code from the Iceberg Project
(http://iceberg.apache.org/), licensed by the Apache
+ * Software Foundation (ASF) under the Apache License, Version 2.0. See the
NOTICE file distributed with this work for
+ * additional information regarding copyright ownership. */
+
+/**
+ * Coercion of procedure arguments rule.
+ *
+ * <p>Most of the content of this class is referenced from Iceberg's
ProcedureArgumentCoercion.
+ */
+object CoerceArguments extends Rule[LogicalPlan] {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperators {
+ case call @ CallCommand(procedure, arguments) if call.resolved =>
+ val parameters = procedure.parameters
+ val newArguments = arguments.zipWithIndex.map {
+ case (argument, index) =>
+ val parameter = parameters(index)
+ val parameterType = parameter.dataType
+ val argumentType = argument.dataType
+ if (parameterType != argumentType && !Cast.canUpCast(argumentType,
parameterType)) {
+ throw new AnalysisException(
+ s"Cannot cast $argumentType to $parameterType of
${parameter.name}.")
+ }
+ if (parameterType != argumentType) {
+ Cast(argument, parameterType)
+ } else {
+ argument
+ }
+ }
+
+ if (newArguments != arguments) {
+ call.copy(args = newArguments)
+ } else {
+ call
+ }
+ }
+}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonAnalysis.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/analysis/PaimonAnalysis.scala
similarity index 85%
rename from
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonAnalysis.scala
rename to
paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/analysis/PaimonAnalysis.scala
index 9b1587fc4..04cf36fd1 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonAnalysis.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/analysis/PaimonAnalysis.scala
@@ -15,15 +15,15 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.paimon.spark
+package org.apache.spark.sql.catalyst.analysis
+import org.apache.paimon.spark.SparkTable
import org.apache.paimon.table.FileStoreTable
import org.apache.spark.sql.SparkSession
-import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan,
OverwritePartitionsDynamic}
+import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan,
OverwritePartitionsDynamic, PaimonDynamicPartitionOverwriteCommand,
PaimonTableValuedFunctions, PaimonTableValueFunction}
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
-import
org.apache.spark.sql.paimon.commands.PaimonDynamicPartitionOverwriteCommand
class PaimonAnalysis(session: SparkSession) extends Rule[LogicalPlan] {
@@ -32,12 +32,12 @@ class PaimonAnalysis(session: SparkSession) extends
Rule[LogicalPlan] {
case func: PaimonTableValueFunction if func.args.forall(_.resolved) =>
PaimonTableValuedFunctions.resolvePaimonTableValuedFunction(session,
func)
- case o @ OverwritePartitionsDynamicPaimon(r, d) if o.resolved =>
+ case o @ PaimonDynamicPartitionOverwrite(r, d) if o.resolved =>
PaimonDynamicPartitionOverwriteCommand(r, d, o.query, o.writeOptions,
o.isByName)
}
}
-object OverwritePartitionsDynamicPaimon {
+object PaimonDynamicPartitionOverwrite {
def unapply(o: OverwritePartitionsDynamic): Option[(DataSourceV2Relation,
FileStoreTable)] = {
if (o.query.resolved) {
o.table match {
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveProcedures.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveProcedures.scala
new file mode 100644
index 000000000..27d4777ce
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveProcedures.scala
@@ -0,0 +1,213 @@
+/*
+ * 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.analysis
+
+import org.apache.paimon.spark.catalog.ProcedureCatalog
+import org.apache.paimon.spark.procedure.ProcedureParameter
+
+import org.apache.spark.sql.{AnalysisException, SparkSession}
+import org.apache.spark.sql.catalyst.expressions.{Expression, Literal}
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.connector.catalog.{CatalogManager, CatalogPlugin,
LookupCatalog}
+
+import java.util.Locale
+
+/* This file is based on source code from the Iceberg Project
(http://iceberg.apache.org/), licensed by the Apache
+ * Software Foundation (ASF) under the Apache License, Version 2.0. See the
NOTICE file distributed with this work for
+ * additional information regarding copyright ownership. */
+
+/**
+ * Resolution of stored procedures rule.
+ *
+ * <p>Most of the content of this class is referenced from Iceberg's
ResolveProcedures.
+ *
+ * @param sparkSession
+ * The Spark session.
+ */
+case class ResolveProcedures(sparkSession: SparkSession)
+ extends Rule[LogicalPlan]
+ with LookupCatalog {
+
+ protected lazy val catalogManager: CatalogManager =
sparkSession.sessionState.catalogManager
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperators {
+ case CallStatement(CatalogAndIdentifier(catalog, identifier), arguments) =>
+ val procedure = catalog.asProcedureCatalog.loadProcedure(identifier)
+ val parameters = procedure.parameters
+ val normalizedParameters = normalizeParameters(parameters)
+ validateParameters(normalizedParameters)
+ val normalizedArguments = normalizeArguments(arguments)
+ CallCommand(
+ procedure,
+ args = buildArgumentExpressions(normalizedParameters,
normalizedArguments))
+ }
+
+ /**
+ * Normalizes input parameters of stored procedure.
+ *
+ * @param parameters
+ * Input parameters of stored procedure.
+ * @return
+ * Normalized input parameters.
+ */
+ private def normalizeParameters(parameters: Seq[ProcedureParameter]):
Seq[ProcedureParameter] = {
+ parameters.map {
+ case parameter if parameter.required =>
+ val normalizedName = parameter.name.toLowerCase(Locale.ROOT)
+ ProcedureParameter.required(normalizedName, parameter.dataType)
+ case param =>
+ val normalizedName = param.name.toLowerCase(Locale.ROOT)
+ ProcedureParameter.optional(normalizedName, param.dataType)
+ }
+ }
+
+ /**
+ * Validates input parameters of stored procedure.
+ *
+ * @param parameters
+ * Input parameters of stored procedure.
+ */
+ private def validateParameters(parameters: Seq[ProcedureParameter]): Unit = {
+ val duplicateParamNames = parameters.groupBy(_.name).collect {
+ case (name, matchingParams) if matchingParams.length > 1 => name
+ }
+ if (duplicateParamNames.nonEmpty) {
+ throw new AnalysisException(
+ s"Parameter names ${duplicateParamNames.mkString("[", ",", "]")} are
duplicated.")
+ }
+ parameters.sliding(2).foreach {
+ case Seq(previousParam, currentParam) if !previousParam.required &&
currentParam.required =>
+ throw new AnalysisException(
+ s"Optional parameters should be after required ones but
$currentParam is after $previousParam.")
+ case _ =>
+ }
+ }
+
+ /**
+ * Normalizes arguments of stored procedure.
+ *
+ * @param arguments
+ * Arguments of stored procedure.
+ * @return
+ * Normalized arguments.
+ */
+ private def normalizeArguments(arguments: Seq[CallArgument]):
Seq[CallArgument] = {
+ arguments.map {
+ case a @ NamedArgument(name, _) => a.copy(name =
name.toLowerCase(Locale.ROOT))
+ case other => other
+ }
+ }
+
+ /**
+ * Builds argument expressions.
+ *
+ * @param parameters
+ * Normalized input parameters.
+ * @param arguments
+ * Normalized arguments.
+ * @return
+ * Argument expressions.
+ */
+ private def buildArgumentExpressions(
+ parameters: Seq[ProcedureParameter],
+ arguments: Seq[CallArgument]): Seq[Expression] = {
+ val nameToPositionMap = parameters.map(_.name).zipWithIndex.toMap
+ val nameToArgumentMap = buildNameToArgumentMap(parameters, arguments,
nameToPositionMap)
+ val missingParamNames = parameters.filter(_.required).collect {
+ case parameter if !nameToArgumentMap.contains(parameter.name) =>
parameter.name
+ }
+ if (missingParamNames.nonEmpty) {
+ throw new AnalysisException(
+ s"Required parameters ${missingParamNames.mkString("[", ",", "]")} is
missed.")
+ }
+ val argumentExpressions = new Array[Expression](parameters.size)
+ nameToArgumentMap.foreach {
+ case (name, argument) => argumentExpressions(nameToPositionMap(name)) =
argument.expr
+ }
+ parameters.foreach {
+ case parameter if !parameter.required &&
!nameToArgumentMap.contains(parameter.name) =>
+ argumentExpressions(nameToPositionMap(parameter.name)) =
+ Literal.create(null, parameter.dataType)
+ case _ =>
+ }
+ argumentExpressions
+ }
+
+ /**
+ * Builds mapping between name and position.
+ *
+ * @param parameters
+ * Normalized input parameters.
+ * @param arguments
+ * Normalized arguments.
+ * @param nameToPositionMap
+ * Mapping between name and position.
+ * @return
+ * Mapping between name and position.
+ */
+ private def buildNameToArgumentMap(
+ parameters: Seq[ProcedureParameter],
+ arguments: Seq[CallArgument],
+ nameToPositionMap: Map[String, Int]): Map[String, CallArgument] = {
+ val isNamedArgument = arguments.exists(_.isInstanceOf[NamedArgument])
+ val isPositionalArgument =
arguments.exists(_.isInstanceOf[PositionalArgument])
+
+ if (isNamedArgument && isPositionalArgument) {
+ throw new AnalysisException("Cannot mix named and positional arguments.")
+ }
+
+ if (isNamedArgument) {
+ val namedArguments = arguments.asInstanceOf[Seq[NamedArgument]]
+ val validationErrors = namedArguments.groupBy(_.name).collect {
+ case (name, procedureArguments) if procedureArguments.size > 1 =>
+ s"Procedure argument $name is duplicated."
+ case (name, _) if !nameToPositionMap.contains(name) => s"Argument
$name is unknown."
+ }
+ if (validationErrors.nonEmpty) {
+ throw new AnalysisException(
+ s"Builds name to argument map ${validationErrors.mkString(", ")}
error.")
+ }
+ namedArguments.map(arg => arg.name -> arg).toMap
+ } else {
+ if (arguments.size > parameters.size) {
+ throw new AnalysisException("Too many arguments for procedure")
+ }
+ arguments.zipWithIndex.map {
+ case (argument, position) =>
+ val param = parameters(position)
+ param.name -> argument
+ }.toMap
+ }
+ }
+
+ /**
+ * Procedure catalog validator whether catalog plugin is procedure catalog.
+ *
+ * @param catalogPlugin
+ * Spark catalog plugin.
+ */
+ implicit class CatalogValidator(catalogPlugin: CatalogPlugin) {
+ def asProcedureCatalog: ProcedureCatalog = catalogPlugin match {
+ case procedureCatalog: ProcedureCatalog =>
+ procedureCatalog
+ case _ =>
+ throw new AnalysisException(s"${catalogPlugin.name} is not a
ProcedureCatalog.")
+ }
+ }
+}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/PaimonSparkSqlExtensionsParser.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/PaimonSparkSqlExtensionsParser.scala
new file mode 100644
index 000000000..f42cbab55
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/PaimonSparkSqlExtensionsParser.scala
@@ -0,0 +1,263 @@
+/*
+ * 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.antlr.v4.runtime._
+import org.antlr.v4.runtime.atn.PredictionMode
+import org.antlr.v4.runtime.misc.{Interval, ParseCancellationException}
+import org.antlr.v4.runtime.tree.TerminalNodeImpl
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.{FunctionIdentifier, TableIdentifier}
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.catalyst.parser.{ParseException, ParserInterface}
+import
org.apache.spark.sql.catalyst.parser.extensions.PaimonSqlExtensionsParser.{NonReservedContext,
QuotedIdentifierContext}
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.trees.Origin
+import org.apache.spark.sql.internal.VariableSubstitution
+import org.apache.spark.sql.types.{DataType, StructType}
+
+import java.util.Locale
+
+/* This file is based on source code from the Iceberg Project
(http://iceberg.apache.org/), licensed by the Apache
+ * Software Foundation (ASF) under the Apache License, Version 2.0. See the
NOTICE file distributed with this work for
+ * additional information regarding copyright ownership. */
+
+/**
+ * The implementation of [[ParserInterface]] that parsers the sql extension.
+ *
+ * <p>Most of the content of this class is referenced from Iceberg's
+ * IcebergSparkSqlExtensionsParser.
+ *
+ * @param delegate
+ * The extension parser.
+ */
+class PaimonSparkSqlExtensionsParser(delegate: ParserInterface)
+ extends ParserInterface
+ with Logging {
+
+ private lazy val substitutor = new VariableSubstitution()
+ private lazy val astBuilder = new PaimonSqlExtensionsAstBuilder(delegate)
+
+ /** Parses a string to a LogicalPlan. */
+ override def parsePlan(sqlText: String): LogicalPlan = {
+ val sqlTextAfterSubstitution = substitutor.substitute(sqlText)
+ if (isCommand(sqlTextAfterSubstitution)) {
+ parse(sqlTextAfterSubstitution)(parser =>
astBuilder.visit(parser.singleStatement()))
+ .asInstanceOf[LogicalPlan]
+ } else {
+ delegate.parsePlan(sqlText)
+ }
+ }
+
+ /** Parses a string to an Expression. */
+ override def parseExpression(sqlText: String): Expression =
+ delegate.parseExpression(sqlText)
+
+ /** Parses a string to a TableIdentifier. */
+ override def parseTableIdentifier(sqlText: String): TableIdentifier =
+ delegate.parseTableIdentifier(sqlText)
+
+ /** Parses a string to a FunctionIdentifier. */
+ override def parseFunctionIdentifier(sqlText: String): FunctionIdentifier =
+ delegate.parseFunctionIdentifier(sqlText)
+
+ /**
+ * Creates StructType for a given SQL string, which is a comma separated
list of field definitions
+ * which will preserve the correct Hive metadata.
+ */
+ override def parseTableSchema(sqlText: String): StructType =
+ delegate.parseTableSchema(sqlText)
+
+ /** Parses a string to a DataType. */
+ override def parseDataType(sqlText: String): DataType =
+ delegate.parseDataType(sqlText)
+
+ /** Parses a string to a multi-part identifier. */
+ override def parseMultipartIdentifier(sqlText: String): Seq[String] =
+ delegate.parseMultipartIdentifier(sqlText)
+
+ /** Returns whether SQL text is command. */
+ private def isCommand(sqlText: String): Boolean = {
+ val normalized = sqlText
+ .toLowerCase(Locale.ROOT)
+ .trim()
+ .replaceAll("--.*?\\n", " ")
+ .replaceAll("\\s+", " ")
+ .replaceAll("/\\*.*?\\*/", " ")
+ .trim()
+ normalized.startsWith("call")
+ }
+
+ protected def parse[T](command: String)(toResult: PaimonSqlExtensionsParser
=> T): T = {
+ val lexer = new PaimonSqlExtensionsLexer(
+ new UpperCaseCharStream(CharStreams.fromString(command)))
+ lexer.removeErrorListeners()
+ lexer.addErrorListener(PaimonParseErrorListener)
+
+ val tokenStream = new CommonTokenStream(lexer)
+ val parser = new PaimonSqlExtensionsParser(tokenStream)
+ parser.addParseListener(PaimonSqlExtensionsPostProcessor)
+ parser.removeErrorListeners()
+ parser.addErrorListener(PaimonParseErrorListener)
+
+ try {
+ try {
+ parser.getInterpreter.setPredictionMode(PredictionMode.SLL)
+ toResult(parser)
+ } catch {
+ case _: ParseCancellationException =>
+ tokenStream.seek(0)
+ parser.reset()
+ parser.getInterpreter.setPredictionMode(PredictionMode.LL)
+ toResult(parser)
+ }
+ } catch {
+ case e: PaimonParseException if e.command.isDefined =>
+ throw e
+ case e: PaimonParseException =>
+ throw e.withCommand(command)
+ case e: AnalysisException =>
+ val position = Origin(e.line, e.startPosition)
+ throw new PaimonParseException(Option(command), e.message, position,
position)
+ }
+ }
+
+ def parseQuery(sqlText: String): LogicalPlan =
+ parsePlan(sqlText)
+}
+
+/* Copied from Apache Spark's to avoid dependency on Spark Internals */
+class UpperCaseCharStream(wrapped: CodePointCharStream) extends CharStream {
+ override def consume(): Unit = wrapped.consume()
+ override def getSourceName: String = wrapped.getSourceName
+ override def index(): Int = wrapped.index
+ override def mark(): Int = wrapped.mark
+ override def release(marker: Int): Unit = wrapped.release(marker)
+ override def seek(where: Int): Unit = wrapped.seek(where)
+ override def size(): Int = wrapped.size
+
+ override def getText(interval: Interval): String = wrapped.getText(interval)
+
+ // scalastyle:off
+ override def LA(i: Int): Int = {
+ val la = wrapped.LA(i)
+ if (la == 0 || la == IntStream.EOF) la
+ else Character.toUpperCase(la)
+ }
+ // scalastyle:on
+}
+
+/** The post-processor validates & cleans-up the parse tree during the parse
process. */
+case object PaimonSqlExtensionsPostProcessor extends
PaimonSqlExtensionsBaseListener {
+
+ /** Removes the back ticks from an Identifier. */
+ override def exitQuotedIdentifier(ctx: QuotedIdentifierContext): Unit = {
+ replaceTokenByIdentifier(ctx, 1) {
+ token =>
+ // Remove the double back ticks in the string.
+ token.setText(token.getText.replace("``", "`"))
+ token
+ }
+ }
+
+ /** Treats non-reserved keywords as Identifiers. */
+ override def exitNonReserved(ctx: NonReservedContext): Unit = {
+ replaceTokenByIdentifier(ctx, 0)(identity)
+ }
+
+ private def replaceTokenByIdentifier(ctx: ParserRuleContext, stripMargins:
Int)(
+ f: CommonToken => CommonToken = identity): Unit = {
+ val parent = ctx.getParent
+ parent.removeLastChild()
+ val token = ctx.getChild(0).getPayload.asInstanceOf[Token]
+ val newToken = new CommonToken(
+ new org.antlr.v4.runtime.misc.Pair(token.getTokenSource,
token.getInputStream),
+ PaimonSqlExtensionsParser.IDENTIFIER,
+ token.getChannel,
+ token.getStartIndex + stripMargins,
+ token.getStopIndex - stripMargins)
+ parent.addChild(new TerminalNodeImpl(f(newToken)))
+ }
+}
+
+/* Partially copied from Apache Spark's Parser to avoid dependency on Spark
Internals */
+case object PaimonParseErrorListener extends BaseErrorListener {
+ override def syntaxError(
+ recognizer: Recognizer[_, _],
+ offendingSymbol: scala.Any,
+ line: Int,
+ charPositionInLine: Int,
+ msg: String,
+ e: RecognitionException): Unit = {
+ val (start, stop) = offendingSymbol match {
+ case token: CommonToken =>
+ val start = Origin(Some(line), Some(token.getCharPositionInLine))
+ val length = token.getStopIndex - token.getStartIndex + 1
+ val stop = Origin(Some(line), Some(token.getCharPositionInLine +
length))
+ (start, stop)
+ case _ =>
+ val start = Origin(Some(line), Some(charPositionInLine))
+ (start, start)
+ }
+ throw new PaimonParseException(None, msg, start, stop)
+ }
+}
+
+/**
+ * Copied from Apache Spark A [[ParseException]] is an [[AnalysisException]]
that is thrown during
+ * the parse process. It contains fields and an extended error message that
make reporting and
+ * diagnosing errors easier.
+ */
+class PaimonParseException(
+ val command: Option[String],
+ message: String,
+ val start: Origin,
+ val stop: Origin)
+ extends AnalysisException(message, start.line, start.startPosition) {
+
+ def this(message: String, ctx: ParserRuleContext) =
+ this(
+ Option(PaimonParserUtils.command(ctx)),
+ message,
+ PaimonParserUtils.position(ctx.getStart),
+ PaimonParserUtils.position(ctx.getStop))
+
+ override def getMessage: String = {
+ val builder = new StringBuilder
+ builder ++= "\n" ++= message
+ start match {
+ case Origin(Some(l), Some(p), Some(_), Some(_), Some(_), Some(_),
Some(_)) =>
+ builder ++= s"(line $l, pos $p)\n"
+ command.foreach {
+ cmd =>
+ val (above, below) = cmd.split("\n").splitAt(l)
+ builder ++= "\n== SQL ==\n"
+ above.foreach(builder ++= _ += '\n')
+ builder ++= (0 until p).map(_ => "-").mkString("") ++= "^^^\n"
+ below.foreach(builder ++= _ += '\n')
+ }
+ case _ =>
+ command.foreach(cmd => builder ++= "\n== SQL ==\n" ++= cmd)
+ }
+ builder.toString
+ }
+
+ def withCommand(cmd: String): PaimonParseException =
+ new PaimonParseException(Option(cmd), message, start, stop)
+}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/PaimonSqlExtensionsAstBuilder.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/PaimonSqlExtensionsAstBuilder.scala
new file mode 100644
index 000000000..1b8d9d8dd
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/PaimonSqlExtensionsAstBuilder.scala
@@ -0,0 +1,132 @@
+/*
+ * 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.antlr.v4.runtime._
+import org.antlr.v4.runtime.misc.Interval
+import org.antlr.v4.runtime.tree.{ParseTree, TerminalNode}
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.catalyst.parser.ParserInterface
+import
org.apache.spark.sql.catalyst.parser.extensions.PaimonParserUtils.withOrigin
+import
org.apache.spark.sql.catalyst.parser.extensions.PaimonSqlExtensionsParser._
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.trees.{CurrentOrigin, Origin}
+
+import scala.collection.JavaConverters._
+
+/* This file is based on source code from the Iceberg Project
(http://iceberg.apache.org/), licensed by the Apache
+ * Software Foundation (ASF) under the Apache License, Version 2.0. See the
NOTICE file distributed with this work for
+ * additional information regarding copyright ownership. */
+
+/**
+ * The AST Builder provides an implementation of
[[PaimonSqlExtensionsBaseVisitor]], which can be
+ * extended to create a visitor which only needs to handle a subset of the
available methods.
+ *
+ * <p>Most of the content of this class is referenced from Iceberg's
IcebergSqlExtensionsAstBuilder.
+ *
+ * @param delegate
+ * The extension parser.
+ */
+class PaimonSqlExtensionsAstBuilder(delegate: ParserInterface)
+ extends PaimonSqlExtensionsBaseVisitor[AnyRef]
+ with Logging {
+
+ /** Creates a single statement of extension statements. */
+ override def visitSingleStatement(ctx: SingleStatementContext): LogicalPlan
= withOrigin(ctx) {
+ visit(ctx.statement).asInstanceOf[LogicalPlan]
+ }
+
+ /** Creates a [[CallStatement]] for a stored procedure call. */
+ override def visitCall(ctx: CallContext): CallStatement = withOrigin(ctx) {
+ val name = toSeq(ctx.multipartIdentifier.parts).map(_.getText)
+ val args = toSeq(ctx.callArgument).map(typedVisit[CallArgument])
+ CallStatement(name, args)
+ }
+
+ /** Creates a positional argument in a stored procedure call. */
+ override def visitPositionalArgument(ctx: PositionalArgumentContext):
CallArgument =
+ withOrigin(ctx) {
+ val expression = typedVisit[Expression](ctx.expression)
+ PositionalArgument(expression)
+ }
+
+ /** Creates a named argument in a stored procedure call. */
+ override def visitNamedArgument(ctx: NamedArgumentContext): CallArgument =
withOrigin(ctx) {
+ val name = ctx.identifier.getText
+ val expression = typedVisit[Expression](ctx.expression)
+ NamedArgument(name, expression)
+ }
+
+ /** Creates a [[Expression]] in a positional and named argument. */
+ override def visitExpression(ctx: ExpressionContext): Expression = {
+ // reconstruct the SQL string and parse it using the main Spark parser
+ // while we can avoid the logic to build Spark expressions, we still have
to parse them
+ // we cannot call ctx.getText directly since it will not render spaces
correctly
+ // that's why we need to recurse down the tree in reconstructSqlString
+ val sqlString = reconstructSqlString(ctx)
+ delegate.parseExpression(sqlString)
+ }
+
+ /** Returns a multi-part identifier as Seq[String]. */
+ override def visitMultipartIdentifier(ctx: MultipartIdentifierContext):
Seq[String] =
+ withOrigin(ctx) {
+ ctx.parts.asScala.map(_.getText)
+ }
+
+ private def toBuffer[T](list: java.util.List[T]) = list.asScala
+
+ private def toSeq[T](list: java.util.List[T]) = toBuffer(list)
+
+ private def reconstructSqlString(ctx: ParserRuleContext): String = {
+ toBuffer(ctx.children)
+ .map {
+ case c: ParserRuleContext => reconstructSqlString(c)
+ case t: TerminalNode => t.getText
+ }
+ .mkString(" ")
+ }
+
+ private def typedVisit[T](ctx: ParseTree): T =
+ ctx.accept(this).asInstanceOf[T]
+}
+
+/* Partially copied from Apache Spark's Parser to avoid dependency on Spark
Internals */
+object PaimonParserUtils {
+
+ private[sql] def withOrigin[T](ctx: ParserRuleContext)(f: => T): T = {
+ val current = CurrentOrigin.get
+ CurrentOrigin.set(position(ctx.getStart))
+ try {
+ f
+ } finally {
+ CurrentOrigin.set(current)
+ }
+ }
+
+ private[sql] def position(token: Token): Origin = {
+ val opt = Option(token)
+ Origin(opt.map(_.getLine), opt.map(_.getCharPositionInLine))
+ }
+
+ /** Gets the command which created the token. */
+ private[sql] def command(ctx: ParserRuleContext): String = {
+ val stream = ctx.getStart.getInputStream
+ stream.getText(Interval.of(0, stream.size() - 1))
+ }
+}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonSparkSessionExtension.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/CallCommand.scala
similarity index 57%
copy from
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonSparkSessionExtension.scala
copy to
paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/CallCommand.scala
index 9d6c8bfd5..1fade4a9f 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonSparkSessionExtension.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/CallCommand.scala
@@ -15,21 +15,19 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.paimon.spark
+package org.apache.spark.sql.catalyst.plans.logical
-import org.apache.spark.sql.SparkSessionExtensions
+import org.apache.paimon.spark.procedure.Procedure
-class PaimonSparkSessionExtension extends (SparkSessionExtensions => Unit) {
+import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression}
+import org.apache.spark.sql.catalyst.util.truncatedString
- override def apply(extensions: SparkSessionExtensions): Unit = {
+/** A CALL command that resolves stored procedure from SQL. */
+case class CallCommand(procedure: Procedure, args: Seq[Expression]) extends
LeafCommand {
- extensions.injectResolutionRule(sparkSession => new
PaimonAnalysis(sparkSession))
+ override lazy val output: Seq[Attribute] = procedure.outputType.toAttributes
- PaimonTableValuedFunctions.supportedFnNames.foreach {
- fnName =>
- extensions.injectTableFunction(
- PaimonTableValuedFunctions.getTableValueFunctionInjection(fnName))
- }
+ override def simpleString(maxFields: Int): String = {
+ s"Call${truncatedString(output, "[", ", ", "]", maxFields)}
${procedure.description}"
}
-
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonSparkSessionExtension.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/CallStatement.scala
similarity index 58%
rename from
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonSparkSessionExtension.scala
rename to
paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/CallStatement.scala
index 9d6c8bfd5..823743e56 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonSparkSessionExtension.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/CallStatement.scala
@@ -15,21 +15,21 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.paimon.spark
+package org.apache.spark.sql.catalyst.plans.logical
-import org.apache.spark.sql.SparkSessionExtensions
+import org.apache.spark.sql.catalyst.expressions.Expression
-class PaimonSparkSessionExtension extends (SparkSessionExtensions => Unit) {
+/** A CALL statement parsed from SQL. */
+case class CallStatement(name: Seq[String], args: Seq[CallArgument]) extends
LeafParsedStatement
- override def apply(extensions: SparkSessionExtensions): Unit = {
+/** An argument of a CALL statement. */
+sealed trait CallArgument {
- extensions.injectResolutionRule(sparkSession => new
PaimonAnalysis(sparkSession))
+ def expr: Expression
+}
- PaimonTableValuedFunctions.supportedFnNames.foreach {
- fnName =>
- extensions.injectTableFunction(
- PaimonTableValuedFunctions.getTableValueFunctionInjection(fnName))
- }
- }
+/** An argument identified by name. */
+case class NamedArgument(name: String, expr: Expression) extends CallArgument
-}
+/** An argument identified by position. */
+case class PositionalArgument(expr: Expression) extends CallArgument
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/paimon/commands/PaimonDynamicPartitionOverwriteCommand.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/PaimonDynamicPartitionOverwriteCommand.scala
similarity index 95%
rename from
paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/paimon/commands/PaimonDynamicPartitionOverwriteCommand.scala
rename to
paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/PaimonDynamicPartitionOverwriteCommand.scala
index 842472a2e..730a136dc 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/paimon/commands/PaimonDynamicPartitionOverwriteCommand.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/PaimonDynamicPartitionOverwriteCommand.scala
@@ -15,7 +15,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.spark.sql.paimon.commands
+package org.apache.spark.sql.catalyst.plans.logical
import org.apache.paimon.options.Options
import org.apache.paimon.spark.DynamicOverWrite
@@ -24,7 +24,6 @@ import org.apache.paimon.table.FileStoreTable
import org.apache.spark.sql.{Dataset, Row, SparkSession}
import org.apache.spark.sql.catalyst.analysis.NamedRelation
-import org.apache.spark.sql.catalyst.plans.logical.{Command, LogicalPlan,
V2WriteCommand}
import org.apache.spark.sql.execution.command.RunnableCommand
import scala.collection.convert.ImplicitConversions._
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonTableValuedFunctions.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/PaimonTableValuedFunctions.scala
similarity index 97%
rename from
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonTableValuedFunctions.scala
rename to
paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/PaimonTableValuedFunctions.scala
index 414f3a84b..a751a2d0a 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonTableValuedFunctions.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/PaimonTableValuedFunctions.scala
@@ -15,9 +15,10 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.paimon.spark
+package org.apache.spark.sql.catalyst.plans.logical
import org.apache.paimon.CoreOptions
+import org.apache.paimon.spark.SparkCatalog
import org.apache.paimon.spark.catalog.Catalogs
import org.apache.spark.sql.SparkSession
@@ -25,7 +26,6 @@ import org.apache.spark.sql.catalyst.FunctionIdentifier
import org.apache.spark.sql.catalyst.analysis.FunctionRegistryBase
import
org.apache.spark.sql.catalyst.analysis.TableFunctionRegistry.TableFunctionBuilder
import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression,
ExpressionInfo}
-import org.apache.spark.sql.catalyst.plans.logical.{LeafNode, LogicalPlan}
import org.apache.spark.sql.connector.catalog.Identifier
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
import org.apache.spark.sql.util.CaseInsensitiveStringMap
diff --git
a/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/extensions/CallStatementParserTest.java
b/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/extensions/CallStatementParserTest.java
new file mode 100644
index 000000000..ae7c0e067
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/extensions/CallStatementParserTest.java
@@ -0,0 +1,164 @@
+/*
+ * 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.extensions;
+
+import org.apache.spark.sql.SparkSession;
+import org.apache.spark.sql.catalyst.expressions.Literal$;
+import org.apache.spark.sql.catalyst.parser.ParseException;
+import org.apache.spark.sql.catalyst.parser.ParserInterface;
+import org.apache.spark.sql.catalyst.parser.extensions.PaimonParseException;
+import org.apache.spark.sql.catalyst.plans.logical.CallArgument;
+import org.apache.spark.sql.catalyst.plans.logical.CallStatement;
+import org.apache.spark.sql.catalyst.plans.logical.NamedArgument;
+import org.apache.spark.sql.catalyst.plans.logical.PositionalArgument;
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.DataTypes;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.math.BigDecimal;
+import java.sql.Timestamp;
+import java.time.Instant;
+import java.util.Arrays;
+
+import scala.collection.JavaConverters;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Test for {@link CallStatement} of {@link PaimonSparkSessionExtensions}. */
+public class CallStatementParserTest {
+
+ private SparkSession spark = null;
+ private ParserInterface parser = null;
+
+ @BeforeEach
+ public void startSparkSession() {
+ SparkSession.clearActiveSession();
+ spark =
+ SparkSession.builder()
+ .master("local[2]")
+ .config(
+ "spark.sql.extensions",
+ PaimonSparkSessionExtensions.class.getName())
+ .getOrCreate();
+ parser = spark.sessionState().sqlParser();
+ }
+
+ @AfterEach
+ public void stopSparkSession() {
+ if (spark != null) {
+ spark.stop();
+ spark = null;
+ parser = null;
+ }
+ }
+
+ @Test
+ public void testCallWithNamedArguments() throws ParseException {
+ CallStatement callStatement =
+ (CallStatement)
+ parser.parsePlan(
+ "CALL catalog.system.named_args_func(arg1 =>
1, arg2 => 'test', arg3 => true)");
+ assertThat(JavaConverters.seqAsJavaList(callStatement.name()))
+ .isEqualTo(Arrays.asList("catalog", "system",
"named_args_func"));
+ assertThat(callStatement.args().size()).isEqualTo(3);
+ assertArgument(callStatement, 0, "arg1", 1, DataTypes.IntegerType);
+ assertArgument(callStatement, 1, "arg2", "test", DataTypes.StringType);
+ assertArgument(callStatement, 2, "arg3", true, DataTypes.BooleanType);
+ }
+
+ @Test
+ public void testCallWithPositionalArguments() throws ParseException {
+ CallStatement callStatement =
+ (CallStatement)
+ parser.parsePlan(
+ "CALL catalog.system.positional_args_func(1,
'${spark.sql.extensions}', 2L, true, 3.0D, 4"
+ + ".0e1,500e-1BD, "
+ + "TIMESTAMP
'2017-02-03T10:37:30.00Z')");
+ assertThat(JavaConverters.seqAsJavaList(callStatement.name()))
+ .isEqualTo(Arrays.asList("catalog", "system",
"positional_args_func"));
+ assertThat(callStatement.args().size()).isEqualTo(8);
+ assertArgument(callStatement, 0, 1, DataTypes.IntegerType);
+ assertArgument(
+ callStatement,
+ 1,
+ PaimonSparkSessionExtensions.class.getName(),
+ DataTypes.StringType);
+ assertArgument(callStatement, 2, 2L, DataTypes.LongType);
+ assertArgument(callStatement, 3, true, DataTypes.BooleanType);
+ assertArgument(callStatement, 4, 3.0D, DataTypes.DoubleType);
+ assertArgument(callStatement, 5, 4.0e1, DataTypes.DoubleType);
+ assertArgument(
+ callStatement, 6, new BigDecimal("500e-1"),
DataTypes.createDecimalType(3, 1));
+ assertArgument(
+ callStatement,
+ 7,
+ Timestamp.from(Instant.parse("2017-02-03T10:37:30.00Z")),
+ DataTypes.TimestampType);
+ }
+
+ @Test
+ public void testCallWithMixedArguments() throws ParseException {
+ CallStatement callStatement =
+ (CallStatement)
+ parser.parsePlan("CALL
catalog.system.mixed_function(arg1 => 1, 'test')");
+ assertThat(JavaConverters.seqAsJavaList(callStatement.name()))
+ .isEqualTo(Arrays.asList("catalog", "system",
"mixed_function"));
+ assertThat(callStatement.args().size()).isEqualTo(2);
+ assertArgument(callStatement, 0, "arg1", 1, DataTypes.IntegerType);
+ assertArgument(callStatement, 1, "test", DataTypes.StringType);
+ }
+
+ @Test
+ public void testCallWithParseException() {
+ assertThatThrownBy(() -> parser.parsePlan("CALL catalog.system func
abc"))
+ .isInstanceOf(PaimonParseException.class)
+ .hasMessageContaining("missing '(' at 'func'");
+ }
+
+ private void assertArgument(
+ CallStatement call, int index, Object expectedValue, DataType
expectedType) {
+ assertArgument(call, index, null, expectedValue, expectedType);
+ }
+
+ private void assertArgument(
+ CallStatement callStatement,
+ int index,
+ String expectedName,
+ Object expectedValue,
+ DataType expectedType) {
+ if (expectedName == null) {
+ CallArgument callArgument = callStatement.args().apply(index);
+ assertCast(callArgument, PositionalArgument.class);
+ } else {
+ NamedArgument namedArgument =
+ assertCast(callStatement.args().apply(index),
NamedArgument.class);
+ assertThat(namedArgument.name()).isEqualTo(expectedName);
+ }
+ assertThat(callStatement.args().apply(index).expr())
+ .isEqualTo(Literal$.MODULE$.create(expectedValue,
expectedType));
+ }
+
+ private <T> T assertCast(Object value, Class<T> expectedClass) {
+ assertThat(value).isInstanceOf(expectedClass);
+ return expectedClass.cast(value);
+ }
+}
diff --git
a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/PaimonSparkTestBase.scala
b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/PaimonSparkTestBase.scala
index 5152b506d..10977ee7d 100644
---
a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/PaimonSparkTestBase.scala
+++
b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/PaimonSparkTestBase.scala
@@ -20,6 +20,7 @@ package org.apache.paimon.spark
import org.apache.paimon.catalog.{Catalog, CatalogContext, CatalogFactory,
Identifier}
import org.apache.paimon.options.Options
import org.apache.paimon.spark.catalog.Catalogs
+import org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions
import org.apache.paimon.spark.sql.WithTableOptions
import org.apache.paimon.table.AbstractFileStoreTable
@@ -45,7 +46,7 @@ class PaimonSparkTestBase extends QueryTest with
SharedSparkSession with WithTab
super.sparkConf
.set("spark.sql.catalog.paimon", classOf[SparkCatalog].getName)
.set("spark.sql.catalog.paimon.warehouse", tempDBDir.getCanonicalPath)
- .set("spark.sql.extensions",
classOf[PaimonSparkSessionExtension].getName)
+ .set("spark.sql.extensions",
classOf[PaimonSparkSessionExtensions].getName)
}
override protected def beforeAll(): Unit = {
diff --git
a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/TableValuedFunctionsTest.scala
b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/TableValuedFunctionsTest.scala
index 16d0d797a..0697d2f27 100644
---
a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/TableValuedFunctionsTest.scala
+++
b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/TableValuedFunctionsTest.scala
@@ -18,7 +18,8 @@
package org.apache.paimon.spark.sql
import org.apache.paimon.WriteMode.CHANGE_LOG
-import org.apache.paimon.spark.{PaimonSparkSessionExtension,
PaimonSparkTestBase}
+import org.apache.paimon.spark.PaimonSparkTestBase
+import org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions
import org.apache.spark.SparkConf
import org.apache.spark.sql.{DataFrame, Row}
@@ -28,7 +29,7 @@ class TableValuedFunctionsTest extends PaimonSparkTestBase {
override protected def sparkConf: SparkConf = {
super.sparkConf
.set("spark.sql.catalog.spark_catalog",
"org.apache.paimon.spark.SparkGenericCatalog")
- .set("spark.sql.extensions",
classOf[PaimonSparkSessionExtension].getName)
+ .set("spark.sql.extensions",
classOf[PaimonSparkSessionExtensions].getName)
}
writeModes.foreach {
diff --git a/pom.xml b/pom.xml
index bf5d808d2..eb2d040ee 100644
--- a/pom.xml
+++ b/pom.xml
@@ -126,6 +126,7 @@ under the License.
<avro.version>1.11.1</avro.version>
<kafka.version>3.2.3</kafka.version>
<scala-maven-plugin.version>3.2.2</scala-maven-plugin.version>
+ <antlr4-maven-plugin.version>4.8</antlr4-maven-plugin.version>
<jsoup.version>1.15.3</jsoup.version>
</properties>
