cloud-fan commented on code in PR #59665:
URL: https://github.com/apache/spark/pull/59665#discussion_r4336683685
##########
sql/core/src/test/scala/org/apache/spark/sql/SQLQueryTestSuite.scala:
##########
@@ -496,8 +705,24 @@ class SQLQueryTestSuite extends SharedSparkSession
schema = None,
output = normalizeTestResults(output.mkString("\n")))
case _ =>
+ val parsedPlan = tryParsePlan(localSparkSession, sql)
Review Comment:
**Non-blocking (P2):** `tryParsePlan` now runs here, outside
`handleExceptions`, for every statement of every execution golden, and it
catches only `ParseException`. Some parse-time errors are not
`ParseException`s. For `select interval 10 nanoseconds` (interval.sql:150,
which nonansi/interval.sql also runs via `--IMPORT interval.sql`),
`AstBuilder.visitMultiUnitsInterval` rethrows a
`SparkIllegalArgumentException`. `AbstractParser.parse` converts only
`SparkThrowable with WithOrigin`, which this exception is not, so it escapes
`parsePlan`, `tryParsePlan`, and then `runQueries`. Before this change, the
same exception was raised inside
`handleExceptions(getNormalizedQueryExecutionResult(...))` and recorded as the
statement's golden output (results/interval.sql.out:1444-1456). Now the
interval.sql and nonansi/interval.sql execution cases abort instead of
comparing against their goldens, and regenerating those goldens aborts too.
Suggested fix: move the pre-parse and the flow dispatch into a single
`handleExceptions` block, so a failure during the pre-parse is recorded exactly
as before. Leave `tryParsePlan` itself unchanged, since `runDebugQueries`
shares it and debug mode intentionally lets these errors propagate.
**Verification:**
- **Regression:** Run the regular and non-ANSI interval execution golden
cases and confirm they match their unchanged goldens, including the
SparkIllegalArgumentException recorded for 'select interval 10 nanoseconds'.
- **Behavior:** Run the four streaming execution golden cases and confirm
their recorded outputs are unchanged.
##########
sql/core/src/test/resources/sql-tests/inputs/streaming/available-now-window-aggregation.sql:
##########
@@ -0,0 +1,42 @@
+DROP TABLE IF EXISTS streaming_sql_golden_aggregation_events;
+DROP TABLE IF EXISTS streaming_sql_golden_aggregation_sink;
+
+CREATE TABLE streaming_sql_golden_aggregation_events (
+ event_time TIMESTAMP, category STRING, amount INT) USING parquet;
+INSERT INTO streaming_sql_golden_aggregation_events VALUES
+ (TIMESTAMP '2024-01-01 00:00:01', 'books', 2),
+ (TIMESTAMP '2024-01-01 00:00:05', 'books', 3),
+ (TIMESTAMP '2024-01-01 00:01:02', 'games', 7);
+
+CREATE TABLE streaming_sql_golden_aggregation_sink (
+ window_start TIMESTAMP,
+ window_end TIMESTAMP,
+ category STRING,
+ total BIGINT,
+ event_count BIGINT) USING parquet;
+
+CREATE FLOW streaming_sql_golden_aggregation_flow AS
+INSERT INTO streaming_sql_golden_aggregation_sink BY NAME
+SELECT
+ window(event_time, '1 minute').start AS window_start,
+ window(event_time, '1 minute').end AS window_end,
+ category,
+ sum(amount) AS total,
+ count(*) AS event_count
+FROM STREAM(streaming_sql_golden_aggregation_events)
+ WATERMARK event_time DELAY OF INTERVAL 10 SECONDS
+GROUP BY window(event_time, '1 minute'), category;
+
+REFRESH FLOW streaming_sql_golden_aggregation_flow;
Review Comment:
**Nit (P3):** Nothing in this file would catch a checkpoint-reuse
regression. This is the only golden with two `REFRESH FLOW` statements, and it
reads the sink only after the second one (line 37). If each refresh got a fresh
checkpoint, the second refresh would reprocess all four events, reach the same
00:09:50 watermark, and emit the same two windows (books 5/2, games 7/1). If
the windows were wrongly emitted on the first refresh, the append-only sink
would end in the same state. So the golden observes neither the declared
checkpoint reuse nor watermark-gated emission.
Suggested fix: add a `SELECT` from `streaming_sql_golden_aggregation_sink`
between the two refreshes, expected to be empty. Also add a stateless
refresh/insert/refresh case like the `copy_source` example in the class doc,
whose final sink must contain each source row exactly once; that is the shape
where losing the checkpoint would duplicate rows. Then regenerate the goldens.
**Verification:**
- **Regression:** Temporarily give each REFRESH a fresh checkpoint and
confirm that the stateless multi-refresh golden fails with duplicated rows;
then restore checkpoint reuse and confirm it passes.
- **Behavior:** Run the streaming golden cases and confirm the intermediate
empty-sink result and the final windows match the regenerated goldens.
##########
sql/core/src/test/scala/org/apache/spark/sql/SQLQueryTestSuite.scala:
##########
@@ -496,8 +705,24 @@ class SQLQueryTestSuite extends SharedSparkSession
schema = None,
output = normalizeTestResults(output.mkString("\n")))
case _ =>
+ val parsedPlan = tryParsePlan(localSparkSession, sql)
val (schema, output) =
-
handleExceptions(getNormalizedQueryExecutionResult(localSparkSession, sql))
+ parseTriggeredFlowStatement(localSparkSession, sql, parsedPlan)
match {
Review Comment:
**Nit (P3):** This flow dispatch exists only in the execution branch.
`listTestCases` also creates an `_analyzer_test` copy of each new streaming
input, and the unchanged `AnalyzerTest` branch above sends every statement to
`getNormalizedQueryAnalysisResult`. As a result, all four
`analyzer-results/streaming/*.sql.out` files record `CREATE FLOW` as
`ExtendedAnalysisException` `_LEGACY_ERROR_TEMP_3102` ("Queries with streaming
sources must be executed with writeStream.start()...") and `REFRESH FLOW` as a
`ParseException` (`INVALID_SQL_SYNTAX.INVALID_REFRESH_RESOURCE_PATH`). That
contradicts class doc item 7, which describes both statements as harness syntax
without limiting that to execution goldens.
As a result, the analyzer goldens cover none of the analysis of the flow
queries: the watermarked window aggregate, the stream-static join, and the
streaming union. They will also need regeneration whenever unrelated
streaming-checker or REFRESH-grammar error text changes. And because no golden
contains an analyzed streaming plan, the new `EventTimeWatermark` UUID
normalization in `SQLQueryTestHelper` never matches anything.
**Recommended change:** Teach the analyzer-golden path the two harness flow
statements and regenerate the streaming analyzer golden files. For CREATE FLOW,
register the flow and record the normalized analyzed plan of its INSERT ... BY
NAME query. For REFRESH FLOW, record empty output without starting a streaming
query. Every other statement stays on the existing analysis path, and all of it
runs inside the existing per-statement error capture.
**Why this works:** Recognizing the harness statements before the generic
analysis call stops the analyzer copy from analyzing CREATE FLOW as a batch
command and from parsing REFRESH FLOW as Spark SQL. The copy then records the
analyzed streaming plans the inputs are written to cover: the windowed
aggregation with a watermark, the stream-static join, and the streaming union.
Because that output passes through the existing normalization, the watermark
plan appears with its UUID masked.
**Scope:** Give the analyzer golden runner analyzer-mode handling of CREATE
FLOW and REFRESH FLOW, and regenerate the four streaming analyzer goldens.
**Compatibility:** Analyzer golden output for every statement other than
CREATE FLOW and REFRESH FLOW stays exactly as it is today.
**Risks:** Analyzed streaming plans may print other run-specific identifiers
besides the EventTimeWatermark UUID, which would make the regenerated goldens
nondeterministic; two consecutive regenerations must produce identical output.
**Constraints:** Keep the new analyzer-mode dispatch inside the
per-statement error capture, so that a parse-time failure is recorded instead
of aborting the analyzer copy.
**Success:** The streaming analyzer goldens contain the analyzed plan of
each flow query and empty output for REFRESH FLOW, with no
_LEGACY_ERROR_TEMP_3102 or INVALID_REFRESH_RESOURCE_PATH entries. The
window-aggregation analyzer golden shows EventTimeWatermark with its UUID
masked. Analyzer goldens for inputs that contain no flow statements are
unchanged.
**Verification:**
- **Behavior:** Regenerate and then re-run the streaming analyzer golden
cases; confirm they pass, record the flow queries' analyzed plans, and show the
masked watermark UUID.
- **Regression:** Regenerate the streaming analyzer goldens twice and
confirm the output is identical, so no run-specific identifier leaks into them.
- **Compatibility:** Run the complete analyzer golden suite and confirm that
no analyzer golden for a non-streaming input changes.
##########
sql/core/src/test/scala/org/apache/spark/sql/SQLQueryTestSuite.scala:
##########
@@ -366,6 +394,173 @@ class SQLQueryTestSuite extends SharedSparkSession
localSparkSession
}
+ private def triggeredFlowQueryName(
+ testCase: TestCase,
+ sparkConfigSet: Seq[(String, String)],
+ flowName: Seq[String]): String = {
+ val identity =
+ Seq(testCase.name, sparkConfigSet.mkString(","),
formatFlowName(flowName)).mkString("\u0000")
+ "sql-query-test-" +
UUID.nameUUIDFromBytes(identity.getBytes(StandardCharsets.UTF_8))
+ }
+
+ private case class TriggeredFlow(
+ flowName: Seq[String],
+ query: LogicalPlan,
+ targetTable: String)
+
+ private sealed trait TriggeredFlowStatement
+ private case class CreateTriggeredFlow(command: CreateFlowCommand)
+ extends TriggeredFlowStatement
+ private case class RefreshTriggeredFlow(flowName: Seq[String])
+ extends TriggeredFlowStatement
+
+ private def formatFlowName(flowName: Seq[String]): String =
+ flowName.map(quoteIdentifier).mkString(".")
+
+ private def sameFlowName(
+ session: SparkSession,
+ left: Seq[String],
+ right: Seq[String]): Boolean = {
+ left.length == right.length && left.zip(right).forall { case (l, r) =>
+ session.sessionState.conf.resolver(l, r)
+ }
+ }
+
+ private def parseCreateTriggeredFlow(command: CreateFlowCommand):
TriggeredFlow = {
+ val flowName = command.name match {
+ case identifier: UnresolvedIdentifier => identifier.nameParts
+ case _ =>
+ throw new IllegalArgumentException(
+ "SQL golden triggered flows require a static flow name.")
+ }
+ val insert = command.flowOperation match {
+ case insert: UnresolvedInsert => insert
+ case _ =>
+ throw new IllegalArgumentException(
+ "SQL golden triggered flows require CREATE FLOW AS " +
+ "INSERT INTO ... BY NAME SELECT.")
+ }
+ require(insert.byName && !insert.overwrite && insert.partitionSpec.isEmpty
&&
+ insert.userSpecifiedCols.isEmpty && insert.replaceCriteriaOpt.isEmpty
&&
+ !insert.withSchemaEvolution,
+ "SQL golden triggered flows require CREATE FLOW AS INSERT INTO ... BY
NAME SELECT.")
+
+ val targetTable = insert.table match {
+ case target: UnresolvedInsertTarget =>
+ target.multipartIdentifier.map(quoteIdentifier).mkString(".")
+ case _ =>
+ throw new IllegalArgumentException(
+ "SQL golden triggered flows require a static table identifier as the
INSERT target.")
+ }
+ TriggeredFlow(flowName, insert.query, targetTable)
+ }
+
+ private def registerTriggeredFlow(
+ session: SparkSession,
+ command: CreateFlowCommand,
+ flows: mutable.Map[Seq[String], TriggeredFlow]): (String, Seq[String]) =
{
+ val flow = parseCreateTriggeredFlow(command)
+ require(!flows.keys.exists(sameFlowName(session, _, flow.flowName)),
+ s"Triggered flow ${formatFlowName(flow.flowName)} is already defined in
this test run.")
+ flows += flow.flowName -> flow
+ (emptySchema, Seq.empty)
+ }
+
+ private def resolveTriggeredFlow(
+ session: SparkSession,
+ flowName: Seq[String],
+ flows: mutable.Map[Seq[String], TriggeredFlow]): TriggeredFlow = {
+ flows.collectFirst {
+ case (registeredName, flow) if sameFlowName(session, registeredName,
flowName) => flow
+ }.getOrElse {
+ throw new IllegalArgumentException(
+ s"Triggered flow ${formatFlowName(flowName)} has not been defined in
this test run.")
+ }
+ }
+
+ private def tryParsePlan(session: SparkSession, sql: String):
Option[LogicalPlan] = {
+ try {
+ Some(session.sessionState.sqlParser.parsePlan(sql))
+ } catch {
+ case _: ParseException => None
+ }
+ }
+
+ private def parseRefreshTriggeredFlowStatement(
+ session: SparkSession,
+ sql: String): Option[RefreshTriggeredFlow] = {
+ val tokens = sql.trim.split("\\s+", 3)
+ if (tokens.length == 3 && tokens(0).equalsIgnoreCase("REFRESH") &&
+ tokens(1).equalsIgnoreCase("FLOW")) {
+ try {
+ Some(RefreshTriggeredFlow(
+ session.sessionState.sqlParser.parseMultipartIdentifier(tokens(2))))
+ } catch {
+ case _: ParseException => None
+ }
+ } else {
+ None
+ }
+ }
+
+ private def parseTriggeredFlowStatement(
+ session: SparkSession,
+ sql: String,
+ parsedPlan: Option[LogicalPlan]): Option[TriggeredFlowStatement] = {
+ parsedPlan match {
+ case Some(command: CreateFlowCommand) =>
Some(CreateTriggeredFlow(command))
+ case _ => parseRefreshTriggeredFlowStatement(session, sql)
+ }
+ }
+
+ /** Runs one REFRESH FLOW statement to completion. */
+ private def runTriggeredFlow(
+ session: SparkSession,
+ sql: String,
+ flowName: Seq[String],
+ testCase: TestCase,
+ sparkConfigSet: Seq[(String, String)],
+ flows: mutable.Map[Seq[String], TriggeredFlow]): (String, Seq[String]) =
{
+ val flow = resolveTriggeredFlow(session, flowName, flows)
+ val df =
classic.Dataset.ofRows(session.asInstanceOf[classic.SparkSession], flow.query)
+ require(df.isStreaming,
+ s"SQL golden triggered flows require a streaming query, but '$sql' is
batch SQL.")
Review Comment:
**Nit (P3):** `sql` here is the `REFRESH FLOW` statement that `runQueries` /
`runDebugQueries` pass in, not the query captured by `CREATE FLOW`. For a flow
defined as `CREATE FLOW f AS INSERT INTO sink BY NAME SELECT * FROM source`,
with no `STREAM(...)`, refreshing it reports "... but 'REFRESH FLOW f' is batch
SQL." That statement isn't batch SQL, and the test author has to go find the
definition themselves. Consider naming the flow instead, for example `s"SQL
golden triggered flow ${formatFlowName(flow.flowName)} requires a streaming
query, but its CREATE FLOW query reads no streaming source."`.
**Verification:**
- **Behavior:** Define a flow over a batch SELECT, refresh it, and assert
that the error names the flow and does not describe the REFRESH statement as
batch SQL.
--
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]
