[SPARK-52110][SDP][SQL] Implement SQL syntax support for pipelines

### What changes were proposed in this pull request?
Introduces SQL syntax support needed to implement Spark Declarative Pipelines ([SPIP](https://issues.apache.org/jira/browse/SPARK-51727)).

Spark Declarative Pipelines is a spark sub-project that allows users to declaratively define data processing pipelines, using Python (Pyspark) and/or SQL (Spark SQL). This PR implements the necessary Spark SQL syntax and commands to support the pipelines' SQL interface.

## Creating Materialized Views
```
CREATE MATERIALIZED VIEW [materialized view identifier] [column definitions]? [create table options]? AS [query]
```

Syntax to create a materialized view with identifier `materialized view identifier`, optionally specified column definitions (i.e schema) `column definitions`, optionally specified table options `create table options`, and the query that defines the data that will populate this materialized view `query`.

This syntax is parsed into the `CreateMaterializedViewAsSelect` logical plan, which **cannot** be executed by spark's query engine, and will throw an exception if attempted. Instead, the intention is for the logical plan to be _interpreted_ by the pipelines module, in a future PR.

## Creating Streaming Tables
```
CREATE STREAMING TABLE [streaming table identifier] [column definitions]? [create table options]? (AS [query])?
```

Syntax to create a streaming table with identifier `streaming table identifier`, optionally specified column definitions (i.e schema) `column definitions`, optionally specified table options `create table options`, and optionally the query that defines the data that will populate this streaming table `query`.

It is allowed for a streaming table to be defined without a query, as streaming tables' data can be backed by standalone flows (see below). During a pipeline execution, it will be validated that a streaming table has at least one standalone flow writing to the table, if no query is specified in the create statement itself.

This syntax is parsed into either the `CreateStreamingTableAsSelect` or `CreateStreamingTable` logical plan, which **cannot** be executed by spark's query engine, and will throw an exception if attempted. Instead, the intention is for the logical plan to be _interpreted_ by the pipelines module, in a future PR.

## Creating Append Flows
```
CREATE FLOW [flow identifier] AS INSERT INTO BY NAME [flow destination identifier] [query]
```

Syntax to create an append (insert into) flow with identifier `flow identifier`, that inserts data defined by `query` into some destination dataset with identifier `flow destination identifier`.

This syntax is parsed into the `CreateFlowCommand` logical plan, which **cannot** be executed by spark's query engine, and will throw an exception if attempted. Instead, the intention is for the logical plan to be _interpreted_ by the pipelines module, in a future PR.

## Annotating Logical Plans as Streaming
```
STREAM ([relation | TVF])
STREAM [relation | TVF]
```

Syntax to mark an [unresolved] relation or table valued function as streaming in SQL. Since streaming tables must read from streaming sources in their query, the `STREAM` keyword assists with marking batch data sources as streaming, so that streaming tables can treat them as streaming data sources.

### Why are the changes needed?
These changes introduce SQL API for the pipelines project.

### Does this PR introduce _any_ user-facing change?
Yes, this change introduces new Spark SQL syntax that can be parsed by the Spark SQL parser, but cannot be executed by Spark's query engine - the corresponding logical plans will instead be interpreted by the Spark Declarative Pipelines project.

### How was this patch tested?
`org.apache.spark.sql.catalyst.analysis.StreamRelationSuite`
`org.apache.spark.sql.execution.command.CreateFlowCommandSuite`
`org.apache.spark.sql.execution.command.CreateStreamingTableAsSelectParserSuite`
`org.apache.spark.sql.execution.command.CreateMaterializedViewAsSelectParserSuite`
`org.apache.spark.sql.execution.command.v2.CreateStreamingTableParserSuite`

### Was this patch authored or co-authored using generative AI tooling?

Closes #50875 from AnishMahto/add_spark_declarative_pipelines_sql_syntax_support.

Lead-authored-by: anishm-db <anish.mahto@databricks.com>
Co-authored-by: Anish Mahto <anish.mahto@databricks.com>
Co-authored-by: AnishMahto <anish.mahto99@gmail.com>
Co-authored-by: Gengliang Wang <gengliang@apache.org>
Signed-off-by: Gengliang Wang <gengliang@apache.org>

Author: 3 people

common/utils/src/main/resources/error/error-conditions.json

@@ -5876,6 +5876,16 @@
           "CONTINUE exception handler is not supported. Use EXIT handler."
         ]
       },
+      "CREATE_FLOW_QUERY_EXECUTION" : {
+        "message" : [
+          "Direct execution of a CREATE FLOW statement is not supported. To create a flow, create and execute a pipeline with a SQL file containing your statement attached in the pipeline spec."
+        ]
+      },
+      "CREATE_PIPELINE_DATASET_QUERY_EXECUTION" : {
+        "message" : [
+          "Direct execution of a CREATE ... <pipelineDatasetType> query. To create a pipeline dataset, create and execute a pipeline with a SQL file containing your query attached in the pipeline definition."
+        ]
+      },
       "DESC_TABLE_COLUMN_JSON" : {
         "message" : [
           "DESC TABLE COLUMN AS JSON not supported for individual columns."
@@ -6349,6 +6359,12 @@
     ],
     "sqlState" : "0A000"
   },
+  "UNSUPPORTED_STREAMING_TABLE_VALUED_FUNCTION" : {
+    "message" : [
+      "The function <funcName> does not support streaming. Please remove the STREAM keyword"
+    ],
+    "sqlState" : "42000"
+  },
   "UNSUPPORTED_SUBQUERY_EXPRESSION_CATEGORY" : {
     "message" : [
       "Unsupported subquery expression:"

docs/sql-ref-ansi-compliance.md

@@ -532,6 +532,7 @@ Below is a list of all the keywords in Spark SQL.
 |FILEFORMAT|non-reserved|non-reserved|non-reserved|
 |FIRST|non-reserved|non-reserved|non-reserved|
 |FLOAT|non-reserved|non-reserved|reserved|
+|FLOW|non-reserved|non-reserved|non-reserved|
 |FOLLOWING|non-reserved|non-reserved|non-reserved|
 |FOR|reserved|non-reserved|reserved|
 |FOREIGN|reserved|non-reserved|reserved|
@@ -604,6 +605,7 @@ Below is a list of all the keywords in Spark SQL.
 |MACRO|non-reserved|non-reserved|non-reserved|
 |MAP|non-reserved|non-reserved|non-reserved|
 |MATCHED|non-reserved|non-reserved|non-reserved|
+|MATERIALIZED|non-reserved|non-reserved|non-reserved|
 |MAX|non-reserved|non-reserved|non-reserved|
 |MERGE|non-reserved|non-reserved|non-reserved|
 |MICROSECOND|non-reserved|non-reserved|non-reserved|
@@ -723,6 +725,8 @@ Below is a list of all the keywords in Spark SQL.
 |STATISTICS|non-reserved|non-reserved|non-reserved|
 |STORED|non-reserved|non-reserved|non-reserved|
 |STRATIFY|non-reserved|non-reserved|non-reserved|
+|STREAM|non-reserved|non-reserved|non-reserved|
+|STREAMING|non-reserved|non-reserved|non-reserved|
 |STRING|non-reserved|non-reserved|non-reserved|
 |STRUCT|non-reserved|non-reserved|non-reserved|
 |SUBSTR|non-reserved|non-reserved|non-reserved|

sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseLexer.g4

@@ -245,6 +245,7 @@ FILTER: 'FILTER';
 FILEFORMAT: 'FILEFORMAT';
 FIRST: 'FIRST';
 FLOAT: 'FLOAT';
+FLOW: 'FLOW';
 FOLLOWING: 'FOLLOWING';
 FOR: 'FOR';
 FOREIGN: 'FOREIGN';
@@ -318,6 +319,7 @@ LOOP: 'LOOP';
 MACRO: 'MACRO';
 MAP: 'MAP' {incComplexTypeLevelCounter();};
 MATCHED: 'MATCHED';
+MATERIALIZED: 'MATERIALIZED';
 MAX: 'MAX';
 MERGE: 'MERGE';
 MICROSECOND: 'MICROSECOND';
@@ -436,6 +438,8 @@ START: 'START';
 STATISTICS: 'STATISTICS';
 STORED: 'STORED';
 STRATIFY: 'STRATIFY';
+STREAM: 'STREAM';
+STREAMING: 'STREAMING';
 STRING: 'STRING';
 STRUCT: 'STRUCT' {incComplexTypeLevelCounter();};
 SUBSTR: 'SUBSTR';

sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4

@@ -349,6 +349,30 @@ statement
         (functionArgument (COMMA functionArgument)*)?
         RIGHT_PAREN                                                    #call
     | unsupportedHiveNativeCommands .*?                                #failNativeCommand
+    | createPipelineDatasetHeader (LEFT_PAREN tableElementList? RIGHT_PAREN)? tableProvider?
+        createTableClauses
+        (AS query)?                                                    #createPipelineDataset
+    | createPipelineFlowHeader insertInto query                        #createPipelineInsertIntoFlow
+    ;
+
+materializedView
+    : MATERIALIZED VIEW
+    ;
+
+streamingTable
+    : STREAMING TABLE
+    ;
+
+createPipelineDatasetHeader
+    : CREATE
+      (materializedView | streamingTable)
+      (IF errorCapturingNot EXISTS)?
+      identifierReference
+    ;
+
+streamRelationPrimary
+    : STREAM multipartIdentifier tableAlias optionsClause?                           #streamTableName
+    | STREAM LEFT_PAREN multipartIdentifier RIGHT_PAREN tableAlias optionsClause?    #streamTableName
     ;
 
 setResetStatement
@@ -514,6 +538,10 @@ partitionVal
     | identifier EQ DEFAULT
     ;
 
+createPipelineFlowHeader
+    : CREATE FLOW flowName=identifierReference (commentSpec)? AS
+    ;
+
 namespace
     : NAMESPACE
     | DATABASE
@@ -973,7 +1001,8 @@ identifierComment
     ;
 
 relationPrimary
-    : identifierReference temporalClause?
+    : streamRelationPrimary                                 #streamRelation
+    | identifierReference temporalClause?
       optionsClause? sample? tableAlias                     #tableName
     | LEFT_PAREN query RIGHT_PAREN sample? tableAlias       #aliasedQuery
     | LEFT_PAREN relation RIGHT_PAREN sample? tableAlias    #aliasedRelation
@@ -1769,6 +1798,7 @@ ansiNonReserved
     | FILEFORMAT
     | FIRST
     | FLOAT
+    | FLOW
     | FOLLOWING
     | FORMAT
     | FORMATTED
@@ -1825,6 +1855,7 @@ ansiNonReserved
     | MACRO
     | MAP
     | MATCHED
+    | MATERIALIZED
     | MAX
     | MERGE
     | MICROSECOND
@@ -1927,6 +1958,8 @@ ansiNonReserved
     | STRATIFY
     | STRING
     | STRUCT
+    | STREAM
+    | STREAMING
     | SUBSTR
     | SUBSTRING
     | SYNC
@@ -2139,6 +2172,7 @@ nonReserved
     | FILEFORMAT
     | FIRST
     | FLOAT
+    | FLOW
     | FOLLOWING
     | FOR
     | FOREIGN
@@ -2206,6 +2240,7 @@ nonReserved
     | MACRO
     | MAP
     | MATCHED
+    | MATERIALIZED
     | MAX
     | MERGE
     | MICROSECOND
@@ -2319,6 +2354,8 @@ nonReserved
     | STATISTICS
     | STORED
     | STRATIFY
+    | STREAM
+    | STREAMING
     | STRING
     | STRUCT
     | SUBSTR

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationChecker.scala

@@ -40,7 +40,8 @@ object UnsupportedOperationChecker extends Logging {
   def checkForBatch(plan: LogicalPlan): Unit = {
     plan.foreachUp {
       case p if p.isStreaming =>
-        throwError("Queries with streaming sources must be executed with writeStream.start()")(p)
+        throwError("Queries with streaming sources must be executed with writeStream.start(), or " +
+          "from a streaming table or flow definition within a Spark Declarative Pipeline.")(p)
 
       case d: DeduplicateWithinWatermark =>
         throwError("dropDuplicatesWithinWatermark is not supported with batch " +

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala

@@ -862,7 +862,7 @@ class AstBuilder extends DataTypeAstBuilder
    * }}}
    * operation to logical plan
    */
-  private def withInsertInto(
+  protected def withInsertInto(
       ctx: InsertIntoContext,
       query: LogicalPlan): LogicalPlan = withOrigin(ctx) {
     ctx match {
@@ -878,7 +878,8 @@ class AstBuilder extends DataTypeAstBuilder
               ctx = insertParams.relationCtx,
               ident = ident,
               optionsClause = insertParams.options,
-              writePrivileges = Seq(TableWritePrivilege.INSERT)),
+              writePrivileges = Seq(TableWritePrivilege.INSERT),
+              isStreaming = false),
             partitionSpec = insertParams.partitionSpec,
             userSpecifiedCols = insertParams.userSpecifiedCols,
             query = otherPlans.head,
@@ -894,7 +895,8 @@ class AstBuilder extends DataTypeAstBuilder
               ctx = insertParams.relationCtx,
               ident = ident,
               optionsClause = insertParams.options,
-              writePrivileges = Seq(TableWritePrivilege.INSERT, TableWritePrivilege.DELETE)),
+              writePrivileges = Seq(TableWritePrivilege.INSERT, TableWritePrivilege.DELETE),
+              isStreaming = false),
             partitionSpec = insertParams.partitionSpec,
             userSpecifiedCols = insertParams.userSpecifiedCols,
             query = otherPlans.head,
@@ -907,7 +909,7 @@ class AstBuilder extends DataTypeAstBuilder
         withIdentClause(ctx.identifierReference, Seq(query), (ident, otherPlans) => {
           OverwriteByExpression.byPosition(
             createUnresolvedRelation(ctx.identifierReference, ident, options,
-              Seq(TableWritePrivilege.INSERT, TableWritePrivilege.DELETE)),
+              Seq(TableWritePrivilege.INSERT, TableWritePrivilege.DELETE), isStreaming = false),
             otherPlans.head,
             expression(ctx.whereClause().booleanExpression()))
         })
@@ -2385,6 +2387,17 @@ class AstBuilder extends DataTypeAstBuilder
       })
   }
 
+  override def visitStreamTableName(ctx: StreamTableNameContext): LogicalPlan = {
+    val ident = visitMultipartIdentifier(ctx.multipartIdentifier)
+    val tableStreamingRelation = createUnresolvedRelation(
+      ctx = ctx,
+      ident = ident,
+      optionsClause = Option(ctx.optionsClause),
+      writePrivileges = Seq.empty,
+      isStreaming = true)
+    mayApplyAliasPlan(ctx.tableAlias, tableStreamingRelation)
+  }
+
   /**
    * Create an inline table (a virtual table in Hive parlance).
    */
@@ -3635,9 +3648,10 @@ class AstBuilder extends DataTypeAstBuilder
       ctx: ParserRuleContext,
       ident: Seq[String],
       optionsClause: Option[OptionsClauseContext],
-      writePrivileges: Seq[TableWritePrivilege]): UnresolvedRelation = withOrigin(ctx) {
+      writePrivileges: Seq[TableWritePrivilege],
+      isStreaming: Boolean): UnresolvedRelation = withOrigin(ctx) {
     val options = resolveOptions(optionsClause)
-    val relation = new UnresolvedRelation(ident, options, isStreaming = false)
+    val relation = new UnresolvedRelation(ident, options, isStreaming)
     relation.requireWritePrivileges(writePrivileges)
   }
 
@@ -4791,7 +4805,7 @@ class AstBuilder extends DataTypeAstBuilder
     (fileFormatSerdeInfo ++ rowFormatSerdeInfo).reduceLeftOption((l, r) => l.merge(r))
   }
 
-  private def partitionExpressions(
+  protected[sql] def partitionExpressions(
       partTransforms: Seq[Transform],
       partCols: Seq[ColumnDefinition],
       ctx: ParserRuleContext): Seq[Transform] = {
@@ -5779,7 +5793,12 @@ class AstBuilder extends DataTypeAstBuilder
         CacheTableAsSelect(ident.head, children.head, source(ctx.query()), isLazy, options)
       } else {
         CacheTable(
-          createUnresolvedRelation(ctx.identifierReference, ident, None, writePrivileges = Nil),
+          createUnresolvedRelation(
+            ctx.identifierReference,
+            ident,
+            None,
+            writePrivileges = Nil,
+            isStreaming = false),
           ident, isLazy, options)
       }
     })