[spark] Support parser of Spark call procedure command

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

@@ -41,10 +41,34 @@
             <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>
+                    <listener>true</listener>
+                    <sourceDirectory>src/main/antlr4</sourceDirectory>
+                </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/FlussSqlExtension.g4

@@ -0,0 +1,230 @@
+/*
+ * 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 FlussSqlExtension;
+
+@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
+    : booleanExpression
+    ;
+
+booleanExpression
+    : predicated
+    | NOT booleanExpression
+    | EXISTS '(' query ')'
+    | booleanExpression AND booleanExpression
+    | booleanExpression OR booleanExpression
+    ;
+
+predicated
+    : valueExpression
+    ;
+
+valueExpression
+    : primaryExpression
+    ;
+
+primaryExpression
+    : constant                                                        #constantDefault
+    | functionName '(' (expression (',' expression)*)? ')'            #functionCall
+    | '(' expression ')'                                              #parenthesizedExpression
+    ;
+
+functionName
+    : multipartIdentifier
+    | identifier
+    ;
+
+query
+    : .+?
+    ;
+
+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 | NOT | AND | OR | EXISTS
+    ;
+
+// Keywords
+CALL: 'CALL';
+TRUE: 'TRUE';
+FALSE: 'FALSE';
+NOT: 'NOT';
+AND: 'AND';
+OR: 'OR';
+EXISTS: 'EXISTS';
+
+// 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/FlussSparkSessionExtensions.scala

@@ -0,0 +1,39 @@
+/*
+ * 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.fluss.spark
+
+import org.apache.fluss.spark.catalyst.analysis.FlussProcedureResolver
+import org.apache.fluss.spark.execution.FlussStrategy
+
+import org.apache.spark.sql.SparkSessionExtensions
+import org.apache.spark.sql.catalyst.parser.FlussSparkSqlParser
+
+/** Spark session extensions for Fluss. */
+class FlussSparkSessionExtensions extends (SparkSessionExtensions => Unit) {
+
+  override def apply(extensions: SparkSessionExtensions): Unit = {
+    // parser extensions
+    extensions.injectParser { case (_, parser) => new FlussSparkSqlParser(parser) }
+
+    // analyzer extensions
+    extensions.injectResolutionRule(spark => FlussProcedureResolver(spark))
+
+    // planner extensions
+    extensions.injectPlannerStrategy(spark => FlussStrategy(spark))
+  }
+}

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.catalog.{SupportsFlussNamespaces, SupportsProcedures, WithFlussAdmin}
+import org.apache.fluss.spark.exception.NoSuchProcedureException
+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 SupportsProcedures {
 
   private var catalogName: String = "fluss"
+  private val SYSTEM_NAMESPACE = "sys"
 
   override def listTables(namespace: Array[String]): Array[Identifier] = {
     doNamespaceOperator(namespace) {
@@ -116,6 +123,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(s"Procedure not found: $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)

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

@@ -0,0 +1,40 @@
+/*
+ * 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.fluss.spark
+
+import org.apache.fluss.spark.procedure.{GetClusterConfigsProcedure, ProcedureBuilder}
+
+import java.util.Locale
+
+object SparkProcedures {
+
+  private val BUILDERS: Map[String, () => ProcedureBuilder] = initProcedureBuilders()
+
+  def newBuilder(name: String): ProcedureBuilder = {
+    val builderSupplier = BUILDERS.get(name.toLowerCase(Locale.ROOT))
+    builderSupplier.map(_()).orNull
+  }
+
+  def names(): Set[String] = BUILDERS.keySet
+
+  private def initProcedureBuilders(): Map[String, () => ProcedureBuilder] = {
+    Map(
+      "get_cluster_configs" -> (() => GetClusterConfigsProcedure.builder())
+    )
+  }
+}