[spark] Support parser of Spark call procedure command #2408

XuQianJin-Stars · 2026-01-18T06:27:39Z

Purpose

Linked issue: close #2406

This PR introduces the parser and execution framework for Spark's CALL procedure command, allowing users to invoke stored procedures using SQL syntax like CALL sys.procedure_name(args). This provides a foundation for implementing various administrative and maintenance operations.

All implementations are in Scala for better integration with Spark's ecosystem.

Brief change log

Core Framework (Scala):

Added Procedure trait in fluss-spark-common/src/main/scala/org/apache/fluss/spark/procedure/Procedure.scala
Added ProcedureParameter trait and case class implementation for parameter definitions in ProcedureParameter.scala
Added BaseProcedure abstract class providing common utilities in BaseProcedure.scala
Added ProcedureBuilder trait for procedure instantiation in ProcedureBuilder.scala
Added ProcedureCatalog trait for catalog integration in catalog/ProcedureCatalog.scala

Parser & SQL Extensions:

Created ANTLR grammar FlussSqlExtensions.g4 for CALL statement syntax
Implemented FlussSparkSqlParser extending Spark's ParserInterface
Implemented FlussSqlExtensionsAstBuilder to convert ANTLR parse tree to logical plans
Added custom Origin and CurrentOrigin handling for source position tracking
Added Maven ANTLR4 plugin configuration to fluss-spark-common/pom.xml

Logical & Physical Plans:

Created FlussCallStatement (unresolved) and FlussCallCommand (resolved) logical plan nodes
Created FlussCallArgument, FlussPositionalArgument, and FlussNamedArgument for argument representation
Implemented CallProcedureExec physical plan node for execution

Analysis & Execution:

Implemented FlussProcedureResolver analyzer rule for procedure resolution and validation
Implemented FlussStrategy planner strategy to inject CallProcedureExec
Created FlussSparkSessionExtensions to register all custom components

Catalog Integration:

Modified SparkCatalog to implement ProcedureCatalog
Updated FlussSparkTestBase to enable SQL extensions in test environment

Procedure Registry (Scala):

Created SparkProcedures object in fluss-spark-common/src/main/scala/org/apache/fluss/spark/SparkProcedures.scala for managing procedure builders
Added NoSuchProcedureException class in analysis/NoSuchProcedureException.scala for error handling

Example Implementation (Scala):

Implemented CompactProcedure in procedure/CompactProcedure.scala as a sample procedure (skeleton implementation)

Documentation & Tests (Scala):

Added PROCEDURES.md documenting the new feature
Added CallStatementParserTest.scala in fluss-spark-ut/src/test/scala with comprehensive parser tests

Tests

Unit Tests (ScalaTest):

CallStatementParserTest: Tests parsing of CALL statements
- testCallWithBackticks: Tests backtick-quoted identifiers
- testCallWithNamedArguments: Tests named argument syntax
- testCallWithPositionalArguments: Tests positional arguments with various data types
- testCallWithMixedArguments: Tests mixed named and positional arguments
- testCallSimpleProcedure: Tests simple procedure call

All existing tests in fluss-spark-ut module pass successfully.

API and Format

New Public APIs (Scala):

Procedure trait: Defines contract for stored procedures
ProcedureParameter trait: Defines procedure parameters with companion object factory methods
ProcedureCatalog trait: Extends Spark's TableCatalog with procedure loading capability

Modified APIs:

SparkCatalog now implements ProcedureCatalog trait

No changes to storage format.

Documentation

New feature introduced: Spark CALL procedure command support

Documentation added:

fluss-spark/PROCEDURES.md: Comprehensive guide on using the CALL procedure feature
- Syntax examples
- Available procedures
- Usage guidelines
- Extension points for custom procedures

Configuration required:
Users need to configure Spark session with:
spark.sql.extensions = org.apache.fluss.spark.extensions.FlussSparkSessionExtensions

XuQianJin-Stars · 2026-01-18T12:45:06Z

@wuchong @YannByron hi, Please help review when you got some time.

YannByron · 2026-01-23T15:31:45Z

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

+ * limitations under the License.
+ */
+
+grammar FlussSqlExtensions;


Rename to FlussSqlParser.

YannByron · 2026-01-23T15:35:58Z

...park/fluss-spark-common/src/main/scala/org/apache/fluss/spark/catalog/ProcedureCatalog.scala

+
+import org.apache.spark.sql.connector.catalog.Identifier
+
+trait ProcedureCatalog {


Rename to SupportsProcedures to align spark SupportsNamespaces and SupportsPartitionManagement fashion.

YannByron · 2026-01-23T15:39:16Z

...s-spark-common/src/main/scala/org/apache/fluss/spark/analysis/NoSuchProcedureException.scala

+ * limitations under the License.
+ */
+
+package org.apache.fluss.spark.analysis


Move to org.apache.fluss.exception in fluss-common.

YannByron · 2026-01-23T15:42:06Z

...rk-common/src/main/scala/org/apache/fluss/spark/extensions/FlussSparkSessionExtensions.scala

+ * limitations under the License.
+ */
+
+package org.apache.fluss.spark.extensions


I personally want to put this in org.apache.fluss.spark.

YannByron · 2026-01-23T15:47:28Z

...ain/scala/org/apache/spark/sql/catalyst/parser/extensions/FlussSqlExtensionsAstBuilder.scala

+ * @param delegate
+ *   The extension parser.
+ */
+class FlussSqlExtensionsAstBuilder(delegate: ParserInterface)


Rename to FlussSqlAstBuilder and move it in org.apache.spark.sql.catalyst.parser

YannByron · 2026-01-23T15:48:49Z

...s-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/FlussSparkSqlParser.scala

+ * @param delegate
+ *   The main Spark SQL parser.
+ */
+class FlussSparkSqlParser(delegate: ParserInterface) extends ParserInterface {


Move it in org.apache.spark.sql.catalyst.parser.

YannByron · 2026-01-23T15:51:25Z

.../fluss-spark-common/src/main/scala/org/apache/fluss/spark/procedure/ProcedureParameter.scala

+    ProcedureParameterImpl(name, dataType, isRequired = false)
+}
+
+private case class ProcedureParameterImpl(


Why we have to define a ProcedureParameter trait if only ProcedureParameterImpl extends it.

YannByron · 2026-01-23T15:59:54Z

...rk/fluss-spark-common/src/main/scala/org/apache/fluss/spark/procedure/CompactProcedure.scala

+    val sparkTable = loadSparkTable(tableIdent)
+
+    try {
+      val tablePath = toTablePath(tableIdent)


So this procedure do nothing for now, then throw exception, explain it will be supported soon and link a issue.

YannByron · 2026-01-29T15:22:49Z

+1

wuchong

Thanks, @XuQianJin-Stars!

It looks like this pull request doesn’t yet deliver a complete feature. I suggest rounding it out by implementing a few basic procedures and adding the corresponding documentation.

Additionally, please add necessary Javadoc comments for the class and its methods to improve code clarity and maintainability.

wuchong · 2026-02-01T13:21:29Z

fluss-spark/PROCEDURES.md

@@ -0,0 +1,96 @@
+# Fluss Spark Procedures


This is not the appropriate place for documentation—please move it to the website/ directory.

Specifically:

Create a new section titled “Engine Spark” under “Engine Flink” in the documentation sidebar.

Within “Engine Spark,” add a page named “Procedures”.

Please follow the structure and style of the Flink Procedures page as a reference. The Spark Procedures page should include, for each supported procedure:

Syntax

Parameters

Return value(s)

Example usage

Additionally, ensure that all procedure names are listed in the right-side table of contents (TOC) for easy navigation.

wuchong · 2026-02-01T13:49:28Z

...rk/fluss-spark-common/src/main/scala/org/apache/fluss/spark/procedure/CompactProcedure.scala

+import org.apache.spark.sql.connector.catalog.TableCatalog
+import org.apache.spark.sql.types.{DataTypes, Metadata, StructField, StructType}
+
+class CompactProcedure(tableCatalog: TableCatalog) extends BaseProcedure(tableCatalog) {


Fluss doesn't support compact, and will not support it in the future. So providing an empty compact procedure looks strange to users, and will be backward in-compatible when we removing it.

Could you remove this in the PR, and introduce xxx_cluster_configs as first procedures, like Flink procedures https://fluss.apache.org/docs/next/engine-flink/procedures/#get_cluster_configs?

wuchong · 2026-02-01T13:51:57Z

...k/fluss-spark-common/src/main/scala/org/apache/fluss/spark/execution/CallProcedureExec.scala

+
+/** Physical plan node for executing a stored procedure. */
+case class CallProcedureExec(output: Seq[Attribute], procedure: Procedure, args: Seq[Expression])
+  extends SparkPlan {


It seems Paimon implements procedure exec by extending Spark LeafV2CommandExec which seems much simpler (not relying on RDD). Is there any reason for us to extending SparkPlan?

wuchong · 2026-02-01T13:55:56Z

fluss-common/src/main/java/org/apache/fluss/exception/NoSuchProcedureException.java

+ * @since 0.9
+ */
+@PublicEvolving
+public class NoSuchProcedureException extends ApiException {


ApiException is used for the communitcation between Fluss server and clients. However, NoSuchProcedureException is only a client-side exception in spark connector. I suggest to move it to package org.apache.fluss.spark.exception in fluss-spark-common module.

XuQianJin-Stars · 2026-02-02T07:20:36Z

@wuchong @YannByron Hi, i already updated the pr. Please help review when you got some time.

…to source code

wuchong

@XuQianJin-Stars I make some changes to the pull request.

renamed the parser to FlussSqlExtension like how Iceberg and Paimon name it.
added the generated classes into source code, so IDE can recognize the generated classes
improved some code and tests

XuQianJin-Stars force-pushed the feature/issue-2406-procedure branch from ff9735b to b9d21ed Compare January 18, 2026 10:31

YannByron reviewed Jan 23, 2026

View reviewed changes

XuQianJin-Stars force-pushed the feature/issue-2406-procedure branch from 0b2788c to ef4c33f Compare January 24, 2026 03:03

XuQianJin-Stars requested a review from YannByron January 26, 2026 10:41

XuQianJin-Stars force-pushed the feature/issue-2406-procedure branch from 113ca9d to e119c14 Compare January 28, 2026 05:30

wuchong reviewed Feb 1, 2026

View reviewed changes

XuQianJin-Stars force-pushed the feature/issue-2406-procedure branch from 193b107 to 16533c1 Compare February 2, 2026 05:24

XuQianJin-Stars requested a review from wuchong February 2, 2026 07:18

XuQianJin-Stars and others added 2 commits February 3, 2026 15:07

[spark] Support Spark CALL procedure command framework (apache#2408)

29b6fed

[spark] Rename parser to FlussSqlExtension and add generated class in…

59dc299

…to source code

wuchong approved these changes Feb 3, 2026

View reviewed changes

wuchong force-pushed the feature/issue-2406-procedure branch from cc2dfb3 to 59dc299 Compare February 3, 2026 07:11


		import org.apache.spark.sql.connector.catalog.Identifier

		trait ProcedureCatalog {

[spark] Support parser of Spark call procedure command #2408

Are you sure you want to change the base?

[spark] Support parser of Spark call procedure command #2408

Uh oh!

Conversation

XuQianJin-Stars commented Jan 18, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Purpose

Brief change log

Tests

API and Format

Documentation

Uh oh!

XuQianJin-Stars commented Jan 18, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

YannByron commented Jan 29, 2026

Uh oh!

wuchong left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

XuQianJin-Stars commented Feb 2, 2026

Uh oh!

wuchong left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

XuQianJin-Stars commented Jan 18, 2026 •

edited

Loading