[spark] Support parser of Spark call procedure command

fluss-spark/PROCEDURES.md

@@ -0,0 +1,96 @@
+# Fluss Spark Procedures
+
+This document describes the stored procedures available in Fluss for Spark.
+
+## Overview
+
+Fluss provides stored procedures to perform administrative and management operations through Spark SQL. All procedures are located in the `sys` namespace and can be invoked using the `CALL` statement.
+
+## Configuration
+
+To enable Fluss procedures in Spark, you need to configure the Spark session extensions:
+
+```scala
+spark.conf.set("spark.sql.extensions", "org.apache.fluss.spark.extensions.FlussSparkSessionExtensions")
+```
+
+Or in `spark-defaults.conf`:
+
+```properties
+spark.sql.extensions=org.apache.fluss.spark.extensions.FlussSparkSessionExtensions
+```
+
+## Syntax
+
+The general syntax for calling a procedure is:
+
+```sql
+CALL [catalog_name.]sys.procedure_name(
+  parameter_name => 'value',
+  another_parameter => 'value'
+)
+```
+
+### Argument Passing
+
+Procedures support two ways to pass arguments:
+
+1. **Named Arguments** (recommended):
+   ```sql
+   CALL catalog.sys.procedure_name(parameter => 'value')
+   ```
+
+2. **Positional Arguments**:
+   ```sql
+   CALL catalog.sys.procedure_name('value')
+   ```
+
+Note: You cannot mix named and positional arguments in a single procedure call.
+
+## Available Procedures
+
+Currently, no procedures are implemented in this PR. This section will be updated when procedures are added.
+
+## Error Handling
+
+Procedures will throw exceptions in the following cases:
+
+- **Missing Required Parameters**: If a required parameter is not provided
+- **Invalid Table Name**: If the specified table does not exist
+- **Type Mismatch**: If a parameter value cannot be converted to the expected type
+- **Permission Denied**: If the user does not have permission to perform the operation
+
+## Examples
+
+### Basic Usage
+
+```scala
+// Start Spark with Fluss extensions
+val spark = SparkSession.builder()
+  .config("spark.sql.extensions", "org.apache.fluss.spark.extensions.FlussSparkSessionExtensions")
+  .config("spark.sql.catalog.fluss_catalog", "org.apache.fluss.spark.SparkCatalog")
+  .config("spark.sql.catalog.fluss_catalog.bootstrap.servers", "localhost:9092")
+  .getOrCreate()
+
+// Create a table
+spark.sql("""
+  CREATE TABLE fluss_catalog.my_db.my_table (
+    id INT,
+    name STRING,
+    age INT
+  ) USING fluss
+""")
+
+// Procedures will be added here when implemented
+```
+
+## Implementation Notes
+
+- Procedures are executed synchronously and return results immediately
+- The `sys` namespace is reserved for system procedures
+- Custom procedures can be added by implementing the `Procedure` interface
+
+## See Also
+
+- [Fluss Spark Connector Documentation](../spark-connector.md)
+- [Fluss Admin API](../admin-api.md)

fluss-spark/fluss-spark-common/pom.xml

@@ -41,10 +41,32 @@
             <artifactId>spark-catalyst_${scala.binary.version}</artifactId>
             <version>${spark.version}</version>
         </dependency>
+
+        <dependency>
+            <groupId>org.antlr</groupId>
+            <artifactId>antlr4-runtime</artifactId>
+            <version>4.9.3</version>
+        </dependency>
     </dependencies>
 
     <build>
         <plugins>
+            <plugin>
+                <groupId>org.antlr</groupId>
+                <artifactId>antlr4-maven-plugin</artifactId>
+                <version>4.9.3</version>
+                <executions>
+                    <execution>
+                        <goals>
+                            <goal>antlr4</goal>
+                        </goals>
+                    </execution>
+                </executions>
+                <configuration>
+                    <visitor>true</visitor>
+                </configuration>
+            </plugin>
+
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-shade-plugin</artifactId>

fluss-spark/fluss-spark-common/src/main/antlr4/org.apache.spark.sql.catalyst.parser.extensions/FlussSqlExtensions.g4

@@ -0,0 +1,195 @@
+/*
+ * 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.
+ */
+
+grammar FlussSqlExtensions;
+
+@lexer::members {
+  public boolean isValidDecimal() {
+    int nextChar = _input.LA(1);
+    if (nextChar >= 'A' && nextChar <= 'Z' || nextChar >= '0' && nextChar <= '9' ||
+      nextChar == '_') {
+      return false;
+    } else {
+      return true;
+    }
+  }
+
+  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
+    ;
+
+constant
+    : number                          #numericLiteral
+    | booleanValue                    #booleanLiteral
+    | STRING+                         #stringLiteral
+    | identifier STRING               #typeConstructor
+    ;
+
+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
+    ;
+
+// Keywords
+CALL: 'CALL';
+TRUE: 'TRUE';
+FALSE: 'FALSE';
+
+// Operators
+MINUS: '-';
+
+// Literals
+STRING
+    : '\'' ( ~('\''|'\\') | ('\\' .) )* '\''
+    | '"' ( ~('"'|'\\') | ('\\' .) )* '"'
+    ;
+
+BIGINT_LITERAL
+    : INTEGER_VALUE 'L'
+    ;
+
+SMALLINT_LITERAL
+    : INTEGER_VALUE 'S'
+    ;
+
+TINYINT_LITERAL
+    : INTEGER_VALUE '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]
+    ;
+
+// Whitespace and comments
+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 any characters we didn't match
+UNRECOGNIZED
+    : .
+    ;

fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/SparkCatalog.scala

@@ -19,7 +19,9 @@ package org.apache.fluss.spark
 
 import org.apache.fluss.exception.{DatabaseNotExistException, TableAlreadyExistException, TableNotExistException}
 import org.apache.fluss.metadata.TablePath
-import org.apache.fluss.spark.catalog.{SupportsFlussNamespaces, WithFlussAdmin}
+import org.apache.fluss.spark.analysis.NoSuchProcedureException
+import org.apache.fluss.spark.catalog.{ProcedureCatalog, SupportsFlussNamespaces, WithFlussAdmin}
+import org.apache.fluss.spark.procedure.{Procedure, ProcedureBuilder}
 
 import org.apache.spark.sql.catalyst.analysis.{NoSuchNamespaceException, NoSuchTableException, TableAlreadyExistsException}
 import org.apache.spark.sql.connector.catalog.{Identifier, Table, TableCatalog, TableChange}
@@ -32,9 +34,14 @@ import java.util.concurrent.ExecutionException
 
 import scala.collection.JavaConverters._
 
-class SparkCatalog extends TableCatalog with SupportsFlussNamespaces with WithFlussAdmin {
+class SparkCatalog
+  extends TableCatalog
+  with SupportsFlussNamespaces
+  with WithFlussAdmin
+  with ProcedureCatalog {
 
   private var catalogName: String = "fluss"
+  private val SYSTEM_NAMESPACE = "sys"
 
   override def listTables(namespace: Array[String]): Array[Identifier] = {
     doNamespaceOperator(namespace) {
@@ -104,6 +111,20 @@ class SparkCatalog extends TableCatalog with SupportsFlussNamespaces with WithFl
 
   override def name(): String = catalogName
 
+  override def loadProcedure(identifier: Identifier): Procedure = {
+    if (isSystemNamespace(identifier.namespace)) {
+      val builder: ProcedureBuilder = SparkProcedures.newBuilder(identifier.name)
+      if (builder != null) {
+        return builder.withTableCatalog(this).build()
+      }
+    }
+    throw new NoSuchProcedureException(identifier)
+  }
+
+  private def isSystemNamespace(namespace: Array[String]): Boolean = {
+    namespace.length == 1 && namespace(0).equalsIgnoreCase(SYSTEM_NAMESPACE)
+  }
+
   private def toTablePath(ident: Identifier): TablePath = {
     assert(ident.namespace().length == 1, "Only single namespace is supported")
     TablePath.of(ident.namespace().head, ident.name)