szehon-ho commented on code in PR #57612:
URL: https://github.com/apache/spark/pull/57612#discussion_r3670414872
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/AutoCdcParserSuite.scala:
##########
@@ -592,70 +594,151 @@ class AutoCdcParserSuite extends CommandSuiteBase with
AnalysisTest {
//
---------------------------------------------------------------------------
test("CREATE FLOW AS AUTO CDC INTO - SEQUENCE BY is required") {
- checkError(
- intercept[ParseException] {
- parser.parsePlan(
- """CREATE FLOW f AS AUTO CDC INTO target
- |FROM STREAM(source)
- |KEYS (id)""".stripMargin)
- },
- condition = "PARSE_SYNTAX_ERROR",
- sqlState = "42601",
- parameters = Map("error" -> "end of input", "hint" -> "")
- )
+ val ex = intercept[ParseException] {
+ parser.parsePlan(
+ """CREATE FLOW f AS AUTO CDC INTO target
+ |FROM STREAM(source)
+ |KEYS (id)""".stripMargin)
+ }
+ assert(ex.getMessage.contains("AUTO CDC requires a SEQUENCE BY clause."))
Review Comment:
This suite asserts parse errors with `checkError(condition = ..., parameters
= ..., queryContext = ...)` everywhere else, including the other
`_LEGACY_ERROR_TEMP_0035` cases below. `getMessage.contains` pins neither the
condition, the SQLSTATE, nor the query context, so it would keep passing if the
condition silently changed. Same for the copy at line 614. `checkError` would
also make the rendered double period visible.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala:
##########
@@ -1420,11 +1420,27 @@ class AstBuilder extends DataTypeAstBuilder
}
}
val keys =
visitIdentifierSeq(params.keys).map(UnresolvedAttribute.quoted)
- val deleteCondition = Option(params.autoCdcDeleteClause())
+
+ // The optional clauses may appear in any order after `KEYS (...)`, so
the grammar accepts
+ // each any number of times; reject an accidental repeat here rather
than silently taking
+ // the last occurrence.
+ checkDuplicateClauses(params.autoCdcDeleteClause(), "APPLY AS DELETE
WHEN", params)
+ checkDuplicateClauses(params.autoCdcSequenceByClause(), "SEQUENCE BY",
params)
+ checkDuplicateClauses(params.autoCdcColumnsClause(), "COLUMNS", params)
+ checkDuplicateClauses(params.autoCdcStoredAsClause(), "STORED AS SCD
TYPE", params)
+ checkDuplicateClauses(params.autoCdcTrackHistoryClause(), "TRACK HISTORY
ON", params)
+
+ val deleteCondition = params.autoCdcDeleteClause().asScala.headOption
.map(c => expression(c.deleteCondition))
- val sequencing = expression(params.autoCdcSequenceByClause().sequence)
- val columnsClause = Option(params.autoCdcColumnsClause())
+ // SEQUENCE BY is mandatory; the grammar no longer enforces its presence
(the clauses are an
+ // unordered set), so require it explicitly here with a targeted error.
+ val sequenceByClause =
params.autoCdcSequenceByClause().asScala.headOption.getOrElse {
+ operationNotAllowed("AUTO CDC requires a SEQUENCE BY clause.", params)
Review Comment:
`operationNotAllowed` throws `_LEGACY_ERROR_TEMP_0035`, which has no
`sqlState`. Before this PR the same statement failed with `PARSE_SYNTAX_ERROR`
/ SQLSTATE `42601` -- that's what the removed test asserted -- so this converts
a classified error into an uncategorized one. That's a step back for JDBC/ODBC
clients that branch on SQLSTATE, and `error/README.md` asks that new code not
introduce `_LEGACY_ERROR_TEMP_*`.
`MISSING_CLAUSES_FOR_OPERATION` covers exactly this shape and keeps SQLSTATE
42601:
```
"Missing required clause(s) <clauses> for operation <operation>."
```
It's already used for metric views created without `WITH METRICS`. With
`clauses = "SEQUENCE BY"` and `operation = "AUTO CDC"` it also avoids the
double period the legacy template produces here (`Operation not allowed: AUTO
CDC requires a SEQUENCE BY clause..`).
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/AutoCdcParserSuite.scala:
##########
@@ -592,70 +594,151 @@ class AutoCdcParserSuite extends CommandSuiteBase with
AnalysisTest {
//
---------------------------------------------------------------------------
test("CREATE FLOW AS AUTO CDC INTO - SEQUENCE BY is required") {
- checkError(
- intercept[ParseException] {
- parser.parsePlan(
- """CREATE FLOW f AS AUTO CDC INTO target
- |FROM STREAM(source)
- |KEYS (id)""".stripMargin)
- },
- condition = "PARSE_SYNTAX_ERROR",
- sqlState = "42601",
- parameters = Map("error" -> "end of input", "hint" -> "")
- )
+ val ex = intercept[ParseException] {
+ parser.parsePlan(
+ """CREATE FLOW f AS AUTO CDC INTO target
+ |FROM STREAM(source)
+ |KEYS (id)""".stripMargin)
+ }
+ assert(ex.getMessage.contains("AUTO CDC requires a SEQUENCE BY clause."))
}
test("CREATE STREAMING TABLE FLOW AUTO CDC - SEQUENCE BY is required") {
- checkError(
- intercept[ParseException] {
- parser.parsePlan(
- """CREATE STREAMING TABLE target
- |FLOW AUTO CDC
- |FROM STREAM(source)
- |KEYS (id)""".stripMargin)
- },
- condition = "PARSE_SYNTAX_ERROR",
- sqlState = "42601",
- parameters = Map("error" -> "end of input", "hint" -> "")
- )
+ val ex = intercept[ParseException] {
+ parser.parsePlan(
+ """CREATE STREAMING TABLE target
+ |FLOW AUTO CDC
+ |FROM STREAM(source)
+ |KEYS (id)""".stripMargin)
+ }
+ assert(ex.getMessage.contains("AUTO CDC requires a SEQUENCE BY clause."))
}
//
---------------------------------------------------------------------------
- // Error cases: wrong clause order
+ // Clause ordering: the optional clauses may appear in any order
//
---------------------------------------------------------------------------
- test("SEQUENCE BY before APPLY AS DELETE is not allowed") {
- checkError(
- intercept[ParseException] {
- parser.parsePlan(
- """CREATE FLOW f AS AUTO CDC INTO target
- |FROM STREAM(source)
- |KEYS (id)
- |SEQUENCE BY ts
- |APPLY AS DELETE WHEN a = 1""".stripMargin)
- },
- condition = "PARSE_SYNTAX_ERROR",
- sqlState = "42601",
- parameters = Map("error" -> "'APPLY'", "hint" -> "")
- )
+ test("AUTO CDC - SEQUENCE BY before APPLY AS DELETE is allowed") {
+ val plan = parser.parsePlan(
+ """CREATE FLOW f AS AUTO CDC INTO target
+ |FROM STREAM(source)
+ |KEYS (id)
+ |SEQUENCE BY ts
+ |APPLY AS DELETE WHEN a = 1""".stripMargin)
+
+ val cdc =
plan.asInstanceOf[CreateFlowCommand].flowOperation.asInstanceOf[AutoCdcInto]
+ assert(cdc.sequenceByExpr == UnresolvedAttribute("ts"))
+ assert(cdc.deleteCondition.isDefined)
+ assert(cdc.deleteCondition.get.sql.contains("a"))
}
- test("COLUMNS before SEQUENCE BY is not allowed") {
+ test("AUTO CDC - COLUMNS before SEQUENCE BY is allowed") {
+ val plan = parser.parsePlan(
+ """CREATE FLOW f AS AUTO CDC INTO target
+ |FROM STREAM(source)
+ |KEYS (id)
+ |COLUMNS (a, b)
+ |SEQUENCE BY ts""".stripMargin)
+
+ val cdc =
plan.asInstanceOf[CreateFlowCommand].flowOperation.asInstanceOf[AutoCdcInto]
+ assert(cdc.includeColumns.get.map(_.name) == Seq("a", "b"))
+ assert(cdc.sequenceByExpr == UnresolvedAttribute("ts"))
+ }
+
+ test("AUTO CDC - clauses supplied in fully reversed order are all honored") {
Review Comment:
All the new ordering tests and all five duplicate-clause tests use `CREATE
FLOW ... AS AUTO CDC INTO`; among the new tests only the "SEQUENCE BY is
required" pair covers `CREATE STREAMING TABLE ... FLOW AUTO CDC`. Both share
`parseAutoCdcParams` so risk is low, but the suite otherwise mirrors cases
across both entry points -- one mirrored any-order test would match that
structure.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala:
##########
@@ -1420,11 +1420,27 @@ class AstBuilder extends DataTypeAstBuilder
}
}
val keys =
visitIdentifierSeq(params.keys).map(UnresolvedAttribute.quoted)
- val deleteCondition = Option(params.autoCdcDeleteClause())
+
+ // The optional clauses may appear in any order after `KEYS (...)`, so
the grammar accepts
+ // each any number of times; reject an accidental repeat here rather
than silently taking
+ // the last occurrence.
+ checkDuplicateClauses(params.autoCdcDeleteClause(), "APPLY AS DELETE
WHEN", params)
+ checkDuplicateClauses(params.autoCdcSequenceByClause(), "SEQUENCE BY",
params)
+ checkDuplicateClauses(params.autoCdcColumnsClause(), "COLUMNS", params)
+ checkDuplicateClauses(params.autoCdcStoredAsClause(), "STORED AS SCD
TYPE", params)
+ checkDuplicateClauses(params.autoCdcTrackHistoryClause(), "TRACK HISTORY
ON", params)
+
+ val deleteCondition = params.autoCdcDeleteClause().asScala.headOption
.map(c => expression(c.deleteCondition))
- val sequencing = expression(params.autoCdcSequenceByClause().sequence)
- val columnsClause = Option(params.autoCdcColumnsClause())
+ // SEQUENCE BY is mandatory; the grammar no longer enforces its presence
(the clauses are an
Review Comment:
Nit: "no longer enforces" only parses for a reader who remembers the old
grammar. The parenthetical already carries the durable reason, so consider:
"SEQUENCE BY is mandatory, but the grammar accepts the clauses as an unordered
set, so require it explicitly here."
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
