This is an automated email from the ASF dual-hosted git repository.
szehon-ho pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 65396d2f8d2f [SPARK-57923][SQL][SDP][TESTS] Add CREATE STREAMING TABLE
parser test coverage
65396d2f8d2f is described below
commit 65396d2f8d2f5ecf81e5f3a252f98fbb53efa848
Author: andreas-neumann_data <[email protected]>
AuthorDate: Thu Jul 9 10:55:33 2026 -0700
[SPARK-57923][SQL][SDP][TESTS] Add CREATE STREAMING TABLE parser test
coverage
### What changes were proposed in this pull request?
This PR adds parser-level test coverage for `CREATE STREAMING TABLE`
clauses that were previously untested. No behavior changes — tests only.
- `CreatePipelineDatasetAsSelectParserSuiteBase`: add a `CLUSTER BY` test.
Because this is the shared base for the AS SELECT parser suites, it runs for
both `CreateStreamingTableAsSelectParserSuite` and
`CreateMaterializedViewAsSelectParserSuite`.
- `CreateStreamingTableParserSuite` (the no-subquery `CREATE STREAMING
TABLE` form): add positive tests for `CLUSTER BY`, `USING`, `DEFAULT
COLLATION`, a column list, and `IF NOT EXISTS`, plus unsupported-clause error
tests (column constraints, bucketing, `OPTIONS`, `LOCATION`, `STORED AS`,
SerDe) mirroring the AS SELECT base suite. A local `toExpectedContext` helper
is used to no-op the query-context check on the error cases.
### Why are the changes needed?
While reviewing the AUTO CDC SQL work (SPARK-56249 / #56419), several
`CREATE STREAMING TABLE` clauses were found to have no parser test coverage. In
particular `CLUSTER BY` (liquid clustering) was untested across all three
command forms, and the no-subquery `CreateStreamingTable` form only had a
single `COMMENT` test. This PR closes those gaps so the parsing of these
clauses is verified and protected against regressions.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Added the tests described above. Ran the affected suites locally; all 47
tests pass:
```
build/sbt 'sql/testOnly
org.apache.spark.sql.execution.command.v2.CreateStreamingTableParserSuite
org.apache.spark.sql.execution.command.v2.CreateStreamingTableAsSelectParserSuite
org.apache.spark.sql.execution.command.v2.CreateMaterializedViewAsSelectParserSuite'
```
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8)
Closes #56985 from anew/streaming-table-test-coverage.
Authored-by: andreas-neumann_data <[email protected]>
Signed-off-by: Szehon Ho <[email protected]>
---
...atePipelineDatasetAsSelectParserSuiteBase.scala | 25 ++--
.../execution/command/v1/CommandSuiteBase.scala | 11 ++
.../v2/CreateStreamingTableParserSuite.scala | 163 ++++++++++++++++++++-
3 files changed, 186 insertions(+), 13 deletions(-)
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/CreatePipelineDatasetAsSelectParserSuiteBase.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/CreatePipelineDatasetAsSelectParserSuiteBase.scala
index 4ba4601ca729..25e8c1d87c13 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/CreatePipelineDatasetAsSelectParserSuiteBase.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/CreatePipelineDatasetAsSelectParserSuiteBase.scala
@@ -17,10 +17,9 @@
package org.apache.spark.sql.execution.command
-import org.apache.spark.QueryContext
import org.apache.spark.sql.catalyst.parser.ParseException
import org.apache.spark.sql.catalyst.plans.logical.{ColumnDefinition,
CreatePipelineDatasetAsSelect}
-import org.apache.spark.sql.connector.expressions.{FieldReference,
IdentityTransform}
+import org.apache.spark.sql.connector.expressions.{ClusterByTransform,
FieldReference, IdentityTransform}
import org.apache.spark.sql.execution.SparkSqlParser
import org.apache.spark.sql.execution.command.v1.CommandSuiteBase
import org.apache.spark.sql.types.{IntegerType, MetadataBuilder, StringType,
StructField, StructType}
@@ -86,6 +85,18 @@ trait CreatePipelineDatasetAsSelectParserSuiteBase extends
CommandSuiteBase {
}
}
+ test("Clustering is correctly parsed") {
+ Seq(
+ (s"CREATE $datasetSqlSyntax table1 CLUSTER BY (a) AS SELECT * FROM
input",
+ Seq(ClusterByTransform(Seq(FieldReference(Seq("a")))))),
+ (s"CREATE $datasetSqlSyntax table1 AS SELECT * FROM input", Seq.empty)
+ ).foreach { case (query, partitioning) =>
+ val plan = parser.parsePlan(query)
+ val cmd = plan.asInstanceOf[CreatePipelineDatasetAsSelect]
+ assert(cmd.partitioning == partitioning)
+ }
+ }
+
test("Location is unsupported") {
val ex = intercept[ParseException] {
parser.parsePlan(
@@ -261,14 +272,4 @@ trait CreatePipelineDatasetAsSelectParserSuiteBase extends
CommandSuiteBase {
assert(cmd.tableSpec.collation == collationOpt)
}
}
-
- /** Return an ExpectedContext that is equivalent to the passed QueryContext.
Used to no-op
- * queryContext checks on error validation */
- def toExpectedContext(actualQueryContext: QueryContext): ExpectedContext =
ExpectedContext(
- objectType = actualQueryContext.objectType(),
- objectName = actualQueryContext.objectName(),
- startIndex = actualQueryContext.startIndex(),
- stopIndex = actualQueryContext.stopIndex(),
- fragment = actualQueryContext.fragment()
- )
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v1/CommandSuiteBase.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v1/CommandSuiteBase.scala
index 24f3ac30d628..5316099dbd84 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v1/CommandSuiteBase.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v1/CommandSuiteBase.scala
@@ -17,6 +17,7 @@
package org.apache.spark.sql.execution.command.v1
+import org.apache.spark.QueryContext
import org.apache.spark.sql.catalyst.catalog.CatalogTypes.TablePartitionSpec
import org.apache.spark.sql.connector.catalog.CatalogManager
import org.apache.spark.sql.test.SharedSparkSession
@@ -48,4 +49,14 @@ trait CommandSuiteBase extends SharedSparkSession {
val location =
information.split("\\r?\\n").filter(_.startsWith("Location:")).head
assert(location.endsWith(expected))
}
+
+ /** Return an ExpectedContext that is equivalent to the passed QueryContext.
Used to no-op
+ * queryContext checks on error validation */
+ def toExpectedContext(actualQueryContext: QueryContext): ExpectedContext =
ExpectedContext(
+ objectType = actualQueryContext.objectType(),
+ objectName = actualQueryContext.objectName(),
+ startIndex = actualQueryContext.startIndex(),
+ stopIndex = actualQueryContext.stopIndex(),
+ fragment = actualQueryContext.fragment()
+ )
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/CreateStreamingTableParserSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/CreateStreamingTableParserSuite.scala
index 8782f8579712..29fb490a81f7 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/CreateStreamingTableParserSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/CreateStreamingTableParserSuite.scala
@@ -18,12 +18,15 @@
package org.apache.spark.sql.execution.command.v2
import org.apache.spark.sql.catalyst.analysis.UnresolvedIdentifier
+import org.apache.spark.sql.catalyst.parser.ParseException
import org.apache.spark.sql.catalyst.plans.logical.{CreateStreamingTable,
TableSpec}
+import org.apache.spark.sql.connector.expressions.{ClusterByTransform,
FieldReference, IdentityTransform}
import org.apache.spark.sql.execution.SparkSqlParser
import org.apache.spark.sql.execution.command.v1.CommandSuiteBase
+import org.apache.spark.sql.types.{IntegerType, StringType}
/**
- * The class contains tests for the `CREATE STREAMING TABLE ... AS ...` command
+ * The class contains tests for the `CREATE STREAMING TABLE ...` command
(without a subquery).
*/
class CreateStreamingTableParserSuite extends CommandSuiteBase {
protected lazy val parser = new SparkSqlParser()
@@ -54,4 +57,162 @@ class CreateStreamingTableParserSuite extends
CommandSuiteBase {
checkAnalysis = false
)
}
+
+ test("CREATE STREAMING TABLE - PARTITIONED BY is correctly parsed") {
+ val cmd = parser
+ .parsePlan("CREATE STREAMING TABLE st PARTITIONED BY (a)")
+ .asInstanceOf[CreateStreamingTable]
+ assert(cmd.partitioning ==
Seq(IdentityTransform(FieldReference(Seq("a")))))
+ }
+
+ test("CREATE STREAMING TABLE - TBLPROPERTIES are correctly parsed") {
+ val cmd = parser
+ .parsePlan("CREATE STREAMING TABLE st TBLPROPERTIES ('key' = 'value',
'num' = '1')")
+ .asInstanceOf[CreateStreamingTable]
+ assert(cmd.tableSpec.properties == Map("key" -> "value", "num" -> "1"))
+ }
+
+ test("CREATE STREAMING TABLE - CLUSTER BY is correctly parsed") {
+ val cmd = parser
+ .parsePlan("CREATE STREAMING TABLE st CLUSTER BY (a)")
+ .asInstanceOf[CreateStreamingTable]
+ assert(cmd.partitioning ==
Seq(ClusterByTransform(Seq(FieldReference(Seq("a"))))))
+ }
+
+ test("CREATE STREAMING TABLE - USING provider is correctly parsed") {
+ val cmd = parser
+ .parsePlan("CREATE STREAMING TABLE st USING parquet")
+ .asInstanceOf[CreateStreamingTable]
+ assert(cmd.tableSpec.provider == Some("parquet"))
+ }
+
+ test("CREATE STREAMING TABLE - DEFAULT COLLATION is correctly parsed") {
+ val cmd = parser
+ .parsePlan("CREATE STREAMING TABLE st DEFAULT COLLATION UTF8_LCASE")
+ .asInstanceOf[CreateStreamingTable]
+ assert(cmd.tableSpec.collation == Some("UTF8_LCASE"))
+ }
+
+ test("CREATE STREAMING TABLE - column list is correctly parsed") {
+ val cmd = parser
+ .parsePlan("CREATE STREAMING TABLE st (id INT, name STRING)")
+ .asInstanceOf[CreateStreamingTable]
+ assert(cmd.columns.map(_.name) == Seq("id", "name"))
+ assert(cmd.columns.head.dataType == IntegerType)
+ assert(cmd.columns(1).dataType == StringType)
+ }
+
+ test("CREATE STREAMING TABLE - IF NOT EXISTS is correctly parsed") {
+ Seq(
+ ("CREATE STREAMING TABLE IF NOT EXISTS st", true),
+ ("CREATE STREAMING TABLE st", false)
+ ).foreach { case (query, ifNotExists) =>
+ val cmd = parser.parsePlan(query).asInstanceOf[CreateStreamingTable]
+ assert(cmd.ifNotExists == ifNotExists)
+ }
+ }
+
+ test("CREATE STREAMING TABLE - PARTITIONED BY, COMMENT, TBLPROPERTIES
combined") {
+ val cmd = parser
+ .parsePlan(
+ """CREATE STREAMING TABLE st
+ |PARTITIONED BY (a)
+ |COMMENT 'my streaming table'
+ |TBLPROPERTIES ('key' = 'value')""".stripMargin)
+ .asInstanceOf[CreateStreamingTable]
+ assert(cmd.partitioning ==
Seq(IdentityTransform(FieldReference(Seq("a")))))
+ assert(cmd.tableSpec.comment == Some("my streaming table"))
+ assert(cmd.tableSpec.properties == Map("key" -> "value"))
+ }
+
+ //
---------------------------------------------------------------------------
+ // Error cases: unsupported table features
+ //
---------------------------------------------------------------------------
+
+ test("CREATE STREAMING TABLE - column constraints are unsupported") {
+ val ex = intercept[ParseException] {
+ parser.parsePlan("CREATE STREAMING TABLE st (id INT PRIMARY KEY)")
+ }
+ checkError(
+ exception = ex,
+ condition = "_LEGACY_ERROR_TEMP_0035",
+ parameters = Map("message" ->
+ ("Pipeline datasets do not currently support column constraints. " +
+ "Please remove any CHECK, UNIQUE, PK, and FK constraints " +
+ "specified on the pipeline dataset.")),
+ queryContext = ex.getQueryContext.map(toExpectedContext)
+ )
+ }
+
+ test("CREATE STREAMING TABLE - bucketing is unsupported") {
+ val ex = intercept[ParseException] {
+ parser.parsePlan("CREATE STREAMING TABLE st CLUSTERED BY (id) INTO 4
BUCKETS")
+ }
+ checkError(
+ exception = ex,
+ condition = "_LEGACY_ERROR_TEMP_0035",
+ parameters = Map("message" ->
+ ("Bucketing is not supported for CREATE STREAMING TABLE statements. " +
+ "Please remove any bucket spec specified in the statement.")),
+ queryContext = ex.getQueryContext.map(toExpectedContext)
+ )
+ }
+
+ test("CREATE STREAMING TABLE - options are unsupported") {
+ val ex = intercept[ParseException] {
+ parser.parsePlan("CREATE STREAMING TABLE st OPTIONS (key = 'value')")
+ }
+ checkError(
+ exception = ex,
+ condition = "_LEGACY_ERROR_TEMP_0035",
+ parameters = Map("message" ->
+ ("Options are not supported for CREATE STREAMING TABLE statements. " +
+ "Please remove any OPTIONS lists specified in the statement.")),
+ queryContext = ex.getQueryContext.map(toExpectedContext)
+ )
+ }
+
+ test("CREATE STREAMING TABLE - location is unsupported") {
+ val ex = intercept[ParseException] {
+ parser.parsePlan("CREATE STREAMING TABLE st LOCATION '/tmp/data'")
+ }
+ checkError(
+ exception = ex,
+ condition = "_LEGACY_ERROR_TEMP_0035",
+ parameters = Map("message" ->
+ ("Specifying location is not supported for CREATE STREAMING TABLE
statements. " +
+ "The storage location for a pipeline dataset is managed by the
pipeline itself.")),
+ queryContext = ex.getQueryContext.map(toExpectedContext)
+ )
+ }
+
+ test("CREATE STREAMING TABLE - STORED AS syntax is unsupported") {
+ val ex = intercept[ParseException] {
+ parser.parsePlan("CREATE STREAMING TABLE st STORED AS TEXTFILE")
+ }
+ checkError(
+ exception = ex,
+ condition = "_LEGACY_ERROR_TEMP_0035",
+ parameters = Map("message" ->
+ ("The STORED AS syntax is not supported for CREATE STREAMING TABLE
statements. " +
+ "Consider using the Data Source based USING clause instead.")),
+ queryContext = ex.getQueryContext.map(toExpectedContext)
+ )
+ }
+
+ test("CREATE STREAMING TABLE - SerDe information is unsupported") {
+ val ex = intercept[ParseException] {
+ parser.parsePlan(
+ """CREATE STREAMING TABLE st
+ |ROW FORMAT SERDE
'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe'""".stripMargin)
+ }
+ checkError(
+ exception = ex,
+ condition = "_LEGACY_ERROR_TEMP_0035",
+ parameters = Map("message" ->
+ ("Hive SerDe format options are not supported for " +
+ "CREATE STREAMING TABLE statements.")),
+ queryContext = ex.getQueryContext.map(toExpectedContext)
+ )
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
