cloud-fan commented on code in PR #59665: URL: https://github.com/apache/spark/pull/59665#discussion_r4355824212
########## sql/core/src/test/scala/org/apache/spark/sql/StreamingSQLQueryTestSuite.scala: ########## @@ -0,0 +1,594 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql + +import java.nio.charset.StandardCharsets +import java.util.UUID +import java.util.concurrent.TimeUnit + +import scala.collection.mutable + +import org.apache.hadoop.fs.Path + +import org.apache.spark.sql.catalyst.analysis.{TableOutputResolver, UnresolvedIdentifier, + UnresolvedInsertTarget} +import org.apache.spark.sql.catalyst.catalog.CatalogTable +import org.apache.spark.sql.catalyst.parser.ParseException +import org.apache.spark.sql.catalyst.plans.logical.{CreateFlowCommand, LogicalPlan, UnresolvedInsert} +import org.apache.spark.sql.catalyst.util.QuotingUtils.quoteIdentifier +import org.apache.spark.sql.connector.catalog.{V1Table, V2TableWithV1Fallback} +import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._ +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.streaming.{OutputMode, StreamingQueryException, Trigger} +import org.apache.spark.tags.ExtendedSQLTest +import org.apache.spark.util.Utils + +// scalastyle:off line.size.limit +/** + * SQL golden tests for Structured Streaming flows. + * + * CREATE FLOW registers a named flow definition without running it. REFRESH FLOW runs the + * remembered definition with Trigger.AvailableNow. The flow must use INSERT INTO <table> BY NAME, + * and the target table must already exist. Runs of the same named flow reuse its checkpoint. The + * target table must be written only by the flow because batch writes are not tracked by the + * streaming sink's metadata log, and each target table may be written by only one flow. Dropping + * and recreating the target table while reusing a flow is unsupported and should not be tested. + * REFRESH FLOW is test-harness syntax; it is not part of the Spark SQL parser. + * + * To run the suite: + * {{{ + * build/sbt "sql/testOnly org.apache.spark.sql.StreamingSQLQueryTestSuite" + * }}} + * + * To run one input file: + * {{{ + * build/sbt "sql/testOnly org.apache.spark.sql.StreamingSQLQueryTestSuite -- -z available-now-to-table.sql" + * }}} + * + * For example: + * {{{ + * CREATE TABLE source (id INT, value STRING) USING parquet; + * INSERT INTO source VALUES (1, 'a'), (2, 'b'); + * CREATE TABLE sink (id INT, value STRING) USING parquet; + * CREATE FLOW copy_source AS + * INSERT INTO sink BY NAME + * SELECT * FROM STREAM(source); + * REFRESH FLOW copy_source; + * SELECT * FROM sink ORDER BY id; + * }}} + */ +// scalastyle:on line.size.limit +@ExtendedSQLTest +class StreamingSQLQueryTestSuite extends SQLQueryTestSuite { + + override protected def ignoreList: Set[String] = super.ignoreList - "streaming/" + + override protected lazy val listTestCases: Seq[TestCase] = + createTestCases().filter(_.name.startsWith("streaming/")) + + 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 val triggeredFlowTimeoutMs = TimeUnit.MINUTES.toMillis(10) + + 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 loadV1Table(session: SparkSession, tableName: String): Option[CatalogTable] = { + val analyzer = session.sessionState.analyzer + import analyzer.CatalogAndIdentifier + + val multipartIdentifier = session.sessionState.sqlParser.parseMultipartIdentifier(tableName) + val CatalogAndIdentifier(catalog, identifier) = multipartIdentifier + catalog.asTableCatalog.loadTable(identifier) match { + case table: V1Table => Some(table.v1Table) + case table: V2TableWithV1Fallback => Some(table.v1Table) + case _ => None + } + } + + 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]): TriggeredFlow = { + session.withActive { + 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.values.find { registeredFlow => + session.sessionState.conf.resolver(registeredFlow.targetTable, flow.targetTable) Review Comment: **Nit (P3):** This compares the quoted, dot-joined spellings of the two INSERT targets. Spellings that differ but name one table, such as `sink`, `default.sink`, and `spark_catalog.default.sink`, count as different targets, so both flows register. Both file sinks then write under the table's location and share its `_spark_metadata` log. `FileStreamSink.addBatch` skips any batch ID at or below that log's latest batch (FileStreamSink.scala:153-155). So the second flow's first refresh drops its rows with no error, while its checkpoint marks the input as processed, and regenerating the golden records the missing rows as expected output. This only matters for a golden input that already breaks the class doc's one-flow-per-target rule using different spellings, and the five current inputs use distinct targets. But turning that mistake into a clear error is exactly this guard's job. An earlier revision of this PR compared catalog-resolved names (`canonicalTableName` via `CatalogAndIdentifier`). Commit 14395dd0d9a replaced that with this string comparison. Suggested fix: resolve each target through `CatalogAndIdentifier`, as `loadV1Table` already does, and compare catalog, namespace, and table name with the resolver. Keep the stored spelling for messages. That resolution does not need the table to exist yet, so the check can stay at CREATE FLOW time. **Verification:** - **Regression:** For target spellings that resolve to the same table (unqualified, database-qualified, and catalog-qualified forms), registering a second flow into any of them after a first flow targets another of them is rejected with the one-flow-per-target error instead of registering both flows. - **Behavior:** The existing streaming golden cases, including the chained multiple-flows case whose flows write distinct tables, and the existing same-spelling shared-target rejection still pass unchanged. ########## sql/core/src/test/scala/org/apache/spark/sql/StreamingSQLQueryTestSuite.scala: ########## @@ -0,0 +1,594 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql + +import java.nio.charset.StandardCharsets +import java.util.UUID +import java.util.concurrent.TimeUnit + +import scala.collection.mutable + +import org.apache.hadoop.fs.Path + +import org.apache.spark.sql.catalyst.analysis.{TableOutputResolver, UnresolvedIdentifier, + UnresolvedInsertTarget} +import org.apache.spark.sql.catalyst.catalog.CatalogTable +import org.apache.spark.sql.catalyst.parser.ParseException +import org.apache.spark.sql.catalyst.plans.logical.{CreateFlowCommand, LogicalPlan, UnresolvedInsert} +import org.apache.spark.sql.catalyst.util.QuotingUtils.quoteIdentifier +import org.apache.spark.sql.connector.catalog.{V1Table, V2TableWithV1Fallback} +import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._ +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.streaming.{OutputMode, StreamingQueryException, Trigger} +import org.apache.spark.tags.ExtendedSQLTest +import org.apache.spark.util.Utils + +// scalastyle:off line.size.limit +/** + * SQL golden tests for Structured Streaming flows. + * + * CREATE FLOW registers a named flow definition without running it. REFRESH FLOW runs the + * remembered definition with Trigger.AvailableNow. The flow must use INSERT INTO <table> BY NAME, + * and the target table must already exist. Runs of the same named flow reuse its checkpoint. The + * target table must be written only by the flow because batch writes are not tracked by the + * streaming sink's metadata log, and each target table may be written by only one flow. Dropping + * and recreating the target table while reusing a flow is unsupported and should not be tested. + * REFRESH FLOW is test-harness syntax; it is not part of the Spark SQL parser. + * + * To run the suite: + * {{{ + * build/sbt "sql/testOnly org.apache.spark.sql.StreamingSQLQueryTestSuite" + * }}} + * + * To run one input file: + * {{{ + * build/sbt "sql/testOnly org.apache.spark.sql.StreamingSQLQueryTestSuite -- -z available-now-to-table.sql" + * }}} + * + * For example: + * {{{ + * CREATE TABLE source (id INT, value STRING) USING parquet; + * INSERT INTO source VALUES (1, 'a'), (2, 'b'); + * CREATE TABLE sink (id INT, value STRING) USING parquet; + * CREATE FLOW copy_source AS + * INSERT INTO sink BY NAME + * SELECT * FROM STREAM(source); + * REFRESH FLOW copy_source; + * SELECT * FROM sink ORDER BY id; + * }}} + */ +// scalastyle:on line.size.limit +@ExtendedSQLTest +class StreamingSQLQueryTestSuite extends SQLQueryTestSuite { + + override protected def ignoreList: Set[String] = super.ignoreList - "streaming/" + + override protected lazy val listTestCases: Seq[TestCase] = + createTestCases().filter(_.name.startsWith("streaming/")) + + 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 val triggeredFlowTimeoutMs = TimeUnit.MINUTES.toMillis(10) + + 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 loadV1Table(session: SparkSession, tableName: String): Option[CatalogTable] = { + val analyzer = session.sessionState.analyzer + import analyzer.CatalogAndIdentifier + + val multipartIdentifier = session.sessionState.sqlParser.parseMultipartIdentifier(tableName) + val CatalogAndIdentifier(catalog, identifier) = multipartIdentifier + catalog.asTableCatalog.loadTable(identifier) match { + case table: V1Table => Some(table.v1Table) + case table: V2TableWithV1Fallback => Some(table.v1Table) + case _ => None + } + } + + 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]): TriggeredFlow = { + session.withActive { + 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.values.find { registeredFlow => + session.sessionState.conf.resolver(registeredFlow.targetTable, flow.targetTable) + }.foreach { registeredFlow => + throw new IllegalArgumentException( + s"SQL golden triggered flow target table '${flow.targetTable}' is already written by " + + s"flow ${formatFlowName(registeredFlow.flowName)}.") + } + flows += flow.flowName -> flow + flow + } + } + + 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 parseRefreshTriggeredFlowStatement( + session: SparkSession, + name: String): Option[RefreshTriggeredFlow] = { + try { + Some(RefreshTriggeredFlow(session.sessionState.sqlParser.parseMultipartIdentifier(name))) + } catch { + case _: ParseException => None Review Comment: **Nit (P3):** A statement that starts with `REFRESH FLOW` reaches this helper only because Spark's parser already rejected it. `REFRESH (stringLit | .*?)` treats the rest as an unquoted resource path, and whitespace in that path is an error. If `parseMultipartIdentifier` then also rejects the name, as with `REFRESH FLOW my-flow` or `REFRESH FLOW a b`, returning `None` sends the statement back to ordinary analysis or execution. There it is re-parsed, and the golden, in either analyzer or execution mode, records `INVALID_SQL_SYNTAX.INVALID_REFRESH_RESOURCE_PATH` (REFRESH statements cannot contain whitespace inside unquoted resource paths). That error mentions neither the flow nor the bad identifier, and regenerating the golden saves it as the expected output. Suggested fix: once a statement is recognized as REFRESH FLOW, raise an `IllegalArgumentException` that names the invalid flow name, or let the identifier `ParseException` propagate, instead of returning `None`. The per-statement `handleExceptions` in `runQueries` then records that error instead. **Verification:** - **Regression:** REFRESH FLOW followed by a name that is not a valid identifier, such as an unquoted hyphenated name or two whitespace-separated names, fails with an error about the flow name rather than an invalid refresh resource path, in both analyzer and execution golden modes. - **Behavior:** Quoted multipart REFRESH FLOW names still resolve to their registered flows, a bare REFRESH FLOW still reports the missing flow name, and ordinary statements such as a nanosecond interval literal are still not claimed by the harness. ########## sql/core/src/test/scala/org/apache/spark/sql/StreamingSQLQueryTestSuite.scala: ########## @@ -0,0 +1,594 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql + +import java.nio.charset.StandardCharsets +import java.util.UUID +import java.util.concurrent.TimeUnit + +import scala.collection.mutable + +import org.apache.hadoop.fs.Path + +import org.apache.spark.sql.catalyst.analysis.{TableOutputResolver, UnresolvedIdentifier, + UnresolvedInsertTarget} +import org.apache.spark.sql.catalyst.catalog.CatalogTable +import org.apache.spark.sql.catalyst.parser.ParseException +import org.apache.spark.sql.catalyst.plans.logical.{CreateFlowCommand, LogicalPlan, UnresolvedInsert} +import org.apache.spark.sql.catalyst.util.QuotingUtils.quoteIdentifier +import org.apache.spark.sql.connector.catalog.{V1Table, V2TableWithV1Fallback} +import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._ +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.streaming.{OutputMode, StreamingQueryException, Trigger} +import org.apache.spark.tags.ExtendedSQLTest +import org.apache.spark.util.Utils + +// scalastyle:off line.size.limit +/** + * SQL golden tests for Structured Streaming flows. + * + * CREATE FLOW registers a named flow definition without running it. REFRESH FLOW runs the + * remembered definition with Trigger.AvailableNow. The flow must use INSERT INTO <table> BY NAME, + * and the target table must already exist. Runs of the same named flow reuse its checkpoint. The + * target table must be written only by the flow because batch writes are not tracked by the + * streaming sink's metadata log, and each target table may be written by only one flow. Dropping + * and recreating the target table while reusing a flow is unsupported and should not be tested. + * REFRESH FLOW is test-harness syntax; it is not part of the Spark SQL parser. + * + * To run the suite: + * {{{ + * build/sbt "sql/testOnly org.apache.spark.sql.StreamingSQLQueryTestSuite" + * }}} + * + * To run one input file: + * {{{ + * build/sbt "sql/testOnly org.apache.spark.sql.StreamingSQLQueryTestSuite -- -z available-now-to-table.sql" + * }}} + * + * For example: + * {{{ + * CREATE TABLE source (id INT, value STRING) USING parquet; + * INSERT INTO source VALUES (1, 'a'), (2, 'b'); + * CREATE TABLE sink (id INT, value STRING) USING parquet; + * CREATE FLOW copy_source AS + * INSERT INTO sink BY NAME + * SELECT * FROM STREAM(source); + * REFRESH FLOW copy_source; + * SELECT * FROM sink ORDER BY id; + * }}} + */ +// scalastyle:on line.size.limit +@ExtendedSQLTest +class StreamingSQLQueryTestSuite extends SQLQueryTestSuite { + + override protected def ignoreList: Set[String] = super.ignoreList - "streaming/" + + override protected lazy val listTestCases: Seq[TestCase] = + createTestCases().filter(_.name.startsWith("streaming/")) + + 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 val triggeredFlowTimeoutMs = TimeUnit.MINUTES.toMillis(10) + + 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 loadV1Table(session: SparkSession, tableName: String): Option[CatalogTable] = { + val analyzer = session.sessionState.analyzer + import analyzer.CatalogAndIdentifier + + val multipartIdentifier = session.sessionState.sqlParser.parseMultipartIdentifier(tableName) + val CatalogAndIdentifier(catalog, identifier) = multipartIdentifier + catalog.asTableCatalog.loadTable(identifier) match { + case table: V1Table => Some(table.v1Table) + case table: V2TableWithV1Fallback => Some(table.v1Table) + case _ => None + } + } + + 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]): TriggeredFlow = { + session.withActive { + 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.values.find { registeredFlow => + session.sessionState.conf.resolver(registeredFlow.targetTable, flow.targetTable) + }.foreach { registeredFlow => + throw new IllegalArgumentException( + s"SQL golden triggered flow target table '${flow.targetTable}' is already written by " + + s"flow ${formatFlowName(registeredFlow.flowName)}.") + } + flows += flow.flowName -> flow + flow + } + } + + 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 parseRefreshTriggeredFlowStatement( + session: SparkSession, + name: String): Option[RefreshTriggeredFlow] = { + try { + Some(RefreshTriggeredFlow(session.sessionState.sqlParser.parseMultipartIdentifier(name))) + } catch { + case _: ParseException => None + } + } + + private def parseTriggeredFlowStatement( + session: SparkSession, + sql: String, + parsedPlan: => Option[LogicalPlan]): Option[TriggeredFlowStatement] = { + session.withActive { + val tokens = sql.trim.split("\\s+", 3) + val hasFlowCommandPrefix = tokens.length >= 2 && tokens(1).equalsIgnoreCase("FLOW") && + (tokens(0).equalsIgnoreCase("CREATE") || tokens(0).equalsIgnoreCase("REFRESH")) + if (hasFlowCommandPrefix) { + if (tokens.length == 2 && tokens(0).equalsIgnoreCase("REFRESH")) { + throw new IllegalArgumentException("REFRESH FLOW requires a flow name.") + } + parsedPlan match { + case Some(command: CreateFlowCommand) => Some(CreateTriggeredFlow(command)) + case None if tokens.length == 3 && tokens(0).equalsIgnoreCase("REFRESH") => + parseRefreshTriggeredFlowStatement(session, tokens(2)) + case _ => None + } + } else { + None + } + } + } + + private def parseTriggeredFlowStatement( + session: SparkSession, + sql: String): Option[TriggeredFlowStatement] = { + parseTriggeredFlowStatement(session, sql, parsePlan(session, sql)) + } + + /** Runs one REFRESH FLOW statement to completion. */ + private def runTriggeredFlow( + session: SparkSession, + flowName: Seq[String], + testCase: TestCase, + sparkConfigSet: Seq[(String, String)], + flows: mutable.Map[Seq[String], TriggeredFlow]): (String, Seq[String]) = { + session.withActive { + val flow = resolveTriggeredFlow(session, flowName, flows) + val df = classic.Dataset.ofRows( + session.asInstanceOf[classic.SparkSession], flow.query.clone()) + require(df.isStreaming, + s"SQL golden triggered flow ${formatFlowName(flow.flowName)} requires a streaming query, " + + "but its CREATE FLOW query reads no streaming source.") + require(session.catalog.tableExists(flow.targetTable), + s"SQL golden triggered flow target table '${flow.targetTable}' does not exist.") + val queryName = triggeredFlowQueryName(testCase, sparkConfigSet, flow.flowName) + val targetTable = loadV1Table(session, flow.targetTable) + val alignedQuery = TableOutputResolver.resolveOutputColumns( + flow.targetTable, + session.table(flow.targetTable).queryExecution.analyzed.output, + df.queryExecution.analyzed, + byName = true, + session.sessionState.conf, + TableOutputResolver.DefaultValueFillMode.FILL) + val alignedDf = classic.Dataset.ofRows( + session.asInstanceOf[classic.SparkSession], alignedQuery) + + val writer = alignedDf.writeStream + .outputMode(OutputMode.Append()) + .trigger(Trigger.AvailableNow()) + .queryName(queryName) + .option( + "checkpointLocation", + new Path(session.sessionState.conf.warehousePath, queryName).toString) + targetTable.foreach { table => + writer.format(table.provider.getOrElse { + throw new IllegalArgumentException( + s"SQL golden triggered flow target table '${flow.targetTable}' has no provider.") + }) + if (table.partitionColumnNames.nonEmpty) { + writer.partitionBy(table.partitionColumnNames: _*) + } + } + val query = writer.toTable(flow.targetTable) + Utils.tryWithSafeFinally { + try { + if (!query.awaitTermination(triggeredFlowTimeoutMs)) { + throw new IllegalStateException( + s"SQL golden triggered flow ${formatFlowName(flow.flowName)} did not terminate " + + s"within ${TimeUnit.MILLISECONDS.toMinutes(triggeredFlowTimeoutMs)} minutes.") + } + } catch { + case e: StreamingQueryException if e.cause != null => throw e.cause + } + } { + query.stop() + } + + (emptySchema, Seq.empty) + } + } + + override protected def createCustomQueryExecutor( + session: SparkSession, + testCase: TestCase, + sparkConfigSet: Seq[(String, String)], + analyzerOnly: Boolean): + (String, () => Option[LogicalPlan]) => Option[(String, Seq[String])] = { + val flows = mutable.Map.empty[Seq[String], TriggeredFlow] + (sql, getParsedPlan) => parseTriggeredFlowStatement(session, sql, getParsedPlan()).map { + case CreateTriggeredFlow(command) => + val flow = registerTriggeredFlow(session, command, flows) + if (analyzerOnly) { + val df = classic.Dataset.ofRows( + session.asInstanceOf[classic.SparkSession], flow.query) + getNormalizedQueryAnalysisResult(df) + } else { + (emptySchema, Seq.empty) + } + case RefreshTriggeredFlow(flowName) => + if (analyzerOnly) { + resolveTriggeredFlow(session, flowName, flows) + (emptySchema, Seq.empty) + } else { + runTriggeredFlow(session, flowName, testCase, sparkConfigSet, flows) + } + } + } + + test("test triggered flow parsing") { + def querySpecs(input: String): Seq[(String, Boolean)] = { + val (comments, code) = splitCommentsAndCodes(input) + getQueriesWithDebugFlag(code, comments, Nil) + } + + def parseTriggeredFlow(sql: String): Option[TriggeredFlowStatement] = { + parseTriggeredFlowStatement(spark, sql) + } + + def createFlowCommand(sql: String): CreateFlowCommand = { + parseTriggeredFlow(sql) match { + case Some(CreateTriggeredFlow(command)) => command + case statement => fail(s"Expected CREATE FLOW, found $statement") + } + } + + def refreshFlowName(sql: String): Seq[String] = { + parseTriggeredFlow(sql) match { + case Some(RefreshTriggeredFlow(flowName)) => flowName + case statement => fail(s"Expected REFRESH FLOW, found $statement") + } + } + + val specs = querySpecs( + """-- ordinary comment + |SELECT 1; + |CREATE FLOW test_flow AS + |INSERT INTO `catalog`.`schema`.`sink-table` BY NAME + |SELECT * FROM STREAM(source); + |REFRESH FLOW test_flow; + |SELECT 2; + |""".stripMargin) + assert(specs.map(_._1) == Seq( + "SELECT 1", + """CREATE FLOW test_flow AS + |INSERT INTO `catalog`.`schema`.`sink-table` BY NAME + |SELECT * FROM STREAM(source)""".stripMargin, + "REFRESH FLOW test_flow", + "SELECT 2")) + + val flows = mutable.Map.empty[Seq[String], TriggeredFlow] + registerTriggeredFlow(spark, createFlowCommand(specs(1)._1), flows) + val flowName = refreshFlowName(specs(2)._1) + val flow = resolveTriggeredFlow(spark, flowName, flows) + assert(flow.targetTable == "`catalog`.`schema`.`sink-table`") + assert(resolveTriggeredFlow(spark, flowName, flows) eq flow) + assert(parseTriggeredFlow(specs.head._1).isEmpty) + assert(parseTriggeredFlow("select interval 10 nanoseconds").isEmpty) + assert(parseTriggeredFlow(specs(1)._1) + .exists(_.isInstanceOf[CreateTriggeredFlow])) + assert(parseTriggeredFlow(specs(2)._1) + .exists(_.isInstanceOf[RefreshTriggeredFlow])) + assert(refreshFlowName("refresh flow `catalog`.`flow-name`") == Seq("catalog", "flow-name")) + val missingRefreshNameError = intercept[IllegalArgumentException] { + parseTriggeredFlow("REFRESH FLOW") + } + assert(missingRefreshNameError.getMessage.contains("requires a flow name")) + + val debugSpec = querySpecs( + """--DEBUG + |CREATE FLOW debug_flow AS + |INSERT INTO sink BY NAME SELECT * FROM STREAM(source); + |""".stripMargin).head + assert(debugSpec._2) + + val nonByNameError = intercept[IllegalArgumentException] { + registerTriggeredFlow( + spark, + createFlowCommand("CREATE FLOW f AS INSERT INTO sink SELECT * FROM STREAM(source)"), + mutable.Map.empty) + } + assert(nonByNameError.getMessage.contains("INSERT INTO ... BY NAME SELECT")) + + val undefinedFlowError = intercept[IllegalArgumentException] { + resolveTriggeredFlow(spark, Seq("missing_flow"), mutable.Map.empty) + } + assert(undefinedFlowError.getMessage.contains("has not been defined in this test run")) + + val batchFlows = mutable.Map.empty[Seq[String], TriggeredFlow] + registerTriggeredFlow( + spark, + createFlowCommand( + "CREATE FLOW batch_flow AS INSERT INTO sink BY NAME SELECT 1 AS id"), + batchFlows) + val batchFlowError = intercept[IllegalArgumentException] { + runTriggeredFlow( + spark, + Seq("batch_flow"), + RegularTestCase("batch-flow-error", "", ""), + Nil, + batchFlows) + } + assert(batchFlowError.getMessage.contains("triggered flow `batch_flow`")) + assert(!batchFlowError.getMessage.contains("REFRESH FLOW")) + + val sharedTargetFlows = mutable.Map.empty[Seq[String], TriggeredFlow] + registerTriggeredFlow( + spark, + createFlowCommand( + "CREATE FLOW first_flow AS INSERT INTO shared_sink BY NAME " + + "SELECT * FROM STREAM(first_source)"), + sharedTargetFlows) + val sharedTargetError = intercept[IllegalArgumentException] { + registerTriggeredFlow( + spark, + createFlowCommand( + "CREATE FLOW second_flow AS INSERT INTO shared_sink BY NAME " + + "SELECT * FROM STREAM(second_source)"), + sharedTargetFlows) + } + assert(sharedTargetError.getMessage.contains("already written by flow `first_flow`")) + + withTable("by_name_source", "by_name_sink") { + spark.sql("CREATE TABLE by_name_source (second INT, first INT) USING parquet") + spark.sql("INSERT INTO by_name_source VALUES (2, 1)") + spark.sql("CREATE TABLE by_name_sink (first INT, second INT) USING parquet") + val byNameFlows = mutable.Map.empty[Seq[String], TriggeredFlow] + registerTriggeredFlow( + spark, + createFlowCommand( + """CREATE FLOW by_name_flow AS + |INSERT INTO by_name_sink BY NAME + |SELECT second, first FROM STREAM(by_name_source) + |""".stripMargin), + byNameFlows) + runTriggeredFlow( + spark, + Seq("by_name_flow"), + RegularTestCase("by-name", "", ""), + Nil, + byNameFlows) + checkAnswer(spark.table("by_name_sink"), Row(1, 2)) + } + + withTable("incompatible_source", "incompatible_sink") { + spark.sql("CREATE TABLE incompatible_source (id INT) USING parquet") + spark.sql("CREATE TABLE incompatible_sink (other INT) USING parquet") + val incompatibleFlows = mutable.Map.empty[Seq[String], TriggeredFlow] + registerTriggeredFlow( + spark, + createFlowCommand( + """CREATE FLOW incompatible_flow AS + |INSERT INTO incompatible_sink BY NAME + |SELECT id FROM STREAM(incompatible_source) + |""".stripMargin), + incompatibleFlows) + checkError( + exception = intercept[AnalysisException] { + runTriggeredFlow( + spark, + Seq("incompatible_flow"), + RegularTestCase("incompatible", "", ""), + Nil, + incompatibleFlows) + }, + condition = "INCOMPATIBLE_DATA_FOR_TABLE.EXTRA_COLUMNS", + parameters = Map("tableName" -> "`incompatible_sink`", "extraColumns" -> "`id`")) + } + + withSQLConf(SQLConf.ANSI_ENABLED.key -> "true") { + withTable("runtime_error_source", "runtime_error_sink") { + spark.sql("CREATE TABLE runtime_error_source (value STRING) USING parquet") + spark.sql("INSERT INTO runtime_error_source VALUES ('not_an_int')") + spark.sql("CREATE TABLE runtime_error_sink (value INT) USING parquet") + val runtimeErrorFlows = mutable.Map.empty[Seq[String], TriggeredFlow] + registerTriggeredFlow( + spark, + createFlowCommand( + """CREATE FLOW runtime_error_flow AS + |INSERT INTO runtime_error_sink BY NAME + |SELECT CAST(value AS INT) AS value FROM STREAM(runtime_error_source) + |""".stripMargin), + runtimeErrorFlows) + val (_, runtimeErrorOutput) = handleExceptions { + runTriggeredFlow( + spark, + Seq("runtime_error_flow"), + RegularTestCase("runtime-error", "", ""), + Nil, + runtimeErrorFlows) + } + assert(runtimeErrorOutput.exists(_.contains("CAST_INVALID_INPUT"))) + assert(!runtimeErrorOutput.exists(_.contains("STREAM_FAILED"))) + assert(!runtimeErrorOutput.exists(_.contains("runId"))) + } + } + } + + test("test triggered flow execution semantics") { + def createFlowCommand(session: SparkSession, sql: String): CreateFlowCommand = { + parseTriggeredFlowStatement(session, sql) match { + case Some(CreateTriggeredFlow(command)) => command + case statement => fail(s"Expected CREATE FLOW, found $statement") + } + } + + withTable("provider_source", "provider_sink") { + spark.sql("CREATE TABLE provider_source (id INT, part STRING) USING parquet") + spark.sql("INSERT INTO provider_source VALUES (1, 'a'), (2, 'b')") + spark.sql( + "CREATE TABLE provider_sink (id INT, part STRING) USING JSON PARTITIONED BY (part)") + val flows = mutable.Map.empty[Seq[String], TriggeredFlow] + registerTriggeredFlow( + spark, + createFlowCommand( + spark, + "CREATE FLOW provider_flow AS INSERT INTO provider_sink BY NAME " + + "SELECT id, part FROM STREAM(provider_source)"), + flows) + runTriggeredFlow( + spark, + Seq("provider_flow"), + RegularTestCase("provider", "", ""), + Nil, + flows) + checkAnswer(spark.table("provider_sink"), Seq(Row(1, "a"), Row(2, "b"))) Review Comment: **Nit (P3):** This case is the only coverage for `runTriggeredFlow`'s `partitionBy(table.partitionColumnNames: _*)` branch, but it checks only the rows read back. If that call were dropped, `createV1Sink` would get no partition columns, because it uses only the writer's partitioning. The sink would then write `part` inline into JSON files at the table root. Reading `provider_sink` takes `DataSource.resolveRelation`'s `_spark_metadata` branch, which comes before any catalog-partition handling. That branch reads `id` and `part` from the files using the catalog schema, so it still returns `(1, a)` and `(2, b)` and the test still passes. The provider half of this case does fail on a regression, since a format mismatch is rejected. The partition layout has no such check. Suggested fix: also assert the on-disk layout. For example, check that the table location contains `part=a` and `part=b` directories, or that each row's `input_file_name()` is under its partition directory. **Verification:** - **Regression:** With the table's partition columns no longer passed to the streaming writer, the partitioned-target test fails on its layout assertion; with them restored, it passes. - **Behavior:** The test still exercises the provider requirement and still checks the rows read back from the partitioned JSON target. -- 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]
